diff --git a/packages/app/src/contexts/session-stream-reducers.test.ts b/packages/app/src/contexts/session-stream-reducers.test.ts index ccd2c9757..ce021a52b 100644 --- a/packages/app/src/contexts/session-stream-reducers.test.ts +++ b/packages/app/src/contexts/session-stream-reducers.test.ts @@ -19,6 +19,7 @@ import { function makeTimelineEntry(seq: number, text: string, type: string = "assistant_message") { return { seqStart: seq, + seqEnd: seq, provider: "claude", item: { type, text }, timestamp: new Date(1000 + seq).toISOString(), @@ -318,6 +319,39 @@ describe("processTimelineResponse", () => { expect(result.cursorChanged).toBe(false); }); + it("requests canonical catch-up when a projected entry overlaps unseen seqs", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 5, + }; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + entries: [ + { + ...makeTimelineEntry(4, "merged assistant message"), + seqEnd: 8, + }, + ], + }, + }); + + expect(result.tail).toBe(baseTimelineInput.currentTail); + expect(result.cursorChanged).toBe(false); + expect(result.sideEffects).toContainEqual({ + type: "catch_up", + cursor: { + epoch: "epoch-1", + endSeq: 5, + }, + }); + }); + it("drops entries with epoch mismatch", () => { const existingCursor: TimelineCursor = { epoch: "epoch-1", diff --git a/packages/app/src/contexts/session-stream-reducers.ts b/packages/app/src/contexts/session-stream-reducers.ts index ab31c8b80..e74bf8803 100644 --- a/packages/app/src/contexts/session-stream-reducers.ts +++ b/packages/app/src/contexts/session-stream-reducers.ts @@ -53,6 +53,7 @@ type InitRequestDirection = "tail" | "after"; type TimelineResponseEntry = { seqStart: number; + seqEnd: number; provider: string; item: Record; timestamp: string; @@ -122,6 +123,7 @@ export function processTimelineResponse( // ------------------------------------------------------------------ const timelineUnits = payload.entries.map((entry) => ({ seq: entry.seqStart, + seqEnd: entry.seqEnd, event: { type: "timeline", provider: entry.provider, @@ -200,7 +202,14 @@ export function processTimelineResponse( gapCursor = cursor ? { epoch: cursor.epoch, endSeq: cursor.endSeq } : null; break; } - if (decision === "drop_stale" || decision === "drop_epoch") { + if (decision === "drop_stale") { + if (cursor && unit.seqEnd > cursor.endSeq) { + gapCursor = { epoch: cursor.epoch, endSeq: cursor.endSeq }; + break; + } + continue; + } + if (decision === "drop_epoch") { continue; } @@ -209,7 +218,7 @@ export function processTimelineResponse( cursor = { epoch: payload.epoch, startSeq: unit.seq, - endSeq: unit.seq, + endSeq: unit.seqEnd, }; continue; } @@ -218,7 +227,7 @@ export function processTimelineResponse( } cursor = { ...cursor, - endSeq: unit.seq, + endSeq: unit.seqEnd, }; }