From 2b2012e8e6b3a9eb5560659ad8ef7c395cab6c25 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 1 Aug 2026 01:22:45 +0200 Subject: [PATCH] refactor(timeline): remove submission timing state --- docs/timeline-sync.md | 20 ++++++----- packages/app/src/composer/actions.test.ts | 1 - .../app/src/composer/draft/create-flow.ts | 1 - .../app/src/composer/draft/workspace-tab.tsx | 2 +- .../app/src/composer/submission/model.test.ts | 35 +++++++------------ packages/app/src/composer/submission/model.ts | 1 - packages/app/src/stores/session-store.ts | 4 +-- .../app/src/timeline/turn-liveness.test.ts | 6 +--- packages/app/src/timeline/turn-liveness.ts | 5 ++- .../src/utils/agent-directory-sync.test.ts | 1 - 10 files changed, 30 insertions(+), 46 deletions(-) diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 6e9ac61e44..dc56a268f5 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -138,15 +138,17 @@ footer, and permanently drops attachments because the daemon does not echo them A submitted prompt is one `UserMessageItem` row. That row is the authoritative local presentation: its stable identity, text, timestamp, images, and attachments do not change when the provider acknowledges it. Submission lifecycle is a separate record keyed by agent, not another row shape. -The transaction registry exists only until the daemon accepts or rejects the send. An accepted RPC -settles it because capable daemons guarantee the prompt is already durable. Canonical acknowledgement -can arrive first and prevents a later transport error from rolling back a prompt already observed. +The transaction registry records two independent settlement facts: canonical acknowledgement and RPC +settlement. Canonical acknowledgement retires optimistic activity immediately. When both facts are +known, whichever arrives second deletes the record. A canonical acknowledgement that arrives first +prevents a later transport error from rolling back a prompt already observed. The daemon's accepted response waits for the correlated run start and guarantees that the canonical submitted row has been recorded. It publishes the accepted turn's liveness before that row, so the -client applies authoritative activity before retiring optimistic activity on RPC acceptance. Timeline -render batching does not delay lifecycle application. Directory status never settles a submission. -Overlapping sends settle independently rather than collapsing to one newest pending message. +client applies authoritative activity before canonical acknowledgement retires optimistic activity. +Timeline render batching does not delay lifecycle application. Directory status never settles a +submission. Overlapping sends settle independently rather than collapsing to one newest pending +message. Daemons advertising `server_info.features.canonicalSubmittedPrompts` guarantee that every accepted prompt carrying a client message id is recorded and streamed as a canonical `user_message` with that @@ -167,9 +169,9 @@ remain a separate pre-turn registry and retire on canonical acknowledgement. The compatibility boundary for older daemons is snapshot normalization: running/idle status becomes an anonymous active turn or idle state once, and downstream code consumes the same activity shape. The app does not combine anonymous lifecycle events, timestamps, timeline rows, and resume coverage to infer a -second running state. Disconnect and replica removal remain destructive close boundaries. Running time -comes from the daemon snapshot/event or the pending submission, never from whichever timeline rows happen -to be mounted. +second running state. Disconnect and replica removal remain destructive close boundaries. Elapsed time +comes only from turn liveness, never from submission records or whichever timeline rows happen to be +mounted. The daemon records one canonical submitted user row with Paseo's `clientMessageId` as soon as the provider accepts the turn. A correlated provider echo enriches that same sequence with its provider diff --git a/packages/app/src/composer/actions.test.ts b/packages/app/src/composer/actions.test.ts index 3923ef9eb3..d1aff546a2 100644 --- a/packages/app/src/composer/actions.test.ts +++ b/packages/app/src/composer/actions.test.ts @@ -215,7 +215,6 @@ function createFakeStream(initialHead: Map = new Map()): F ...stream, submissions: beginMessageSubmission(current.submissions, { clientMessageId: message.clientMessageId!, - submittedAt: message.timestamp, }), }); }, diff --git a/packages/app/src/composer/draft/create-flow.ts b/packages/app/src/composer/draft/create-flow.ts index 88046e7e60..89a237e786 100644 --- a/packages/app/src/composer/draft/create-flow.ts +++ b/packages/app/src/composer/draft/create-flow.ts @@ -159,7 +159,6 @@ export function useDraftAgentCreateFlow({ return [ { clientMessageId: machine.attempt.clientMessageId, - submittedAt: machine.attempt.timestamp, }, ]; }, [machine]); diff --git a/packages/app/src/composer/draft/workspace-tab.tsx b/packages/app/src/composer/draft/workspace-tab.tsx index 97f19a7741..a078ea0c9f 100644 --- a/packages/app/src/composer/draft/workspace-tab.tsx +++ b/packages/app/src/composer/draft/workspace-tab.tsx @@ -545,7 +545,7 @@ export function WorkspaceDraftAgentTab({ }, }); const turnPresentation = useMemo( - () => resolveTurnPresentation(TURN_LIVENESS_IDLE, pendingMessageSubmissions), + () => resolveTurnPresentation(TURN_LIVENESS_IDLE, pendingMessageSubmissions.length > 0), [pendingMessageSubmissions], ); useCommandCenterActions({ diff --git a/packages/app/src/composer/submission/model.test.ts b/packages/app/src/composer/submission/model.test.ts index 47ed42c88a..147ccce954 100644 --- a/packages/app/src/composer/submission/model.test.ts +++ b/packages/app/src/composer/submission/model.test.ts @@ -8,15 +8,10 @@ import { rejectMessageSubmission, } from "./model"; -const submittedAt = new Date("2026-07-26T10:00:00.000Z"); - describe("message submission transactions", () => { it("tracks every in-flight submission independently", () => { - const first = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); - const both = beginMessageSubmission(first, { - clientMessageId: "client-2", - submittedAt: new Date(submittedAt.getTime() + 1), - }); + const first = beginMessageSubmission([], { clientMessageId: "client-1" }); + const both = beginMessageSubmission(first, { clientMessageId: "client-2" }); expect(getActiveMessageSubmissions(both).map((item) => item.clientMessageId)).toEqual([ "client-1", @@ -27,15 +22,14 @@ describe("message submission transactions", () => { it("settles only the accepted transaction on RPC acceptance", () => { const both = beginMessageSubmission( - beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }), - { clientMessageId: "client-2", submittedAt }, + beginMessageSubmission([], { clientMessageId: "client-1" }), + { clientMessageId: "client-2" }, ); const acknowledged = observeMessageSubmissionCanonical(both, ["client-1"]); expect(acceptMessageSubmission(acknowledged, "client-1")).toEqual([ { clientMessageId: "client-2", - submittedAt, providerAcknowledged: false, rpcSettled: false, }, @@ -43,13 +37,12 @@ describe("message submission transactions", () => { }); it("keeps activity until an accepted RPC's canonical row is observed", () => { - const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); + const sending = beginMessageSubmission([], { clientMessageId: "client-1" }); const accepted = acceptMessageSubmission(sending, "client-1"); expect(accepted).toEqual([ { clientMessageId: "client-1", - submittedAt, providerAcknowledged: false, rpcSettled: true, }, @@ -62,21 +55,19 @@ describe("message submission transactions", () => { it("records provider acknowledgement without settling another transaction", () => { const both = beginMessageSubmission( - beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }), - { clientMessageId: "client-2", submittedAt }, + beginMessageSubmission([], { clientMessageId: "client-1" }), + { clientMessageId: "client-2" }, ); const observed = observeMessageSubmissionCanonical(both, ["client-1"]); expect(observed).toEqual([ { clientMessageId: "client-1", - submittedAt, providerAcknowledged: true, rpcSettled: false, }, { clientMessageId: "client-2", - submittedAt, providerAcknowledged: false, rpcSettled: false, }, @@ -85,7 +76,7 @@ describe("message submission transactions", () => { }); it("does not roll back a provider-acknowledged prompt on a later transport error", () => { - const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); + const sending = beginMessageSubmission([], { clientMessageId: "client-1" }); const observed = observeMessageSubmissionCanonical(sending, ["client-1"]); expect(rejectMessageSubmission(observed, "client-1")).toEqual({ @@ -95,7 +86,7 @@ describe("message submission transactions", () => { }); it("rejects an unacknowledged transaction", () => { - const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); + const sending = beginMessageSubmission([], { clientMessageId: "client-1" }); expect(rejectMessageSubmission(sending, "client-1")).toEqual({ outcome: "rejected", @@ -104,10 +95,10 @@ describe("message submission transactions", () => { }); it("does not create duplicate transaction identity", () => { - const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); + const sending = beginMessageSubmission([], { clientMessageId: "client-1" }); - expect(() => - beginMessageSubmission(sending, { clientMessageId: "client-1", submittedAt }), - ).toThrow("Message submission already exists"); + expect(() => beginMessageSubmission(sending, { clientMessageId: "client-1" })).toThrow( + "Message submission already exists", + ); }); }); diff --git a/packages/app/src/composer/submission/model.ts b/packages/app/src/composer/submission/model.ts index 4246f6a938..26febae04f 100644 --- a/packages/app/src/composer/submission/model.ts +++ b/packages/app/src/composer/submission/model.ts @@ -1,6 +1,5 @@ export interface PendingMessageSubmission { clientMessageId: string; - submittedAt: Date; } export type MessageSubmissionRecord = PendingMessageSubmission & { diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index b7102c9d68..fc529273ef 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -387,7 +387,7 @@ export function selectAgentTurnPresentation( ): TurnPresentation { return resolveTurnPresentation( session?.agentTurnLiveness.get(agentId) ?? TURN_LIVENESS_IDLE, - getActiveMessageSubmissions(session?.messageSubmissions.get(agentId)), + getActiveMessageSubmissions(session?.messageSubmissions.get(agentId)).length > 0, ); } @@ -1237,7 +1237,7 @@ export const useSessionStore = create()( }); const submissions = beginMessageSubmission( session.messageSubmissions.get(agentId) ?? [], - { clientMessageId: message.clientMessageId, submittedAt: message.timestamp }, + { clientMessageId: message.clientMessageId }, ); const messageSubmissions = new Map(session.messageSubmissions); messageSubmissions.set(agentId, submissions); diff --git a/packages/app/src/timeline/turn-liveness.test.ts b/packages/app/src/timeline/turn-liveness.test.ts index 2146b50615..b3a39742d3 100644 --- a/packages/app/src/timeline/turn-liveness.test.ts +++ b/packages/app/src/timeline/turn-liveness.test.ts @@ -68,11 +68,7 @@ describe("turn activity", () => { }); it("keeps submission-only activity untimed", () => { - expect( - resolveTurnPresentation(TURN_LIVENESS_IDLE, [ - { clientMessageId: "message-1", submittedAt: startedAt }, - ]), - ).toEqual({ + expect(resolveTurnPresentation(TURN_LIVENESS_IDLE, true)).toEqual({ isActive: true, isCancelling: false, startedAt: null, diff --git a/packages/app/src/timeline/turn-liveness.ts b/packages/app/src/timeline/turn-liveness.ts index a817af38c3..deadee830e 100644 --- a/packages/app/src/timeline/turn-liveness.ts +++ b/packages/app/src/timeline/turn-liveness.ts @@ -56,9 +56,8 @@ export interface TurnPresentation { export function resolveTurnPresentation( liveness: TurnLiveness, - submissions: readonly { clientMessageId: string; submittedAt: Date }[], + hasActiveSubmission: boolean, ): TurnPresentation { - const latestSubmission = submissions.at(-1); if (liveness.phase === "open") { return { isActive: true, @@ -68,7 +67,7 @@ export function resolveTurnPresentation( }; } return { - isActive: latestSubmission !== undefined, + isActive: hasActiveSubmission, isCancelling: liveness.cancellationRequestId !== null, startedAt: null, turnId: null, diff --git a/packages/app/src/utils/agent-directory-sync.test.ts b/packages/app/src/utils/agent-directory-sync.test.ts index 94bda7d622..71fb4cc350 100644 --- a/packages/app/src/utils/agent-directory-sync.test.ts +++ b/packages/app/src/utils/agent-directory-sync.test.ts @@ -157,7 +157,6 @@ describe("message submission authority", () => { expect(useSessionStore.getState().sessions[serverId]?.messageSubmissions.get(agentId)).toEqual([ { clientMessageId, - submittedAt: new Date("2026-07-27T10:00:00.000Z"), providerAcknowledged: false, rpcSettled: false, },