diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 7e57c53b96..6aa9f72b10 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -258,63 +258,6 @@ const baseStreamInput: ProcessAgentStreamEventInput = { // --------------------------------------------------------------------------- describe("processTimelineResponse", () => { - it("applies a missed same-sequence provider identity revision from an unchanged resume tail", () => { - const canonical = createUserMessage({ - id: "canonical-prompt", - clientMessageId: "client-message", - text: "local prompt", - timestamp: new Date(1000), - timelineCursor: { epoch: "epoch-1", seq: 5 }, - }); - const unrelatedTail = makeAssistantItem("unrelated tail", "unrelated-tail"); - const currentTail = [canonical, unrelatedTail]; - const currentHead = [makeAssistantItem("live head", "live-head")]; - const currentCursor: TimelineCursor = { epoch: "epoch-1", startSeq: 1, endSeq: 8 }; - - const result = processTimelineResponse({ - ...baseTimelineInput, - currentTail, - currentHead, - currentCursor, - payload: { - ...baseTimelineInput.payload, - direction: "tail", - window: { minSeq: 1, maxSeq: 8, nextSeq: 9 }, - startCursor: { seq: 5 }, - endCursor: { seq: 8 }, - entries: [ - { - ...makeTimelineEntry(5, "provider prompt", "user_message"), - item: { - type: "user_message", - text: "provider prompt", - messageId: "provider-message", - clientMessageId: "client-message", - }, - }, - ], - }, - }); - - expect(result.commit).toBe("apply"); - expect(result.tail).not.toBe(currentTail); - expect(result.tail[0]).toEqual( - expect.objectContaining({ - id: "canonical-prompt", - clientMessageId: "client-message", - messageId: "provider-message", - text: "local prompt", - timestamp: canonical.timestamp, - timelineCursor: { epoch: "epoch-1", seq: 5 }, - }), - ); - expect(result.tail[1]).toBe(unrelatedTail); - expect(result.head).toBe(currentHead); - expect(result.cursor).toBe(currentCursor); - expect(result.cursorChanged).toBe(false); - expect(result.acknowledgedClientMessageIds).toEqual(["client-message"]); - }); - it("discards an unchanged resume tail without replacing timeline state", () => { const canonical = createUserMessage({ id: "canonical-prompt", @@ -357,7 +300,7 @@ describe("processTimelineResponse", () => { expect(result.tail).toBe(currentTail); expect(result.head).toBe(currentHead); expect(result.cursorChanged).toBe(false); - expect(result.acknowledgedClientMessageIds).toEqual(["client-message"]); + expect(result.acknowledgedClientMessageIds).toEqual([]); expect(result.sideEffects).toEqual([{ type: "flush_pending_updates" }]); }); @@ -365,6 +308,7 @@ describe("processTimelineResponse", () => { const loaded = createUserMessage({ id: "local-prompt-30", clientMessageId: "client-message-30", + messageId: "client-message-30", text: "local presentation", timestamp: new Date(1030), timelineCursor: { epoch: "epoch-1", seq: 30 }, @@ -407,7 +351,7 @@ describe("processTimelineResponse", () => { }); expect(result.commit).toBe("apply"); - expect(result.tail[0]).toEqual({ ...loaded, messageId: "provider-message-30" }); + expect(result.tail[0]).toBe(loaded); expect(result.tail[1]).toBe(unrelated); expect( [...result.tail, ...result.head].filter( @@ -419,7 +363,7 @@ describe("processTimelineResponse", () => { ]); 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.acknowledgedClientMessageIds).toEqual([]); expect(result.sideEffects).toEqual([{ type: "flush_pending_updates" }]); }); @@ -3242,48 +3186,6 @@ describe("processAgentStreamEvent", () => { expect(result.sideEffects).toEqual([]); }); - it("applies a same-sequence provider identity revision to its canonical prompt", () => { - const canonical = createUserMessage({ - id: "canonical-prompt", - clientMessageId: "client-message-1", - text: "Revise this prompt", - timestamp: new Date(1000), - timelineCursor: { epoch: "epoch-1", seq: 5 }, - }); - const currentHead = [makeAssistantItem("Still streaming")]; - - const result = processAgentStreamEvent({ - ...baseStreamInput, - currentTail: [canonical], - currentHead, - currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 8 }, - event: { - type: "timeline", - provider: "claude", - item: { - type: "user_message", - text: "Revise this prompt", - clientMessageId: "client-message-1", - messageId: "provider-message-1", - }, - }, - seq: 5, - epoch: "epoch-1", - }); - - expect(result.tail).toEqual([ - expect.objectContaining({ - id: "canonical-prompt", - clientMessageId: "client-message-1", - messageId: "provider-message-1", - timelineCursor: { epoch: "epoch-1", seq: 5 }, - }), - ]); - expect(result.head).toBe(currentHead); - expect(result.cursorChanged).toBe(false); - expect(result.sideEffects).toEqual([]); - }); - it("drops timeline event with epoch mismatch", () => { const existingCursor: TimelineCursor = { epoch: "epoch-1", diff --git a/packages/app/src/timeline/session-stream-reducers.ts b/packages/app/src/timeline/session-stream-reducers.ts index 047026a6d0..cc2c8e90c7 100644 --- a/packages/app/src/timeline/session-stream-reducers.ts +++ b/packages/app/src/timeline/session-stream-reducers.ts @@ -4,7 +4,6 @@ import type { AssistantMessageItem, StreamItem } from "@/types/stream"; import type { TurnLivenessTransition } from "@/timeline/turn-liveness"; import { applyStreamEvent, - createUserMessage, flushHeadToTail, hydrateStreamState, isAgentToolCallItem, @@ -185,8 +184,8 @@ function deriveBootstrapTailTimelinePolicy({ type ResumeTailPolicy = | { kind: "not_resume" } - | { kind: "discard"; coveredEndSeq: number } - | { kind: "append"; coveredEndSeq: number } + | { kind: "discard" } + | { kind: "append" } | { kind: "replace"; preserveContinuity: boolean }; function deriveResumeTailPolicy(input: { @@ -205,13 +204,13 @@ function deriveResumeTailPolicy(input: { return { kind: "replace", preserveContinuity: false }; } if (input.windowMaxSeq === input.currentCursor.endSeq) { - return { kind: "discard", coveredEndSeq: input.currentCursor.endSeq }; + return { kind: "discard" }; } if (input.windowMaxSeq < input.currentCursor.endSeq) { return { kind: "replace", preserveContinuity: false }; } if (input.pageStartSeq !== null && input.pageStartSeq <= input.currentCursor.endSeq + 1) { - return { kind: "append", coveredEndSeq: input.currentCursor.endSeq }; + return { kind: "append" }; } return { kind: "replace", preserveContinuity: true }; } @@ -746,19 +745,24 @@ function applyAcceptedForwardTimelineUnits(params: { }); let tail = reconciled.tail; let head = reconciled.head; - const acknowledgedClientMessageIds = new Set(); for (const unit of params.units) { if (reconciled.reconciledUnits.has(unit)) continue; const applied = applyCanonicalForwardUnit({ tail, head, unit, epoch: params.epoch }); tail = applied.tail; head = applied.head; - for (const clientMessageId of applied.acknowledgedClientMessageIds) { - acknowledgedClientMessageIds.add(clientMessageId); - } } - return { tail, head, acknowledgedClientMessageIds: [...acknowledgedClientMessageIds] }; + return { + tail, + head, + acknowledgedClientMessageIds: deriveCanonicalAcknowledgements({ + units: params.units, + epoch: params.epoch, + currentTail: params.currentTail, + currentHead: params.currentHead, + }), + }; } function deriveCanonicalAcknowledgements(params: { @@ -782,20 +786,6 @@ function deriveCanonicalAcknowledgements(params: { return [...acknowledged]; } -function shouldDiscardUnchangedTimelineResult(params: { - discardPolicy: boolean; - currentTail: StreamItem[]; - currentHead: StreamItem[]; - nextTail: StreamItem[]; - nextHead: StreamItem[]; -}): boolean { - return ( - params.discardPolicy && - params.nextTail === params.currentTail && - params.nextHead === params.currentHead - ); -} - function applyTimelineIncrementalPath(args: { timelineUnits: TimelineUnit[]; payload: ProcessTimelineResponseInput["payload"]; @@ -976,31 +966,11 @@ 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) { timelineResult = { - tail: revision.tail, - head: revision.head, + tail: currentTail, + head: currentHead, cursor: currentCursor, cursorChanged: false, older: "unchanged", @@ -1031,20 +1001,11 @@ export function processTimelineResponse( timelineResult = applyTimelineIncrementalPath({ timelineUnits: incrementalUnits, payload, - currentTail: revision.tail, - currentHead: revision.head, + currentTail, + currentHead, currentCursor, }); } - timelineResult = { - ...timelineResult, - acknowledgedClientMessageIds: [ - ...new Set([ - ...coveredRevisionAcknowledgements, - ...timelineResult.acknowledgedClientMessageIds, - ]), - ], - }; const nextTail = timelineResult.tail; const nextHead = timelineResult.head; @@ -1074,15 +1035,7 @@ export function processTimelineResponse( timelineResponseComplete; const initResolution: "resolve" | "reject" | null = shouldResolveDeferredInit ? "resolve" : null; - const commit = shouldDiscardUnchangedTimelineResult({ - discardPolicy: discard, - currentTail, - currentHead, - nextTail, - nextHead, - }) - ? "discard" - : "apply"; + const commit = discard ? "discard" : "apply"; return { commit, @@ -1140,55 +1093,6 @@ interface TimelineSequencingGateResult { sideEffects: AgentStreamReducerSideEffect[]; } -function applyCanonicalUserMessageRevision(input: { - event: AgentStreamEventPayload; - seq: number | undefined; - epoch: string | undefined; - timestamp: Date; - currentTail: StreamItem[]; - currentHead: StreamItem[]; - currentCursor: TimelineCursor | undefined; -}): ReturnType | null { - const { event, seq, epoch, currentCursor } = input; - if ( - event.type !== "timeline" || - event.item.type !== "user_message" || - !event.item.clientMessageId || - !event.item.messageId || - typeof seq !== "number" || - typeof epoch !== "string" || - currentCursor?.epoch !== epoch || - seq > currentCursor.endSeq - ) { - return null; - } - const clientMessageId = event.item.clientMessageId; - const existing = [...input.currentTail, ...input.currentHead].find( - (item) => - item.kind === "user_message" && - item.clientMessageId === clientMessageId && - item.timelineCursor?.epoch === epoch && - item.timelineCursor.seq === seq, - ); - if (!existing || existing.kind !== "user_message") { - return null; - } - return upsertUserMessageAcrossStream({ - tail: input.currentTail, - head: input.currentHead, - message: createUserMessage({ - id: existing.id, - clientMessageId, - messageId: event.item.messageId, - text: event.item.text, - timestamp: input.timestamp, - timelineCursor: existing.timelineCursor, - }), - insert: "none", - presentation: "existing", - }); -} - export interface ProcessAgentStreamEventsInput { events: AgentStreamReducerEvent[]; currentTail: StreamItem[]; @@ -1306,30 +1210,6 @@ export function processAgentStreamEvent( hasAuthoritativeBaseline = true, } = input; - const revision = applyCanonicalUserMessageRevision({ - event, - seq, - epoch, - timestamp, - currentTail, - currentHead, - currentCursor, - }); - if (revision) { - return { - tail: revision.tail, - head: revision.head, - changedTail: revision.changedTail, - changedHead: revision.changedHead, - cursor: null, - cursorChanged: false, - acknowledgedClientMessageIds: revision.location?.message.clientMessageId - ? [revision.location.message.clientMessageId] - : [], - sideEffects: [], - }; - } - const sequencing = processTimelineSequencingGate({ event, seq,