diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 2beddbb96..b881744f5 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -87,11 +87,10 @@ its completion advances `seqEnd`, followed by a merged assistant message. The ap remaining page through the existing stream reducer. It must not append full projected text to a live prefix. -Optimistic user prompts are presentation state rather than canonical history. Incremental catch-up -temporarily separates them, applies canonical entries, lets canonical user rows reconcile through -the existing optimistic-message rules, then restores any unmatched prompts after the caught-up -history. This keeps late history before a newly submitted prompt without duplicating an -acknowledged prompt. +Optimistic user prompts occupy stable timeline slots. Catch-up never extracts, delays, or reinserts +them. A canonical user row replaces its matching slot in place; an unmatched prompt stays exactly +where the user submitted it. Other canonical rows are applied after the already-present timeline +instead of relocating visible user messages around newly fetched history. Canonical submitted user rows carry the provider's `messageId` and Paseo's optional `clientMessageId`. Clients reconcile optimistic prompts by `clientMessageId`. Content matching is diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 1b84d2e15..491da6763 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -415,6 +415,82 @@ describe("processTimelineResponse", () => { expect(result.tail).toEqual([optimistic]); }); + it("does not move an unmatched submission during timeline replacement", () => { + const unmatched = makeOptimisticUserMessage("first submission", "client-first"); + const acknowledged: StreamItem[] = [ + { + kind: "user_message", + id: "provider-second", + clientMessageId: "client-second", + text: "second submission", + timestamp: new Date(2000), + }, + { + kind: "user_message", + id: "provider-third", + clientMessageId: "client-third", + text: "third submission", + timestamp: new Date(3000), + }, + ]; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [unmatched, ...acknowledged], + payload: { + ...baseTimelineInput.payload, + reset: true, + startCursor: { seq: 2 }, + endCursor: { seq: 4 }, + entries: [ + { + ...makeTimelineEntry(2, "second submission", "user_message"), + item: { + type: "user_message", + text: "second submission", + messageId: "provider-second", + clientMessageId: "client-second", + }, + }, + { + ...makeTimelineEntry(3, "third submission", "user_message"), + item: { + type: "user_message", + text: "third submission", + messageId: "provider-third", + clientMessageId: "client-third", + }, + }, + { + ...makeTimelineEntry(4, "response to all three submissions"), + item: { + type: "assistant_message", + text: "response to all three submissions", + messageId: "assistant-response", + }, + }, + ], + }, + }); + + expect( + result.tail.map((item) => ({ + kind: item.kind, + id: item.id, + text: "text" in item ? item.text : undefined, + })), + ).toEqual([ + { kind: "user_message", id: "client-first", text: "first submission" }, + { kind: "user_message", id: "provider-second", text: "second submission" }, + { kind: "user_message", id: "provider-third", text: "third submission" }, + { + kind: "assistant_message", + id: "assistant-response", + text: "response to all three submissions", + }, + ]); + }); + it("sets cursor to null when reset=true but no cursors in payload", () => { const existingTail: StreamItem[] = [ { @@ -923,7 +999,7 @@ describe("processTimelineResponse", () => { expect(thoughts[0]?.text).toBe("Thinking"); }); - it("keeps delayed catch-up history before a newly submitted prompt", () => { + it("does not move a submitted prompt when catch-up history arrives", () => { const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); const result = processTimelineResponse({ @@ -944,10 +1020,10 @@ describe("processTimelineResponse", () => { [...result.tail, ...result.head] .filter((item) => item.kind === "assistant_message" || item.kind === "user_message") .map((item) => item.text), - ).toEqual(["Earlier answer", "Missed answer", "New prompt"]); + ).toEqual(["Earlier answer", "New prompt", "Missed answer"]); }); - it("keeps delayed catch-up history between a live head and its unmatched head prompt", () => { + it("does not move an unmatched head prompt when catch-up history arrives", () => { const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); const result = processTimelineResponse({ @@ -971,12 +1047,12 @@ describe("processTimelineResponse", () => { expect([...result.tail, ...result.head].map((item) => item.kind)).toEqual([ "assistant_message", - "tool_call", "user_message", + "tool_call", ]); }); - it("keeps delayed catch-up history between a live head and its acknowledged head prompt", () => { + it("acknowledges a head prompt in place while catch-up history arrives", () => { const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); const result = processTimelineResponse({ @@ -1008,8 +1084,8 @@ describe("processTimelineResponse", () => { expect([...result.tail, ...result.head].map((item) => item.kind)).toEqual([ "assistant_message", - "tool_call", "user_message", + "tool_call", ]); expect( [...result.tail, ...result.head] @@ -1018,7 +1094,7 @@ describe("processTimelineResponse", () => { ).toEqual([undefined]); }); - it("keeps unrelated delayed history before the prompt and its live response", () => { + it("does not move a prompt around unrelated catch-up history", () => { const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); const result = processTimelineResponse({ @@ -1037,7 +1113,14 @@ describe("processTimelineResponse", () => { startCursor: { seq: 3 }, endCursor: { seq: 4 }, entries: [ - makeTimelineEntry(3, "Missed answer"), + { + ...makeTimelineEntry(3, "Missed answer"), + item: { + type: "assistant_message", + text: "Missed answer", + messageId: "missed-answer", + }, + }, { ...makeTimelineEntry(4, "Remote prompt", "user_message"), item: { @@ -1054,10 +1137,10 @@ describe("processTimelineResponse", () => { [...result.tail, ...result.head] .filter((item) => item.kind === "assistant_message" || item.kind === "user_message") .map((item) => item.text), - ).toEqual(["Earlier answer", "Missed answer", "Remote prompt", "New prompt", "Live response"]); + ).toEqual(["Earlier answer", "New prompt", "Live response", "Missed answer", "Remote prompt"]); }); - it("keeps delayed history before a prompt whose live answer has promoted blocks", () => { + it("does not move a prompt or its live answer around catch-up history", () => { const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); const live = processAgentStreamEvents({ events: [ @@ -1082,7 +1165,16 @@ describe("processTimelineResponse", () => { epoch: "epoch-1", startCursor: { seq: 3 }, endCursor: { seq: 3 }, - entries: [makeTimelineEntry(3, "Missed answer")], + entries: [ + { + ...makeTimelineEntry(3, "Missed answer"), + item: { + type: "assistant_message", + text: "Missed answer", + messageId: "missed-answer", + }, + }, + ], }, }); @@ -1090,10 +1182,10 @@ describe("processTimelineResponse", () => { [...result.tail, ...result.head] .filter((item) => item.kind === "assistant_message" || item.kind === "user_message") .map((item) => item.text), - ).toEqual(["Missed answer", "New prompt", "First paragraph.", "Second paragraph"]); + ).toEqual(["New prompt", "First paragraph.", "Second paragraph", "Missed answer"]); }); - it("keeps delayed tool history before a prompt whose live answer has promoted blocks", () => { + it("does not move a prompt or its live answer around catch-up tool history", () => { const prompt = makeOptimisticUserMessage("New prompt", "new-prompt"); const live = processAgentStreamEvents({ events: [ @@ -1128,14 +1220,73 @@ describe("processTimelineResponse", () => { }); expect([...result.tail, ...result.head].map((item) => item.kind)).toEqual([ - "tool_call", "user_message", "assistant_message", "assistant_message", + "tool_call", ]); }); - it("matches a local optimistic prompt after an unrelated remote user row", () => { + it("never moves submitted messages behind a later assistant response", () => { + const unmatched = makeOptimisticUserMessage("first submission", "client-first"); + const acknowledged: StreamItem[] = [ + { + kind: "user_message", + id: "provider-second", + clientMessageId: "client-second", + text: "second submission", + timestamp: new Date(2000), + }, + { + kind: "user_message", + id: "provider-third", + clientMessageId: "client-third", + text: "third submission", + timestamp: new Date(3000), + }, + ]; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentTail: [unmatched, ...acknowledged], + currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 3 }, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + startCursor: { seq: 4 }, + endCursor: { seq: 4 }, + entries: [ + { + ...makeTimelineEntry(4, "response to all three submissions"), + item: { + type: "assistant_message", + text: "response to all three submissions", + messageId: "assistant-response", + }, + }, + ], + }, + }); + + expect( + [...result.tail, ...result.head].map((item) => ({ + kind: item.kind, + id: item.id, + text: "text" in item ? item.text : undefined, + })), + ).toEqual([ + { kind: "user_message", id: "client-first", text: "first submission" }, + { kind: "user_message", id: "provider-second", text: "second submission" }, + { kind: "user_message", id: "provider-third", text: "third submission" }, + { + kind: "assistant_message", + id: "assistant-response", + text: "response to all three submissions", + }, + ]); + }); + + it("acknowledges a local prompt in place when a remote user row also arrives", () => { const prompt = makeOptimisticUserMessage("Local prompt", "local-prompt"); const result = processTimelineResponse({ @@ -1173,8 +1324,8 @@ describe("processTimelineResponse", () => { .filter((item) => item.kind === "user_message") .map((item) => ({ text: item.text, optimistic: item.optimistic })), ).toEqual([ - { text: "Remote prompt", optimistic: undefined }, { text: "Local prompt", optimistic: undefined }, + { text: "Remote prompt", optimistic: undefined }, ]); }); @@ -1208,8 +1359,8 @@ describe("processTimelineResponse", () => { .filter((item) => item.kind === "user_message") .map((item) => ({ text: item.text, optimistic: item.optimistic })), ).toEqual([ - { text: "Remote prompt", optimistic: undefined }, { text: "Local prompt", optimistic: true }, + { text: "Remote prompt", optimistic: undefined }, ]); }); @@ -1244,8 +1395,8 @@ describe("processTimelineResponse", () => { .filter((item) => item.kind === "user_message") .map((item) => ({ id: item.id, optimistic: item.optimistic })), ).toEqual([ - { id: "remote-prompt", optimistic: undefined }, { id: "local-prompt", optimistic: true }, + { id: "remote-prompt", optimistic: undefined }, ]); }); diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 1ae2708f6..7458e855d 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -328,7 +328,7 @@ interface CanonicalUserMessageIdentity { text: string; } -function matchesOptimisticUserMessageIdentity( +function matchesLocalUserMessageIdentity( canonical: CanonicalUserMessageIdentity, optimistic: UserMessageItem, ): boolean { @@ -363,61 +363,55 @@ function reconcileLocalUserPresentationAfterReplace(params: { } }); - let changed = false; const nextTail = [...params.canonicalTail]; - let searchFromOrdinal = 0; + const claimedCanonicalIndexes = new Set(); const unmatched: UserMessageItem[] = []; for (const local of localUsers) { - const canonicalOrdinal = canonicalUserIndexes.findIndex((index, ordinal) => { - if (ordinal < searchFromOrdinal) { - return false; - } - if (local.item.optimistic) { - const canonical = params.canonicalTail[index]; - if (canonical?.kind !== "user_message") { - return false; - } - const identityMatches = matchesOptimisticUserMessageIdentity( + const exactIndex = canonicalUserIndexes.find((index) => { + if (claimedCanonicalIndexes.has(index)) return false; + const canonical = params.canonicalTail[index]; + return ( + canonical?.kind === "user_message" && + matchesLocalUserMessageIdentity( { messageId: canonical.id, clientMessageId: canonical.clientMessageId, text: canonical.text, }, local.item, - ); - if (canonical.clientMessageId !== undefined || identityMatches) { - return identityMatches; - } - return ordinal >= local.ordinal; - } - return params.canonicalTail[index]?.id === local.item.id; + ) + ); }); - if (canonicalOrdinal < 0) { - if (local.item.optimistic) { - unmatched.push(local.item); - } - continue; - } - - const canonicalIndex = canonicalUserIndexes[canonicalOrdinal]; - const canonicalItem = canonicalIndex !== undefined ? nextTail[canonicalIndex] : undefined; - if (!canonicalItem || canonicalItem.kind !== "user_message") { + const ordinalIndex = canonicalUserIndexes[local.ordinal]; + const ordinalItem = ordinalIndex === undefined ? undefined : params.canonicalTail[ordinalIndex]; + const canonicalIndex = + exactIndex ?? + (ordinalIndex !== undefined && + !claimedCanonicalIndexes.has(ordinalIndex) && + ordinalItem?.kind === "user_message" && + ordinalItem.clientMessageId === undefined + ? ordinalIndex + : undefined); + const canonicalItem = canonicalIndex === undefined ? undefined : nextTail[canonicalIndex]; + if (canonicalIndex === undefined || !canonicalItem || canonicalItem.kind !== "user_message") { if (local.item.optimistic) { unmatched.push(local.item); } continue; } nextTail[canonicalIndex] = mergeCanonicalUserWithLocalPresentation(canonicalItem, local.item); - searchFromOrdinal = canonicalOrdinal + 1; - changed = true; + claimedCanonicalIndexes.add(canonicalIndex); } - if (unmatched.length === 0) { - return changed ? nextTail : params.canonicalTail; + for (const item of unmatched) { + const insertionIndex = nextTail.findIndex( + (canonical) => canonical.timestamp.getTime() > item.timestamp.getTime(), + ); + nextTail.splice(insertionIndex < 0 ? nextTail.length : insertionIndex, 0, item); } - return [...nextTail, ...unmatched]; + return nextTail; } interface IncrementalAcceptResult { @@ -582,62 +576,6 @@ function replaceLiveAssistantWithProjectedText(params: { return next; } -interface OptimisticUserMessagePosition { - item: UserMessageItem; - placement: "tail" | "head"; - index: number; -} - -function findOptimisticUserMessage(params: { - tail: StreamItem[]; - head: StreamItem[]; -}): OptimisticUserMessagePosition | null { - const tailIndex = params.tail.findIndex( - (item) => item.kind === "user_message" && item.optimistic, - ); - const tailItem = params.tail[tailIndex]; - if (tailItem?.kind === "user_message") { - return { item: tailItem, placement: "tail", index: tailIndex }; - } - - const headIndex = params.head.findIndex( - (item) => item.kind === "user_message" && item.optimistic, - ); - const headItem = params.head[headIndex]; - return headItem?.kind === "user_message" - ? { item: headItem, placement: "head", index: headIndex } - : null; -} - -function replaceOptimisticUserMessage(params: { - tail: StreamItem[]; - head: StreamItem[]; - position: OptimisticUserMessagePosition; - precedingItems: StreamItem[]; - replacement: StreamItem[]; -}): { tail: StreamItem[]; head: StreamItem[] } { - const { position } = params; - const items = [...params.precedingItems, ...params.replacement]; - if (position.placement === "tail") { - return { - tail: [ - ...params.tail.slice(0, position.index), - ...items, - ...params.tail.slice(position.index + 1), - ], - head: params.head, - }; - } - return { - tail: params.tail, - head: [ - ...params.head.slice(0, position.index), - ...items, - ...params.head.slice(position.index + 1), - ], - }; -} - function reconcileOverlappingProjectedAssistant(params: { tail: StreamItem[]; head: StreamItem[]; @@ -802,40 +740,6 @@ function reconcileOverlappingProjectedStreamItems(params: { return { tail, head, reconciledUnits }; } -function matchesOptimisticUserMessage(params: { - unit: TimelineUnit; - optimistic: UserMessageItem; -}): boolean { - const { event } = params.unit; - if (event.type !== "timeline" || event.item.type !== "user_message") { - return false; - } - return matchesOptimisticUserMessageIdentity(event.item, params.optimistic); -} - -function acknowledgeOptimisticUserMessage(params: { - tail: StreamItem[]; - head: StreamItem[]; - unit: TimelineUnit; - epoch: string; - optimistic: OptimisticUserMessagePosition; - precedingItems: StreamItem[]; -}): { tail: StreamItem[]; head: StreamItem[] } { - const { event, timestamp, seqEnd } = params.unit; - const timelineCursor = { epoch: params.epoch, seq: seqEnd }; - const acknowledged = reduceStreamUpdate([params.optimistic.item], event, timestamp, { - source: "canonical", - timelineCursor, - }); - return replaceOptimisticUserMessage({ - tail: params.tail, - head: params.head, - position: params.optimistic, - precedingItems: params.precedingItems, - replacement: acknowledged, - }); -} - function applyCanonicalForwardUnit(params: { tail: StreamItem[]; head: StreamItem[]; @@ -861,6 +765,25 @@ function applyCanonicalForwardUnit(params: { }); if (replacedHead) return { tail: params.tail, head: replacedHead }; + const activeAssistant = params.head.findLast( + (item): item is Extract => + item.kind === "assistant_message", + ); + if ( + event.type === "timeline" && + event.item.type === "assistant_message" && + event.item.messageId !== undefined && + event.item.messageId !== activeAssistant?.messageId + ) { + return { + tail: flushHeadToTail(params.tail, params.head), + head: reduceStreamUpdate([], event, timestamp, { + source: "canonical", + timelineCursor, + }), + }; + } + const applied = applyStreamEvent({ tail: params.tail, head: params.head, @@ -888,67 +811,15 @@ function applyAcceptedForwardTimelineUnits(params: { }); let tail = reconciled.tail; let head = reconciled.head; - let delayedHistoryTail: StreamItem[] = []; - let delayedHistoryHead: StreamItem[] = []; for (const unit of params.units) { if (reconciled.reconciledUnits.has(unit)) continue; - const nextOptimistic = findOptimisticUserMessage({ tail, head }); - if (nextOptimistic && matchesOptimisticUserMessage({ unit, optimistic: nextOptimistic.item })) { - const precedingItems = flushHeadToTail(delayedHistoryTail, delayedHistoryHead); - delayedHistoryTail = []; - delayedHistoryHead = []; - const applied = acknowledgeOptimisticUserMessage({ - tail, - head, - unit, - epoch: params.epoch, - optimistic: nextOptimistic, - precedingItems, - }); - tail = applied.tail; - head = applied.head; - continue; - } - if (nextOptimistic) { - if ( - unit.event.type === "timeline" && - (unit.event.item.type === "assistant_message" || unit.event.item.type === "reasoning") - ) { - delayedHistoryHead = reduceStreamUpdate(delayedHistoryHead, unit.event, unit.timestamp, { - source: "canonical", - timelineCursor: { epoch: params.epoch, seq: unit.seqEnd }, - }); - continue; - } - const applied = applyStreamEvent({ - tail: delayedHistoryTail, - head: delayedHistoryHead, - event: unit.event, - timestamp: unit.timestamp, - source: "canonical", - timelineCursor: { epoch: params.epoch, seq: unit.seqEnd }, - }); - delayedHistoryTail = applied.tail; - delayedHistoryHead = applied.head; - continue; - } const applied = applyCanonicalForwardUnit({ tail, head, unit, epoch: params.epoch }); tail = applied.tail; head = applied.head; } - const remainingOptimistic = findOptimisticUserMessage({ tail, head }); - if (!remainingOptimistic) return { tail, head }; - const precedingItems = flushHeadToTail(delayedHistoryTail, delayedHistoryHead); - if (precedingItems.length === 0) return { tail, head }; - return replaceOptimisticUserMessage({ - tail, - head, - position: remainingOptimistic, - precedingItems, - replacement: [remainingOptimistic.item], - }); + return { tail, head }; } function applyTimelineIncrementalPath(args: { diff --git a/packages/app/src/types/stream.test.ts b/packages/app/src/types/stream.test.ts index 6d3cc93e2..fb68d7a02 100644 --- a/packages/app/src/types/stream.test.ts +++ b/packages/app/src/types/stream.test.ts @@ -1242,6 +1242,7 @@ describe("turn lifecycle events", () => { type: "user_message", text: "server-rendered attachment text", messageId: "provider-owned-canonical", + clientMessageId: optimistic.id, }, }, new Date("2025-01-01T15:03:11Z"), diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index edd1ab36d..76407d84c 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -354,6 +354,7 @@ function appendUserMessage( state: StreamItem[], text: string, timestamp: Date, + source: StreamUpdateSource, messageId?: string, clientMessageId?: string, ): StreamItem[] { @@ -368,7 +369,9 @@ function appendUserMessage( (entry) => entry.kind === "user_message" && entry.optimistic && - (clientMessageId === undefined || entry.id === clientMessageId), + (clientMessageId !== undefined + ? entry.id === clientMessageId + : source === "live" || entry.id === messageId || entry.text === chunk), ); const optimistic = optimisticIndex >= 0 ? (state[optimisticIndex] as UserMessageItem) : null; @@ -843,7 +846,14 @@ function reduceTimelineEvent( switch (item.type) { case "user_message": return finalizeActiveThoughts( - appendUserMessage(state, item.text, timestamp, item.messageId, item.clientMessageId), + appendUserMessage( + state, + item.text, + timestamp, + source, + item.messageId, + item.clientMessageId, + ), ); case "assistant_message": return finalizeActiveThoughts(