From 3d4e6bb71ebc55eb7b7bb789fd6c6fdf5dc23e07 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 27 Jul 2026 15:40:23 +0200 Subject: [PATCH] fix(app): close canonical submission races --- .../app/src/composer/submission/model.test.ts | 39 +++++++ packages/app/src/composer/submission/model.ts | 38 +++++- packages/app/src/contexts/session-context.tsx | 41 ++----- packages/app/src/stores/session-store.ts | 29 ++++- .../timeline/session-stream-reducers.test.ts | 33 ++++++ .../src/timeline/session-stream-reducers.ts | 4 + packages/app/src/types/stream.test.ts | 53 ++++++++- packages/app/src/types/stream.ts | 108 ++++++++++-------- .../src/utils/agent-directory-sync.test.ts | 33 ++++++ 9 files changed, 295 insertions(+), 83 deletions(-) diff --git a/packages/app/src/composer/submission/model.test.ts b/packages/app/src/composer/submission/model.test.ts index a1ef7ea1b6..a972e44f90 100644 --- a/packages/app/src/composer/submission/model.test.ts +++ b/packages/app/src/composer/submission/model.test.ts @@ -5,6 +5,7 @@ import { beginMessageSubmission, getPendingMessageSubmission, observeMessageSubmissionRunning, + observeMessageSubmissionCanonical, rejectMessageSubmission, type MessageSubmissionState, } from "./model"; @@ -40,6 +41,44 @@ describe("message submission model", () => { expect(runningObserved).toEqual([]); }); + it("closes RPC-accepted pending state when canonical history proves acceptance", () => { + const message = createUserMessage({ + clientMessageId: "client-1", + text: "hello", + timestamp: new Date("2026-07-26T10:00:00.000Z"), + }); + const begun = beginMessageSubmission(initialState(), { + message, + submittedAt: message.timestamp, + acceptance: "rpc-and-running", + }); + const accepted = acceptMessageSubmission(begun, "client-1"); + + expect(observeMessageSubmissionCanonical(accepted.submissions, ["client-1"])).toEqual([]); + }); + + it("remembers canonical acceptance until an in-flight RPC rejection settles", () => { + const message = createUserMessage({ + clientMessageId: "client-1", + text: "hello", + timestamp: new Date("2026-07-26T10:00:00.000Z"), + }); + const begun = beginMessageSubmission(initialState(), { + message, + submittedAt: message.timestamp, + acceptance: "rpc-and-running", + }); + const canonicalObserved = observeMessageSubmissionCanonical(begun.submissions, ["client-1"]); + + expect(getPendingMessageSubmission(canonicalObserved)).toBeNull(); + expect( + rejectMessageSubmission({ ...begun, submissions: canonicalObserved }, "client-1"), + ).toEqual({ + outcome: "accepted", + state: { ...begun, submissions: [] }, + }); + }); + it("rejects a row when an unrelated running transition precedes its RPC error", () => { const message = createUserMessage({ clientMessageId: "client-1", diff --git a/packages/app/src/composer/submission/model.ts b/packages/app/src/composer/submission/model.ts index c9dc60e9e5..f7b2ad67ae 100644 --- a/packages/app/src/composer/submission/model.ts +++ b/packages/app/src/composer/submission/model.ts @@ -11,12 +11,13 @@ export type MessageSubmissionRecord = PendingMessageSubmission & ( | { phase: "waiting-for-rpc"; acceptance: MessageSubmissionAcceptance } | { phase: "waiting-for-running" } + | { phase: "canonical-observed" } ); export function getPendingMessageSubmission( submissions: readonly MessageSubmissionRecord[] | null | undefined, ): PendingMessageSubmission | null { - return submissions?.at(-1) ?? null; + return submissions?.findLast((submission) => submission.phase !== "canonical-observed") ?? null; } export interface MessageSubmissionState { @@ -75,6 +76,11 @@ export function acceptMessageSubmission( ); if (index === -1) return state; const submission = state.submissions[index]; + if (submission.phase === "canonical-observed") { + const submissions = state.submissions.slice(); + submissions.splice(index, 1); + return { ...state, submissions }; + } if (submission.phase === "waiting-for-running") return state; const nextSubmission: MessageSubmissionRecord | null = submission.acceptance === "rpc-only" || isAgentRunning @@ -97,6 +103,34 @@ export function observeMessageSubmissionRunning( return next.length === submissions.length ? (submissions as MessageSubmissionRecord[]) : next; } +export function observeMessageSubmissionCanonical( + submissions: readonly MessageSubmissionRecord[], + clientMessageIds: readonly string[], +): MessageSubmissionRecord[] { + if (clientMessageIds.length === 0) return submissions as MessageSubmissionRecord[]; + const observed = new Set(clientMessageIds); + let changed = false; + const next = submissions.flatMap((submission): MessageSubmissionRecord[] => { + if (!observed.has(submission.clientMessageId)) return [submission]; + if (submission.phase === "waiting-for-running") { + changed = true; + return []; + } + if (submission.phase === "waiting-for-rpc") { + changed = true; + return [ + { + clientMessageId: submission.clientMessageId, + submittedAt: submission.submittedAt, + phase: "canonical-observed", + }, + ]; + } + return [submission]; + }); + return changed ? next : (submissions as MessageSubmissionRecord[]); +} + function removeSubmittedMessage(items: StreamItem[], clientMessageId: string): StreamItem[] { const next = items.filter( (item) => item.kind !== "user_message" || item.clientMessageId !== clientMessageId, @@ -115,7 +149,7 @@ export function rejectMessageSubmission( const submission = state.submissions[index]; const submissions = state.submissions.slice(); submissions.splice(index, 1); - if (submission.phase === "waiting-for-running") { + if (submission.phase === "waiting-for-running" || submission.phase === "canonical-observed") { return { outcome: "accepted", state: { ...state, submissions } }; } return { diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 964ecf150f..0ac2a16699 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -192,9 +192,7 @@ type WorkspaceSetupProgressPayload = Extract< type SessionStoreActions = ReturnType; type SetInitializingAgents = SessionStoreActions["setInitializingAgents"]; -type SetAgentStreamTail = SessionStoreActions["setAgentStreamTail"]; -type SetAgentStreamHead = SessionStoreActions["setAgentStreamHead"]; -type ClearAgentStreamHead = SessionStoreActions["clearAgentStreamHead"]; +type SetAgentStreamState = SessionStoreActions["setAgentStreamState"]; type SetAgentTimelineCursor = SessionStoreActions["setAgentTimelineCursor"]; type MarkAgentHistorySynchronized = SessionStoreActions["markAgentHistorySynchronized"]; type SetAgentAuthoritativeHistoryApplied = @@ -237,9 +235,7 @@ function applyTimelineStreamPatches(input: { serverId: string; currentTail: StreamItem[]; currentHead: StreamItem[]; - setAgentStreamTail: SetAgentStreamTail; - setAgentStreamHead: SetAgentStreamHead; - clearAgentStreamHead: ClearAgentStreamHead; + setAgentStreamState: SetAgentStreamState; setAgentTimelineCursor: SetAgentTimelineCursor; }): void { const { @@ -248,32 +244,17 @@ function applyTimelineStreamPatches(input: { serverId, currentTail, currentHead, - setAgentStreamTail, - setAgentStreamHead, - clearAgentStreamHead, + setAgentStreamState, setAgentTimelineCursor, } = input; - if (result.tail !== currentTail) { - setAgentStreamTail(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, result.tail); - return next; + if (result.tail !== currentTail || result.head !== currentHead) { + setAgentStreamState(serverId, agentId, { + ...(result.tail !== currentTail ? { tail: result.tail } : {}), + ...(result.head !== currentHead ? { head: result.head } : {}), }); } - if (result.head !== currentHead) { - if (result.head.length === 0) { - clearAgentStreamHead(serverId, agentId); - } else { - setAgentStreamHead(serverId, (prev) => { - const next = new Map(prev); - next.set(agentId, result.head); - return next; - }); - } - } - if (result.cursorChanged) { setAgentTimelineCursor(serverId, (prev) => { const current = prev.get(agentId); @@ -701,9 +682,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider serverId, currentTail, currentHead, - setAgentStreamTail, - setAgentStreamHead, - clearAgentStreamHead, + setAgentStreamState, setAgentTimelineCursor, }); @@ -725,13 +704,11 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider }); }, [ - clearAgentStreamHead, markAgentHistorySynchronized, recoverTimelineGap, serverId, setAgentAuthoritativeHistoryApplied, - setAgentStreamHead, - setAgentStreamTail, + setAgentStreamState, setAgentTimelineCursor, setAgentTimelineHasOlder, setInitializingAgents, diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 71f53ce8f8..c86c8e2ecb 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -13,6 +13,7 @@ import { acceptMessageSubmission, beginMessageSubmission, observeMessageSubmissionRunning, + observeMessageSubmissionCanonical, rejectMessageSubmission, type MessageSubmissionRecord, type MessageSubmissionRejectionOutcome, @@ -1105,10 +1106,35 @@ export const useSessionStore = create()( } } - if (!changedTail && !changedHead) { + const canonicalClientMessageIds = [ + ...(state.tail ?? session.agentStreamTail.get(agentId) ?? []), + ...(state.head ?? session.agentStreamHead.get(agentId) ?? []), + ].flatMap((item) => + item.kind === "user_message" && item.clientMessageId && item.messageId + ? [item.clientMessageId] + : [], + ); + const currentSubmissions = session.messageSubmissions.get(agentId) ?? []; + const observedSubmissions = observeMessageSubmissionCanonical( + currentSubmissions, + canonicalClientMessageIds, + ); + const changedSubmissions = observedSubmissions !== currentSubmissions; + + if (!changedTail && !changedHead && !changedSubmissions) { return prev; } + let messageSubmissions = session.messageSubmissions; + if (changedSubmissions) { + messageSubmissions = new Map(session.messageSubmissions); + if (observedSubmissions.length > 0) { + messageSubmissions.set(agentId, observedSubmissions); + } else { + messageSubmissions.delete(agentId); + } + } + return { ...prev, sessions: { @@ -1117,6 +1143,7 @@ export const useSessionStore = create()( ...session, agentStreamTail: nextTail, agentStreamHead: nextHead, + messageSubmissions, }, }, }; diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index d3604caeca..c53b24419e 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -330,6 +330,39 @@ describe("processTimelineResponse", () => { expect(result.head).toEqual([{ ...liveAssistant, text: "Live" }, submitted]); }); + it("preserves newer live head items when canonical replacement ends in a tool call", () => { + const liveThought: StreamItem = { + kind: "thought", + id: "live-thought", + text: "newer reasoning", + timestamp: new Date(3000), + status: "loading", + }; + const liveAssistant = makeAssistantItem("newer answer", "live-answer"); + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentHead: [liveThought, liveAssistant], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 1 }, + payload: { + ...baseTimelineInput.payload, + reset: true, + epoch: "epoch-1", + startCursor: { seq: 1 }, + endCursor: { seq: 1 }, + entries: [ + makeToolCallTimelineEntry(1, "canonical-call", "completed", { + type: "read", + filePath: "/tmp/older.ts", + }), + ], + }, + }); + + expect(result.tail.map((item) => item.kind)).toEqual(["tool_call"]); + expect(result.head).toEqual([liveThought, liveAssistant]); + }); + it("uses the timeline entry timestamp as canonical", () => { const result = processTimelineResponse({ ...baseTimelineInput, diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 533896e95c..43c96e2802 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -233,6 +233,7 @@ function applyTimelineReplacePath(args: { currentHead: StreamItem[]; pendingClientMessageId: string | null; replacementPolicy: "preserve-local-submissions" | "replace-history"; + preserveLiveHead: boolean; toHydratedEvents: ( units: TimelineUnit[], ) => Array<{ event: AgentStreamEventPayload; timestamp: Date }>; @@ -245,6 +246,7 @@ function applyTimelineReplacePath(args: { currentHead, pendingClientMessageId, replacementPolicy, + preserveLiveHead, toHydratedEvents, } = args; const hydratedTail = hydrateStreamState(toHydratedEvents(timelineUnits), { source: "canonical" }); @@ -254,6 +256,7 @@ function applyTimelineReplacePath(args: { previousHead: currentHead, pendingClientMessageId, policy: replacementPolicy, + preserveLiveHead, }); const cursor: TimelineCursor | null = payload.startCursor && payload.endCursor @@ -867,6 +870,7 @@ export function processTimelineResponse( currentHead, pendingClientMessageId, replacementPolicy, + preserveLiveHead: currentCursor?.epoch === payload.epoch, toHydratedEvents, }) : applyTimelineIncrementalPath({ diff --git a/packages/app/src/types/stream.test.ts b/packages/app/src/types/stream.test.ts index 36c277b52f..00d73b7f65 100644 --- a/packages/app/src/types/stream.test.ts +++ b/packages/app/src/types/stream.test.ts @@ -1486,7 +1486,7 @@ describe("turn lifecycle events", () => { ); }); - it("reconciles a submitted user message that was pending in the streaming head", () => { + it("flushes an interrupted head when its submitted prompt becomes canonical", () => { const submitted: StreamItem = { kind: "user_message", id: "msg_head_submitted", @@ -1511,13 +1511,60 @@ describe("turn lifecycle events", () => { source: "live", }); - assert.strictEqual(result.tail.length, 0); - const userMessages = result.head.filter((item) => item.kind === "user_message"); + assert.deepStrictEqual(result.head, []); + const userMessages = result.tail.filter((item) => item.kind === "user_message"); assert.strictEqual(userMessages.length, 1); assert.strictEqual(userMessages[0]?.id, "msg_head_submitted"); assert.strictEqual(userMessages[0]?.messageId, "provider-owned-head"); }); + it("keeps a replacement assistant separate after an interrupted prompt is reconciled", () => { + const interruptedAssistant: StreamItem = { + kind: "assistant_message", + id: "interrupted", + text: "old answer", + timestamp: new Date("2025-01-01T15:03:01Z"), + }; + const submitted = createUserMessage({ + clientMessageId: "msg_interrupt", + text: "replacement prompt", + timestamp: new Date("2025-01-01T15:03:02Z"), + }); + const reconciled = applyStreamEvent({ + tail: [], + head: [interruptedAssistant, submitted], + event: { + type: "timeline", + provider: "opencode", + item: { + type: "user_message", + text: submitted.text, + messageId: "provider-prompt", + clientMessageId: submitted.clientMessageId, + }, + }, + timestamp: new Date("2025-01-01T15:03:03Z"), + }); + const replacement = applyStreamEvent({ + tail: reconciled.tail, + head: reconciled.head, + event: { + type: "timeline", + provider: "opencode", + item: { type: "assistant_message", text: "new answer" }, + }, + timestamp: new Date("2025-01-01T15:03:04Z"), + }); + + expect(replacement.tail.map((item) => item.kind)).toEqual([ + "assistant_message", + "user_message", + ]); + expect(replacement.head).toEqual([ + expect.objectContaining({ kind: "assistant_message", text: "new answer" }), + ]); + }); + it("replaces multiple submitted user messages in FIFO order", () => { const submittedTimestamp = new Date("2025-01-01T15:04:00Z"); const serverTimestamp = new Date("2025-01-01T15:04:01Z"); diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index f7410c2b2b..c2e42af376 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -320,6 +320,7 @@ export interface CanonicalStreamReplacementInput { previousHead: StreamItem[]; pendingClientMessageId: string | null; policy: "preserve-local-submissions" | "replace-history"; + preserveLiveHead: boolean; } export interface CanonicalStreamReplacementResult { @@ -334,15 +335,23 @@ function removeUserMessageAt(items: UserMessageItem[], index: number): UserMessa function preserveReplacementHead( tail: StreamItem[], currentHead: StreamItem[], + preserveLiveHead: boolean, ): CanonicalStreamReplacementResult { - const liveAssistantIndex = currentHead.findLastIndex((item) => item.kind === "assistant_message"); - const submittedRows = currentHead.filter( - (item): item is UserMessageItem => - item.kind === "user_message" && item.clientMessageId !== undefined, + const retainedHead = preserveLiveHead + ? currentHead + : currentHead.filter( + (item) => item.kind === "user_message" && item.clientMessageId !== undefined, + ); + const tailIds = new Set(tail.map((item) => item.id)); + const unreconciledHead = retainedHead.filter( + (item) => item.kind === "assistant_message" || !tailIds.has(item.id), ); - if (liveAssistantIndex < 0) return { tail, head: submittedRows }; + const liveAssistantIndex = unreconciledHead.findLastIndex( + (item) => item.kind === "assistant_message", + ); + if (liveAssistantIndex < 0) return { tail, head: unreconciledHead }; - const liveAssistant = currentHead[liveAssistantIndex]; + const liveAssistant = unreconciledHead[liveAssistantIndex]; const tailAssistant = tail.at(-1); if ( liveAssistant.kind !== "assistant_message" || @@ -350,19 +359,14 @@ function preserveReplacementHead( tailAssistant.kind !== "assistant_message" || !liveAssistant.text.startsWith(tailAssistant.text) ) { - return { tail, head: submittedRows }; + return { tail, head: unreconciledHead }; } - const head: StreamItem[] = []; - currentHead.forEach((item, index) => { - if (index === liveAssistantIndex) { - head.push({ ...liveAssistant, text: tailAssistant.text }); - return; - } - if (item.kind === "user_message" && item.clientMessageId) { - head.push(item); - } - }); + const head = [ + ...unreconciledHead.slice(0, liveAssistantIndex), + { ...liveAssistant, text: tailAssistant.text }, + ...unreconciledHead.slice(liveAssistantIndex + 1), + ]; return { tail: tail.slice(0, -1), head }; } @@ -423,7 +427,7 @@ export function replaceWithCanonicalStream( ); }); - return preserveReplacementHead(nextTail, nextHead); + return preserveReplacementHead(nextTail, nextHead, input.preserveLiveHead); } export interface AssistantMessageItem { @@ -1407,6 +1411,45 @@ export interface ApplyStreamEventResult { changedHead: boolean; } +function applyCanonicalUserMessageEvent(params: { + tail: StreamItem[]; + head: StreamItem[]; + event: AgentStreamEventPayload; + timestamp: Date; +}): ApplyStreamEventResult | null { + const { tail, head, event, timestamp } = params; + if (event.type !== "timeline" || event.item.type !== "user_message") return null; + const normalized = normalizeChunk(event.item.text); + if (!normalized.hasContent) { + return { tail, head, changedTail: false, changedHead: false }; + } + + const flushedTail = head.length > 0 ? flushHeadToTail(tail, head) : tail; + const flushedHead = head.length > 0 ? [] : head; + const canonical = createUserMessage({ + id: + event.item.messageId ?? + createUniqueTimelineId([...tail, ...head], "user", normalized.chunk.trim(), timestamp), + messageId: event.item.messageId, + clientMessageId: event.item.clientMessageId, + text: normalized.chunk, + timestamp, + }); + const reconciled = upsertUserMessageAcrossStream({ + tail: flushedTail, + head: flushedHead, + message: canonical, + insert: "tail", + presentation: "existing", + }); + return { + tail: reconciled.tail, + head: reconciled.head, + changedTail: flushedTail !== tail || reconciled.changedTail, + changedHead: flushedHead !== head || reconciled.changedHead, + }; +} + /** * Apply a stream event using head/tail model. * @@ -1427,38 +1470,13 @@ export function applyStreamEvent(params: { timelineCursor?: TimelinePosition; }): ApplyStreamEventResult { const { tail, head, event, timestamp } = params; + const canonicalUserResult = applyCanonicalUserMessageEvent({ tail, head, event, timestamp }); + if (canonicalUserResult) return canonicalUserResult; const source = params.source ?? "live"; let nextTail = tail; let nextHead = head; let changedTail = false; let changedHead = false; - - if (event.type === "timeline" && event.item.type === "user_message") { - const normalized = normalizeChunk(event.item.text); - if (!normalized.hasContent) { - return { tail, head, changedTail: false, changedHead: false }; - } - const canonical = createUserMessage({ - id: - event.item.messageId ?? - createUniqueTimelineId([...tail, ...head], "user", normalized.chunk.trim(), timestamp), - messageId: event.item.messageId, - clientMessageId: event.item.clientMessageId, - text: normalized.chunk, - timestamp, - }); - const reconciled = upsertUserMessageAcrossStream({ - tail, - head, - message: canonical, - insert: "none", - presentation: "existing", - }); - if (reconciled.location) { - return reconciled; - } - } - const flushHead = () => { if (nextHead.length === 0) { return; diff --git a/packages/app/src/utils/agent-directory-sync.test.ts b/packages/app/src/utils/agent-directory-sync.test.ts index 18301c9270..948a8c2619 100644 --- a/packages/app/src/utils/agent-directory-sync.test.ts +++ b/packages/app/src/utils/agent-directory-sync.test.ts @@ -134,6 +134,39 @@ describe("message submission lifecycle ordering", () => { ).toBeUndefined(); store.clearSession(serverId); }); + + it("clears accepted pending state when canonical history arrives after a missed run", () => { + const serverId = "server-canonical-after-missed-run"; + const agentId = "agent-1"; + const store = useSessionStore.getState(); + store.initializeSession(serverId, null as unknown as DaemonClient); + applyAgentStatus({ + serverId, + agentId, + status: "idle", + updatedAt: "2026-07-27T10:00:00.000Z", + }); + const clientMessageId = beginPendingSubmission(serverId, agentId); + store.acceptAgentMessageSubmission(serverId, agentId, clientMessageId); + + store.setAgentStreamState(serverId, agentId, { + tail: [ + createUserMessage({ + id: "provider-message", + messageId: "provider-message", + clientMessageId, + text: "Run this", + timestamp: new Date("2026-07-27T10:00:01.000Z"), + }), + ], + head: [], + }); + + expect( + useSessionStore.getState().sessions[serverId]?.messageSubmissions.get(agentId), + ).toBeUndefined(); + store.clearSession(serverId); + }); }); describe("replaceFetchedAgentDirectory", () => {