diff --git a/packages/app/src/contexts/session-stream-reducers.test.ts b/packages/app/src/contexts/session-stream-reducers.test.ts index 4eb3279b3..5a65c597a 100644 --- a/packages/app/src/contexts/session-stream-reducers.test.ts +++ b/packages/app/src/contexts/session-stream-reducers.test.ts @@ -67,6 +67,23 @@ function makeStreamReducerEvent( }; } +function makeAssistantItem(text: string, id = `assistant-${text.length}`): StreamItem { + return { + kind: "assistant_message", + id, + text, + timestamp: new Date(1000), + }; +} + +function getAssistantTexts(items: StreamItem[]): string[] { + return items + .filter((item): item is Extract => { + return item.kind === "assistant_message"; + }) + .map((item) => item.text); +} + const baseTimelineInput: ProcessTimelineResponseInput = { payload: { agentId: "agent-1", @@ -263,6 +280,34 @@ describe("processTimelineResponse", () => { expect(result.error).toBe(null); }); + it("keeps an active assistant head live when an incremental fetch accepts same-turn assistant text", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 2, + }; + const currentHead = [makeAssistantItem("This is a par")]; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentHead, + currentCursor: existingCursor, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + entries: [makeTimelineEntry(3, "agraph")], + }, + }); + + expect(getAssistantTexts(result.tail)).toEqual([]); + expect(getAssistantTexts(result.head)).toEqual(["This is a paragraph"]); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 3, + }); + }); + it("detects gap and emits catch-up side effect", () => { const existingCursor: TimelineCursor = { epoch: "epoch-1", @@ -968,6 +1013,59 @@ describe("createAgentStreamReducerQueue", () => { expect(scheduler.size).toBe(0); }); + it("keeps a live paragraph in one assistant item when canonical fetch interleaves with queued stream chunks", () => { + const scheduler = createManualScheduler(); + let currentTail: StreamItem[] = []; + let currentHead: StreamItem[] = []; + let currentCursor: TimelineCursor | undefined; + + const queue = createAgentStreamReducerQueue({ + getSnapshot: () => ({ + currentTail, + currentHead, + currentCursor, + currentAgent: null, + }), + commit: (_agentId, result) => { + currentTail = result.tail; + currentHead = result.head; + currentCursor = result.cursor ?? undefined; + }, + handleSideEffects: () => {}, + scheduleFlush: scheduler.schedule, + cancelFlush: scheduler.cancel, + }); + + queue.enqueue("agent-1", makeStreamReducerEvent(makeTimelineEvent("This is a par"), 2)); + queue.flushAgent("agent-1"); + + const timelineResult = processTimelineResponse({ + ...baseTimelineInput, + currentTail, + currentHead, + currentCursor, + payload: { + ...baseTimelineInput.payload, + epoch: "epoch-1", + entries: [makeTimelineEntry(3, "agraph")], + }, + }); + currentTail = timelineResult.tail; + currentHead = timelineResult.head; + currentCursor = timelineResult.cursor ?? undefined; + + queue.enqueue("agent-1", makeStreamReducerEvent(makeTimelineEvent(" continues."), 4)); + queue.flushAgent("agent-1"); + + expect(getAssistantTexts(currentTail)).toEqual([]); + expect(getAssistantTexts(currentHead)).toEqual(["This is a paragraph continues."]); + expect(currentCursor).toEqual({ + epoch: "epoch-1", + startSeq: 2, + endSeq: 4, + }); + }); + it("flushes queued events synchronously before disposal", () => { const scheduler = createManualScheduler(); const commits: string[] = []; diff --git a/packages/app/src/contexts/session-stream-reducers.ts b/packages/app/src/contexts/session-stream-reducers.ts index e24494790..1617cd7a6 100644 --- a/packages/app/src/contexts/session-stream-reducers.ts +++ b/packages/app/src/contexts/session-stream-reducers.ts @@ -3,12 +3,7 @@ import type { AgentLifecycleStatus } from "@server/shared/agent-lifecycle"; import type { Agent } from "@/stores/session-store"; import { useSessionStore } from "@/stores/session-store"; import type { StreamItem } from "@/types/stream"; -import { - applyStreamEvent, - flushHeadToTail, - hydrateStreamState, - reduceStreamUpdate, -} from "@/types/stream"; +import { applyStreamEvent, hydrateStreamState, reduceStreamUpdate } from "@/types/stream"; import { classifySessionTimelineSeq, type SessionTimelineSeqDecision, @@ -206,16 +201,25 @@ function applyTimelineIncrementalPath(args: { }); if (acceptedUnits.length > 0) { - const baseTail = - currentHead.length > 0 ? flushHeadToTail(currentTail, currentHead) : currentTail; if (currentHead.length > 0) { - nextHead = []; + for (const { event, timestamp } of acceptedUnits) { + const applied = applyStreamEvent({ + tail: nextTail, + head: nextHead, + event, + timestamp, + source: "canonical", + }); + nextTail = applied.tail; + nextHead = applied.head; + } + } else { + nextTail = acceptedUnits.reduce( + (state, { event, timestamp }) => + reduceStreamUpdate(state, event, timestamp, { source: "canonical" }), + currentTail, + ); } - nextTail = acceptedUnits.reduce( - (state, { event, timestamp }) => - reduceStreamUpdate(state, event, timestamp, { source: "canonical" }), - baseTail, - ); } if (