diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 88e512abea..7e57c53b96 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -362,18 +362,28 @@ describe("processTimelineResponse", () => { }); it("appends only rows newer than the cursor from an overlapping resume tail", () => { - const loaded: StreamItem = { - kind: "user_message", - id: "loaded-30", - text: "already loaded", + const loaded = createUserMessage({ + id: "local-prompt-30", + clientMessageId: "client-message-30", + text: "local presentation", timestamp: new Date(1030), timelineCursor: { epoch: "epoch-1", seq: 30 }, - }; + attachments: [{ type: "text", mimeType: "text/plain", text: "local attachment" }], + }); + const unrelated = createUserMessage({ + id: "unrelated-row", + messageId: "unrelated-provider-message", + text: "unrelated", + timestamp: new Date(1035), + timelineCursor: { epoch: "epoch-1", seq: 35 }, + }); + const currentTail = [loaded, unrelated]; + const currentCursor: TimelineCursor = { epoch: "epoch-1", startSeq: 1, endSeq: 40 }; const result = processTimelineResponse({ ...baseTimelineInput, - currentTail: [loaded], - currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 40 }, + currentTail, + currentCursor, payload: { ...baseTimelineInput.payload, direction: "tail", @@ -381,17 +391,35 @@ describe("processTimelineResponse", () => { startCursor: { seq: 30 }, endCursor: { seq: 42 }, entries: [ - makeTimelineEntry(30, "already loaded", "user_message"), - makeTimelineEntry(41, "new 41", "user_message"), - makeTimelineEntry(42, "new 42", "user_message"), + { + ...makeTimelineEntry(30, "provider presentation", "user_message"), + item: { + type: "user_message", + text: "provider presentation", + clientMessageId: "client-message-30", + messageId: "provider-message-30", + }, + }, + makeTimelineEntry(41, "new output 41"), + makeTimelineEntry(42, "new output 42"), ], }, }); expect(result.commit).toBe("apply"); - expect(result.tail[0]).toBe(loaded); - expect(getUserTexts(result.tail)).toEqual(["already loaded", "new 41", "new 42"]); + expect(result.tail[0]).toEqual({ ...loaded, messageId: "provider-message-30" }); + expect(result.tail[1]).toBe(unrelated); + expect( + [...result.tail, ...result.head].filter( + (item) => item.kind === "user_message" && item.clientMessageId === "client-message-30", + ), + ).toHaveLength(1); + expect(getAssistantTexts([...result.tail, ...result.head])).toEqual([ + "new output 41new output 42", + ]); expect(result.cursor).toEqual({ epoch: "epoch-1", startSeq: 1, endSeq: 42 }); + expect(result.cursorChanged).toBe(true); + expect(result.acknowledgedClientMessageIds).toEqual(["client-message-30"]); expect(result.sideEffects).toEqual([{ type: "flush_pending_updates" }]); }); diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 5a694584b5..047026a6d0 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -185,8 +185,8 @@ function deriveBootstrapTailTimelinePolicy({ type ResumeTailPolicy = | { kind: "not_resume" } - | { kind: "discard" } - | { kind: "append" } + | { kind: "discard"; coveredEndSeq: number } + | { kind: "append"; coveredEndSeq: number } | { kind: "replace"; preserveContinuity: boolean }; function deriveResumeTailPolicy(input: { @@ -205,13 +205,13 @@ function deriveResumeTailPolicy(input: { return { kind: "replace", preserveContinuity: false }; } if (input.windowMaxSeq === input.currentCursor.endSeq) { - return { kind: "discard" }; + return { kind: "discard", coveredEndSeq: input.currentCursor.endSeq }; } if (input.windowMaxSeq < input.currentCursor.endSeq) { return { kind: "replace", preserveContinuity: false }; } if (input.pageStartSeq !== null && input.pageStartSeq <= input.currentCursor.endSeq + 1) { - return { kind: "append" }; + return { kind: "append", coveredEndSeq: input.currentCursor.endSeq }; } return { kind: "replace", preserveContinuity: true }; } @@ -976,19 +976,28 @@ export function processTimelineResponse( bootstrapReplace: replace, }); const discard = resumeTailPolicy.kind === "discard"; + const coveredUnits = timelineUnits.filter( + (unit) => "coveredEndSeq" in resumeTailPolicy && unit.seqEnd <= resumeTailPolicy.coveredEndSeq, + ); + const revision = processAgentStreamEvents({ + events: coveredUnits.map((unit) => ({ + event: unit.event, + seq: unit.seqEnd, + epoch: payload.epoch, + timestamp: unit.timestamp, + })), + currentTail, + currentHead, + currentCursor, + }); + const coveredRevisionAcknowledgements = deriveCanonicalAcknowledgements({ + units: coveredUnits, + epoch: payload.epoch, + currentTail, + currentHead, + }); let timelineResult: TimelinePathResult; if (discard) { - const revision = processAgentStreamEvents({ - events: timelineUnits.map((unit) => ({ - event: unit.event, - seq: unit.seqEnd, - epoch: payload.epoch, - timestamp: unit.timestamp, - })), - currentTail, - currentHead, - currentCursor, - }); timelineResult = { tail: revision.tail, head: revision.head, @@ -996,12 +1005,7 @@ export function processTimelineResponse( cursorChanged: false, older: "unchanged", sideEffects: [], - acknowledgedClientMessageIds: deriveCanonicalAcknowledgements({ - units: timelineUnits, - epoch: payload.epoch, - currentTail, - currentHead, - }), + acknowledgedClientMessageIds: [], }; } else if (replace || resumeTailPolicy.kind === "replace") { const isResumeReplacement = resumeTailPolicy.kind === "replace"; @@ -1027,11 +1031,20 @@ export function processTimelineResponse( timelineResult = applyTimelineIncrementalPath({ timelineUnits: incrementalUnits, payload, - currentTail, - currentHead, + currentTail: revision.tail, + currentHead: revision.head, currentCursor, }); } + timelineResult = { + ...timelineResult, + acknowledgedClientMessageIds: [ + ...new Set([ + ...coveredRevisionAcknowledgements, + ...timelineResult.acknowledgedClientMessageIds, + ]), + ], + }; const nextTail = timelineResult.tail; const nextHead = timelineResult.head;