diff --git a/docs/STORAGE_REVAMP_PLAN.md b/docs/STORAGE_REVAMP_PLAN.md index e9c14947cc..2c997f7b76 100644 --- a/docs/STORAGE_REVAMP_PLAN.md +++ b/docs/STORAGE_REVAMP_PLAN.md @@ -92,9 +92,9 @@ The room has converged on a simplified contract: - `timeline` event with no `seq` means provisional live-only state and is never returned by catch-up fetches - committed rows must already be in canonical display/history shape, not raw provider chunk shape - `fetch-after-seq` returns committed rows only -- subscribe may emit the current provisional head once if a turn is active -- that provisional seed must contain current full provisional items, not replayed deltas -- that provisional seed must arrive before later provisional patches for that subscription +- Phase 0 policy: reconnect does not use a subscribe-time provisional seed +- if a turn is still active after reconnect, the client resumes from future live provisional updates only +- provisional-seed ordering is therefore out of scope for Phase 0 until product explicitly chooses to add that policy - at most one active provisional assistant stream and one active provisional reasoning stream exist per turn - tool progress continues to correlate by existing `callId` - reasoning is live-only for v1 unless product explicitly changes that call later @@ -170,6 +170,8 @@ It is not: If emitted, it must arrive before later provisional patch events for that subscription. +Phase 0 chooses the simpler policy: no subscribe-time provisional seed for now. + ### Reconnect scenarios Concrete reconnect example: @@ -193,7 +195,7 @@ Case B: the assistant message has not finished by the time the client reconnects 2. The client calls `fetch-after-seq(agentId, 120)`. 3. The server may legitimately return nothing new yet, because committed row `121` does not exist yet. 4. Live streaming resumes. -5. A subscribe-time provisional seed may repopulate the current in-progress state if that policy is enabled; otherwise the client continues from fresh live updates only. +5. In Phase 0, the client continues from fresh live updates only. 6. When the assistant message finally completes, it becomes one committed row `121`. In both cases: @@ -498,9 +500,9 @@ Dependencies: Tests required: - shared message parsing tests for the simplified timeline payloads -- app reducer tests for committed append, provisional merge, and subscribe-time provisional seeding -- session/WebSocket tests for `live -> disconnect -> fetch-after-seq -> subscribe -> provisional seed -> resume` -- explicit subscribe-time provisional-seed ordering test +- app reducer tests for committed append, provisional merge, and reconnect without provisional seeding +- session/WebSocket tests for `live -> disconnect -> fetch-after-seq -> subscribe -> resume` +- explicit no-provisional-seed reconnect test proving that reconnect resumes from committed catch-up plus future live events only Exit gate: The client and server can exchange committed and provisional timeline updates without `epoch`, `reset`, or projection metadata, and all timeline catch-up tests pass with the new contract. diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index b9d139ec74..6d917712f8 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -525,9 +525,8 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider void client .fetchAgentTimeline(agentId, { direction: "after", - cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, + cursor: { seq: cursor.endSeq }, limit: 0, - projection: "canonical", }) .catch((error) => { console.warn("[Session] failed to fetch catch-up timeline on resume", agentId, error); @@ -753,13 +752,12 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider ); const requestCanonicalCatchUp = useCallback( - (agentId: string, cursor: { epoch: string; endSeq: number }) => { + (agentId: string, cursor: { endSeq: number }) => { void client .fetchAgentTimeline(agentId, { direction: "after", - cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, + cursor: { seq: cursor.endSeq }, limit: 0, - projection: "canonical", }) .catch((error) => { console.warn("[Session] failed to fetch canonical catch-up timeline", agentId, error); @@ -862,7 +860,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider } if ( current && - current.epoch === result.cursor.epoch && current.startSeq === result.cursor.startSeq && current.endSeq === result.cursor.endSeq ) { @@ -967,7 +964,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const unsubAgentStream = client.on("agent_stream", (message) => { if (message.type !== "agent_stream") return; - const { agentId, event, timestamp, seq, epoch } = message.payload; + const { agentId, event, timestamp, seq } = message.payload; const parsedTimestamp = new Date(timestamp); const streamEvent = event as AgentStreamEventPayload; if ( @@ -1009,7 +1006,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const result = processAgentStreamEvent({ event: streamEvent, seq, - epoch, currentTail, currentHead, currentCursor, @@ -1033,8 +1029,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider if ( current && typeof seq === "number" && - typeof epoch === "string" && - current.epoch === epoch && seq >= current.startSeq && seq <= current.endSeq ) { @@ -1043,7 +1037,6 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider } if ( current && - current.epoch === nextCursor.epoch && current.startSeq === nextCursor.startSeq && current.endSeq === nextCursor.endSeq ) { diff --git a/packages/app/src/contexts/session-stream-reducers.test.ts b/packages/app/src/contexts/session-stream-reducers.test.ts index 580189da30..138baf8a87 100644 --- a/packages/app/src/contexts/session-stream-reducers.test.ts +++ b/packages/app/src/contexts/session-stream-reducers.test.ts @@ -9,13 +9,9 @@ import { type TimelineCursor, } from "./session-stream-reducers"; -// --------------------------------------------------------------------------- -// Test helpers -// --------------------------------------------------------------------------- - function makeTimelineEntry(seq: number, text: string, type: string = "assistant_message") { return { - seqStart: seq, + seq, provider: "claude", item: { type, text }, timestamp: new Date(1000 + seq).toISOString(), @@ -33,22 +29,30 @@ function makeTimelineEvent( } as AgentStreamEventPayload; } -function makeUserTimelineEvent(text: string): AgentStreamEventPayload { +function makeToolCallEvent(status: "running" | "completed"): AgentStreamEventPayload { return { type: "timeline", provider: "claude", - item: { type: "user_message", text }, - } as AgentStreamEventPayload; + item: { + type: "tool_call", + callId: "call-1", + name: "shell", + status, + detail: { + type: "shell", + command: "pwd", + }, + error: null, + }, + }; } const baseTimelineInput: ProcessTimelineResponseInput = { payload: { agentId: "agent-1", direction: "after", - reset: false, - epoch: "epoch-1", - startCursor: null, - endCursor: null, + startSeq: null, + endSeq: null, entries: [], error: null, }, @@ -63,7 +67,6 @@ const baseTimelineInput: ProcessTimelineResponseInput = { const baseStreamInput: ProcessAgentStreamEventInput = { event: makeTimelineEvent("hello"), seq: undefined, - epoch: undefined, currentTail: [], currentHead: [], currentCursor: undefined, @@ -71,10 +74,6 @@ const baseStreamInput: ProcessAgentStreamEventInput = { timestamp: new Date(2000), }; -// --------------------------------------------------------------------------- -// processTimelineResponse -// --------------------------------------------------------------------------- - describe("processTimelineResponse", () => { it("returns error path when payload.error is set", () => { const result = processTimelineResponse({ @@ -93,35 +92,10 @@ describe("processTimelineResponse", () => { expect(result.tail).toBe(baseTimelineInput.currentTail); expect(result.head).toBe(baseTimelineInput.currentHead); expect(result.cursorChanged).toBe(false); - expect(result.sideEffects).toEqual([]); }); - it("returns error with no init resolution when no deferred exists", () => { - const result = processTimelineResponse({ - ...baseTimelineInput, - isInitializing: true, - hasActiveInitDeferred: false, - payload: { - ...baseTimelineInput.payload, - error: "timeout", - }, - }); - - expect(result.error).toBe("timeout"); - expect(result.initResolution).toBe(null); - expect(result.clearInitializing).toBe(true); - }); - - it("replaces tail and clears head when reset=true", () => { - const existingTail: StreamItem[] = [ - { - kind: "user_message", - id: "old", - text: "old message", - timestamp: new Date(500), - }, - ]; - const existingHead: StreamItem[] = [ + it("replaces tail during bootstrap tail init and schedules committed catch-up", () => { + const provisionalHead: StreamItem[] = [ { kind: "assistant_message", id: "head-1", @@ -132,518 +106,280 @@ describe("processTimelineResponse", () => { const result = processTimelineResponse({ ...baseTimelineInput, - currentTail: existingTail, - currentHead: existingHead, - payload: { - ...baseTimelineInput.payload, - reset: true, - startCursor: { seq: 1 }, - endCursor: { seq: 3 }, - entries: [ - makeTimelineEntry(1, "first"), - makeTimelineEntry(2, "second"), - makeTimelineEntry(3, "third"), - ], - }, - }); - - expect(result.tail).not.toBe(existingTail); - expect(result.tail.length).toBeGreaterThan(0); - expect(result.head).toEqual([]); - expect(result.cursorChanged).toBe(true); - expect(result.cursor).toEqual({ - epoch: "epoch-1", - startSeq: 1, - endSeq: 3, - }); - expect(result.error).toBe(null); - expect(result.sideEffects.some((e) => e.type === "flush_pending_updates")).toBe(true); - }); - - it("sets cursor to null when reset=true but no cursors in payload", () => { - const result = processTimelineResponse({ - ...baseTimelineInput, - currentCursor: { epoch: "epoch-1", startSeq: 1, endSeq: 5 }, - payload: { - ...baseTimelineInput.payload, - reset: true, - entries: [], - }, - }); - - expect(result.cursor).toBe(null); - expect(result.cursorChanged).toBe(true); - }); - - it("performs bootstrap tail init with catch-up side effect", () => { - const result = processTimelineResponse({ - ...baseTimelineInput, + currentHead: provisionalHead, isInitializing: true, hasActiveInitDeferred: true, initRequestDirection: "tail", payload: { ...baseTimelineInput.payload, direction: "tail", - epoch: "epoch-1", - startCursor: { seq: 1 }, - endCursor: { seq: 5 }, + startSeq: 1, + endSeq: 5, entries: [makeTimelineEntry(1, "first"), makeTimelineEntry(5, "last")], }, }); - // Bootstrap tail replaces expect(result.tail.length).toBeGreaterThan(0); expect(result.head).toEqual([]); expect(result.cursorChanged).toBe(true); expect(result.cursor).toEqual({ - epoch: "epoch-1", startSeq: 1, endSeq: 5, }); - // Should have catch-up side effect - const catchUp = result.sideEffects.find((e) => e.type === "catch_up"); - expect(catchUp).toBeDefined(); - expect(catchUp!.type === "catch_up" && catchUp!.cursor).toEqual({ - epoch: "epoch-1", - endSeq: 5, + const catchUp = result.sideEffects.find((effect) => effect.type === "catch_up"); + expect(catchUp).toEqual({ + type: "catch_up", + cursor: { endSeq: 5 }, }); }); - it("appends incrementally for contiguous seqs", () => { - const existingCursor: TimelineCursor = { - epoch: "epoch-1", - startSeq: 1, - endSeq: 3, - }; - - const result = processTimelineResponse({ - ...baseTimelineInput, - currentCursor: existingCursor, - payload: { - ...baseTimelineInput.payload, - epoch: "epoch-1", - entries: [makeTimelineEntry(4, "next-1"), makeTimelineEntry(5, "next-2")], + it("prepends older committed history for before pagination", () => { + const currentTail: StreamItem[] = [ + { + kind: "assistant_message", + id: "tail-3", + text: "newer", + timestamp: new Date(3000), }, - }); - - expect(result.tail.length).toBeGreaterThan(0); - expect(result.cursorChanged).toBe(true); - expect(result.cursor).toEqual({ - epoch: "epoch-1", - startSeq: 1, - endSeq: 5, - }); - expect(result.error).toBe(null); - }); - - it("detects gap and emits catch-up side effect", () => { - const existingCursor: TimelineCursor = { - epoch: "epoch-1", - startSeq: 1, - endSeq: 3, - }; + ]; + const currentCursor: TimelineCursor = { startSeq: 3, endSeq: 4 }; const result = processTimelineResponse({ ...baseTimelineInput, - currentCursor: existingCursor, - payload: { - ...baseTimelineInput.payload, - epoch: "epoch-1", - entries: [makeTimelineEntry(10, "far ahead")], - }, - }); - - // Gap should trigger catch-up - const catchUp = result.sideEffects.find((e) => e.type === "catch_up"); - expect(catchUp).toBeDefined(); - expect(catchUp!.type === "catch_up" && catchUp!.cursor).toEqual({ - epoch: "epoch-1", - endSeq: 3, - }); - }); - - it("drops stale entries silently", () => { - const existingCursor: TimelineCursor = { - epoch: "epoch-1", - startSeq: 1, - endSeq: 8, - }; - - const result = processTimelineResponse({ - ...baseTimelineInput, - currentCursor: existingCursor, - payload: { - ...baseTimelineInput.payload, - epoch: "epoch-1", - entries: [makeTimelineEntry(5, "old"), makeTimelineEntry(7, "also old")], - }, - }); - - // No new items appended (all dropped as stale) - expect(result.tail).toBe(baseTimelineInput.currentTail); - expect(result.cursorChanged).toBe(false); - }); - - it("drops entries with epoch mismatch", () => { - const existingCursor: TimelineCursor = { - epoch: "epoch-1", - startSeq: 1, - endSeq: 5, - }; - - const result = processTimelineResponse({ - ...baseTimelineInput, - currentCursor: existingCursor, - payload: { - ...baseTimelineInput.payload, - epoch: "epoch-2", - entries: [makeTimelineEntry(6, "different epoch")], - }, - }); - - expect(result.tail).toBe(baseTimelineInput.currentTail); - expect(result.cursorChanged).toBe(false); - }); - - it("resolves init when deferred matches direction", () => { - const result = processTimelineResponse({ - ...baseTimelineInput, - isInitializing: true, - hasActiveInitDeferred: true, - initRequestDirection: "after", - payload: { - ...baseTimelineInput.payload, - direction: "after", - entries: [], - }, - }); - - expect(result.initResolution).toBe("resolve"); - expect(result.clearInitializing).toBe(true); - }); - - it("does not resolve init when directions differ (before vs after)", () => { - const result = processTimelineResponse({ - ...baseTimelineInput, - isInitializing: true, - hasActiveInitDeferred: true, - initRequestDirection: "after", + currentTail, + currentCursor, payload: { ...baseTimelineInput.payload, direction: "before", - entries: [], + startSeq: 1, + endSeq: 2, + entries: [ + makeTimelineEntry(1, "hello", "user_message"), + makeTimelineEntry(2, "older"), + ], }, }); - // "before" direction doesn't match "after" initRequestDirection, - // and "before" is not a bootstrap tail path, so init should NOT resolve - expect(result.initResolution).toBe(null); - expect(result.clearInitializing).toBe(false); + expect(result.cursorChanged).toBe(true); + expect(result.cursor).toEqual({ + startSeq: 1, + endSeq: 4, + }); + expect(result.tail).toHaveLength(3); + expect(result.tail[0]?.kind).toBe("user_message"); + expect(result.tail[1]?.kind).toBe("assistant_message"); + expect(result.tail[2]).toBe(currentTail[0]); }); - it("clears initializing even without deferred", () => { + it("replaces stale provisional assistant UI when fetch-after returns committed row 121", () => { + const currentHead: StreamItem[] = [ + { + kind: "assistant_message", + id: "head-assistant", + text: "partial", + timestamp: new Date(120000), + }, + ]; + const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 }; + const result = processTimelineResponse({ ...baseTimelineInput, - isInitializing: true, - hasActiveInitDeferred: false, + currentHead, + currentCursor, payload: { ...baseTimelineInput.payload, direction: "after", - entries: [], - }, - }); - - expect(result.clearInitializing).toBe(true); - expect(result.initResolution).toBe(null); - }); - - it("always includes flush_pending_updates side effect on success", () => { - const result = processTimelineResponse({ - ...baseTimelineInput, - payload: { - ...baseTimelineInput.payload, - entries: [], - }, - }); - - expect(result.sideEffects.some((e) => e.type === "flush_pending_updates")).toBe(true); - }); - - it("initializes cursor when no existing cursor on first entries", () => { - const result = processTimelineResponse({ - ...baseTimelineInput, - currentCursor: undefined, - payload: { - ...baseTimelineInput.payload, - epoch: "epoch-1", - entries: [makeTimelineEntry(1, "first"), makeTimelineEntry(2, "second")], + startSeq: 121, + endSeq: 121, + entries: [makeTimelineEntry(121, "finalized reply")], }, }); + expect(result.head).toEqual([]); expect(result.cursorChanged).toBe(true); expect(result.cursor).toEqual({ - epoch: "epoch-1", startSeq: 1, - endSeq: 2, + endSeq: 121, + }); + expect(result.tail[result.tail.length - 1]).toMatchObject({ + kind: "assistant_message", + text: "finalized reply", + }); + }); + + it("keeps provisional head when reconnect catch-up has no new committed rows yet", () => { + const currentHead: StreamItem[] = [ + { + kind: "assistant_message", + id: "head-assistant", + text: "still streaming", + timestamp: new Date(120000), + }, + ]; + const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 }; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentHead, + currentCursor, + payload: { + ...baseTimelineInput.payload, + direction: "after", + startSeq: null, + endSeq: null, + entries: [], + }, + }); + + expect(result.head).toBe(currentHead); + expect(result.cursorChanged).toBe(false); + expect(result.tail).toBe(baseTimelineInput.currentTail); + }); + + it("requests catch-up when committed rows arrive with a forward gap", () => { + const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 }; + + const result = processTimelineResponse({ + ...baseTimelineInput, + currentCursor, + payload: { + ...baseTimelineInput.payload, + direction: "after", + startSeq: 125, + endSeq: 125, + entries: [makeTimelineEntry(125, "far ahead")], + }, + }); + + expect(result.cursorChanged).toBe(false); + expect(result.tail).toBe(baseTimelineInput.currentTail); + expect(result.sideEffects).toContainEqual({ + type: "catch_up", + cursor: { endSeq: 120 }, }); }); }); -// --------------------------------------------------------------------------- -// processAgentStreamEvent -// --------------------------------------------------------------------------- - describe("processAgentStreamEvent", () => { - it("passes through non-timeline events without cursor changes", () => { - const turnEvent: AgentStreamEventPayload = { - type: "turn_completed", - provider: "claude", - }; - + it("treats seq-less timeline events as provisional head updates", () => { const result = processAgentStreamEvent({ ...baseStreamInput, - event: turnEvent, + event: makeTimelineEvent("partial"), seq: undefined, - epoch: undefined, }); + expect(result.changedHead).toBe(true); + expect(result.changedTail).toBe(false); + expect(result.head).toHaveLength(1); + expect(result.head[0]).toMatchObject({ + kind: "assistant_message", + text: "partial", + }); expect(result.cursorChanged).toBe(false); - expect(result.cursor).toBe(null); - expect(result.sideEffects).toEqual([]); }); - it("accepts timeline event with cursor advance", () => { - const existingCursor: TimelineCursor = { - epoch: "epoch-1", - startSeq: 1, - endSeq: 4, - }; + it("appends committed live rows to tail and clears superseded provisional assistant state", () => { + const currentHead: StreamItem[] = [ + { + kind: "assistant_message", + id: "head-assistant", + text: "partial", + timestamp: new Date(1000), + }, + ]; + const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 }; const result = processAgentStreamEvent({ ...baseStreamInput, - event: makeTimelineEvent("new chunk"), - seq: 5, - epoch: "epoch-1", - currentCursor: existingCursor, + event: makeTimelineEvent("finalized reply"), + seq: 121, + currentHead, + currentCursor, }); + expect(result.changedTail).toBe(true); + expect(result.changedHead).toBe(true); + expect(result.head).toEqual([]); expect(result.cursorChanged).toBe(true); expect(result.cursor).toEqual({ - epoch: "epoch-1", startSeq: 1, - endSeq: 5, + endSeq: 121, + }); + expect(result.tail[result.tail.length - 1]).toMatchObject({ + kind: "assistant_message", + text: "finalized reply", }); - expect(result.sideEffects).toEqual([]); }); - it("detects gap and emits catch-up side effect", () => { - const existingCursor: TimelineCursor = { - epoch: "epoch-1", - startSeq: 1, - endSeq: 4, - }; + it("replaces provisional tool progress when the committed tool row arrives", () => { + const provisional = processAgentStreamEvent({ + ...baseStreamInput, + event: makeToolCallEvent("running"), + seq: undefined, + }); + const committed = processAgentStreamEvent({ + ...baseStreamInput, + event: makeToolCallEvent("completed"), + seq: 8, + currentHead: provisional.head, + currentTail: provisional.tail, + currentCursor: { startSeq: 1, endSeq: 7 }, + }); + + expect(committed.head).toEqual([]); + expect(committed.tail).toHaveLength(1); + expect(committed.tail[0]).toMatchObject({ + kind: "tool_call", + payload: { + source: "agent", + data: { + callId: "call-1", + status: "completed", + }, + }, + }); + }); + + it("requests catch-up when a committed live row skips ahead", () => { const result = processAgentStreamEvent({ ...baseStreamInput, event: makeTimelineEvent("far ahead"), - seq: 10, - epoch: "epoch-1", - currentCursor: existingCursor, + seq: 125, + currentCursor: { startSeq: 1, endSeq: 120 }, }); - expect(result.cursorChanged).toBe(false); expect(result.changedTail).toBe(false); expect(result.changedHead).toBe(false); - - const catchUp = result.sideEffects.find((e) => e.type === "catch_up"); - expect(catchUp).toBeDefined(); - expect(catchUp!.cursor).toEqual({ - epoch: "epoch-1", - endSeq: 4, - }); - }); - - it("drops stale timeline event", () => { - const existingCursor: TimelineCursor = { - epoch: "epoch-1", - startSeq: 1, - endSeq: 8, - }; - - const result = processAgentStreamEvent({ - ...baseStreamInput, - event: makeTimelineEvent("old"), - seq: 5, - epoch: "epoch-1", - currentCursor: existingCursor, - }); - expect(result.cursorChanged).toBe(false); + expect(result.sideEffects).toContainEqual({ + type: "catch_up", + cursor: { endSeq: 120 }, + }); + }); + + it("clears provisional head on terminal turn events without committing it to tail", () => { + const result = processAgentStreamEvent({ + ...baseStreamInput, + event: { + type: "turn_completed", + provider: "claude", + }, + currentHead: [ + { + kind: "thought", + id: "reasoning-1", + text: "thinking", + timestamp: new Date(1000), + status: "loading", + }, + ], + }); + + expect(result.changedHead).toBe(true); expect(result.changedTail).toBe(false); - expect(result.changedHead).toBe(false); - expect(result.sideEffects).toEqual([]); - }); - - it("drops timeline event with epoch mismatch", () => { - const existingCursor: TimelineCursor = { - epoch: "epoch-1", - startSeq: 1, - endSeq: 5, - }; - - const result = processAgentStreamEvent({ - ...baseStreamInput, - event: makeTimelineEvent("wrong epoch"), - seq: 6, - epoch: "epoch-2", - currentCursor: existingCursor, - }); - - expect(result.cursorChanged).toBe(false); - expect(result.changedTail).toBe(false); - expect(result.changedHead).toBe(false); - expect(result.sideEffects).toEqual([]); - }); - - it("initializes cursor when none exists", () => { - const result = processAgentStreamEvent({ - ...baseStreamInput, - event: makeTimelineEvent("first"), - seq: 1, - epoch: "epoch-1", - currentCursor: undefined, - }); - - expect(result.cursorChanged).toBe(true); - expect(result.cursor).toEqual({ - epoch: "epoch-1", - startSeq: 1, - endSeq: 1, - }); - }); - - it("derives optimistic idle status on turn_completed for running agent", () => { - const turnCompletedEvent: AgentStreamEventPayload = { - type: "turn_completed", - provider: "claude", - }; - - const result = processAgentStreamEvent({ - ...baseStreamInput, - event: turnCompletedEvent, - currentAgent: { - status: "running", - updatedAt: new Date(1000), - lastActivityAt: new Date(1000), - }, - timestamp: new Date(2000), - }); - - expect(result.agentChanged).toBe(true); - expect(result.agent).not.toBe(null); - expect(result.agent!.status).toBe("idle"); - expect(result.agent!.updatedAt.getTime()).toBe(2000); - expect(result.agent!.lastActivityAt.getTime()).toBe(2000); - }); - - it("derives optimistic error status on turn_failed for running agent", () => { - const turnFailedEvent: AgentStreamEventPayload = { - type: "turn_failed", - provider: "claude", - error: "something broke", - }; - - const result = processAgentStreamEvent({ - ...baseStreamInput, - event: turnFailedEvent, - currentAgent: { - status: "running", - updatedAt: new Date(1000), - lastActivityAt: new Date(1000), - }, - timestamp: new Date(2000), - }); - - expect(result.agentChanged).toBe(true); - expect(result.agent!.status).toBe("error"); - }); - - it("does not change agent when status is not running", () => { - const turnCompletedEvent: AgentStreamEventPayload = { - type: "turn_completed", - provider: "claude", - }; - - const result = processAgentStreamEvent({ - ...baseStreamInput, - event: turnCompletedEvent, - currentAgent: { - status: "idle", - updatedAt: new Date(1000), - lastActivityAt: new Date(1000), - }, - timestamp: new Date(2000), - }); - - expect(result.agentChanged).toBe(false); - expect(result.agent).toBe(null); - }); - - it("does not change agent when no agent is provided", () => { - const turnCompletedEvent: AgentStreamEventPayload = { - type: "turn_completed", - provider: "claude", - }; - - const result = processAgentStreamEvent({ - ...baseStreamInput, - event: turnCompletedEvent, - currentAgent: null, - timestamp: new Date(2000), - }); - - expect(result.agentChanged).toBe(false); - expect(result.agent).toBe(null); - }); - - it("preserves updatedAt when agent timestamp is newer than event", () => { - const turnCompletedEvent: AgentStreamEventPayload = { - type: "turn_completed", - provider: "claude", - }; - - const result = processAgentStreamEvent({ - ...baseStreamInput, - event: turnCompletedEvent, - currentAgent: { - status: "running", - updatedAt: new Date(5000), - lastActivityAt: new Date(5000), - }, - timestamp: new Date(2000), - }); - - expect(result.agentChanged).toBe(true); - expect(result.agent!.updatedAt.getTime()).toBe(5000); - expect(result.agent!.lastActivityAt.getTime()).toBe(5000); - }); - - it("does not produce agent patch for non-terminal events", () => { - const result = processAgentStreamEvent({ - ...baseStreamInput, - event: makeTimelineEvent("just text"), - currentAgent: { - status: "running", - updatedAt: new Date(1000), - lastActivityAt: new Date(1000), - }, - seq: 1, - epoch: "epoch-1", - timestamp: new Date(2000), - }); - - expect(result.agentChanged).toBe(false); - expect(result.agent).toBe(null); + expect(result.head).toEqual([]); + expect(result.tail).toEqual([]); }); }); diff --git a/packages/app/src/contexts/session-stream-reducers.ts b/packages/app/src/contexts/session-stream-reducers.ts index 6623fba943..5600d57ac7 100644 --- a/packages/app/src/contexts/session-stream-reducers.ts +++ b/packages/app/src/contexts/session-stream-reducers.ts @@ -1,7 +1,7 @@ import type { AgentStreamEventPayload } from "@server/shared/messages"; import type { AgentLifecycleStatus } from "@server/shared/agent-lifecycle"; import type { StreamItem } from "@/types/stream"; -import { applyStreamEvent, hydrateStreamState, reduceStreamUpdate } from "@/types/stream"; +import { hydrateStreamState, reduceStreamUpdate } from "@/types/stream"; import { classifySessionTimelineSeq, type SessionTimelineSeqDecision, @@ -12,38 +12,25 @@ import { } from "@/contexts/session-timeline-bootstrap-policy"; import { deriveOptimisticLifecycleStatus } from "@/contexts/session-stream-lifecycle"; -// --------------------------------------------------------------------------- -// Shared cursor type -// --------------------------------------------------------------------------- - export type TimelineCursor = { - epoch: string; startSeq: number; endSeq: number; }; -// --------------------------------------------------------------------------- -// Side-effect discriminated unions -// --------------------------------------------------------------------------- - export type TimelineReducerSideEffect = - | { type: "catch_up"; cursor: { epoch: string; endSeq: number } } + | { type: "catch_up"; cursor: { endSeq: number } } | { type: "flush_pending_updates" }; export type AgentStreamReducerSideEffect = { type: "catch_up"; - cursor: { epoch: string; endSeq: number }; + cursor: { endSeq: number }; }; -// --------------------------------------------------------------------------- -// processTimelineResponse -// --------------------------------------------------------------------------- - type TimelineDirection = "tail" | "before" | "after"; type InitRequestDirection = "tail" | "after"; type TimelineResponseEntry = { - seqStart: number; + seq: number; provider: string; item: Record; timestamp: string; @@ -53,10 +40,8 @@ export interface ProcessTimelineResponseInput { payload: { agentId: string; direction: TimelineDirection; - reset: boolean; - epoch: string; - startCursor: { seq: number } | null; - endCursor: { seq: number } | null; + startSeq: number | null; + endSeq: number | null; entries: TimelineResponseEntry[]; error: string | null; }; @@ -79,205 +64,9 @@ export interface ProcessTimelineResponseOutput { sideEffects: TimelineReducerSideEffect[]; } -export function processTimelineResponse( - input: ProcessTimelineResponseInput, -): ProcessTimelineResponseOutput { - const { - payload, - currentTail, - currentHead, - currentCursor, - isInitializing, - hasActiveInitDeferred, - initRequestDirection, - } = input; - - // ------------------------------------------------------------------ - // Error path: reject init and leave stream state unchanged - // ------------------------------------------------------------------ - if (payload.error) { - return { - tail: currentTail, - head: currentHead, - cursor: currentCursor, - cursorChanged: false, - initResolution: hasActiveInitDeferred ? "reject" : null, - clearInitializing: isInitializing, - error: payload.error, - sideEffects: [], - }; - } - - // ------------------------------------------------------------------ - // Convert entries to timeline units - // ------------------------------------------------------------------ - const timelineUnits = payload.entries.map((entry) => ({ - seq: entry.seqStart, - event: { - type: "timeline", - provider: entry.provider, - item: entry.item, - } as AgentStreamEventPayload, - timestamp: new Date(entry.timestamp), - })); - - const toHydratedEvents = ( - units: typeof timelineUnits, - ): Array<{ event: AgentStreamEventPayload; timestamp: Date }> => - units.map(({ event, timestamp }) => ({ event, timestamp })); - - // ------------------------------------------------------------------ - // Derive bootstrap policy (replace vs incremental) - // ------------------------------------------------------------------ - const bootstrapPolicy = deriveBootstrapTailTimelinePolicy({ - direction: payload.direction, - reset: payload.reset, - epoch: payload.epoch, - endCursor: payload.endCursor, - isInitializing, - hasActiveInitDeferred, - }); - const replace = bootstrapPolicy.replace; - - let nextTail = currentTail; - let nextHead = currentHead; - let nextCursor: TimelineCursor | null | undefined = currentCursor; - let cursorChanged = false; - const sideEffects: TimelineReducerSideEffect[] = []; - - if (replace) { - // ---------------------------------------------------------------- - // Replace path: full hydration from scratch - // ---------------------------------------------------------------- - nextTail = hydrateStreamState(toHydratedEvents(timelineUnits), { - source: "canonical", - }); - nextHead = []; - - if (payload.startCursor && payload.endCursor) { - nextCursor = { - epoch: payload.epoch, - startSeq: payload.startCursor.seq, - endSeq: payload.endCursor.seq, - }; - cursorChanged = true; - } else { - nextCursor = null; - cursorChanged = true; - } - - if (bootstrapPolicy.catchUpCursor) { - sideEffects.push({ - type: "catch_up", - cursor: bootstrapPolicy.catchUpCursor, - }); - } - } else if (timelineUnits.length > 0) { - // ---------------------------------------------------------------- - // Incremental append path - // ---------------------------------------------------------------- - const acceptedUnits: typeof timelineUnits = []; - let cursor = currentCursor; - let gapCursor: { epoch: string; endSeq: number } | null = null; - - for (const unit of timelineUnits) { - const decision: SessionTimelineSeqDecision = classifySessionTimelineSeq({ - cursor: cursor ? { epoch: cursor.epoch, endSeq: cursor.endSeq } : null, - epoch: payload.epoch, - seq: unit.seq, - }); - - if (decision === "gap") { - gapCursor = cursor ? { epoch: cursor.epoch, endSeq: cursor.endSeq } : null; - break; - } - if (decision === "drop_stale" || decision === "drop_epoch") { - continue; - } - - acceptedUnits.push(unit); - if (decision === "init") { - cursor = { - epoch: payload.epoch, - startSeq: unit.seq, - endSeq: unit.seq, - }; - continue; - } - if (!cursor) { - continue; - } - cursor = { - ...cursor, - endSeq: unit.seq, - }; - } - - if (acceptedUnits.length > 0) { - nextTail = acceptedUnits.reduce( - (state, { event, timestamp }) => - reduceStreamUpdate(state, event, timestamp, { - source: "canonical", - }), - currentTail, - ); - } - - if ( - cursor && - (!currentCursor || - currentCursor.epoch !== cursor.epoch || - currentCursor.startSeq !== cursor.startSeq || - currentCursor.endSeq !== cursor.endSeq) - ) { - nextCursor = cursor; - cursorChanged = true; - } - - if (gapCursor) { - sideEffects.push({ type: "catch_up", cursor: gapCursor }); - } - } - - // ------------------------------------------------------------------ - // Flush pending agent updates side effect - // ------------------------------------------------------------------ - sideEffects.push({ type: "flush_pending_updates" }); - - // ------------------------------------------------------------------ - // Init resolution - // ------------------------------------------------------------------ - const shouldResolveDeferredInit = shouldResolveTimelineInit({ - hasActiveInitDeferred, - isInitializing, - initRequestDirection, - responseDirection: payload.direction, - reset: payload.reset, - }); - const clearInitializing = shouldResolveDeferredInit || (isInitializing && !hasActiveInitDeferred); - - const initResolution: "resolve" | "reject" | null = shouldResolveDeferredInit ? "resolve" : null; - - return { - tail: nextTail, - head: nextHead, - cursor: nextCursor, - cursorChanged, - initResolution, - clearInitializing, - error: null, - sideEffects, - }; -} - -// --------------------------------------------------------------------------- -// processAgentStreamEvent -// --------------------------------------------------------------------------- - export interface ProcessAgentStreamEventInput { event: AgentStreamEventPayload; seq: number | undefined; - epoch: string | undefined; currentTail: StreamItem[]; currentHead: StreamItem[]; currentCursor: TimelineCursor | undefined; @@ -307,75 +96,250 @@ export interface ProcessAgentStreamEventOutput { sideEffects: AgentStreamReducerSideEffect[]; } +function cursorsEqual( + left: TimelineCursor | null | undefined, + right: TimelineCursor | null | undefined, +): boolean { + if (!left || !right) { + return left === right; + } + return left.startSeq === right.startSeq && left.endSeq === right.endSeq; +} + +function removeSupersededProvisionalItems( + head: StreamItem[], + event: AgentStreamEventPayload, +): StreamItem[] { + if (head.length === 0 || event.type !== "timeline") { + return head; + } + + let nextHead = head; + if (event.item.type === "assistant_message") { + nextHead = head.filter((item) => item.kind !== "assistant_message"); + } else if (event.item.type === "tool_call") { + const committedToolCall = event.item; + nextHead = head.filter( + (item) => + item.kind !== "tool_call" || + item.payload.source !== "agent" || + item.payload.data.callId !== committedToolCall.callId, + ); + } + + return nextHead.length === head.length ? head : nextHead; +} + +export function processTimelineResponse( + input: ProcessTimelineResponseInput, +): ProcessTimelineResponseOutput { + const { + payload, + currentTail, + currentHead, + currentCursor, + isInitializing, + hasActiveInitDeferred, + initRequestDirection, + } = input; + + if (payload.error) { + return { + tail: currentTail, + head: currentHead, + cursor: currentCursor, + cursorChanged: false, + initResolution: hasActiveInitDeferred ? "reject" : null, + clearInitializing: isInitializing, + error: payload.error, + sideEffects: [], + }; + } + + const timelineUnits = payload.entries.map((entry) => ({ + seq: entry.seq, + event: { + type: "timeline", + provider: entry.provider, + item: entry.item, + } as AgentStreamEventPayload, + timestamp: new Date(entry.timestamp), + })); + + const bootstrapPolicy = deriveBootstrapTailTimelinePolicy({ + direction: payload.direction, + endSeq: payload.endSeq, + isInitializing, + hasActiveInitDeferred, + }); + + let nextTail = currentTail; + let nextHead = currentHead; + let nextCursor: TimelineCursor | null | undefined = currentCursor; + let cursorChanged = false; + const sideEffects: TimelineReducerSideEffect[] = []; + + if (bootstrapPolicy.replace) { + nextTail = hydrateStreamState( + timelineUnits.map(({ event, timestamp }) => ({ event, timestamp })), + { source: "canonical" }, + ); + nextHead = []; + nextCursor = + typeof payload.startSeq === "number" && typeof payload.endSeq === "number" + ? { + startSeq: payload.startSeq, + endSeq: payload.endSeq, + } + : null; + cursorChanged = !cursorsEqual(currentCursor, nextCursor); + + if (bootstrapPolicy.catchUpCursor) { + sideEffects.push({ + type: "catch_up", + cursor: bootstrapPolicy.catchUpCursor, + }); + } + } else if (payload.direction === "before") { + const prepended = hydrateStreamState( + timelineUnits.map(({ event, timestamp }) => ({ event, timestamp })), + { source: "canonical" }, + ); + nextTail = prepended.length > 0 ? [...prepended, ...currentTail] : currentTail; + const derivedCursor = + typeof payload.startSeq === "number" + ? { + startSeq: payload.startSeq, + endSeq: currentCursor?.endSeq ?? payload.endSeq ?? payload.startSeq, + } + : currentCursor; + nextCursor = derivedCursor; + cursorChanged = !cursorsEqual(currentCursor, derivedCursor); + } else if (timelineUnits.length > 0) { + const acceptedUnits: typeof timelineUnits = []; + let cursor = currentCursor; + let gapCursor: { endSeq: number } | null = null; + + for (const unit of timelineUnits) { + const decision: SessionTimelineSeqDecision = classifySessionTimelineSeq({ + cursor: cursor ? { endSeq: cursor.endSeq } : null, + seq: unit.seq, + }); + + if (decision === "gap") { + gapCursor = cursor ? { endSeq: cursor.endSeq } : null; + break; + } + if (decision === "drop_stale") { + continue; + } + + acceptedUnits.push(unit); + cursor = + decision === "init" + ? { startSeq: unit.seq, endSeq: unit.seq } + : { ...(cursor ?? { startSeq: unit.seq, endSeq: unit.seq }), endSeq: unit.seq }; + nextHead = removeSupersededProvisionalItems(nextHead, unit.event); + } + + if (acceptedUnits.length > 0) { + nextTail = acceptedUnits.reduce( + (state, { event, timestamp }) => + reduceStreamUpdate(state, event, timestamp, { + source: "canonical", + }), + currentTail, + ); + } + + if (cursor && !cursorsEqual(currentCursor, cursor)) { + nextCursor = cursor; + cursorChanged = true; + } + + if (gapCursor) { + sideEffects.push({ type: "catch_up", cursor: gapCursor }); + } + } + + sideEffects.push({ type: "flush_pending_updates" }); + + const shouldResolveDeferredInit = shouldResolveTimelineInit({ + hasActiveInitDeferred, + isInitializing, + initRequestDirection, + responseDirection: payload.direction, + }); + const clearInitializing = shouldResolveDeferredInit || (isInitializing && !hasActiveInitDeferred); + + return { + tail: nextTail, + head: nextHead, + cursor: nextCursor, + cursorChanged, + initResolution: shouldResolveDeferredInit ? "resolve" : null, + clearInitializing, + error: null, + sideEffects, + }; +} + export function processAgentStreamEvent( input: ProcessAgentStreamEventInput, ): ProcessAgentStreamEventOutput { - const { event, seq, epoch, currentTail, currentHead, currentCursor, currentAgent, timestamp } = - input; + const { event, seq, currentTail, currentHead, currentCursor, currentAgent, timestamp } = input; - let shouldApplyStreamEvent = true; + let nextTail = currentTail; + let nextHead = currentHead; + let changedTail = false; + let changedHead = false; let nextTimelineCursor: TimelineCursor | null = null; let cursorChanged = false; const sideEffects: AgentStreamReducerSideEffect[] = []; - // ------------------------------------------------------------------ - // Timeline sequencing gate - // ------------------------------------------------------------------ - if (event.type === "timeline" && typeof seq === "number" && typeof epoch === "string") { + if (event.type === "timeline" && typeof seq === "number") { const decision = classifySessionTimelineSeq({ - cursor: currentCursor ? { epoch: currentCursor.epoch, endSeq: currentCursor.endSeq } : null, - epoch, + cursor: currentCursor ? { endSeq: currentCursor.endSeq } : null, seq, }); - if (decision === "init") { - nextTimelineCursor = { epoch, startSeq: seq, endSeq: seq }; - cursorChanged = true; - } else if (decision === "accept") { - nextTimelineCursor = { - ...(currentCursor ?? { epoch, startSeq: seq, endSeq: seq }), - epoch, - endSeq: seq, - }; - cursorChanged = true; - } else if (decision === "gap") { - shouldApplyStreamEvent = false; + if (decision === "gap") { if (currentCursor) { sideEffects.push({ type: "catch_up", - cursor: { - epoch: currentCursor.epoch, - endSeq: currentCursor.endSeq, - }, + cursor: { endSeq: currentCursor.endSeq }, }); } - } else { - // drop_stale or drop_epoch - shouldApplyStreamEvent = false; + } else if (decision !== "drop_stale") { + nextTail = reduceStreamUpdate(currentTail, event, timestamp, { + source: "canonical", + }); + changedTail = nextTail !== currentTail; + + nextHead = removeSupersededProvisionalItems(currentHead, event); + changedHead = nextHead !== currentHead; + + nextTimelineCursor = + decision === "init" + ? { startSeq: seq, endSeq: seq } + : { ...(currentCursor ?? { startSeq: seq, endSeq: seq }), endSeq: seq }; + cursorChanged = !cursorsEqual(currentCursor, nextTimelineCursor); } + } else if (event.type === "timeline") { + nextHead = reduceStreamUpdate(currentHead, event, timestamp, { + source: "live", + }); + changedHead = nextHead !== currentHead; + } else if ( + (event.type === "turn_completed" || + event.type === "turn_canceled" || + event.type === "turn_failed") && + currentHead.length > 0 + ) { + nextHead = []; + changedHead = true; } - // ------------------------------------------------------------------ - // Apply stream event to tail/head - // ------------------------------------------------------------------ - const { tail, head, changedTail, changedHead } = shouldApplyStreamEvent - ? applyStreamEvent({ - tail: currentTail, - head: currentHead, - event, - timestamp, - source: "live", - }) - : { - tail: currentTail, - head: currentHead, - changedTail: false, - changedHead: false, - }; - - // ------------------------------------------------------------------ - // Optimistic lifecycle status - // ------------------------------------------------------------------ let agentPatch: AgentPatch | null = null; let agentChanged = false; @@ -402,8 +366,8 @@ export function processAgentStreamEvent( } return { - tail, - head, + tail: nextTail, + head: nextHead, changedTail, changedHead, cursor: nextTimelineCursor, diff --git a/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts b/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts index e233b87421..cc8b98ce03 100644 --- a/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts +++ b/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts @@ -2,26 +2,42 @@ import { describe, expect, it } from "vitest"; import { classifySessionTimelineSeq } from "./session-timeline-seq-gate"; import { deriveBootstrapTailTimelinePolicy, + deriveInitialTimelineRequest, shouldResolveTimelineInit, } from "./session-timeline-bootstrap-policy"; -describe("deriveBootstrapTailTimelinePolicy", () => { - it("always replaces on explicit reset without catch-up cursor", () => { - const policy = deriveBootstrapTailTimelinePolicy({ - direction: "after", - reset: true, - epoch: "epoch-1", - endCursor: { seq: 200 }, - isInitializing: false, - hasActiveInitDeferred: false, +describe("deriveInitialTimelineRequest", () => { + it("uses tail bootstrap when history has not synced yet", () => { + expect( + deriveInitialTimelineRequest({ + cursor: { seq: 42 }, + hasAuthoritativeHistory: false, + initialTimelineLimit: 200, + }), + ).toEqual({ + direction: "tail", + limit: 200, }); - - expect(policy.replace).toBe(true); - expect(policy.catchUpCursor).toBeNull(); }); + it("uses catch-up after the committed cursor once history is synced", () => { + expect( + deriveInitialTimelineRequest({ + cursor: { seq: 42 }, + hasAuthoritativeHistory: true, + initialTimelineLimit: 200, + }), + ).toEqual({ + direction: "after", + cursor: { seq: 42 }, + limit: 0, + }); + }); +}); + +describe("deriveBootstrapTailTimelinePolicy", () => { it("forces baseline replace and canonical catch-up for init tail race", () => { - const advancedCursor = { epoch: "epoch-1", endSeq: 205 }; + const advancedCursor = { endSeq: 205 }; const tailSeqStart = 101; const tailSeqEnd = 200; @@ -29,7 +45,6 @@ describe("deriveBootstrapTailTimelinePolicy", () => { for (let seq = tailSeqStart; seq <= tailSeqEnd; seq += 1) { const decision = classifySessionTimelineSeq({ cursor: advancedCursor, - epoch: "epoch-1", seq, }); if (decision === "accept" || decision === "init") { @@ -40,16 +55,13 @@ describe("deriveBootstrapTailTimelinePolicy", () => { const policy = deriveBootstrapTailTimelinePolicy({ direction: "tail", - reset: false, - epoch: "epoch-1", - endCursor: { seq: 200 }, + endSeq: 200, isInitializing: true, hasActiveInitDeferred: true, }); expect(policy.replace).toBe(true); expect(policy.catchUpCursor).toEqual({ - epoch: "epoch-1", endSeq: 200, }); }); @@ -57,9 +69,7 @@ describe("deriveBootstrapTailTimelinePolicy", () => { it("does not replace non-bootstrap, non-reset responses", () => { const policy = deriveBootstrapTailTimelinePolicy({ direction: "tail", - reset: false, - epoch: "epoch-1", - endCursor: { seq: 200 }, + endSeq: 200, isInitializing: false, hasActiveInitDeferred: false, }); @@ -77,7 +87,6 @@ describe("shouldResolveTimelineInit", () => { isInitializing: true, initRequestDirection: "tail", responseDirection: "tail", - reset: false, }), ).toBe(true); }); @@ -89,7 +98,6 @@ describe("shouldResolveTimelineInit", () => { isInitializing: true, initRequestDirection: "tail", responseDirection: "after", - reset: false, }), ).toBe(false); }); @@ -101,7 +109,6 @@ describe("shouldResolveTimelineInit", () => { isInitializing: true, initRequestDirection: "after", responseDirection: "after", - reset: false, }), ).toBe(true); }); diff --git a/packages/app/src/contexts/session-timeline-bootstrap-policy.ts b/packages/app/src/contexts/session-timeline-bootstrap-policy.ts index 706ab5eeee..c7ab9ded83 100644 --- a/packages/app/src/contexts/session-timeline-bootstrap-policy.ts +++ b/packages/app/src/contexts/session-timeline-bootstrap-policy.ts @@ -6,7 +6,6 @@ type BootstrapTailCursor = { } | null; type InitialTimelineCursor = { - epoch: string; seq: number; } | null; @@ -20,48 +19,37 @@ export function deriveInitialTimelineRequest({ initialTimelineLimit: number; }): { direction: "tail" | "after"; - cursor?: { epoch: string; seq: number }; + cursor?: { seq: number }; limit: number; - projection: "canonical"; } { if (!hasAuthoritativeHistory || !cursor) { return { direction: "tail", limit: initialTimelineLimit, - projection: "canonical", }; } return { direction: "after", - cursor: { epoch: cursor.epoch, seq: cursor.seq }, + cursor: { seq: cursor.seq }, limit: 0, - projection: "canonical", }; } export function deriveBootstrapTailTimelinePolicy({ direction, - reset, - epoch, - endCursor, + endSeq, isInitializing, hasActiveInitDeferred, }: { direction: TimelineDirection; - reset: boolean; - epoch: string; - endCursor: BootstrapTailCursor; + endSeq: number | null; isInitializing: boolean; hasActiveInitDeferred: boolean; }): { replace: boolean; - catchUpCursor: { epoch: string; endSeq: number } | null; + catchUpCursor: { endSeq: number } | null; } { - if (reset) { - return { replace: true, catchUpCursor: null }; - } - const isBootstrapTailInit = direction === "tail" && isInitializing && hasActiveInitDeferred; if (!isBootstrapTailInit) { return { replace: false, catchUpCursor: null }; @@ -69,7 +57,7 @@ export function deriveBootstrapTailTimelinePolicy({ return { replace: true, - catchUpCursor: endCursor ? { epoch, endSeq: endCursor.seq } : null, + catchUpCursor: typeof endSeq === "number" ? { endSeq } : null, }; } @@ -78,19 +66,14 @@ export function shouldResolveTimelineInit({ isInitializing, initRequestDirection, responseDirection, - reset, }: { hasActiveInitDeferred: boolean; isInitializing: boolean; initRequestDirection: InitRequestDirection; responseDirection: TimelineDirection; - reset: boolean; }): boolean { if (!hasActiveInitDeferred || !isInitializing) { return false; } - if (reset) { - return true; - } return responseDirection === initRequestDirection; } diff --git a/packages/app/src/contexts/session-timeline-seq-gate.test.ts b/packages/app/src/contexts/session-timeline-seq-gate.test.ts index d18af910a4..3f827dce85 100644 --- a/packages/app/src/contexts/session-timeline-seq-gate.test.ts +++ b/packages/app/src/contexts/session-timeline-seq-gate.test.ts @@ -5,8 +5,7 @@ describe("classifySessionTimelineSeq", () => { it("accepts contiguous forward seq", () => { expect( classifySessionTimelineSeq({ - cursor: { epoch: "epoch-1", endSeq: 4 }, - epoch: "epoch-1", + cursor: { endSeq: 4 }, seq: 5, }), ).toBe("accept"); @@ -15,8 +14,7 @@ describe("classifySessionTimelineSeq", () => { it("drops stale seq older than the current end", () => { expect( classifySessionTimelineSeq({ - cursor: { epoch: "epoch-1", endSeq: 8 }, - epoch: "epoch-1", + cursor: { endSeq: 8 }, seq: 7, }), ).toBe("drop_stale"); @@ -25,28 +23,16 @@ describe("classifySessionTimelineSeq", () => { it("drops duplicate replay seq equal to the current end", () => { expect( classifySessionTimelineSeq({ - cursor: { epoch: "epoch-1", endSeq: 8 }, - epoch: "epoch-1", + cursor: { endSeq: 8 }, seq: 8, }), ).toBe("drop_stale"); }); - it("drops epoch mismatch", () => { - expect( - classifySessionTimelineSeq({ - cursor: { epoch: "epoch-1", endSeq: 4 }, - epoch: "epoch-2", - seq: 5, - }), - ).toBe("drop_epoch"); - }); - it("initializes when cursor is null", () => { expect( classifySessionTimelineSeq({ cursor: null, - epoch: "epoch-1", seq: 1, }), ).toBe("init"); @@ -55,8 +41,7 @@ describe("classifySessionTimelineSeq", () => { it("classifies forward gaps", () => { expect( classifySessionTimelineSeq({ - cursor: { epoch: "epoch-1", endSeq: 4 }, - epoch: "epoch-1", + cursor: { endSeq: 4 }, seq: 9, }), ).toBe("gap"); diff --git a/packages/app/src/contexts/session-timeline-seq-gate.ts b/packages/app/src/contexts/session-timeline-seq-gate.ts index e7e35b5675..1fe15597e9 100644 --- a/packages/app/src/contexts/session-timeline-seq-gate.ts +++ b/packages/app/src/contexts/session-timeline-seq-gate.ts @@ -1,28 +1,22 @@ export type SessionTimelineSeqCursor = | { - epoch: string; endSeq: number; } | null | undefined; -export type SessionTimelineSeqDecision = "accept" | "drop_stale" | "drop_epoch" | "gap" | "init"; +export type SessionTimelineSeqDecision = "accept" | "drop_stale" | "gap" | "init"; export function classifySessionTimelineSeq({ cursor, - epoch, seq, }: { cursor: SessionTimelineSeqCursor; - epoch: string; seq: number; }): SessionTimelineSeqDecision { if (!cursor) { return "init"; } - if (cursor.epoch !== epoch) { - return "drop_epoch"; - } if (seq <= cursor.endSeq) { return "drop_stale"; } diff --git a/packages/app/src/hooks/use-agent-initialization.test.ts b/packages/app/src/hooks/use-agent-initialization.test.ts index dd32a15d6e..43da108e30 100644 --- a/packages/app/src/hooks/use-agent-initialization.test.ts +++ b/packages/app/src/hooks/use-agent-initialization.test.ts @@ -2,11 +2,10 @@ import { describe, expect, it } from "vitest"; import { __private__ } from "./use-agent-initialization"; describe("useAgentInitialization timeline request policy", () => { - it("uses canonical tail bootstrap when history has not synced yet", () => { + it("uses committed tail bootstrap when history has not synced yet", () => { expect( __private__.deriveInitialTimelineRequest({ cursor: { - epoch: "epoch-1", seq: 42, }, hasAuthoritativeHistory: false, @@ -15,11 +14,10 @@ describe("useAgentInitialization timeline request policy", () => { ).toEqual({ direction: "tail", limit: 200, - projection: "canonical", }); }); - it("uses canonical tail bootstrap when cursor is missing", () => { + it("uses committed tail bootstrap when cursor is missing", () => { expect( __private__.deriveInitialTimelineRequest({ cursor: null, @@ -29,15 +27,13 @@ describe("useAgentInitialization timeline request policy", () => { ).toEqual({ direction: "tail", limit: 200, - projection: "canonical", }); }); - it("uses canonical catch-up after the current cursor once history is synced", () => { + it("uses committed catch-up after the current cursor once history is synced", () => { expect( __private__.deriveInitialTimelineRequest({ cursor: { - epoch: "epoch-1", seq: 42, }, hasAuthoritativeHistory: true, @@ -45,9 +41,8 @@ describe("useAgentInitialization timeline request policy", () => { }), ).toEqual({ direction: "after", - cursor: { epoch: "epoch-1", seq: 42 }, + cursor: { seq: 42 }, limit: 0, - projection: "canonical", }); }); @@ -61,7 +56,6 @@ describe("useAgentInitialization timeline request policy", () => { ).toEqual({ direction: "tail", limit: 0, - projection: "canonical", }); }); diff --git a/packages/app/src/hooks/use-agent-initialization.ts b/packages/app/src/hooks/use-agent-initialization.ts index 62ffa9a9dd..20df64bcdc 100644 --- a/packages/app/src/hooks/use-agent-initialization.ts +++ b/packages/app/src/hooks/use-agent-initialization.ts @@ -60,7 +60,7 @@ export function useAgentInitialization({ const hasAuthoritativeHistory = session?.agentAuthoritativeHistoryApplied.get(agentId) === true; const timelineRequest = deriveInitialTimelineRequest({ - cursor: cursor ? { epoch: cursor.epoch, seq: cursor.endSeq } : null, + cursor: cursor ? { seq: cursor.endSeq } : null, hasAuthoritativeHistory, initialTimelineLimit, }); @@ -107,7 +107,6 @@ export function useAgentInitialization({ await client.fetchAgentTimeline(agentId, { direction: "tail", limit: initialTimelineLimit, - projection: "canonical", }); } catch (error) { setAgentInitializing(agentId, false); diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 0646658cbe..094662af70 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -191,7 +191,6 @@ export type DaemonServerInfo = { }; export interface AgentTimelineCursorState { - epoch: string; startSeq: number; endSeq: number; } diff --git a/packages/server/src/client/daemon-client.test.ts b/packages/server/src/client/daemon-client.test.ts index eed79a7e00..f705b2bb27 100644 --- a/packages/server/src/client/daemon-client.test.ts +++ b/packages/server/src/client/daemon-client.test.ts @@ -1704,24 +1704,15 @@ describe("DaemonClient", () => { agentId: "agent_cli", agent: null, direction: "tail", - projection: "projected", - epoch: "epoch-1", - reset: false, - staleCursor: false, - gap: false, - window: { minSeq: 1, maxSeq: 1, nextSeq: 2 }, - startCursor: { epoch: "epoch-1", seq: 1 }, - endCursor: { epoch: "epoch-1", seq: 1 }, + startSeq: 1, + endSeq: 1, hasOlder: false, hasNewer: false, entries: [ { timestamp: "2026-02-08T20:20:00.000Z", provider: "codex", - seqStart: 1, - seqEnd: 1, - sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }], - collapsed: [], + seq: 1, item: { type: "tool_call", callId: "call_cli_snapshot", @@ -1798,24 +1789,15 @@ describe("DaemonClient", () => { agentId: "agent_cli", agent: null, direction: "tail", - projection: "projected", - epoch: "epoch-1", - reset: false, - staleCursor: false, - gap: false, - window: { minSeq: 1, maxSeq: 1, nextSeq: 2 }, - startCursor: { epoch: "epoch-1", seq: 1 }, - endCursor: { epoch: "epoch-1", seq: 1 }, + startSeq: 1, + endSeq: 1, hasOlder: false, hasNewer: false, entries: [ { timestamp: "2026-02-08T20:20:00.000Z", provider: "codex", - seqStart: 1, - seqEnd: 1, - sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }], - collapsed: [], + seq: 1, item: { type: "tool_call", callId: "call_cli_invalid", diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index f927f3fa67..22307c92bc 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -130,7 +130,6 @@ export type DaemonEvent = event: AgentStreamEventPayload; timestamp: string; seq?: number; - epoch?: string; } | { type: "status"; payload: { status: string } & Record } | { type: "agent_deleted"; agentId: string } @@ -244,13 +243,13 @@ type KillTerminalPayload = KillTerminalResponse["payload"]; export type FetchAgentTimelinePayload = FetchAgentTimelineResponseMessage["payload"]; export type FetchAgentTimelineDirection = FetchAgentTimelinePayload["direction"]; -export type FetchAgentTimelineProjection = FetchAgentTimelinePayload["projection"]; -export type FetchAgentTimelineCursor = NonNullable; +export type FetchAgentTimelineCursor = NonNullable< + Extract["cursor"] +>; export type FetchAgentTimelineOptions = { direction?: FetchAgentTimelineDirection; cursor?: FetchAgentTimelineCursor; limit?: number; - projection?: FetchAgentTimelineProjection; requestId?: string; }; @@ -1396,7 +1395,6 @@ export class DaemonClient { ...(options.direction ? { direction: options.direction } : {}), ...(options.cursor ? { cursor: options.cursor } : {}), ...(typeof options.limit === "number" ? { limit: options.limit } : {}), - ...(options.projection ? { projection: options.projection } : {}), }); const payload = await this.sendRequest({ @@ -3100,7 +3098,6 @@ export class DaemonClient { event: msg.payload.event, timestamp: msg.payload.timestamp, ...(typeof msg.payload.seq === "number" ? { seq: msg.payload.seq } : {}), - ...(typeof msg.payload.epoch === "string" ? { epoch: msg.payload.epoch } : {}), }; case "status": return { type: "status", payload: msg.payload }; diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 00866731b3..5d2e78a471 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -197,6 +197,119 @@ class TestAgentSession implements AgentSession { async close(): Promise {} } +class StreamingAssistantSession implements AgentSession { + readonly provider = "codex" as const; + readonly capabilities = TEST_CAPABILITIES; + readonly id = randomUUID(); + private subscribers = new Set<(event: AgentStreamEvent) => void>(); + private turnIdCounter = 0; + + constructor(private readonly config: AgentSessionConfig) {} + + async run(): Promise { + return { + sessionId: this.id, + finalText: "", + timeline: [], + }; + } + + async startTurn(): Promise<{ turnId: string }> { + const turnId = `turn-${++this.turnIdCounter}`; + setTimeout(() => { + this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); + this.pushEvent({ + type: "timeline", + provider: this.provider, + turnId, + item: { type: "assistant_message", text: "final " }, + }); + this.pushEvent({ + type: "timeline", + provider: this.provider, + turnId, + item: { type: "assistant_message", text: "reply" }, + }); + this.pushEvent({ type: "turn_completed", provider: this.provider, turnId }); + }, 0); + return { turnId }; + } + + subscribe(callback: (event: AgentStreamEvent) => void): () => void { + this.subscribers.add(callback); + return () => { + this.subscribers.delete(callback); + }; + } + + private pushEvent(event: AgentStreamEvent): void { + for (const callback of this.subscribers) { + callback(event); + } + } + + async *streamHistory(): AsyncGenerator {} + + async getRuntimeInfo() { + return { + provider: this.provider, + sessionId: this.id, + model: this.config.model ?? null, + modeId: this.config.modeId ?? null, + }; + } + + async getAvailableModes() { + return []; + } + + async getCurrentMode() { + return null; + } + + async setMode(): Promise {} + + getPendingPermissions() { + return []; + } + + async respondToPermission(): Promise {} + + describePersistence() { + return { + provider: this.provider, + sessionId: this.id, + }; + } + + async interrupt(): Promise {} + + async close(): Promise {} +} + +class StreamingAssistantClient implements AgentClient { + readonly provider = "codex" as const; + readonly capabilities = TEST_CAPABILITIES; + + async isAvailable(): Promise { + return true; + } + + async createSession(config: AgentSessionConfig): Promise { + return new StreamingAssistantSession(config); + } + + async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + return new StreamingAssistantSession({ + provider: "codex", + cwd: config?.cwd ?? process.cwd(), + }); + } +} + describe("AgentManager", () => { const logger = createTestLogger(); @@ -784,7 +897,7 @@ describe("AgentManager", () => { } }); - test("fetchTimeline returns full timeline with reset when cursor epoch is stale", async () => { + test("fetchTimeline returns committed rows after a known seq without reset metadata", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-stale-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -802,49 +915,51 @@ describe("AgentManager", () => { cwd: workdir, }); - await manager.appendTimelineItem(snapshot.id, { - type: "assistant_message", - text: "one", - }); - await manager.appendTimelineItem(snapshot.id, { - type: "assistant_message", - text: "two", - }); - await manager.appendTimelineItem(snapshot.id, { - type: "assistant_message", - text: "three", - }); + for (let seq = 1; seq <= 120; seq += 1) { + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: `committed row ${seq}`, + }); + } const baseline = manager.fetchTimeline(snapshot.id, { direction: "tail", - limit: 2, + limit: 0, + }); + expect(baseline.rows).toHaveLength(120); + + await manager.emitLiveTimelineItem(snapshot.id, { + type: "assistant_message", + text: "partial reply", + }); + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "finalized reply", }); - expect(baseline.rows).toHaveLength(2); const result = manager.fetchTimeline(snapshot.id, { direction: "after", cursor: { - epoch: "stale-epoch", - seq: baseline.rows[baseline.rows.length - 1]!.seq, + seq: 120, }, - limit: 1, + limit: 0, }); - expect(result.reset).toBe(true); - expect(result.staleCursor).toBe(true); - expect(result.gap).toBe(false); - expect(result.rows).toHaveLength(3); - expect(result.rows[0]?.seq).toBe(1); - expect(result.rows[result.rows.length - 1]?.seq).toBe(3); + expect(result.rows).toHaveLength(1); + expect(result.rows[0]?.seq).toBe(121); + expect(result.rows[0]?.item).toEqual({ + type: "assistant_message", + text: "finalized reply", + }); }); - test("emits live timeline updates without recording canonical timeline rows", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-live-timeline-")); + test("streams assistant chunks provisionally and commits one finalized assistant row", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-provisional-timeline-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); const manager = new AgentManager({ clients: { - codex: new TestAgentClient(), + codex: new StreamingAssistantClient(), }, registry: storage, logger, @@ -858,9 +973,9 @@ describe("AgentManager", () => { const streamEvents: Array<{ seq?: number; - epoch?: string; eventType?: string; itemType?: string; + text?: string; }> = []; manager.subscribe( (event) => { @@ -869,37 +984,61 @@ describe("AgentManager", () => { } streamEvents.push({ seq: event.seq, - epoch: event.epoch, eventType: event.event.type, itemType: event.event.type === "timeline" ? event.event.item.type : undefined, + text: + event.event.type === "timeline" && event.event.item.type === "assistant_message" + ? event.event.item.text + : undefined, }); }, { agentId: snapshot.id, replayState: false }, ); - await manager.emitLiveTimelineItem(snapshot.id, { - type: "assistant_message", - text: "live-only update", - }); + const stream = manager.streamAgent(snapshot.id, "hello"); + while (true) { + const next = await stream.next(); + if (next.done) { + break; + } + } - expect(streamEvents).toHaveLength(1); - expect(streamEvents[0]).toMatchObject({ + const assistantTimelineEvents = streamEvents.filter((event) => event.itemType === "assistant_message"); + expect(assistantTimelineEvents).toHaveLength(3); + expect(assistantTimelineEvents[0]).toMatchObject({ eventType: "timeline", itemType: "assistant_message", + text: "final ", + }); + expect(assistantTimelineEvents[0]?.seq).toBeUndefined(); + expect(assistantTimelineEvents[1]).toMatchObject({ + text: "reply", + }); + expect(assistantTimelineEvents[1]?.seq).toBeUndefined(); + expect(assistantTimelineEvents[2]).toMatchObject({ + text: "final reply", + seq: 1, }); - expect(streamEvents[0]?.seq).toBeUndefined(); - expect(streamEvents[0]?.epoch).toBeUndefined(); - expect(manager.getTimeline(snapshot.id)).toEqual([]); + expect(manager.getTimeline(snapshot.id)).toEqual([ + { + type: "assistant_message", + text: "final reply", + }, + ]); const fetched = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0, }); - expect(fetched.rows).toEqual([]); + expect(fetched.rows).toHaveLength(1); + expect(fetched.rows[0]?.item).toEqual({ + type: "assistant_message", + text: "final reply", + }); }); - test("fetchTimeline returns full timeline with reset when cursor seq falls behind retention window", async () => { - const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-gap-")); + test("fetchTimeline supports older-history pagination with before seq", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-before-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); const manager = new AgentManager({ @@ -908,7 +1047,6 @@ describe("AgentManager", () => { }, registry: storage, logger, - maxTimelineItems: 2, idFactory: () => "00000000-0000-4000-8000-000000000119", }); @@ -933,32 +1071,27 @@ describe("AgentManager", () => { type: "assistant_message", text: "fourth", }); - - const fresh = manager.fetchTimeline(snapshot.id, { - direction: "tail", - limit: 0, + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "fifth", }); - expect(fresh.window.minSeq).toBe(3); - expect(fresh.window.maxSeq).toBe(4); const result = manager.fetchTimeline(snapshot.id, { - direction: "after", + direction: "before", cursor: { - epoch: fresh.epoch, - seq: 1, + seq: 5, }, - limit: 10, + limit: 2, }); - expect(result.reset).toBe(true); - expect(result.staleCursor).toBe(false); - expect(result.gap).toBe(true); expect(result.rows).toHaveLength(2); expect(result.rows[0]?.seq).toBe(3); expect(result.rows[1]?.seq).toBe(4); + expect(result.hasOlder).toBe(true); + expect(result.hasNewer).toBe(true); }); - test("does not trim timeline by default", async () => { + test("does not trim committed history", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-unbounded-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -998,6 +1131,178 @@ describe("AgentManager", () => { expect(fetched.window.maxSeq).toBe(3); }); + test("hydrateTimeline canonicalizes tool-interleaved assistant replay into the committed turn shape", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-canonical-assistant-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + + class ChunkedAssistantHistorySession extends TestAgentSession { + constructor(config: AgentSessionConfig) { + super(config); + } + + async *streamHistory(): AsyncGenerator { + yield { + type: "timeline", + provider: this.provider, + item: { type: "assistant_message", text: "chunk one " }, + }; + yield { + type: "timeline", + provider: this.provider, + item: { type: "assistant_message", text: "chunk two" }, + }; + yield { + type: "timeline", + provider: this.provider, + item: { type: "reasoning", text: "internal" }, + }; + yield { + type: "timeline", + provider: this.provider, + item: { + type: "tool_call", + callId: "call-history-1", + name: "shell", + status: "completed", + detail: { + type: "shell", + command: "echo hi", + output: "hi\n", + exitCode: 0, + }, + error: null, + }, + }; + yield { + type: "timeline", + provider: this.provider, + item: { type: "assistant_message", text: "final answer" }, + }; + } + } + + class ChunkedAssistantHistoryClient implements AgentClient { + readonly provider = "codex" as const; + readonly capabilities = TEST_CAPABILITIES; + + async isAvailable(): Promise { + return true; + } + + async createSession(config: AgentSessionConfig): Promise { + return new ChunkedAssistantHistorySession(config); + } + + async resumeSession(): Promise { + throw new Error("Not used in this test"); + } + } + + const manager = new AgentManager({ + clients: { + codex: new ChunkedAssistantHistoryClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000121", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + + await manager.hydrateTimelineFromProvider(snapshot.id); + + expect(manager.getTimeline(snapshot.id)).toEqual([ + { + type: "tool_call", + callId: "call-history-1", + name: "shell", + status: "completed", + detail: { + type: "shell", + command: "echo hi", + output: "hi\n", + exitCode: 0, + }, + error: null, + }, + { type: "assistant_message", text: "chunk one chunk twofinal answer" }, + ]); + }); + + test("hydrateTimeline canonicalizes reasoning-interleaved assistant replay into one committed assistant row", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-history-reasoning-interleave-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + + class ReasoningInterleavedHistorySession extends TestAgentSession { + constructor(config: AgentSessionConfig) { + super(config); + } + + async *streamHistory(): AsyncGenerator { + yield { + type: "timeline", + provider: this.provider, + item: { type: "assistant_message", text: "before reasoning " }, + }; + yield { + type: "timeline", + provider: this.provider, + item: { type: "reasoning", text: "internal step" }, + }; + yield { + type: "timeline", + provider: this.provider, + item: { type: "assistant_message", text: "after reasoning" }, + }; + } + } + + class ReasoningInterleavedHistoryClient implements AgentClient { + readonly provider = "codex" as const; + readonly capabilities = TEST_CAPABILITIES; + + async isAvailable(): Promise { + return true; + } + + async createSession(config: AgentSessionConfig): Promise { + return new ReasoningInterleavedHistorySession(config); + } + + async resumeSession(): Promise { + throw new Error("Not used in this test"); + } + } + + const manager = new AgentManager({ + clients: { + codex: new ReasoningInterleavedHistoryClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000122", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + + await manager.hydrateTimelineFromProvider(snapshot.id); + + expect(manager.getTimeline(snapshot.id)).toEqual([ + { + type: "assistant_message", + text: "before reasoning after reasoning", + }, + ]); + }); + test("createAgent fails when generated agent ID is not a UUID", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); const storagePath = join(workdir, "agents"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index e0bb8ce357..e8b99dd8f2 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -42,7 +42,6 @@ export type AgentManagerEvent = agentId: string; event: AgentStreamEvent; seq?: number; - epoch?: string; }; export type AgentSubscriber = (event: AgentManagerEvent) => void; @@ -70,7 +69,6 @@ export type ProviderAvailability = { export type AgentManagerOptions = { clients?: Partial>; - maxTimelineItems?: number; idFactory?: () => string; registry?: AgentStorage; onAgentAttention?: AgentAttentionCallback; @@ -99,7 +97,6 @@ export type AgentTimelineRow = { }; export type AgentTimelineCursor = { - epoch: string; seq: number; }; @@ -123,11 +120,7 @@ export type AgentTimelineWindow = { }; export type AgentTimelineFetchResult = { - epoch: string; direction: AgentTimelineFetchDirection; - reset: boolean; - staleCursor: boolean; - gap: boolean; window: AgentTimelineWindow; hasOlder: boolean; hasNewer: boolean; @@ -173,8 +166,8 @@ type ManagedAgentBase = { pendingReplacement: boolean; timeline: AgentTimelineItem[]; timelineRows: AgentTimelineRow[]; - timelineEpoch: string; timelineNextSeq: number; + provisionalAssistantText: string | null; persistence: AgentPersistenceHandle | null; historyPrimed: boolean; lastUserMessageAt: Date | null; @@ -308,7 +301,6 @@ export class AgentManager { private readonly agents = new Map(); private readonly pendingForegroundRuns = new Map(); private readonly subscribers = new Set(); - private readonly maxTimelineItems: number | null; private readonly idFactory: () => string; private readonly registry?: AgentStorage; private readonly previousStatuses = new Map(); @@ -317,13 +309,6 @@ export class AgentManager { private logger: Logger; constructor(options: AgentManagerOptions) { - const maxTimelineItems = options?.maxTimelineItems; - this.maxTimelineItems = - typeof maxTimelineItems === "number" && - Number.isFinite(maxTimelineItems) && - maxTimelineItems >= 0 - ? Math.floor(maxTimelineItems) - : null; this.idFactory = options?.idFactory ?? (() => randomUUID()); this.registry = options?.registry; this.onAgentAttention = options?.onAgentAttention; @@ -525,7 +510,7 @@ export class AgentManager { fetchTimeline(id: string, options?: AgentTimelineFetchOptions): AgentTimelineFetchResult { const agent = this.requireAgent(id); - const { rows, epoch, nextSeq, minSeq, maxSeq } = this.ensureTimelineState(agent); + const { rows, nextSeq, minSeq, maxSeq } = this.ensureTimelineState(agent); const direction = options?.direction ?? "tail"; const requestedLimit = options?.limit; const limit = @@ -536,44 +521,12 @@ export class AgentManager { const window: AgentTimelineWindow = { minSeq, maxSeq, nextSeq }; - if (cursor && cursor.epoch !== epoch) { - return { - epoch, - direction, - reset: true, - staleCursor: true, - gap: false, - window, - hasOlder: false, - hasNewer: false, - rows: rows.map((row) => ({ ...row })), - }; - } - const selectAll = limit === 0; const cloneRows = (items: AgentTimelineRow[]) => items.map((row) => ({ ...row })); - if (direction === "after" && cursor && rows.length > 0 && cursor.seq < minSeq - 1) { - return { - epoch, - direction, - reset: true, - staleCursor: false, - gap: true, - window, - hasOlder: false, - hasNewer: false, - rows: cloneRows(rows), - }; - } - if (rows.length === 0) { return { - epoch, direction, - reset: false, - staleCursor: false, - gap: false, window, hasOlder: false, hasNewer: false, @@ -585,11 +538,7 @@ export class AgentManager { const selected = selectAll || limit >= rows.length ? rows : rows.slice(rows.length - limit); const hasOlder = selected.length > 0 && selected[0]!.seq > minSeq; return { - epoch, direction, - reset: false, - staleCursor: false, - gap: false, window, hasOlder, hasNewer: false, @@ -602,11 +551,7 @@ export class AgentManager { const startIdx = rows.findIndex((row) => row.seq > baseSeq); if (startIdx < 0) { return { - epoch, direction, - reset: false, - staleCursor: false, - gap: false, window, hasOlder: baseSeq >= minSeq, hasNewer: false, @@ -617,11 +562,7 @@ export class AgentManager { const selected = selectAll ? rows.slice(startIdx) : rows.slice(startIdx, startIdx + limit); const lastSelected = selected[selected.length - 1]; return { - epoch, direction, - reset: false, - staleCursor: false, - gap: false, window, hasOlder: selected[0]!.seq > minSeq, hasNewer: Boolean(lastSelected && lastSelected.seq < maxSeq), @@ -640,11 +581,7 @@ export class AgentManager { const hasOlder = selected.length > 0 && selected[0]!.seq > minSeq; const hasNewer = endExclusive >= 0; return { - epoch, direction, - reset: false, - staleCursor: false, - gap: false, window, hasOlder, hasNewer, @@ -722,8 +659,8 @@ export class AgentManager { const timelineState = this.ensureTimelineState(existing); const preservedTimeline = [...existing.timeline]; const preservedTimelineRows = timelineState.rows.map((row) => ({ ...row })); - const preservedTimelineEpoch = timelineState.epoch; const preservedTimelineNextSeq = timelineState.nextSeq; + const preservedProvisionalAssistantText = existing.provisionalAssistantText; const preservedHistoryPrimed = existing.historyPrimed; const preservedLastUsage = existing.lastUsage; const preservedLastError = existing.lastError; @@ -768,8 +705,8 @@ export class AgentManager { lastUserMessageAt: existing.lastUserMessageAt, timeline: preservedTimeline, timelineRows: preservedTimelineRows, - timelineEpoch: preservedTimelineEpoch, timelineNextSeq: preservedTimelineNextSeq, + provisionalAssistantText: preservedProvisionalAssistantText, historyPrimed: preservedHistoryPrimed, lastUsage: preservedLastUsage, lastError: preservedLastError, @@ -964,7 +901,6 @@ export class AgentManager { }, { seq: row.seq, - epoch: this.ensureTimelineState(agent).epoch, }, ); if (options?.emitState !== false) { @@ -985,7 +921,6 @@ export class AgentManager { }, { seq: row.seq, - epoch: this.ensureTimelineState(agent).epoch, }, ); await this.persistSnapshot(agent); @@ -1640,8 +1575,8 @@ export class AgentManager { labels?: Record; timeline?: AgentTimelineItem[]; timelineRows?: AgentTimelineRow[]; - timelineEpoch?: string; timelineNextSeq?: number; + provisionalAssistantText?: string | null; historyPrimed?: boolean; lastUsage?: AgentUsage; lastError?: string; @@ -1689,8 +1624,8 @@ export class AgentManager { unsubscribeSession: null, timeline: initialTimeline, timelineRows: initialTimelineRows, - timelineEpoch: options?.timelineEpoch ?? randomUUID(), timelineNextSeq: derivedNextSeq, + provisionalAssistantText: options?.provisionalAssistantText ?? null, persistence: attachPersistenceCwd(session.describePersistence(), config.cwd), historyPrimed: options?.historyPrimed ?? false, lastUserMessageAt: options?.lastUserMessageAt ?? null, @@ -1841,7 +1776,6 @@ export class AgentManager { private ensureTimelineState(agent: ManagedAgent): { rows: AgentTimelineRow[]; - epoch: string; nextSeq: number; minSeq: number; maxSeq: number; @@ -1853,7 +1787,6 @@ export class AgentManager { return { rows: agent.timelineRows, - epoch: agent.timelineEpoch, nextSeq: agent.timelineNextSeq, minSeq, maxSeq, @@ -1939,14 +1872,63 @@ export class AgentManager { return [[messageId, row.item.text]]; }), ); + const pendingTurnItems: AgentTimelineItem[] = []; + let bufferedAssistantText = ""; + const flushPendingTurn = () => { + for (const item of pendingTurnItems) { + this.recordTimeline(agent, item); + } + pendingTurnItems.length = 0; + if (bufferedAssistantText) { + this.recordTimeline(agent, { + type: "assistant_message", + text: bufferedAssistantText, + }); + bufferedAssistantText = ""; + } + }; try { for await (const event of agent.session.streamHistory()) { - this.handleStreamEvent(agent, event, { - fromHistory: true, - canonicalUserMessagesById: - canonicalUserMessagesById.size > 0 ? canonicalUserMessagesById : undefined, - }); + if (event.type !== "timeline") { + if ( + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled" + ) { + flushPendingTurn(); + } + continue; + } + + if (event.item.type === "user_message") { + flushPendingTurn(); + const eventMessageId = normalizeMessageId(event.item.messageId); + if (eventMessageId) { + const canonicalText = canonicalUserMessagesById.get(eventMessageId); + if (canonicalText === event.item.text) { + continue; + } + } + this.recordTimeline(agent, event.item); + continue; + } + + if (event.item.type === "assistant_message") { + bufferedAssistantText += event.item.text; + continue; + } + + if (event.item.type === "reasoning") { + continue; + } + + if (event.item.type === "tool_call" && event.item.status === "running") { + continue; + } + + pendingTurnItems.push(event.item); } + flushPendingTurn(); } catch { // ignore history failures } @@ -2019,6 +2001,16 @@ export class AgentManager { } } } + if (event.item.type === "assistant_message") { + agent.provisionalAssistantText = `${agent.provisionalAssistantText ?? ""}${event.item.text}`; + break; + } + if (event.item.type === "reasoning") { + break; + } + if (event.item.type === "tool_call" && event.item.status === "running") { + break; + } timelineRow = this.recordTimeline(agent, event.item); if (!options?.fromHistory && event.item.type === "user_message") { agent.lastUserMessageAt = new Date(); @@ -2035,6 +2027,27 @@ export class AgentManager { }, "handleStreamEvent: turn_completed", ); + if (agent.provisionalAssistantText) { + const item: AgentTimelineItem = { + type: "assistant_message", + text: agent.provisionalAssistantText, + }; + timelineRow = this.recordTimeline(agent, item); + if (!options?.fromHistory) { + this.dispatchStream( + agent.id, + { + type: "timeline", + item, + provider: event.provider, + }, + { + seq: timelineRow.seq, + }, + ); + } + agent.provisionalAssistantText = null; + } agent.lastUsage = event.usage; agent.lastError = undefined; // For autonomous turns (not foreground), transition to idle @@ -2058,6 +2071,7 @@ export class AgentManager { }, "handleStreamEvent: turn_failed", ); + agent.provisionalAssistantText = null; // For autonomous turns, set error state directly if (!isForegroundEvent) { agent.lifecycle = "error"; @@ -2094,6 +2108,7 @@ export class AgentManager { }, "handleStreamEvent: turn_canceled", ); + agent.provisionalAssistantText = null; // For autonomous turns, transition to idle // unless a replacement is pending (avoid idle flash during replace) if (!isForegroundEvent && !agent.pendingReplacement) { @@ -2125,6 +2140,7 @@ export class AgentManager { }, "handleStreamEvent: turn_started", ); + agent.provisionalAssistantText = null; // For autonomous turn_started (no foreground match), set running if (!isForegroundEvent) { (agent as ActiveManagedAgent).lifecycle = "running"; @@ -2161,7 +2177,6 @@ export class AgentManager { timelineRow ? { seq: timelineRow.seq, - epoch: this.ensureTimelineState(agent).epoch, } : undefined, ); @@ -2203,7 +2218,6 @@ export class AgentManager { }, { seq: row.seq, - epoch: this.ensureTimelineState(agent).epoch, }, ); } @@ -2234,14 +2248,6 @@ export class AgentManager { agent.timelineNextSeq = timelineState.nextSeq + 1; agent.timeline.push(item); timelineState.rows.push(row); - if ( - typeof this.maxTimelineItems === "number" && - agent.timeline.length > this.maxTimelineItems - ) { - const removeCount = agent.timeline.length - this.maxTimelineItems; - agent.timeline.splice(0, removeCount); - timelineState.rows.splice(0, removeCount); - } return row; } @@ -2337,7 +2343,7 @@ export class AgentManager { private dispatchStream( agentId: string, event: AgentStreamEvent, - metadata?: { seq?: number; epoch?: string }, + metadata?: { seq?: number }, ): void { this.dispatch({ type: "agent_stream", agentId, event, ...metadata }); } diff --git a/packages/server/src/server/daemon-client.e2e.test.ts b/packages/server/src/server/daemon-client.e2e.test.ts index a83c41cb3b..7ab5383717 100644 --- a/packages/server/src/server/daemon-client.e2e.test.ts +++ b/packages/server/src/server/daemon-client.e2e.test.ts @@ -529,7 +529,6 @@ describe("daemon client E2E", () => { const timelineResult = await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 1, - projection: "projected", }); expect(timelineResult.agentId).toBe(agent.id); @@ -756,7 +755,6 @@ describe("daemon client E2E", () => { const timeline = await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "projected", }); expect(timeline.entries.length).toBeGreaterThan(0); diff --git a/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts b/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts index 442d6ae03a..3d2f5f2b92 100644 --- a/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts @@ -71,7 +71,6 @@ describe("daemon E2E", () => { await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 200, - projection: "projected", }); const refreshedResult = await ctx.client.fetchAgent(agent.id); @@ -89,7 +88,6 @@ describe("daemon E2E", () => { await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 200, - projection: "projected", }); const clearResult = await ctx.client.fetchAgent(agent.id); diff --git a/packages/server/src/server/daemon-e2e/claude-autonomous-wake-simple.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/claude-autonomous-wake-simple.real.e2e.test.ts index 6dd2fe6222..6ab4007124 100644 --- a/packages/server/src/server/daemon-e2e/claude-autonomous-wake-simple.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/claude-autonomous-wake-simple.real.e2e.test.ts @@ -60,7 +60,6 @@ describe("daemon E2E (real claude) - autonomous wake simple", () => { const timelineAtIdle = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); const idleAssistantText = timelineAtIdle.entries .filter( @@ -91,7 +90,6 @@ describe("daemon E2E (real claude) - autonomous wake simple", () => { const finalTimeline = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); const finalAssistantText = finalTimeline.entries .filter( diff --git a/packages/server/src/server/daemon-e2e/claude-autonomous-wake.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/claude-autonomous-wake.real.e2e.test.ts index be0657fc52..c492190061 100644 --- a/packages/server/src/server/daemon-e2e/claude-autonomous-wake.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/claude-autonomous-wake.real.e2e.test.ts @@ -390,7 +390,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () = const timelineAtIdle = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); await client.waitForAgentUpsert( @@ -405,7 +404,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () = const timelineAfterWake = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); expect(timelineAfterWake.entries.length).toBeGreaterThanOrEqual( timelineAtIdle.entries.length, @@ -417,7 +415,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () = const nextTimeline = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); sawTimelineGrowth = nextTimeline.entries.length > timelineAtIdle.entries.length; } @@ -605,7 +602,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () = const timelineBeforeWake = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); const summarized = timelineBeforeWake.entries.map(summarizeTimelineEntry); // Required by reproduction request: log timeline at idle edge before autonomous wake. @@ -735,7 +731,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () = const timelineAtWake = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); // eslint-disable-next-line no-console @@ -751,7 +746,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () = const timelineAfterWake = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); expect(timelineAfterWake.entries.length).toBeGreaterThanOrEqual( timelineAtWake.entries.length, @@ -916,7 +910,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () = const timelineAtWake = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); // eslint-disable-next-line no-console @@ -932,7 +925,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () = const timelineAfterWake = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); expect(timelineAfterWake.entries.length).toBeGreaterThanOrEqual( timelineAtWake.entries.length, @@ -1066,7 +1058,6 @@ describe("daemon E2E (real claude) - autonomous wake from background task", () = const timeline = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); assistantTexts = timeline.entries .filter( diff --git a/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts b/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts index 40f36fcd70..980b90a8de 100644 --- a/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts @@ -114,11 +114,14 @@ describe("daemon E2E - persistence", () => { const timeline = await ctx.client.fetchAgentTimeline(agentId, { direction: "tail", limit: 0, - projection: "canonical", }); const timelineItems = timeline.entries.map((entry) => entry.item); expect(timelineItems.length).toBeGreaterThan(0); - expect(timelineItems.some((item) => item.type === "assistant_message")).toBe(true); + const assistantMessages = timelineItems.filter( + (item): item is Extract<(typeof timelineItems)[number], { type: "assistant_message" }> => + item.type === "assistant_message", + ); + expect(assistantMessages).toEqual([{ type: "assistant_message", text: "timeline test" }]); } finally { await ctx.cleanup(); cleaned = true; diff --git a/packages/server/src/server/daemon-e2e/rewind-user-message-dedupe-claude.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/rewind-user-message-dedupe-claude.real.e2e.test.ts index 7c5627db16..1aeed1d4ff 100644 --- a/packages/server/src/server/daemon-e2e/rewind-user-message-dedupe-claude.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/rewind-user-message-dedupe-claude.real.e2e.test.ts @@ -48,7 +48,6 @@ describe("daemon E2E (real claude) - rewind user message dedupe", () => { const timeline = await client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 0, - projection: "canonical", }); const rewindUserMessages = timeline.entries.filter( diff --git a/packages/server/src/server/daemon-e2e/timeline-reconnect-contract.e2e.test.ts b/packages/server/src/server/daemon-e2e/timeline-reconnect-contract.e2e.test.ts new file mode 100644 index 0000000000..26cbf55e2e --- /dev/null +++ b/packages/server/src/server/daemon-e2e/timeline-reconnect-contract.e2e.test.ts @@ -0,0 +1,206 @@ +import { afterEach, beforeEach, describe, expect, test } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; + +import { createDaemonTestContext, type DaemonTestContext, DaemonClient } from "../test-utils/index.js"; +import { createMessageCollector } from "../test-utils/message-collector.js"; +import type { SessionOutboundMessage } from "../messages.js"; + +function tmpCwd(): string { + return mkdtempSync(path.join(tmpdir(), "daemon-e2e-")); +} + +async function waitFor( + predicate: () => boolean, + timeoutMs = 5_000, + intervalMs = 10, +): Promise { + const startedAt = Date.now(); + while (!predicate()) { + if (Date.now() - startedAt > timeoutMs) { + throw new Error(`Timed out after ${timeoutMs}ms waiting for condition`); + } + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } +} + +function isSeqLessAssistantTimeline( + message: SessionOutboundMessage, + agentId: string, + text?: string, +): boolean { + return ( + message.type === "agent_stream" && + message.payload.agentId === agentId && + message.payload.event.type === "timeline" && + message.payload.event.item.type === "assistant_message" && + message.payload.seq === undefined && + (text === undefined || message.payload.event.item.text === text) + ); +} + +describe("daemon E2E - timeline reconnect contract", () => { + let ctx: DaemonTestContext; + + beforeEach(async () => { + ctx = await createDaemonTestContext(); + }); + + afterEach(async () => { + await ctx.cleanup(); + }, 60_000); + + test("reconnect catches up committed rows without replaying a provisional seed", async () => { + const cwd = tmpCwd(); + const primaryCollector = createMessageCollector(ctx.client); + + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "Reconnect Contract Test", + modeId: "full-access", + }); + + for (let seq = 1; seq <= 120; seq += 1) { + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "assistant_message", + text: `committed row ${seq}`, + }); + } + + primaryCollector.clear(); + await ctx.daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, { + type: "assistant_message", + text: "partial before disconnect", + }); + await waitFor(() => + primaryCollector.messages.some((message) => + isSeqLessAssistantTimeline(message, agent.id, "partial before disconnect"), + ), + ); + + await ctx.client.close(); + + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "assistant_message", + text: "finalized while disconnected", + }); + + const reconnectClient = new DaemonClient({ + url: `ws://127.0.0.1:${ctx.daemon.port}/ws`, + }); + await reconnectClient.connect(); + const reconnectCollector = createMessageCollector(reconnectClient); + + try { + await reconnectClient.fetchAgents({ + subscribe: { subscriptionId: "timeline-reconnect-a" }, + }); + + expect( + reconnectCollector.messages.some((message) => + isSeqLessAssistantTimeline(message, agent.id), + ), + ).toBe(false); + + const catchUp = await reconnectClient.fetchAgentTimeline(agent.id, { + direction: "after", + cursor: { seq: 120 }, + limit: 0, + }); + + expect(catchUp.entries).toHaveLength(1); + expect(catchUp.entries[0]?.seq).toBe(121); + expect(catchUp.entries[0]?.item).toEqual({ + type: "assistant_message", + text: "finalized while disconnected", + }); + } finally { + reconnectCollector.unsubscribe(); + await reconnectClient.close(); + } + } finally { + primaryCollector.unsubscribe(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 30_000); + + test("reconnect with no new committed rows resumes from future live provisional updates only", async () => { + const cwd = tmpCwd(); + const primaryCollector = createMessageCollector(ctx.client); + + try { + const agent = await ctx.client.createAgent({ + provider: "codex", + cwd, + title: "Reconnect No Seed Test", + modeId: "full-access", + }); + + for (let seq = 1; seq <= 120; seq += 1) { + await ctx.daemon.daemon.agentManager.appendTimelineItem(agent.id, { + type: "assistant_message", + text: `committed row ${seq}`, + }); + } + + primaryCollector.clear(); + await ctx.daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, { + type: "assistant_message", + text: "partial before disconnect", + }); + await waitFor(() => + primaryCollector.messages.some((message) => + isSeqLessAssistantTimeline(message, agent.id, "partial before disconnect"), + ), + ); + + await ctx.client.close(); + + const reconnectClient = new DaemonClient({ + url: `ws://127.0.0.1:${ctx.daemon.port}/ws`, + }); + await reconnectClient.connect(); + const reconnectCollector = createMessageCollector(reconnectClient); + + try { + await reconnectClient.fetchAgents({ + subscribe: { subscriptionId: "timeline-reconnect-b" }, + }); + + expect( + reconnectCollector.messages.some((message) => + isSeqLessAssistantTimeline(message, agent.id), + ), + ).toBe(false); + + const catchUp = await reconnectClient.fetchAgentTimeline(agent.id, { + direction: "after", + cursor: { seq: 120 }, + limit: 0, + }); + + expect(catchUp.entries).toHaveLength(0); + + reconnectCollector.clear(); + await ctx.daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, { + type: "assistant_message", + text: "fresh live after reconnect", + }); + await waitFor(() => + reconnectCollector.messages.some((message) => + isSeqLessAssistantTimeline(message, agent.id, "fresh live after reconnect"), + ), + ); + } finally { + reconnectCollector.unsubscribe(); + await reconnectClient.close(); + } + } finally { + primaryCollector.unsubscribe(); + rmSync(cwd, { recursive: true, force: true }); + } + }, 30_000); +}); diff --git a/packages/server/src/server/daemon-e2e/timeline-window.e2e.test.ts b/packages/server/src/server/daemon-e2e/timeline-window.e2e.test.ts index af2f557993..da0079d662 100644 --- a/packages/server/src/server/daemon-e2e/timeline-window.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/timeline-window.e2e.test.ts @@ -20,7 +20,7 @@ describe("daemon E2E - timeline window", () => { await ctx.cleanup(); }, 60_000); - test("canonical tail limit keeps assistant chunks intact at the window boundary", async () => { + test("canonical tail limit returns one finalized committed assistant row at the window boundary", async () => { const cwd = tmpCwd(); try { const agent = await ctx.client.createAgent({ @@ -38,16 +38,14 @@ describe("daemon E2E - timeline window", () => { const timeline = await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 1, - projection: "canonical", }); const assistantTexts = timeline.entries .filter((entry) => entry.item.type === "assistant_message") .map((entry) => entry.item.text); - expect(assistantTexts).toHaveLength(2); - expect(assistantTexts.join("")).toBe(expected); - expect(timeline.startCursor?.seq).toBeLessThan(timeline.endCursor?.seq ?? 0); + expect(assistantTexts).toEqual([expected]); + expect(timeline.startSeq).toBe(timeline.endSeq); } finally { rmSync(cwd, { recursive: true, force: true }); } @@ -73,7 +71,6 @@ describe("daemon E2E - timeline window", () => { const timeline = await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 1, - projection: "canonical", }); const assistantTexts = timeline.entries @@ -82,7 +79,7 @@ describe("daemon E2E - timeline window", () => { expect(assistantTexts.join("")).toBe(expected); expect(timeline.hasOlder).toBe(true); - expect(timeline.startCursor?.seq).toBeGreaterThan(1); + expect(timeline.startSeq).toBeGreaterThan(1); } finally { rmSync(cwd, { recursive: true, force: true }); } diff --git a/packages/server/src/server/daemon-e2e/ui-action-stress.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/ui-action-stress.real.e2e.test.ts index f07aa0900c..24aebe7657 100644 --- a/packages/server/src/server/daemon-e2e/ui-action-stress.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/ui-action-stress.real.e2e.test.ts @@ -289,7 +289,6 @@ async function resolveLatestAssistantMessage( const timeline = await client.fetchAgentTimeline(agentId, { direction: "tail", limit: 300, - projection: "canonical", }); for (let idx = timeline.entries.length - 1; idx >= 0; idx -= 1) { const entry = timeline.entries[idx]; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 99196d94cd..df7b9097e1 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -78,11 +78,6 @@ import { appendTimelineItemIfAgentKnown, emitLiveTimelineItemIfAgentKnown, } from "./agent/timeline-append.js"; -import { - projectTimelineRows, - selectTimelineWindowByProjectedLimit, - type TimelineProjectionMode, -} from "./agent/timeline-projection.js"; import { DEFAULT_STRUCTURED_GENERATION_PROVIDERS, StructuredAgentFallbackError, @@ -992,7 +987,6 @@ export class Session { event: serializedEvent, timestamp: new Date().toISOString(), ...(typeof event.seq === "number" ? { seq: event.seq } : {}), - ...(typeof event.epoch === "string" ? { epoch: event.epoch } : {}), } as const; this.emit({ @@ -6409,17 +6403,10 @@ export class Session { msg: Extract, ): Promise { const direction: AgentTimelineFetchDirection = msg.direction ?? (msg.cursor ? "after" : "tail"); - const projection: TimelineProjectionMode = msg.projection ?? "projected"; const requestedLimit = msg.limit; const limit = requestedLimit ?? (direction === "after" ? 0 : undefined); - const shouldLimitByProjectedWindow = - projection === "canonical" && - direction === "tail" && - typeof requestedLimit === "number" && - requestedLimit > 0; const cursor: AgentTimelineCursor | undefined = msg.cursor ? { - epoch: msg.cursor.epoch, seq: msg.cursor.seq, } : undefined; @@ -6427,87 +6414,19 @@ export class Session { try { const snapshot = await this.ensureAgentLoaded(msg.agentId); const agentPayload = await this.buildAgentPayload(snapshot); - - let timeline = this.agentManager.fetchTimeline(msg.agentId, { + const timeline = this.agentManager.fetchTimeline(msg.agentId, { direction, cursor, - limit: - shouldLimitByProjectedWindow && typeof requestedLimit === "number" - ? Math.max(1, Math.floor(requestedLimit)) - : limit, + limit, }); - - let hasOlder = timeline.hasOlder; - let hasNewer = timeline.hasNewer; - let startCursor: { epoch: string; seq: number } | null = null; - let endCursor: { epoch: string; seq: number } | null = null; - let entries: ReturnType; - - if (shouldLimitByProjectedWindow) { - const projectedLimit = Math.max(1, Math.floor(requestedLimit)); - let fetchLimit = projectedLimit; - let projectedWindow = selectTimelineWindowByProjectedLimit({ - rows: timeline.rows, - provider: snapshot.provider, - direction, - limit: projectedLimit, - collapseToolLifecycle: false, - }); - - while (timeline.hasOlder) { - const needsMoreProjectedEntries = - projectedWindow.projectedEntries.length < projectedLimit; - const firstLoadedRow = timeline.rows[0]; - const firstSelectedRow = projectedWindow.selectedRows[0]; - const startsAtLoadedBoundary = - firstLoadedRow != null && - firstSelectedRow != null && - firstSelectedRow.seq === firstLoadedRow.seq; - const boundaryIsAssistantChunk = - startsAtLoadedBoundary && firstLoadedRow.item.type === "assistant_message"; - - if (!needsMoreProjectedEntries && !boundaryIsAssistantChunk) { - break; - } - - const maxRows = Math.max(0, timeline.window.maxSeq - timeline.window.minSeq + 1); - const nextFetchLimit = Math.min(maxRows, fetchLimit * 2); - if (nextFetchLimit <= fetchLimit) { - break; - } - - fetchLimit = nextFetchLimit; - timeline = this.agentManager.fetchTimeline(msg.agentId, { - direction, - cursor, - limit: fetchLimit, - }); - projectedWindow = selectTimelineWindowByProjectedLimit({ - rows: timeline.rows, - provider: snapshot.provider, - direction, - limit: projectedLimit, - collapseToolLifecycle: false, - }); - } - - const selectedRows = projectedWindow.selectedRows; - - entries = projectTimelineRows(selectedRows, snapshot.provider, projection); - - if (projectedWindow.minSeq !== null && projectedWindow.maxSeq !== null) { - startCursor = { epoch: timeline.epoch, seq: projectedWindow.minSeq }; - endCursor = { epoch: timeline.epoch, seq: projectedWindow.maxSeq }; - hasOlder = projectedWindow.minSeq > timeline.window.minSeq; - hasNewer = false; - } - } else { - const firstRow = timeline.rows[0]; - const lastRow = timeline.rows[timeline.rows.length - 1]; - startCursor = firstRow ? { epoch: timeline.epoch, seq: firstRow.seq } : null; - endCursor = lastRow ? { epoch: timeline.epoch, seq: lastRow.seq } : null; - entries = projectTimelineRows(timeline.rows, snapshot.provider, projection); - } + const firstRow = timeline.rows[0]; + const lastRow = timeline.rows[timeline.rows.length - 1]; + const entries = timeline.rows.map((row) => ({ + provider: snapshot.provider, + item: row.item, + timestamp: row.timestamp, + seq: row.seq, + })); this.emit({ type: "fetch_agent_timeline_response", @@ -6516,16 +6435,10 @@ export class Session { agentId: msg.agentId, agent: agentPayload, direction, - projection, - epoch: timeline.epoch, - reset: timeline.reset, - staleCursor: timeline.staleCursor, - gap: timeline.gap, - window: timeline.window, - startCursor, - endCursor, - hasOlder, - hasNewer, + startSeq: firstRow?.seq ?? null, + endSeq: lastRow?.seq ?? null, + hasOlder: timeline.hasOlder, + hasNewer: timeline.hasNewer, entries, error: null, }, @@ -6542,14 +6455,8 @@ export class Session { agentId: msg.agentId, agent: null, direction, - projection, - epoch: "", - reset: false, - staleCursor: false, - gap: false, - window: { minSeq: 0, maxSeq: 0, nextSeq: 0 }, - startCursor: null, - endCursor: null, + startSeq: null, + endSeq: null, hasOlder: false, hasNewer: false, entries: [], diff --git a/packages/server/src/shared/messages.stream-parsing.test.ts b/packages/server/src/shared/messages.stream-parsing.test.ts index 01dc367992..01a55927c7 100644 --- a/packages/server/src/shared/messages.stream-parsing.test.ts +++ b/packages/server/src/shared/messages.stream-parsing.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { AgentStreamMessageSchema, + FetchAgentTimelineRequestMessageSchema, FetchAgentTimelineResponseMessageSchema, SessionInboundMessageSchema, SessionOutboundMessageSchema, @@ -17,14 +18,8 @@ describe("shared messages stream parsing", () => { agentId: "agent_live", agent: null, direction: "tail", - projection: "projected", - epoch: "epoch-1", - reset: false, - staleCursor: false, - gap: false, - window: { minSeq: 1, maxSeq: 2, nextSeq: 3 }, - startCursor: { epoch: "epoch-1", seq: 1 }, - endCursor: { epoch: "epoch-1", seq: 2 }, + startSeq: 1, + endSeq: 2, hasOlder: false, hasNewer: false, entries: [ @@ -32,10 +27,7 @@ describe("shared messages stream parsing", () => { provider: "codex", item: { type: "assistant_message", text: "hello" }, timestamp: "2026-02-08T20:10:00.000Z", - seqStart: 1, - seqEnd: 2, - sourceSeqRanges: [{ startSeq: 1, endSeq: 2 }], - collapsed: ["assistant_merge"], + seq: 2, }, ], error: null, @@ -46,6 +38,51 @@ describe("shared messages stream parsing", () => { expect(parsed.payload.entries[0]?.item.type).toBe("assistant_message"); }); + it("rejects removed fetch timeline request baggage at the parser boundary", () => { + const parsed = FetchAgentTimelineRequestMessageSchema.safeParse({ + type: "fetch_agent_timeline_request", + agentId: "agent_live", + requestId: "req-legacy", + direction: "after", + cursor: { + seq: 12, + epoch: "legacy-epoch", + }, + projection: "canonical", + }); + + expect(parsed.success).toBe(false); + }); + + it("rejects removed fetch timeline response baggage at the parser boundary", () => { + const parsed = FetchAgentTimelineResponseMessageSchema.safeParse({ + type: "fetch_agent_timeline_response", + payload: { + requestId: "req-1", + agentId: "agent_live", + agent: null, + direction: "tail", + startSeq: 1, + endSeq: 2, + hasOlder: false, + hasNewer: false, + reset: false, + startCursor: { seq: 1 }, + entries: [ + { + provider: "codex", + item: { type: "assistant_message", text: "hello" }, + timestamp: "2026-02-08T20:10:00.000Z", + seq: 2, + }, + ], + error: null, + }, + }); + + expect(parsed.success).toBe(false); + }); + it("parses explicit shutdown and restart lifecycle request payloads as distinct message types", () => { const shutdownParsed = SessionInboundMessageSchema.safeParse({ type: "shutdown_server_request", @@ -70,6 +107,7 @@ describe("shared messages stream parsing", () => { payload: { agentId: "agent_live", timestamp: "2026-02-08T20:10:00.000Z", + seq: 12, event: { type: "timeline", provider: "claude", @@ -97,6 +135,27 @@ describe("shared messages stream parsing", () => { } }); + it("rejects removed agent_stream baggage at the parser boundary", () => { + const parsed = AgentStreamMessageSchema.safeParse({ + type: "agent_stream", + payload: { + agentId: "agent_live", + timestamp: "2026-02-08T20:10:00.000Z", + epoch: "legacy-epoch", + event: { + type: "timeline", + provider: "claude", + item: { + type: "assistant_message", + text: "hello", + }, + }, + }, + }); + + expect(parsed.success).toBe(false); + }); + it("parses representative sub_agent tool_call event", () => { const parsed = AgentStreamMessageSchema.parse({ type: "agent_stream", diff --git a/packages/server/src/shared/messages.ts b/packages/server/src/shared/messages.ts index ee51e0b2a9..0e5d980b51 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -722,22 +722,23 @@ export const ShutdownServerRequestMessageSchema = z.object({ requestId: z.string(), }); -export const AgentTimelineCursorSchema = z.object({ - epoch: z.string(), - seq: z.number().int().nonnegative(), -}); +export const AgentTimelineCursorSchema = z + .object({ + seq: z.number().int().nonnegative(), + }) + .strict(); -export const FetchAgentTimelineRequestMessageSchema = z.object({ - type: z.literal("fetch_agent_timeline_request"), - agentId: z.string(), - requestId: z.string(), - direction: z.enum(["tail", "before", "after"]).optional(), - cursor: AgentTimelineCursorSchema.optional(), - // 0 means "all matching rows for this query window". - limit: z.number().int().nonnegative().optional(), - // Default should be projected for app timeline loading. - projection: z.enum(["projected", "canonical"]).optional(), -}); +export const FetchAgentTimelineRequestMessageSchema = z + .object({ + type: z.literal("fetch_agent_timeline_request"), + agentId: z.string(), + requestId: z.string(), + direction: z.enum(["tail", "before", "after"]).optional(), + cursor: AgentTimelineCursorSchema.optional(), + // 0 means "all matching rows for this query window". + limit: z.number().int().nonnegative().optional(), + }) + .strict(); export const SetAgentModeRequestMessageSchema = z.object({ type: z.literal("set_agent_mode_request"), @@ -1545,17 +1546,20 @@ export const AgentUpdateMessageSchema = z.object({ ]), }); -export const AgentStreamMessageSchema = z.object({ - type: z.literal("agent_stream"), - payload: z.object({ - agentId: z.string(), - event: AgentStreamEventPayloadSchema, - timestamp: z.string(), - // Present for timeline events. Maps 1:1 to canonical in-memory timeline rows. - seq: z.number().int().nonnegative().optional(), - epoch: z.string().optional(), - }), -}); +export const AgentStreamMessageSchema = z + .object({ + type: z.literal("agent_stream"), + payload: z + .object({ + agentId: z.string(), + event: AgentStreamEventPayloadSchema, + timestamp: z.string(), + // Present only for committed timeline events. + seq: z.number().int().nonnegative().optional(), + }) + .strict(), + }) + .strict(); export const AgentStatusMessageSchema = z.object({ type: z.literal("agent_status"), @@ -1649,46 +1653,34 @@ export const FetchAgentResponseMessageSchema = z.object({ }), }); -const AgentTimelineSeqRangeSchema = z.object({ - startSeq: z.number().int().nonnegative(), - endSeq: z.number().int().nonnegative(), -}); +export const AgentTimelineEntryPayloadSchema = z + .object({ + provider: AgentProviderSchema, + item: AgentTimelineItemPayloadSchema, + timestamp: z.string(), + seq: z.number().int().nonnegative(), + }) + .strict(); -export const AgentTimelineEntryPayloadSchema = z.object({ - provider: AgentProviderSchema, - item: AgentTimelineItemPayloadSchema, - timestamp: z.string(), - seqStart: z.number().int().nonnegative(), - seqEnd: z.number().int().nonnegative(), - sourceSeqRanges: z.array(AgentTimelineSeqRangeSchema), - collapsed: z.array(z.enum(["assistant_merge", "tool_lifecycle"])), -}); - -export const FetchAgentTimelineResponseMessageSchema = z.object({ - type: z.literal("fetch_agent_timeline_response"), - payload: z.object({ - requestId: z.string(), - agentId: z.string(), - agent: AgentSnapshotPayloadSchema.nullable(), - direction: z.enum(["tail", "before", "after"]), - projection: z.enum(["projected", "canonical"]), - epoch: z.string(), - reset: z.boolean(), - staleCursor: z.boolean(), - gap: z.boolean(), - window: z.object({ - minSeq: z.number().int().nonnegative(), - maxSeq: z.number().int().nonnegative(), - nextSeq: z.number().int().nonnegative(), - }), - startCursor: AgentTimelineCursorSchema.nullable(), - endCursor: AgentTimelineCursorSchema.nullable(), - hasOlder: z.boolean(), - hasNewer: z.boolean(), - entries: z.array(AgentTimelineEntryPayloadSchema), - error: z.string().nullable(), - }), -}); +export const FetchAgentTimelineResponseMessageSchema = z + .object({ + type: z.literal("fetch_agent_timeline_response"), + payload: z + .object({ + requestId: z.string(), + agentId: z.string(), + agent: AgentSnapshotPayloadSchema.nullable(), + direction: z.enum(["tail", "before", "after"]), + startSeq: z.number().int().nonnegative().nullable(), + endSeq: z.number().int().nonnegative().nullable(), + hasOlder: z.boolean(), + hasNewer: z.boolean(), + entries: z.array(AgentTimelineEntryPayloadSchema), + error: z.string().nullable(), + }) + .strict(), + }) + .strict(); export const SendAgentMessageResponseMessageSchema = z.object({ type: z.literal("send_agent_message_response"),