diff --git a/package-lock.json b/package-lock.json index cc64d1ba3..a8b72ec83 100644 --- a/package-lock.json +++ b/package-lock.json @@ -32,8 +32,8 @@ "get-port-cli": "^3.0.0", "knip": "^5.82.1", "patch-package": "^8.0.1", - "react": "19.1.4", - "react-dom": "19.1.4", + "react": "19.1.0", + "react-dom": "19.1.0", "typescript": "^5.9.3" } }, @@ -29122,9 +29122,9 @@ } }, "node_modules/react": { - "version": "19.1.4", - "resolved": "https://registry.npmjs.org/react/-/react-19.1.4.tgz", - "integrity": "sha512-DHINL3PAmPUiK1uszfbKiXqfE03eszdt5BpVSuEAHb5nfmNPwnsy7g39h2t8aXFc/Bv99GH81s+j8dobtD+jOw==", + "version": "19.1.0", + "resolved": "https://registry.npmjs.org/react/-/react-19.1.0.tgz", + "integrity": "sha512-FS+XFBNvn3GTAWq26joslQgWNoFu08F4kl0J4CgdNKADkdSGXQyTCnKteIAJy96Br6YbpEU1LSzV5dYtjMkMDg==", "license": "MIT", "engines": { "node": ">=0.10.0" @@ -29162,15 +29162,15 @@ } }, "node_modules/react-dom": { - "version": "19.1.4", - "resolved": "https://registry.npmjs.org/react-dom/-/react-dom-19.1.4.tgz", - "integrity": "sha512-s2868ab/xo2SI6H4106A7aFI8Mrqa4xC6HZT/pBzYyQ3cBLqa88hu47xYD8xf+uECleN698Awn7RCWlkTiKnqQ==", + "version": "19.1.0", + "resolved": "https://registry.npmjs.org/react-dom/-/react-dom-19.1.0.tgz", + "integrity": "sha512-Xs1hdnE+DyKgeHJeJznQmYMIBG3TKIHJJT95Q58nHLSrElKlGQqDTR2HQ9fx5CN/Gk6Vh/kupBTDLU11/nDk/g==", "license": "MIT", "dependencies": { "scheduler": "^0.26.0" }, "peerDependencies": { - "react": "^19.1.4" + "react": "^19.1.0" } }, "node_modules/react-error-boundary": { @@ -34911,8 +34911,8 @@ "expo-web-browser": "~15.0.8", "lucide-react-native": "^0.546.0", "mnemonic-id": "^3.2.7", - "react": "19.1.4", - "react-dom": "19.1.4", + "react": "19.1.0", + "react-dom": "19.1.0", "react-native": "^0.81.5", "react-native-css": "^3.0.1", "react-native-draggable-flatlist": "^4.0.3", diff --git a/package.json b/package.json index 9f5c23d11..c84601039 100644 --- a/package.json +++ b/package.json @@ -69,8 +69,8 @@ "get-port-cli": "^3.0.0", "knip": "^5.82.1", "patch-package": "^8.0.1", - "react": "19.1.4", - "react-dom": "19.1.4", + "react": "19.1.0", + "react-dom": "19.1.0", "typescript": "^5.9.3" }, "description": "Paseo: voice-controlled development environment with OpenAI Realtime API", @@ -94,8 +94,8 @@ "license": "AGPL-3.0-or-later", "overrides": { "lightningcss": "1.30.1", - "react": "19.1.4", - "react-dom": "19.1.4" + "react": "19.1.0", + "react-dom": "19.1.0" }, "dependencies": { "@anthropic-ai/claude-agent-sdk": "^0.2.11", diff --git a/packages/app/package.json b/packages/app/package.json index 49d6695d4..aeaae5b55 100644 --- a/packages/app/package.json +++ b/packages/app/package.json @@ -81,8 +81,8 @@ "expo-web-browser": "~15.0.8", "lucide-react-native": "^0.546.0", "mnemonic-id": "^3.2.7", - "react": "19.1.4", - "react-dom": "19.1.4", + "react": "19.1.0", + "react-dom": "19.1.0", "react-native": "^0.81.5", "react-native-css": "^3.0.1", "react-native-draggable-flatlist": "^4.0.3", diff --git a/packages/app/src/components/icons/aider-icon.tsx b/packages/app/src/components/icons/aider-icon.tsx deleted file mode 100644 index 6b2fc25e6..000000000 --- a/packages/app/src/components/icons/aider-icon.tsx +++ /dev/null @@ -1,14 +0,0 @@ -import Svg, { Path } from "react-native-svg"; - -interface AiderIconProps { - size?: number; - color?: string; -} - -export function AiderIcon({ size = 16, color = "currentColor" }: AiderIconProps) { - return ( - - - - ); -} diff --git a/packages/app/src/components/icons/amp-icon.tsx b/packages/app/src/components/icons/amp-icon.tsx deleted file mode 100644 index 5b855867d..000000000 --- a/packages/app/src/components/icons/amp-icon.tsx +++ /dev/null @@ -1,29 +0,0 @@ -import Svg, { Path } from "react-native-svg"; - -interface AmpIconProps { - size?: number; - color?: string; -} - -export function AmpIcon({ size = 16, color = "currentColor" }: AmpIconProps) { - return ( - - - - - - - ); -} diff --git a/packages/app/src/components/icons/gemini-icon.tsx b/packages/app/src/components/icons/gemini-icon.tsx deleted file mode 100644 index 05d3822ce..000000000 --- a/packages/app/src/components/icons/gemini-icon.tsx +++ /dev/null @@ -1,17 +0,0 @@ -import Svg, { Path } from "react-native-svg"; - -interface GeminiIconProps { - size?: number; - color?: string; -} - -export function GeminiIcon({ size = 16, color = "currentColor" }: GeminiIconProps) { - return ( - - - - ); -} diff --git a/packages/app/src/components/provider-icons.ts b/packages/app/src/components/provider-icons.ts index efa9c8ffd..6c8a689c5 100644 --- a/packages/app/src/components/provider-icons.ts +++ b/packages/app/src/components/provider-icons.ts @@ -1,9 +1,6 @@ import { Bot } from "lucide-react-native"; -import { AiderIcon } from "@/components/icons/aider-icon"; -import { AmpIcon } from "@/components/icons/amp-icon"; import { ClaudeIcon } from "@/components/icons/claude-icon"; import { CodexIcon } from "@/components/icons/codex-icon"; -import { GeminiIcon } from "@/components/icons/gemini-icon"; import { CopilotIcon } from "@/components/icons/copilot-icon"; import { OpenCodeIcon } from "@/components/icons/opencode-icon"; import { PiIcon } from "@/components/icons/pi-icon"; @@ -11,9 +8,6 @@ import { PiIcon } from "@/components/icons/pi-icon"; const PROVIDER_ICONS: Record = { claude: ClaudeIcon as unknown as typeof Bot, codex: CodexIcon as unknown as typeof Bot, - gemini: GeminiIcon as unknown as typeof Bot, - amp: AmpIcon as unknown as typeof Bot, - aider: AiderIcon as unknown as typeof Bot, copilot: CopilotIcon as unknown as typeof Bot, opencode: OpenCodeIcon as unknown as typeof Bot, pi: PiIcon as unknown as typeof Bot, diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index dccf604b6..241493813 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/packages/app/src/contexts/session-context.tsx @@ -591,8 +591,9 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider void client .fetchAgentTimeline(agentId, { direction: "after", - cursor: { seq: cursor.endSeq }, + cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, limit: 0, + projection: "canonical", }) .catch((error) => { console.warn("[Session] failed to fetch catch-up timeline on resume", agentId, error); @@ -853,12 +854,13 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider ); const requestCanonicalCatchUp = useCallback( - (agentId: string, cursor: { endSeq: number }) => { + (agentId: string, cursor: { epoch: string; endSeq: number }) => { void client .fetchAgentTimeline(agentId, { direction: "after", - cursor: { seq: cursor.endSeq }, + cursor: { epoch: cursor.epoch, seq: cursor.endSeq }, limit: 0, + projection: "canonical", }) .catch((error) => { console.warn("[Session] failed to fetch canonical catch-up timeline", agentId, error); @@ -961,6 +963,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider } if ( current && + current.epoch === result.cursor.epoch && current.startSeq === result.cursor.startSeq && current.endSeq === result.cursor.endSeq ) { @@ -1065,7 +1068,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 } = message.payload; + const { agentId, event, timestamp, seq, epoch } = message.payload; const parsedTimestamp = new Date(timestamp); const streamEvent = event as AgentStreamEventPayload; if ( @@ -1107,6 +1110,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider const result = processAgentStreamEvent({ event: streamEvent, seq, + epoch, currentTail, currentHead, currentCursor, @@ -1130,6 +1134,8 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider if ( current && typeof seq === "number" && + typeof epoch === "string" && + current.epoch === epoch && seq >= current.startSeq && seq <= current.endSeq ) { @@ -1138,6 +1144,7 @@ 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 cac018e35..580189da3 100644 --- a/packages/app/src/contexts/session-stream-reducers.test.ts +++ b/packages/app/src/contexts/session-stream-reducers.test.ts @@ -9,9 +9,13 @@ import { type TimelineCursor, } from "./session-stream-reducers"; +// --------------------------------------------------------------------------- +// Test helpers +// --------------------------------------------------------------------------- + function makeTimelineEntry(seq: number, text: string, type: string = "assistant_message") { return { - seq, + seqStart: seq, provider: "claude", item: { type, text }, timestamp: new Date(1000 + seq).toISOString(), @@ -29,30 +33,22 @@ function makeTimelineEvent( } as AgentStreamEventPayload; } -function makeToolCallEvent(status: "running" | "completed"): AgentStreamEventPayload { +function makeUserTimelineEvent(text: string): AgentStreamEventPayload { return { type: "timeline", provider: "claude", - item: { - type: "tool_call", - callId: "call-1", - name: "shell", - status, - detail: { - type: "shell", - command: "pwd", - }, - error: null, - }, - }; + item: { type: "user_message", text }, + } as AgentStreamEventPayload; } const baseTimelineInput: ProcessTimelineResponseInput = { payload: { agentId: "agent-1", direction: "after", - startSeq: null, - endSeq: null, + reset: false, + epoch: "epoch-1", + startCursor: null, + endCursor: null, entries: [], error: null, }, @@ -67,6 +63,7 @@ const baseTimelineInput: ProcessTimelineResponseInput = { const baseStreamInput: ProcessAgentStreamEventInput = { event: makeTimelineEvent("hello"), seq: undefined, + epoch: undefined, currentTail: [], currentHead: [], currentCursor: undefined, @@ -74,6 +71,10 @@ const baseStreamInput: ProcessAgentStreamEventInput = { timestamp: new Date(2000), }; +// --------------------------------------------------------------------------- +// processTimelineResponse +// --------------------------------------------------------------------------- + describe("processTimelineResponse", () => { it("returns error path when payload.error is set", () => { const result = processTimelineResponse({ @@ -92,10 +93,35 @@ describe("processTimelineResponse", () => { expect(result.tail).toBe(baseTimelineInput.currentTail); expect(result.head).toBe(baseTimelineInput.currentHead); expect(result.cursorChanged).toBe(false); + expect(result.sideEffects).toEqual([]); }); - it("replaces tail during bootstrap tail init and schedules committed catch-up", () => { - const provisionalHead: StreamItem[] = [ + 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[] = [ { kind: "assistant_message", id: "head-1", @@ -106,325 +132,518 @@ describe("processTimelineResponse", () => { const result = processTimelineResponse({ ...baseTimelineInput, - currentHead: provisionalHead, + 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, isInitializing: true, hasActiveInitDeferred: true, initRequestDirection: "tail", payload: { ...baseTimelineInput.payload, direction: "tail", - startSeq: 1, - endSeq: 5, + epoch: "epoch-1", + startCursor: { seq: 1 }, + endCursor: { seq: 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, }); - const catchUp = result.sideEffects.find((effect) => effect.type === "catch_up"); - expect(catchUp).toEqual({ - type: "catch_up", - cursor: { 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, }); }); - it("prepends older committed history for before pagination", () => { - const currentTail: StreamItem[] = [ - { - kind: "assistant_message", - id: "tail-3", - text: "newer", - timestamp: new Date(3000), - }, - ]; - const currentCursor: TimelineCursor = { startSeq: 3, endSeq: 4 }; + it("appends incrementally for contiguous seqs", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 3, + }; const result = processTimelineResponse({ ...baseTimelineInput, - currentTail, - currentCursor, + currentCursor: existingCursor, payload: { ...baseTimelineInput.payload, - direction: "before", - startSeq: 1, - endSeq: 2, - entries: [makeTimelineEntry(1, "hello", "user_message"), makeTimelineEntry(2, "older")], + epoch: "epoch-1", + entries: [makeTimelineEntry(4, "next-1"), makeTimelineEntry(5, "next-2")], }, }); + expect(result.tail.length).toBeGreaterThan(0); expect(result.cursorChanged).toBe(true); expect(result.cursor).toEqual({ + epoch: "epoch-1", startSeq: 1, - endSeq: 4, + endSeq: 5, }); - 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]); + expect(result.error).toBe(null); }); - 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 }; + it("detects gap and emits catch-up side effect", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 3, + }; const result = processTimelineResponse({ ...baseTimelineInput, - currentHead, - currentCursor, + 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", - startSeq: 121, - endSeq: 121, - entries: [makeTimelineEntry(121, "finalized reply")], - }, - }); - - expect(result.head).toEqual([]); - expect(result.cursorChanged).toBe(true); - expect(result.cursor).toEqual({ - startSeq: 1, - 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); + expect(result.initResolution).toBe("resolve"); + expect(result.clearInitializing).toBe(true); }); - it("requests catch-up when committed rows arrive with a forward gap", () => { - const currentCursor: TimelineCursor = { startSeq: 1, endSeq: 120 }; - + it("does not resolve init when directions differ (before vs after)", () => { const result = processTimelineResponse({ ...baseTimelineInput, - currentCursor, + isInitializing: true, + hasActiveInitDeferred: true, + initRequestDirection: "after", + payload: { + ...baseTimelineInput.payload, + direction: "before", + entries: [], + }, + }); + + // "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); + }); + + it("clears initializing even without deferred", () => { + const result = processTimelineResponse({ + ...baseTimelineInput, + isInitializing: true, + hasActiveInitDeferred: false, payload: { ...baseTimelineInput.payload, direction: "after", - startSeq: 125, - endSeq: 125, - entries: [makeTimelineEntry(125, "far ahead")], + entries: [], }, }); - expect(result.cursorChanged).toBe(false); - expect(result.tail).toBe(baseTimelineInput.currentTail); - expect(result.sideEffects).toContainEqual({ - type: "catch_up", - cursor: { endSeq: 120 }, + 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")], + }, + }); + + expect(result.cursorChanged).toBe(true); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 2, }); }); }); +// --------------------------------------------------------------------------- +// processAgentStreamEvent +// --------------------------------------------------------------------------- + describe("processAgentStreamEvent", () => { - it("treats seq-less timeline events as provisional head updates", () => { + it("passes through non-timeline events without cursor changes", () => { + const turnEvent: AgentStreamEventPayload = { + type: "turn_completed", + provider: "claude", + }; + const result = processAgentStreamEvent({ ...baseStreamInput, - event: makeTimelineEvent("partial"), + event: turnEvent, 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("keeps tool call rows anchored in tail across live and committed updates", () => { - const running = processAgentStreamEvent({ + it("accepts timeline event with cursor advance", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 4, + }; + + const result = processAgentStreamEvent({ ...baseStreamInput, - event: makeToolCallEvent("running"), - seq: undefined, - }); - expect(running.head).toEqual([]); - expect(running.tail).toHaveLength(1); - expect(running.tail[0]).toMatchObject({ - kind: "tool_call", - payload: { - source: "agent", - data: { - callId: "call-1", - status: "running", - }, - }, + event: makeTimelineEvent("new chunk"), + seq: 5, + epoch: "epoch-1", + currentCursor: existingCursor, }); - const completedLive = processAgentStreamEvent({ - ...baseStreamInput, - event: makeToolCallEvent("completed"), - seq: undefined, - currentHead: running.head, - currentTail: running.tail, - currentCursor: { startSeq: 1, endSeq: 7 }, - }); - expect(completedLive.head).toEqual([]); - expect(completedLive.tail).toHaveLength(1); - expect(completedLive.tail[0]).toMatchObject({ - kind: "tool_call", - payload: { - source: "agent", - data: { - callId: "call-1", - status: "completed", - }, - }, - }); - - const committed = processAgentStreamEvent({ - ...baseStreamInput, - event: makeToolCallEvent("completed"), - seq: 8, - currentHead: completedLive.head, - currentTail: completedLive.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", - }, - }, + expect(result.cursorChanged).toBe(true); + expect(result.cursor).toEqual({ + epoch: "epoch-1", + startSeq: 1, + endSeq: 5, }); + expect(result.sideEffects).toEqual([]); }); - it("preserves assistant/tool interleaving while a turn is streaming", () => { - const assistantBeforeTool = processAgentStreamEvent({ - ...baseStreamInput, - event: makeTimelineEvent("before"), - seq: undefined, - }); + it("detects gap and emits catch-up side effect", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 4, + }; - const runningTool = processAgentStreamEvent({ - ...baseStreamInput, - event: makeToolCallEvent("running"), - seq: undefined, - currentHead: assistantBeforeTool.head, - currentTail: assistantBeforeTool.tail, - }); - - const assistantAfterTool = processAgentStreamEvent({ - ...baseStreamInput, - event: makeTimelineEvent("after"), - seq: undefined, - currentHead: runningTool.head, - currentTail: runningTool.tail, - }); - - const completedTool = processAgentStreamEvent({ - ...baseStreamInput, - event: makeToolCallEvent("completed"), - seq: undefined, - currentHead: assistantAfterTool.head, - currentTail: assistantAfterTool.tail, - }); - - expect(completedTool.head).toEqual([]); - expect(completedTool.tail.map((item) => item.kind)).toEqual([ - "assistant_message", - "tool_call", - "assistant_message", - ]); - expect( - completedTool.tail[0]?.kind === "assistant_message" ? completedTool.tail[0].text : null, - ).toBe("before"); - expect( - completedTool.tail[2]?.kind === "assistant_message" ? completedTool.tail[2].text : null, - ).toBe("after"); - }); - - it("requests catch-up when a committed live row skips ahead", () => { const result = processAgentStreamEvent({ ...baseStreamInput, event: makeTimelineEvent("far ahead"), - seq: 125, - currentCursor: { startSeq: 1, endSeq: 120 }, + seq: 10, + epoch: "epoch-1", + currentCursor: existingCursor, }); + expect(result.cursorChanged).toBe(false); expect(result.changedTail).toBe(false); expect(result.changedHead).toBe(false); - expect(result.cursorChanged).toBe(false); - expect(result.sideEffects).toContainEqual({ - type: "catch_up", - cursor: { endSeq: 120 }, + + const catchUp = result.sideEffects.find((e) => e.type === "catch_up"); + expect(catchUp).toBeDefined(); + expect(catchUp!.cursor).toEqual({ + epoch: "epoch-1", + endSeq: 4, }); }); - it("flushes provisional head into tail on terminal turn events", () => { - const withHead = processAgentStreamEvent({ - ...baseStreamInput, - event: makeTimelineEvent("streaming"), - seq: undefined, - }); + it("drops stale timeline event", () => { + const existingCursor: TimelineCursor = { + epoch: "epoch-1", + startSeq: 1, + endSeq: 8, + }; const result = processAgentStreamEvent({ ...baseStreamInput, - event: { - type: "turn_completed", - provider: "claude", - }, - currentHead: withHead.head, - currentTail: withHead.tail, + event: makeTimelineEvent("old"), + seq: 5, + epoch: "epoch-1", + currentCursor: existingCursor, }); - expect(result.changedHead).toBe(true); - expect(result.changedTail).toBe(true); - expect(result.head).toEqual([]); - expect(result.tail).toHaveLength(1); - expect(result.tail[0]).toMatchObject({ - kind: "assistant_message", - text: "streaming", + expect(result.cursorChanged).toBe(false); + 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); + }); }); diff --git a/packages/app/src/contexts/session-stream-reducers.ts b/packages/app/src/contexts/session-stream-reducers.ts index 1c2ac4a4d..c7fb61e51 100644 --- a/packages/app/src/contexts/session-stream-reducers.ts +++ b/packages/app/src/contexts/session-stream-reducers.ts @@ -1,7 +1,12 @@ 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 { + applyStreamEvent, + flushHeadToTail, + hydrateStreamState, + reduceStreamUpdate, +} from "@/types/stream"; import { classifySessionTimelineSeq, type SessionTimelineSeqDecision, @@ -12,25 +17,38 @@ 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: { endSeq: number } } + | { type: "catch_up"; cursor: { epoch: string; endSeq: number } } | { type: "flush_pending_updates" }; export type AgentStreamReducerSideEffect = { type: "catch_up"; - cursor: { endSeq: number }; + cursor: { epoch: string; endSeq: number }; }; +// --------------------------------------------------------------------------- +// processTimelineResponse +// --------------------------------------------------------------------------- + type TimelineDirection = "tail" | "before" | "after"; type InitRequestDirection = "tail" | "after"; type TimelineResponseEntry = { - seq: number; + seqStart: number; provider: string; item: Record; timestamp: string; @@ -40,8 +58,10 @@ export interface ProcessTimelineResponseInput { payload: { agentId: string; direction: TimelineDirection; - startSeq: number | null; - endSeq: number | null; + reset: boolean; + epoch: string; + startCursor: { seq: number } | null; + endCursor: { seq: number } | null; entries: TimelineResponseEntry[]; error: string | null; }; @@ -64,9 +84,217 @@ 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) { + // Flush head to tail before appending canonical entries so that + // chronological ordering is preserved. This matters on mobile where + // the server drops live events for backgrounded/unfocused agents: + // when the client catches up, the head may still hold stale live + // items from before the gap that must be committed ahead of the + // canonical gap-fill entries. + const baseTail = + currentHead.length > 0 ? flushHeadToTail(currentTail, currentHead) : currentTail; + if (currentHead.length > 0) { + nextHead = []; + } + + nextTail = acceptedUnits.reduce( + (state, { event, timestamp }) => + reduceStreamUpdate(state, event, timestamp, { + source: "canonical", + }), + baseTail, + ); + } + + 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; @@ -96,229 +324,57 @@ 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, currentTail, currentHead, currentCursor, currentAgent, timestamp } = input; + const { event, seq, epoch, currentTail, currentHead, currentCursor, currentAgent, timestamp } = + input; let shouldApplyStreamEvent = true; let nextTimelineCursor: TimelineCursor | null = null; let cursorChanged = false; const sideEffects: AgentStreamReducerSideEffect[] = []; - if (event.type === "timeline" && typeof seq === "number") { + // ------------------------------------------------------------------ + // Timeline sequencing gate + // ------------------------------------------------------------------ + if (event.type === "timeline" && typeof seq === "number" && typeof epoch === "string") { const decision = classifySessionTimelineSeq({ - cursor: currentCursor ? { endSeq: currentCursor.endSeq } : null, + cursor: currentCursor ? { epoch: currentCursor.epoch, endSeq: currentCursor.endSeq } : null, + epoch, seq, }); - if (decision === "gap") { + 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 (currentCursor) { sideEffects.push({ type: "catch_up", - cursor: { endSeq: currentCursor.endSeq }, + cursor: { + epoch: currentCursor.epoch, + endSeq: currentCursor.endSeq, + }, }); } - } else if (decision === "drop_stale") { - shouldApplyStreamEvent = false; } else { - nextTimelineCursor = - decision === "init" - ? { startSeq: seq, endSeq: seq } - : { ...(currentCursor ?? { startSeq: seq, endSeq: seq }), endSeq: seq }; - cursorChanged = !cursorsEqual(currentCursor, nextTimelineCursor); + // drop_stale or drop_epoch + shouldApplyStreamEvent = false; } } + // ------------------------------------------------------------------ + // Apply stream event to tail/head + // ------------------------------------------------------------------ const { tail, head, changedTail, changedHead } = shouldApplyStreamEvent ? applyStreamEvent({ tail: currentTail, @@ -334,6 +390,9 @@ export function processAgentStreamEvent( changedHead: false, }; + // ------------------------------------------------------------------ + // Optimistic lifecycle status + // ------------------------------------------------------------------ let agentPatch: AgentPatch | null = null; let agentChanged = false; 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 cc8b98ce0..e233b8742 100644 --- a/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts +++ b/packages/app/src/contexts/session-timeline-bootstrap-policy.test.ts @@ -2,42 +2,26 @@ import { describe, expect, it } from "vitest"; import { classifySessionTimelineSeq } from "./session-timeline-seq-gate"; import { deriveBootstrapTailTimelinePolicy, - deriveInitialTimelineRequest, shouldResolveTimelineInit, } from "./session-timeline-bootstrap-policy"; -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, - }); - }); - - 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("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, + }); + + expect(policy.replace).toBe(true); + expect(policy.catchUpCursor).toBeNull(); + }); + it("forces baseline replace and canonical catch-up for init tail race", () => { - const advancedCursor = { endSeq: 205 }; + const advancedCursor = { epoch: "epoch-1", endSeq: 205 }; const tailSeqStart = 101; const tailSeqEnd = 200; @@ -45,6 +29,7 @@ describe("deriveBootstrapTailTimelinePolicy", () => { for (let seq = tailSeqStart; seq <= tailSeqEnd; seq += 1) { const decision = classifySessionTimelineSeq({ cursor: advancedCursor, + epoch: "epoch-1", seq, }); if (decision === "accept" || decision === "init") { @@ -55,13 +40,16 @@ describe("deriveBootstrapTailTimelinePolicy", () => { const policy = deriveBootstrapTailTimelinePolicy({ direction: "tail", - endSeq: 200, + reset: false, + epoch: "epoch-1", + endCursor: { seq: 200 }, isInitializing: true, hasActiveInitDeferred: true, }); expect(policy.replace).toBe(true); expect(policy.catchUpCursor).toEqual({ + epoch: "epoch-1", endSeq: 200, }); }); @@ -69,7 +57,9 @@ describe("deriveBootstrapTailTimelinePolicy", () => { it("does not replace non-bootstrap, non-reset responses", () => { const policy = deriveBootstrapTailTimelinePolicy({ direction: "tail", - endSeq: 200, + reset: false, + epoch: "epoch-1", + endCursor: { seq: 200 }, isInitializing: false, hasActiveInitDeferred: false, }); @@ -87,6 +77,7 @@ describe("shouldResolveTimelineInit", () => { isInitializing: true, initRequestDirection: "tail", responseDirection: "tail", + reset: false, }), ).toBe(true); }); @@ -98,6 +89,7 @@ describe("shouldResolveTimelineInit", () => { isInitializing: true, initRequestDirection: "tail", responseDirection: "after", + reset: false, }), ).toBe(false); }); @@ -109,6 +101,7 @@ 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 c7ab9ded8..706ab5eee 100644 --- a/packages/app/src/contexts/session-timeline-bootstrap-policy.ts +++ b/packages/app/src/contexts/session-timeline-bootstrap-policy.ts @@ -6,6 +6,7 @@ type BootstrapTailCursor = { } | null; type InitialTimelineCursor = { + epoch: string; seq: number; } | null; @@ -19,37 +20,48 @@ export function deriveInitialTimelineRequest({ initialTimelineLimit: number; }): { direction: "tail" | "after"; - cursor?: { seq: number }; + cursor?: { epoch: string; seq: number }; limit: number; + projection: "canonical"; } { if (!hasAuthoritativeHistory || !cursor) { return { direction: "tail", limit: initialTimelineLimit, + projection: "canonical", }; } return { direction: "after", - cursor: { seq: cursor.seq }, + cursor: { epoch: cursor.epoch, seq: cursor.seq }, limit: 0, + projection: "canonical", }; } export function deriveBootstrapTailTimelinePolicy({ direction, - endSeq, + reset, + epoch, + endCursor, isInitializing, hasActiveInitDeferred, }: { direction: TimelineDirection; - endSeq: number | null; + reset: boolean; + epoch: string; + endCursor: BootstrapTailCursor; isInitializing: boolean; hasActiveInitDeferred: boolean; }): { replace: boolean; - catchUpCursor: { endSeq: number } | null; + catchUpCursor: { epoch: string; endSeq: number } | null; } { + if (reset) { + return { replace: true, catchUpCursor: null }; + } + const isBootstrapTailInit = direction === "tail" && isInitializing && hasActiveInitDeferred; if (!isBootstrapTailInit) { return { replace: false, catchUpCursor: null }; @@ -57,7 +69,7 @@ export function deriveBootstrapTailTimelinePolicy({ return { replace: true, - catchUpCursor: typeof endSeq === "number" ? { endSeq } : null, + catchUpCursor: endCursor ? { epoch, endSeq: endCursor.seq } : null, }; } @@ -66,14 +78,19 @@ 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 3f827dce8..d18af910a 100644 --- a/packages/app/src/contexts/session-timeline-seq-gate.test.ts +++ b/packages/app/src/contexts/session-timeline-seq-gate.test.ts @@ -5,7 +5,8 @@ describe("classifySessionTimelineSeq", () => { it("accepts contiguous forward seq", () => { expect( classifySessionTimelineSeq({ - cursor: { endSeq: 4 }, + cursor: { epoch: "epoch-1", endSeq: 4 }, + epoch: "epoch-1", seq: 5, }), ).toBe("accept"); @@ -14,7 +15,8 @@ describe("classifySessionTimelineSeq", () => { it("drops stale seq older than the current end", () => { expect( classifySessionTimelineSeq({ - cursor: { endSeq: 8 }, + cursor: { epoch: "epoch-1", endSeq: 8 }, + epoch: "epoch-1", seq: 7, }), ).toBe("drop_stale"); @@ -23,16 +25,28 @@ describe("classifySessionTimelineSeq", () => { it("drops duplicate replay seq equal to the current end", () => { expect( classifySessionTimelineSeq({ - cursor: { endSeq: 8 }, + cursor: { epoch: "epoch-1", endSeq: 8 }, + epoch: "epoch-1", 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"); @@ -41,7 +55,8 @@ describe("classifySessionTimelineSeq", () => { it("classifies forward gaps", () => { expect( classifySessionTimelineSeq({ - cursor: { endSeq: 4 }, + cursor: { epoch: "epoch-1", endSeq: 4 }, + epoch: "epoch-1", 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 1fe15597e..e7e35b567 100644 --- a/packages/app/src/contexts/session-timeline-seq-gate.ts +++ b/packages/app/src/contexts/session-timeline-seq-gate.ts @@ -1,22 +1,28 @@ export type SessionTimelineSeqCursor = | { + epoch: string; endSeq: number; } | null | undefined; -export type SessionTimelineSeqDecision = "accept" | "drop_stale" | "gap" | "init"; +export type SessionTimelineSeqDecision = "accept" | "drop_stale" | "drop_epoch" | "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 43da108e3..dd32a15d6 100644 --- a/packages/app/src/hooks/use-agent-initialization.test.ts +++ b/packages/app/src/hooks/use-agent-initialization.test.ts @@ -2,10 +2,11 @@ import { describe, expect, it } from "vitest"; import { __private__ } from "./use-agent-initialization"; describe("useAgentInitialization timeline request policy", () => { - it("uses committed tail bootstrap when history has not synced yet", () => { + it("uses canonical tail bootstrap when history has not synced yet", () => { expect( __private__.deriveInitialTimelineRequest({ cursor: { + epoch: "epoch-1", seq: 42, }, hasAuthoritativeHistory: false, @@ -14,10 +15,11 @@ describe("useAgentInitialization timeline request policy", () => { ).toEqual({ direction: "tail", limit: 200, + projection: "canonical", }); }); - it("uses committed tail bootstrap when cursor is missing", () => { + it("uses canonical tail bootstrap when cursor is missing", () => { expect( __private__.deriveInitialTimelineRequest({ cursor: null, @@ -27,13 +29,15 @@ describe("useAgentInitialization timeline request policy", () => { ).toEqual({ direction: "tail", limit: 200, + projection: "canonical", }); }); - it("uses committed catch-up after the current cursor once history is synced", () => { + it("uses canonical catch-up after the current cursor once history is synced", () => { expect( __private__.deriveInitialTimelineRequest({ cursor: { + epoch: "epoch-1", seq: 42, }, hasAuthoritativeHistory: true, @@ -41,8 +45,9 @@ describe("useAgentInitialization timeline request policy", () => { }), ).toEqual({ direction: "after", - cursor: { seq: 42 }, + cursor: { epoch: "epoch-1", seq: 42 }, limit: 0, + projection: "canonical", }); }); @@ -56,6 +61,7 @@ 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 310231a04..dd1f9aafc 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 ? { seq: cursor.endSeq } : null, + cursor: cursor ? { epoch: cursor.epoch, seq: cursor.endSeq } : null, hasAuthoritativeHistory, initialTimelineLimit, }); @@ -107,6 +107,7 @@ 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/provider-recency-store.test.ts b/packages/app/src/stores/provider-recency-store.test.ts deleted file mode 100644 index 3bb8d0b97..000000000 --- a/packages/app/src/stores/provider-recency-store.test.ts +++ /dev/null @@ -1,61 +0,0 @@ -import { beforeEach, describe, expect, it, vi } from "vitest"; - -vi.mock("@react-native-async-storage/async-storage", () => { - const storage = new Map(); - return { - default: { - getItem: vi.fn(async (key: string) => storage.get(key) ?? null), - setItem: vi.fn(async (key: string, value: string) => { - storage.set(key, value); - }), - removeItem: vi.fn(async (key: string) => { - storage.delete(key); - }), - }, - }; -}); - -import { AGENT_PROVIDER_DEFINITIONS } from "@server/server/agent/provider-manifest"; -import { - __providerRecencyStoreTestUtils, - sortProvidersByRecency, - useProviderRecencyStore, -} from "./provider-recency-store"; - -describe("provider-recency-store", () => { - beforeEach(() => { - useProviderRecencyStore.setState({ - recentProviderIds: [], - recordUsage: useProviderRecencyStore.getState().recordUsage, - }); - }); - - it("sorts used providers first and keeps unused providers in default order", () => { - const sorted = sortProvidersByRecency(AGENT_PROVIDER_DEFINITIONS, ["codex"]); - - expect(sorted.map((provider) => provider.id)).toEqual([ - "codex", - ...AGENT_PROVIDER_DEFINITIONS.filter((provider) => provider.id !== "codex").map( - (provider) => provider.id, - ), - ]); - }); - - it("moves the latest provider to the front without duplicating prior entries", () => { - useProviderRecencyStore.getState().recordUsage("codex"); - useProviderRecencyStore.getState().recordUsage("opencode"); - useProviderRecencyStore.getState().recordUsage("codex"); - - expect(useProviderRecencyStore.getState().recentProviderIds).toEqual(["codex", "opencode"]); - }); - - it("filters invalid and duplicate providers during migration", () => { - expect( - __providerRecencyStoreTestUtils.migratePersistedState({ - recentProviderIds: ["codex", "invalid", "codex", "claude"], - }), - ).toEqual({ - recentProviderIds: ["codex", "claude"], - }); - }); -}); diff --git a/packages/app/src/stores/provider-recency-store.ts b/packages/app/src/stores/provider-recency-store.ts deleted file mode 100644 index 5dc48ac58..000000000 --- a/packages/app/src/stores/provider-recency-store.ts +++ /dev/null @@ -1,131 +0,0 @@ -import { useMemo } from "react"; -import AsyncStorage from "@react-native-async-storage/async-storage"; -import type { AgentProvider } from "@server/server/agent/agent-sdk-types"; -import { - AGENT_PROVIDER_DEFINITIONS, - isValidAgentProvider, - type AgentProviderDefinition, -} from "@server/server/agent/provider-manifest"; -import { create } from "zustand"; -import { createJSONStorage, persist } from "zustand/middleware"; - -const PROVIDER_RECENCY_STORE_VERSION = 1; - -interface ProviderRecencyStoreState { - recentProviderIds: AgentProvider[]; - recordUsage: (providerId: AgentProvider) => void; -} - -function sanitizeRecentProviderIds(providerIds: readonly string[] | undefined): AgentProvider[] { - if (!providerIds || providerIds.length === 0) { - return []; - } - - const seen = new Set(); - const sanitized: AgentProvider[] = []; - for (const providerId of providerIds) { - if (!isValidAgentProvider(providerId)) { - continue; - } - if (seen.has(providerId)) { - continue; - } - seen.add(providerId); - sanitized.push(providerId); - } - return sanitized; -} - -export function sortProvidersByRecency( - providers: readonly T[], - recentProviderIds: readonly string[], -): T[] { - if (providers.length <= 1) { - return [...providers]; - } - - const recentRank = new Map(); - for (const providerId of recentProviderIds) { - if (recentRank.has(providerId)) { - continue; - } - recentRank.set(providerId, recentRank.size); - } - - return providers - .map((provider, defaultIndex) => ({ - provider, - defaultIndex, - recentIndex: recentRank.get(provider.id) ?? Number.POSITIVE_INFINITY, - })) - .sort((left, right) => { - if (left.recentIndex !== right.recentIndex) { - return left.recentIndex - right.recentIndex; - } - return left.defaultIndex - right.defaultIndex; - }) - .map((entry) => entry.provider); -} - -function migratePersistedState( - state: unknown, -): Pick { - const record = state as { recentProviderIds?: string[] } | null | undefined; - return { - recentProviderIds: sanitizeRecentProviderIds(record?.recentProviderIds), - }; -} - -export const useProviderRecencyStore = create()( - persist( - (set) => ({ - recentProviderIds: [], - recordUsage: (providerId) => { - if (!isValidAgentProvider(providerId)) { - return; - } - - set((state) => ({ - recentProviderIds: [ - providerId, - ...state.recentProviderIds.filter((id) => id !== providerId), - ], - })); - }, - }), - { - name: "agent-provider-recency", - version: PROVIDER_RECENCY_STORE_VERSION, - storage: createJSONStorage(() => AsyncStorage), - partialize: (state) => ({ - recentProviderIds: state.recentProviderIds, - }), - migrate: (persistedState) => migratePersistedState(persistedState), - }, - ), -); - -export function useProviderRecency( - availableProviders: readonly AgentProviderDefinition[] = AGENT_PROVIDER_DEFINITIONS, -): { - providers: AgentProviderDefinition[]; - recordUsage: (providerId: AgentProvider) => void; -} { - const recentProviderIds = useProviderRecencyStore((state) => state.recentProviderIds); - const recordUsage = useProviderRecencyStore((state) => state.recordUsage); - - const providers = useMemo( - () => sortProvidersByRecency(availableProviders, recentProviderIds), - [availableProviders, recentProviderIds], - ); - - return { - providers, - recordUsage, - }; -} - -export const __providerRecencyStoreTestUtils = { - migratePersistedState, - sanitizeRecentProviderIds, -}; diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index c77e9d720..74bec9bf5 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -212,6 +212,7 @@ export type DaemonServerInfo = { }; export interface AgentTimelineCursorState { + epoch: string; startSeq: number; endSeq: number; } diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 677667a2e..86f979870 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -729,7 +729,7 @@ function finalizeHeadItems(head: StreamItem[]): StreamItem[] { /** * Flush head items to tail, avoiding duplicates. */ -function flushHeadToTail(tail: StreamItem[], head: StreamItem[]): StreamItem[] { +export function flushHeadToTail(tail: StreamItem[], head: StreamItem[]): StreamItem[] { if (head.length === 0) { return tail; } diff --git a/packages/cli/src/utils/timeline.ts b/packages/cli/src/utils/timeline.ts index 2390b00a5..e49b0244e 100644 --- a/packages/cli/src/utils/timeline.ts +++ b/packages/cli/src/utils/timeline.ts @@ -11,6 +11,7 @@ export async function fetchProjectedTimelineItems( const timeline = await input.client.fetchAgentTimeline(input.agentId, { direction: "tail", limit: 0, + projection: "projected", }); return timeline.entries.map((entry) => entry.item); } diff --git a/packages/server/src/client/daemon-client.test.ts b/packages/server/src/client/daemon-client.test.ts index 04cc58d90..e51419cc4 100644 --- a/packages/server/src/client/daemon-client.test.ts +++ b/packages/server/src/client/daemon-client.test.ts @@ -2032,15 +2032,24 @@ describe("DaemonClient", () => { agentId: "agent_cli", agent: null, direction: "tail", - startSeq: 1, - endSeq: 1, + 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 }, hasOlder: false, hasNewer: false, entries: [ { timestamp: "2026-02-08T20:20:00.000Z", provider: "codex", - seq: 1, + seqStart: 1, + seqEnd: 1, + sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }], + collapsed: [], item: { type: "tool_call", callId: "call_cli_snapshot", @@ -2117,15 +2126,24 @@ describe("DaemonClient", () => { agentId: "agent_cli", agent: null, direction: "tail", - startSeq: 1, - endSeq: 1, + 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 }, hasOlder: false, hasNewer: false, entries: [ { timestamp: "2026-02-08T20:20:00.000Z", provider: "codex", - seq: 1, + seqStart: 1, + seqEnd: 1, + sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }], + collapsed: [], 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 81019f555..3ebe2ed9a 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -152,6 +152,7 @@ export type DaemonEvent = event: AgentStreamEventPayload; timestamp: string; seq?: number; + epoch?: string; } | { type: "status"; payload: { status: string } & Record } | { type: "agent_deleted"; agentId: string } @@ -336,13 +337,13 @@ type ScheduleDeletePayload = Extract< export type FetchAgentTimelinePayload = FetchAgentTimelineResponseMessage["payload"]; export type FetchAgentTimelineDirection = FetchAgentTimelinePayload["direction"]; -export type FetchAgentTimelineCursor = NonNullable< - Extract["cursor"] ->; +export type FetchAgentTimelineProjection = FetchAgentTimelinePayload["projection"]; +export type FetchAgentTimelineCursor = NonNullable; export type FetchAgentTimelineOptions = { direction?: FetchAgentTimelineDirection; cursor?: FetchAgentTimelineCursor; limit?: number; + projection?: FetchAgentTimelineProjection; requestId?: string; }; @@ -1673,6 +1674,7 @@ 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({ @@ -3899,6 +3901,7 @@ 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 8aba788a1..6904d529f 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -1273,7 +1273,7 @@ describe("AgentManager", () => { } }); - test("fetchTimeline returns committed rows after a known seq without reset metadata", async () => { + test("fetchTimeline returns full timeline with reset when cursor epoch is stale", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-timeline-stale-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -1291,42 +1291,40 @@ describe("AgentManager", () => { cwd: workdir, }); - for (let seq = 1; seq <= 120; seq += 1) { - await manager.appendTimelineItem(snapshot.id, { - type: "assistant_message", - text: `committed row ${seq}`, - }); - } - - const baseline = await manager.fetchTimeline(snapshot.id, { - direction: "tail", - limit: 0, - }); - expect(baseline.rows).toHaveLength(120); - - await manager.emitLiveTimelineItem(snapshot.id, { + await manager.appendTimelineItem(snapshot.id, { type: "assistant_message", - text: "partial reply", + text: "one", }); await manager.appendTimelineItem(snapshot.id, { type: "assistant_message", - text: "finalized reply", + text: "two", + }); + await manager.appendTimelineItem(snapshot.id, { + type: "assistant_message", + text: "three", }); - const result = await manager.fetchTimeline(snapshot.id, { + const baseline = manager.fetchTimeline(snapshot.id, { + direction: "tail", + limit: 2, + }); + expect(baseline.rows).toHaveLength(2); + + const result = manager.fetchTimeline(snapshot.id, { direction: "after", cursor: { - seq: 120, + epoch: "stale-epoch", + seq: baseline.rows[baseline.rows.length - 1]!.seq, }, - limit: 0, + limit: 1, }); - expect(result.rows).toHaveLength(1); - expect(result.rows[0]?.seq).toBe(121); - expect(result.rows[0]?.item).toEqual({ - type: "assistant_message", - text: "finalized reply", - }); + 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); }); test("getTimelineRows falls back to the in-memory timeline when no durable store is configured", async () => { @@ -1448,6 +1446,7 @@ describe("AgentManager", () => { const streamEvents: Array<{ seq?: number; + epoch?: string; eventType?: string; itemType?: string; text?: string; @@ -1459,6 +1458,7 @@ 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: @@ -1487,12 +1487,14 @@ describe("AgentManager", () => { itemType: "assistant_message", text: "final ", seq: 1, + epoch: expect.any(String), }); expect(assistantTimelineEvents[1]).toMatchObject({ eventType: "timeline", itemType: "assistant_message", text: "reply", seq: 2, + epoch: expect.any(String), }); expect(manager.getTimeline(snapshot.id)).toEqual([ @@ -1510,6 +1512,8 @@ describe("AgentManager", () => { limit: 0, }); expect(fetched.rows).toHaveLength(2); + expect(assistantTimelineEvents[0]?.epoch).toBe(fetched.epoch); + expect(assistantTimelineEvents[1]?.epoch).toBe(fetched.epoch); expect(fetched.rows[0]?.item).toEqual({ type: "assistant_message", text: "final ", diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index c0dbd729d..39f9673a1 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -63,6 +63,7 @@ export type AgentManagerEvent = agentId: string; event: AgentStreamEvent; seq?: number; + epoch?: string; }; export type AgentSubscriber = (event: AgentManagerEvent) => void; @@ -584,18 +585,8 @@ export class AgentManager { return this.timelineStore.getRows(id); } - async fetchTimeline( - id: string, - options?: AgentTimelineFetchOptions, - ): Promise { + fetchTimeline(id: string, options?: AgentTimelineFetchOptions): AgentTimelineFetchResult { this.requireAgent(id); - const agent = this.agents.get(id); - if (agent && !agent.historyPrimed) { - await this.hydrateTimelineFromLegacyProviderHistory(agent); - } - if (this.durableTimelineStore) { - return await this.durableTimelineStore.fetchCommitted(id, options); - } return this.timelineStore.fetch(id, options); } @@ -1053,6 +1044,7 @@ export class AgentManager { }, { seq: row.seq, + epoch: this.timelineStore.getEpoch(agentId), }, ); if (options?.emitState !== false) { @@ -1073,6 +1065,7 @@ export class AgentManager { }, { seq: row.seq, + epoch: this.timelineStore.getEpoch(agentId), }, ); await this.persistSnapshot(agent); @@ -2404,6 +2397,7 @@ export class AgentManager { timelineRow ? { seq: timelineRow.seq, + epoch: this.timelineStore.getEpoch(agent.id), } : undefined, ); @@ -2445,6 +2439,7 @@ export class AgentManager { }, { seq: row.seq, + epoch: this.timelineStore.getEpoch(agent.id), }, ); } @@ -2604,7 +2599,7 @@ export class AgentManager { private dispatchStream( agentId: string, event: AgentStreamEvent, - metadata?: { seq?: number }, + metadata?: { seq?: number; epoch?: string }, ): void { this.dispatch({ type: "agent_stream", agentId, event, ...metadata }); } diff --git a/packages/server/src/server/agent/agent-timeline-store-types.ts b/packages/server/src/server/agent/agent-timeline-store-types.ts index 1543baa18..55774d6fc 100644 --- a/packages/server/src/server/agent/agent-timeline-store-types.ts +++ b/packages/server/src/server/agent/agent-timeline-store-types.ts @@ -7,6 +7,7 @@ export type AgentTimelineRow = { }; export type AgentTimelineCursor = { + epoch: string; seq: number; }; @@ -30,7 +31,11 @@ export type AgentTimelineWindow = { }; export type AgentTimelineFetchResult = { + epoch: string; direction: AgentTimelineFetchDirection; + reset: boolean; + staleCursor: boolean; + gap: boolean; window: AgentTimelineWindow; hasOlder: boolean; hasNewer: boolean; diff --git a/packages/server/src/server/agent/agent-timeline-store.ts b/packages/server/src/server/agent/agent-timeline-store.ts index 4ac8531ed..c3b2ecd01 100644 --- a/packages/server/src/server/agent/agent-timeline-store.ts +++ b/packages/server/src/server/agent/agent-timeline-store.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import type { AgentTimelineItem } from "./agent-sdk-types.js"; import type { AgentTimelineFetchOptions, @@ -8,11 +9,13 @@ import type { export type SeedAgentTimelineOptions = { items?: readonly AgentTimelineItem[]; rows?: readonly AgentTimelineRow[]; + epoch?: string; nextSeq?: number; timestamp?: string; }; type AgentTimelineState = { + epoch: string; rows: AgentTimelineRow[]; nextSeq: number; }; @@ -45,6 +48,7 @@ export class InMemoryAgentTimelineStore { : this.buildRowsFromItems(options?.items ?? [], options?.nextSeq ?? 1, timestamp); const nextSeq = options?.nextSeq ?? (rows.length ? rows[rows.length - 1]!.seq + 1 : 1); this.states.set(agentId, { + epoch: options?.epoch ?? randomUUID(), rows, nextSeq, }); @@ -62,6 +66,10 @@ export class InMemoryAgentTimelineStore { return this.requireState(agentId).rows.map(cloneRow); } + getEpoch(agentId: string): string { + return this.requireState(agentId).epoch; + } + fetch(agentId: string, options?: AgentTimelineFetchOptions): AgentTimelineFetchResult { const state = this.requireState(agentId); const direction = options?.direction ?? "tail"; @@ -81,9 +89,43 @@ export class InMemoryAgentTimelineStore { nextSeq: state.nextSeq, }; + if (cursor && typeof cursor.epoch === "string" && cursor.epoch !== state.epoch) { + return { + epoch: state.epoch, + direction, + reset: true, + staleCursor: true, + gap: false, + window, + hasOlder: false, + hasNewer: false, + rows: state.rows.map(cloneRow), + }; + } + + const cloneRows = (items: AgentTimelineRow[]) => items.map(cloneRow); + + if (direction === "after" && cursor && state.rows.length > 0 && cursor.seq < minSeq - 1) { + return { + epoch: state.epoch, + direction, + reset: true, + staleCursor: false, + gap: true, + window, + hasOlder: false, + hasNewer: false, + rows: cloneRows(state.rows), + }; + } + if (state.rows.length === 0) { return { + epoch: state.epoch, direction, + reset: false, + staleCursor: false, + gap: false, window, hasOlder: false, hasNewer: false, @@ -97,11 +139,15 @@ export class InMemoryAgentTimelineStore { ? state.rows : state.rows.slice(state.rows.length - limit); return { + epoch: state.epoch, direction, + reset: false, + staleCursor: false, + gap: false, window, hasOlder: selected.length > 0 && selected[0]!.seq > minSeq, hasNewer: false, - rows: selected.map(cloneRow), + rows: cloneRows(selected), }; } @@ -110,7 +156,11 @@ export class InMemoryAgentTimelineStore { const startIdx = state.rows.findIndex((row) => row.seq > baseSeq); if (startIdx < 0) { return { + epoch: state.epoch, direction, + reset: false, + staleCursor: false, + gap: false, window, hasOlder: baseSeq >= minSeq, hasNewer: false, @@ -123,11 +173,15 @@ export class InMemoryAgentTimelineStore { : state.rows.slice(startIdx, startIdx + limit); const lastSelected = selected[selected.length - 1]; return { + epoch: state.epoch, direction, + reset: false, + staleCursor: false, + gap: false, window, hasOlder: selected[0]!.seq > minSeq, hasNewer: Boolean(lastSelected && lastSelected.seq < maxSeq), - rows: selected.map(cloneRow), + rows: cloneRows(selected), }; } @@ -139,11 +193,15 @@ export class InMemoryAgentTimelineStore { ? boundedRows : boundedRows.slice(boundedRows.length - limit); return { + epoch: state.epoch, direction, + reset: false, + staleCursor: false, + gap: false, window, hasOlder: selected.length > 0 && selected[0]!.seq > minSeq, hasNewer: endExclusive >= 0, - rows: selected.map(cloneRow), + rows: cloneRows(selected), }; } diff --git a/packages/server/src/server/agent/provider-manifest.ts b/packages/server/src/server/agent/provider-manifest.ts index 4eed75478..5884fd8e9 100644 --- a/packages/server/src/server/agent/provider-manifest.ts +++ b/packages/server/src/server/agent/provider-manifest.ts @@ -141,27 +141,6 @@ export const AGENT_PROVIDER_DEFINITIONS: AgentProviderDefinition[] = [ defaultModel: "gpt-5.4-mini", }, }, - { - id: "gemini", - label: "Gemini CLI", - description: "Google's coding agent CLI", - defaultModeId: null, - modes: [], - }, - { - id: "amp", - label: "AMP", - description: "Sourcegraph's coding agent CLI", - defaultModeId: null, - modes: [], - }, - { - id: "aider", - label: "Aider", - description: "Paul Gauthier's coding assistant CLI", - defaultModeId: null, - modes: [], - }, { id: "copilot", label: "Copilot", diff --git a/packages/server/src/server/agent/provider-registry.ts b/packages/server/src/server/agent/provider-registry.ts index baf42c3a3..0707a8534 100644 --- a/packages/server/src/server/agent/provider-registry.ts +++ b/packages/server/src/server/agent/provider-registry.ts @@ -20,13 +20,9 @@ import type { ProviderProfileModel, ProviderRuntimeSettings, } from "./provider-launch-config.js"; - -import { AiderAgentClient } from "./providers/aider-agent.js"; -import { AmpAgentClient } from "./providers/amp-agent.js"; import { ClaudeAgentClient } from "./providers/claude-agent.js"; import { CodexAppServerAgentClient } from "./providers/codex-app-server-agent.js"; import { CopilotACPAgentClient } from "./providers/copilot-acp-agent.js"; -import { GeminiAgentClient } from "./providers/gemini-agent.js"; import { GenericACPAgentClient } from "./providers/generic-acp-agent.js"; import { OpenCodeAgentClient, OpenCodeServerManager } from "./providers/opencode-agent.js"; import { PiACPAgentClient } from "./providers/pi-acp-agent.js"; @@ -72,9 +68,6 @@ const PROVIDER_CLIENT_FACTORIES: Record = { runtimeSettings, }), codex: (logger, runtimeSettings) => new CodexAppServerAgentClient(logger, runtimeSettings), - gemini: (_logger, runtimeSettings) => new GeminiAgentClient(runtimeSettings), - amp: (_logger, runtimeSettings) => new AmpAgentClient(runtimeSettings), - aider: (_logger, runtimeSettings) => new AiderAgentClient(runtimeSettings), copilot: (logger, runtimeSettings) => new CopilotACPAgentClient({ logger, diff --git a/packages/server/src/server/agent/providers/aider-agent.ts b/packages/server/src/server/agent/providers/aider-agent.ts deleted file mode 100644 index 2b063babb..000000000 --- a/packages/server/src/server/agent/providers/aider-agent.ts +++ /dev/null @@ -1,75 +0,0 @@ -import { existsSync } from "node:fs"; - -import type { - AgentCapabilityFlags, - AgentClient, - AgentLaunchContext, - AgentModelDefinition, - AgentPersistenceHandle, - AgentSession, - AgentSessionConfig, - ListModelsOptions, -} from "../agent-sdk-types.js"; -import { - findExecutable, - isProviderCommandAvailable, - type ProviderRuntimeSettings, -} from "../provider-launch-config.js"; - -const AIDER_PROVIDER = "aider" as const; - -const AIDER_CAPABILITIES: AgentCapabilityFlags = { - supportsStreaming: false, - supportsSessionPersistence: false, - supportsDynamicModes: false, - supportsMcpServers: false, - supportsReasoningStream: false, - supportsToolInvocations: false, -}; - -function resolveAiderBinary(): string { - const found = findExecutable("aider"); - if (found) { - return found; - } - throw new Error( - "Aider binary not found. Install Aider and ensure 'aider' is available in your shell PATH.", - ); -} - -function createUnsupportedSessionError(): Error { - return new Error("Aider does not support session-backed agents in Paseo."); -} - -export class AiderAgentClient implements AgentClient { - readonly provider = AIDER_PROVIDER; - readonly capabilities = AIDER_CAPABILITIES; - - constructor(private readonly runtimeSettings?: ProviderRuntimeSettings) {} - - async createSession( - _config: AgentSessionConfig, - _launchContext?: AgentLaunchContext, - ): Promise { - throw createUnsupportedSessionError(); - } - - async resumeSession( - _handle: AgentPersistenceHandle, - _overrides?: Partial, - _launchContext?: AgentLaunchContext, - ): Promise { - throw createUnsupportedSessionError(); - } - - async listModels(_options?: ListModelsOptions): Promise { - return []; - } - - async isAvailable(): Promise { - if (this.runtimeSettings?.command?.mode === "replace") { - return existsSync(this.runtimeSettings.command.argv[0]); - } - return isProviderCommandAvailable(this.runtimeSettings?.command, resolveAiderBinary); - } -} diff --git a/packages/server/src/server/agent/providers/amp-agent.ts b/packages/server/src/server/agent/providers/amp-agent.ts deleted file mode 100644 index a676db808..000000000 --- a/packages/server/src/server/agent/providers/amp-agent.ts +++ /dev/null @@ -1,75 +0,0 @@ -import { existsSync } from "node:fs"; - -import type { - AgentCapabilityFlags, - AgentClient, - AgentLaunchContext, - AgentModelDefinition, - AgentPersistenceHandle, - AgentSession, - AgentSessionConfig, - ListModelsOptions, -} from "../agent-sdk-types.js"; -import { - findExecutable, - isProviderCommandAvailable, - type ProviderRuntimeSettings, -} from "../provider-launch-config.js"; - -const AMP_PROVIDER = "amp" as const; - -const AMP_CAPABILITIES: AgentCapabilityFlags = { - supportsStreaming: false, - supportsSessionPersistence: false, - supportsDynamicModes: false, - supportsMcpServers: false, - supportsReasoningStream: false, - supportsToolInvocations: false, -}; - -function resolveAmpBinary(): string { - const found = findExecutable("amp"); - if (found) { - return found; - } - throw new Error( - "AMP binary not found. Install AMP and ensure 'amp' is available in your shell PATH.", - ); -} - -function createUnsupportedSessionError(): Error { - return new Error("AMP does not support session-backed agents in Paseo."); -} - -export class AmpAgentClient implements AgentClient { - readonly provider = AMP_PROVIDER; - readonly capabilities = AMP_CAPABILITIES; - - constructor(private readonly runtimeSettings?: ProviderRuntimeSettings) {} - - async createSession( - _config: AgentSessionConfig, - _launchContext?: AgentLaunchContext, - ): Promise { - throw createUnsupportedSessionError(); - } - - async resumeSession( - _handle: AgentPersistenceHandle, - _overrides?: Partial, - _launchContext?: AgentLaunchContext, - ): Promise { - throw createUnsupportedSessionError(); - } - - async listModels(_options?: ListModelsOptions): Promise { - return []; - } - - async isAvailable(): Promise { - if (this.runtimeSettings?.command?.mode === "replace") { - return existsSync(this.runtimeSettings.command.argv[0]); - } - return isProviderCommandAvailable(this.runtimeSettings?.command, resolveAmpBinary); - } -} diff --git a/packages/server/src/server/agent/providers/gemini-agent.ts b/packages/server/src/server/agent/providers/gemini-agent.ts deleted file mode 100644 index 61dd18a17..000000000 --- a/packages/server/src/server/agent/providers/gemini-agent.ts +++ /dev/null @@ -1,75 +0,0 @@ -import { existsSync } from "node:fs"; - -import type { - AgentCapabilityFlags, - AgentClient, - AgentLaunchContext, - AgentModelDefinition, - AgentPersistenceHandle, - AgentSession, - AgentSessionConfig, - ListModelsOptions, -} from "../agent-sdk-types.js"; -import { - findExecutable, - isProviderCommandAvailable, - type ProviderRuntimeSettings, -} from "../provider-launch-config.js"; - -const GEMINI_PROVIDER = "gemini" as const; - -const GEMINI_CAPABILITIES: AgentCapabilityFlags = { - supportsStreaming: false, - supportsSessionPersistence: false, - supportsDynamicModes: false, - supportsMcpServers: false, - supportsReasoningStream: false, - supportsToolInvocations: false, -}; - -function resolveGeminiBinary(): string { - const found = findExecutable("gemini"); - if (found) { - return found; - } - throw new Error( - "Gemini CLI binary not found. Install Gemini CLI and ensure 'gemini' is available in your shell PATH.", - ); -} - -function createUnsupportedSessionError(): Error { - return new Error("Gemini CLI does not support session-backed agents in Paseo."); -} - -export class GeminiAgentClient implements AgentClient { - readonly provider = GEMINI_PROVIDER; - readonly capabilities = GEMINI_CAPABILITIES; - - constructor(private readonly runtimeSettings?: ProviderRuntimeSettings) {} - - async createSession( - _config: AgentSessionConfig, - _launchContext?: AgentLaunchContext, - ): Promise { - throw createUnsupportedSessionError(); - } - - async resumeSession( - _handle: AgentPersistenceHandle, - _overrides?: Partial, - _launchContext?: AgentLaunchContext, - ): Promise { - throw createUnsupportedSessionError(); - } - - async listModels(_options?: ListModelsOptions): Promise { - return []; - } - - async isAvailable(): Promise { - if (this.runtimeSettings?.command?.mode === "replace") { - return existsSync(this.runtimeSettings.command.argv[0]); - } - return isProviderCommandAvailable(this.runtimeSettings?.command, resolveGeminiBinary); - } -} 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 index 5fe1683d7..a90f5852f 100644 --- 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 @@ -29,9 +29,10 @@ async function waitFor( } } -function isSeqLessAssistantTimeline( +function isLiveAssistantTimeline( message: SessionOutboundMessage, agentId: string, + epoch?: string, text?: string, ): boolean { return ( @@ -40,6 +41,8 @@ function isSeqLessAssistantTimeline( message.payload.event.type === "timeline" && message.payload.event.item.type === "assistant_message" && message.payload.seq === undefined && + typeof message.payload.epoch === "string" && + (epoch === undefined || message.payload.epoch === epoch) && (text === undefined || message.payload.event.item.text === text) ); } @@ -73,6 +76,14 @@ describe("daemon E2E - timeline reconnect contract", () => { text: `committed row ${seq}`, }); } + const baseline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + const epoch = baseline.epoch; + expect(epoch).not.toBe(""); + expect(baseline.endCursor?.epoch).toBe(epoch); primaryCollector.clear(); await ctx.daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, { @@ -81,7 +92,7 @@ describe("daemon E2E - timeline reconnect contract", () => { }); await waitFor(() => primaryCollector.messages.some((message) => - isSeqLessAssistantTimeline(message, agent.id, "partial before disconnect"), + isLiveAssistantTimeline(message, agent.id, epoch, "partial before disconnect"), ), ); @@ -105,18 +116,26 @@ describe("daemon E2E - timeline reconnect contract", () => { expect( reconnectCollector.messages.some((message) => - isSeqLessAssistantTimeline(message, agent.id), + isLiveAssistantTimeline(message, agent.id, epoch), ), ).toBe(false); const catchUp = await reconnectClient.fetchAgentTimeline(agent.id, { direction: "after", - cursor: { seq: 120 }, + cursor: { epoch, seq: 120 }, limit: 0, + projection: "canonical", }); + expect(catchUp.epoch).toBe(epoch); + expect(catchUp.reset).toBe(false); + expect(catchUp.staleCursor).toBe(false); + expect(catchUp.gap).toBe(false); expect(catchUp.entries).toHaveLength(1); - expect(catchUp.entries[0]?.seq).toBe(121); + expect(catchUp.startCursor).toEqual({ epoch, seq: 121 }); + expect(catchUp.endCursor).toEqual({ epoch, seq: 121 }); + expect(catchUp.entries[0]?.seqStart).toBe(121); + expect(catchUp.entries[0]?.seqEnd).toBe(121); expect(catchUp.entries[0]?.item).toEqual({ type: "assistant_message", text: "finalized while disconnected", @@ -149,6 +168,14 @@ describe("daemon E2E - timeline reconnect contract", () => { text: `committed row ${seq}`, }); } + const baseline = await ctx.client.fetchAgentTimeline(agent.id, { + direction: "tail", + limit: 0, + projection: "canonical", + }); + const epoch = baseline.epoch; + expect(epoch).not.toBe(""); + expect(baseline.endCursor?.epoch).toBe(epoch); primaryCollector.clear(); await ctx.daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, { @@ -157,7 +184,7 @@ describe("daemon E2E - timeline reconnect contract", () => { }); await waitFor(() => primaryCollector.messages.some((message) => - isSeqLessAssistantTimeline(message, agent.id, "partial before disconnect"), + isLiveAssistantTimeline(message, agent.id, epoch, "partial before disconnect"), ), ); @@ -176,17 +203,24 @@ describe("daemon E2E - timeline reconnect contract", () => { expect( reconnectCollector.messages.some((message) => - isSeqLessAssistantTimeline(message, agent.id), + isLiveAssistantTimeline(message, agent.id, epoch), ), ).toBe(false); const catchUp = await reconnectClient.fetchAgentTimeline(agent.id, { direction: "after", - cursor: { seq: 120 }, + cursor: { epoch, seq: 120 }, limit: 0, + projection: "canonical", }); + expect(catchUp.epoch).toBe(epoch); + expect(catchUp.reset).toBe(false); + expect(catchUp.staleCursor).toBe(false); + expect(catchUp.gap).toBe(false); expect(catchUp.entries).toHaveLength(0); + expect(catchUp.startCursor).toBeNull(); + expect(catchUp.endCursor).toBeNull(); reconnectCollector.clear(); await ctx.daemon.daemon.agentManager.emitLiveTimelineItem(agent.id, { @@ -195,7 +229,7 @@ describe("daemon E2E - timeline reconnect contract", () => { }); await waitFor(() => reconnectCollector.messages.some((message) => - isSeqLessAssistantTimeline(message, agent.id, "fresh live after reconnect"), + isLiveAssistantTimeline(message, agent.id, epoch, "fresh live after reconnect"), ), ); } finally { 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 ae46e9d72..04e1b0369 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 @@ -38,6 +38,7 @@ describe("daemon E2E - timeline window", () => { const timeline = await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 1, + projection: "canonical", }); const assistantTexts = timeline.entries @@ -45,7 +46,7 @@ describe("daemon E2E - timeline window", () => { .map((entry) => entry.item.text); expect(assistantTexts).toEqual([expected]); - expect(timeline.startSeq).toBe(timeline.endSeq); + expect(timeline.startCursor?.seq).toBe(timeline.endCursor?.seq); } finally { rmSync(cwd, { recursive: true, force: true }); } @@ -71,6 +72,7 @@ describe("daemon E2E - timeline window", () => { const timeline = await ctx.client.fetchAgentTimeline(agent.id, { direction: "tail", limit: 1, + projection: "canonical", }); const assistantTexts = timeline.entries @@ -79,7 +81,7 @@ describe("daemon E2E - timeline window", () => { expect(assistantTexts.join("")).toBe(expected); expect(timeline.hasOlder).toBe(true); - expect(timeline.startSeq).toBeGreaterThan(1); + expect(timeline.startCursor?.seq).toBeGreaterThan(1); } finally { rmSync(cwd, { recursive: true, force: true }); } diff --git a/packages/server/src/server/persistence-hooks.test.ts b/packages/server/src/server/persistence-hooks.test.ts index 3878e5648..d1ce731fc 100644 --- a/packages/server/src/server/persistence-hooks.test.ts +++ b/packages/server/src/server/persistence-hooks.test.ts @@ -1,4 +1,5 @@ -import { describe, expect, test } from "vitest"; +import { describe, expect, test, vi } from "vitest"; +import type { ManagedAgent } from "./agent/agent-manager.js"; import type { StoredAgentRecord } from "./agent/agent-storage.js"; import { attachAgentStoragePersistence, @@ -173,16 +174,16 @@ describe("persistence hooks", () => { test("buildSessionConfig accepts providers from the canonical manifest", () => { const record = createRecord({ - provider: "gemini", + provider: "claude", persistence: { - provider: "gemini", + provider: "claude", sessionId: "session-123", }, config: {}, }); expect(buildSessionConfig(record)).toMatchObject({ - provider: "gemini", + provider: "claude", cwd: "/tmp/project", }); }); diff --git a/packages/server/src/server/provider-history-compatibility-boundary.test.ts b/packages/server/src/server/provider-history-compatibility-boundary.test.ts deleted file mode 100644 index df3e50ff9..000000000 --- a/packages/server/src/server/provider-history-compatibility-boundary.test.ts +++ /dev/null @@ -1,15 +0,0 @@ -import { readFileSync } from "node:fs"; -import { describe, expect, test } from "vitest"; - -describe("agent loading boundary", () => { - test("session runtime code does not directly hydrate provider history", () => { - const sessionSource = readFileSync(new URL("./session.ts", import.meta.url), "utf8"); - const agentLoadingSource = readFileSync( - new URL("./agent-loading-service.ts", import.meta.url), - "utf8", - ); - - expect(sessionSource).not.toMatch(/hydrateTimelineFromProvider\s*\(/); - expect(agentLoadingSource).not.toMatch(/hydrateTimelineFromProvider\s*\(/); - }); -}); diff --git a/packages/server/src/server/provider-history-compatibility-service.test.ts b/packages/server/src/server/provider-history-compatibility-service.test.ts deleted file mode 100644 index f4d3829c3..000000000 --- a/packages/server/src/server/provider-history-compatibility-service.test.ts +++ /dev/null @@ -1,333 +0,0 @@ -import { mkdtempSync, rmSync } from "node:fs"; -import os from "node:os"; -import path from "node:path"; -import pino from "pino"; -import { describe, expect, test, vi } from "vitest"; - -import { AgentLoadingService } from "./agent-loading-service.js"; -import { AgentManager } from "./agent/agent-manager.js"; -import { AgentStorage, type StoredAgentRecord } from "./agent/agent-storage.js"; -import { createTestAgentClients } from "./test-utils/fake-agent-client.js"; - -function createStoredAgentRecord(overrides?: Partial): StoredAgentRecord { - const now = "2026-03-25T00:00:00.000Z"; - return { - id: "agent-compat-1", - provider: "codex", - cwd: "/tmp/project", - createdAt: now, - updatedAt: now, - title: null, - labels: {}, - lastStatus: "idle", - config: { - model: "gpt-5.1-codex-mini", - }, - persistence: { - provider: "codex", - sessionId: "provider-session-1", - }, - ...overrides, - }; -} - -function createDeferred() { - let resolve!: (value: T) => void; - let reject!: (error: unknown) => void; - const promise = new Promise((res, rej) => { - resolve = res; - reject = rej; - }); - return { promise, resolve, reject }; -} - -function createCompatibilitySnapshot(overrides?: Partial>) { - return { - id: "agent-compat-1", - provider: "codex", - cwd: "/tmp/project", - persistence: { - provider: "codex", - sessionId: "provider-session-1", - }, - ...overrides, - }; -} - -describe("AgentLoadingService", () => { - test("ensureAgentLoaded succeeds when provider history is absent", async () => { - const workspaceRoot = mkdtempSync(path.join(os.tmpdir(), "provider-history-compat-empty-")); - const logger = pino({ level: "silent" }); - - try { - const storage = new AgentStorage(path.join(workspaceRoot, "agents"), logger); - const manager = new AgentManager({ - clients: createTestAgentClients(), - registry: storage, - logger, - idFactory: () => "00000000-0000-4000-8000-000000000302", - }); - const service = new AgentLoadingService({ - agentManager: manager as any, - agentStorage: storage as any, - logger, - }); - - const snapshot = await manager.createAgent({ - provider: "codex", - cwd: workspaceRoot, - model: "gpt-5.1-codex-mini", - }); - await manager.flush(); - await storage.flush(); - await manager.closeAgent(snapshot.id); - - const loaded = await service.ensureAgentLoaded({ agentId: snapshot.id }); - - expect(loaded.id).toBe(snapshot.id); - expect(manager.getTimeline(snapshot.id)).toEqual([]); - } finally { - rmSync(workspaceRoot, { recursive: true, force: true }); - } - }); - - test("ensureAgentLoaded dedupes concurrent cold-load bootstrap", async () => { - const deferred = createDeferred(); - let currentAgent: any = null; - const snapshot = createCompatibilitySnapshot({ id: "agent-compat-dedupe" }); - const agentStorage = { - get: vi.fn(async () => - createStoredAgentRecord({ - id: "agent-compat-dedupe", - cwd: "/tmp/dedupe", - persistence: { - provider: "codex", - sessionId: "provider-session-dedupe", - }, - }), - ), - }; - const agentManager = { - getAgent: vi.fn(() => currentAgent), - resumeAgentFromPersistence: vi.fn(async () => deferred.promise), - createAgent: vi.fn(), - reloadAgentSession: vi.fn(), - }; - const logger = { - child: () => logger, - info: vi.fn(), - warn: vi.fn(), - }; - const service = new AgentLoadingService({ - agentManager: agentManager as any, - agentStorage: agentStorage as any, - logger: logger as any, - }); - - const firstLoad = service.ensureAgentLoaded({ agentId: "agent-compat-dedupe" }); - const secondLoad = service.ensureAgentLoaded({ agentId: "agent-compat-dedupe" }); - deferred.resolve(snapshot); - - const [firstResult, secondResult] = await Promise.all([firstLoad, secondLoad]); - - expect(firstResult).toEqual(snapshot); - expect(secondResult).toEqual(snapshot); - expect(agentStorage.get).toHaveBeenCalledTimes(1); - expect(agentManager.resumeAgentFromPersistence).toHaveBeenCalledTimes(1); - }); - - test("resumeAgent delegates to manager resume", async () => { - const snapshot = createCompatibilitySnapshot({ id: "agent-compat-resume" }); - const agentManager = { - getAgent: vi.fn(() => null), - resumeAgentFromPersistence: vi.fn(async () => snapshot), - createAgent: vi.fn(), - reloadAgentSession: vi.fn(), - }; - const logger = { - child: () => logger, - info: vi.fn(), - warn: vi.fn(), - }; - const service = new AgentLoadingService({ - agentManager: agentManager as any, - agentStorage: { - get: async () => null, - } as any, - logger: logger as any, - }); - - const result = await service.resumeAgent({ - handle: { - provider: "codex", - sessionId: "provider-session-resume", - }, - overrides: { - model: "gpt-5.4", - }, - }); - - expect(agentManager.resumeAgentFromPersistence).toHaveBeenCalledWith( - { - provider: "codex", - sessionId: "provider-session-resume", - }, - { - model: "gpt-5.4", - }, - ); - expect(result).toEqual(snapshot); - }); - - test("refreshAgent reloads loaded persisted agents", async () => { - const existing = createCompatibilitySnapshot({ id: "agent-compat-refresh-loaded" }); - const reloaded = createCompatibilitySnapshot({ id: "agent-compat-refresh-loaded" }); - let currentAgent: any = existing; - const agentManager = { - getAgent: vi.fn(() => currentAgent), - resumeAgentFromPersistence: vi.fn(), - createAgent: vi.fn(), - reloadAgentSession: vi.fn(async () => { - currentAgent = reloaded; - return reloaded; - }), - }; - const logger = { - child: () => logger, - info: vi.fn(), - warn: vi.fn(), - }; - const service = new AgentLoadingService({ - agentManager: agentManager as any, - agentStorage: { - get: async () => null, - } as any, - logger: logger as any, - }); - - const result = await service.refreshAgent({ agentId: "agent-compat-refresh-loaded" }); - - expect(agentManager.reloadAgentSession).toHaveBeenCalledWith("agent-compat-refresh-loaded"); - expect(result).toEqual(reloaded); - }); - - test("refreshAgent keeps loaded non-persisted agents without reloading", async () => { - const existing = createCompatibilitySnapshot({ - id: "agent-compat-refresh-live", - persistence: null, - }); - const agentManager = { - getAgent: vi.fn(() => existing), - resumeAgentFromPersistence: vi.fn(), - createAgent: vi.fn(), - reloadAgentSession: vi.fn(), - }; - const logger = { - child: () => logger, - info: vi.fn(), - warn: vi.fn(), - }; - const service = new AgentLoadingService({ - agentManager: agentManager as any, - agentStorage: { - get: async () => null, - } as any, - logger: logger as any, - }); - - const result = await service.refreshAgent({ agentId: "agent-compat-refresh-live" }); - - expect(agentManager.reloadAgentSession).not.toHaveBeenCalled(); - expect(result).toEqual(existing); - }); - - test("refreshAgent resumes unloaded persisted agents", async () => { - const snapshot = createCompatibilitySnapshot({ id: "agent-compat-refresh-cold" }); - const record = createStoredAgentRecord({ - id: "agent-compat-refresh-cold", - cwd: "/tmp/refresh-cold", - persistence: { - provider: "codex", - sessionId: "provider-session-refresh-cold", - }, - }); - const agentManager = { - getAgent: vi.fn(() => null), - resumeAgentFromPersistence: vi.fn(async () => snapshot), - createAgent: vi.fn(), - reloadAgentSession: vi.fn(), - }; - const logger = { - child: () => logger, - info: vi.fn(), - warn: vi.fn(), - }; - const service = new AgentLoadingService({ - agentManager: agentManager as any, - agentStorage: { - get: vi.fn(async () => record), - } as any, - logger: logger as any, - }); - - const result = await service.refreshAgent({ agentId: "agent-compat-refresh-cold" }); - - expect(agentManager.resumeAgentFromPersistence).toHaveBeenCalledWith( - { - provider: "codex", - sessionId: "provider-session-refresh-cold", - nativeHandle: undefined, - metadata: undefined, - }, - { - cwd: "/tmp/refresh-cold", - modeId: undefined, - model: "gpt-5.1-codex-mini", - thinkingOptionId: undefined, - title: undefined, - extra: undefined, - systemPrompt: undefined, - mcpServers: undefined, - }, - "agent-compat-refresh-cold", - { - createdAt: new Date("2026-03-25T00:00:00.000Z"), - updatedAt: new Date("2026-03-25T00:00:00.000Z"), - lastUserMessageAt: null, - labels: {}, - }, - ); - expect(result).toEqual(snapshot); - }); - - test("refreshAgent preserves the unloaded no-persistence error", async () => { - const service = new AgentLoadingService({ - agentManager: { - getAgent: vi.fn(() => null), - resumeAgentFromPersistence: vi.fn(), - createAgent: vi.fn(), - reloadAgentSession: vi.fn(), - } as any, - agentStorage: { - get: async () => - createStoredAgentRecord({ - id: "agent-compat-no-persistence", - persistence: null, - }), - } as any, - logger: { - child: () => ({ - child: () => null, - info: vi.fn(), - warn: vi.fn(), - }), - info: vi.fn(), - warn: vi.fn(), - } as any, - }); - - await expect(service.refreshAgent({ agentId: "agent-compat-no-persistence" })).rejects.toThrow( - "Agent agent-compat-no-persistence cannot be refreshed because it lacks persistence", - ); - }); -}); diff --git a/packages/server/src/server/session.provider-history-compatibility-ownership.test.ts b/packages/server/src/server/session.provider-history-compatibility-ownership.test.ts deleted file mode 100644 index 287df24ca..000000000 --- a/packages/server/src/server/session.provider-history-compatibility-ownership.test.ts +++ /dev/null @@ -1,389 +0,0 @@ -import { describe, expect, test, vi } from "vitest"; - -import { Session } from "./session.js"; - -function createStoredAgentRecord(overrides?: Partial>) { - return { - id: "agent-1", - provider: "codex", - cwd: "/tmp/project", - createdAt: "2026-03-24T00:00:00.000Z", - updatedAt: "2026-03-24T00:00:00.000Z", - title: null, - labels: {}, - lastStatus: "idle", - config: null, - persistence: { - provider: "codex", - sessionId: "provider-session-1", - }, - archivedAt: null, - requiresAttention: false, - attentionReason: null, - attentionTimestamp: null, - ...overrides, - }; -} - -function createCompatibilitySnapshot(overrides?: Partial>) { - return { - id: "agent-1", - provider: "codex", - cwd: "/tmp/project", - persistence: { - provider: "codex", - sessionId: "provider-session-1", - }, - ...overrides, - }; -} - -function createSessionForOwnershipTests(options?: { - agentLoadingService?: { - ensureAgentLoaded?: (options: { agentId: string }) => Promise; - resumeAgent?: (options: { - handle: { provider: string; sessionId: string }; - overrides?: Record; - }) => Promise; - refreshAgent?: (options: { agentId: string }) => Promise; - }; - storedRecord?: Record | null; - loadedAgent?: Record | null; - timelineRows?: Array<{ seq: number; item: Record; timestamp: Date }>; -}) { - const emitted: Array<{ type: string; payload: unknown }> = []; - const logger = { - child: () => logger, - trace: vi.fn(), - debug: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - }; - - const agentManager = { - subscribe: () => () => {}, - listAgents: () => [], - getAgent: vi.fn(() => options?.loadedAgent ?? null), - createAgent: vi.fn(async () => { - throw new Error("Session should delegate unloaded bootstrap to AgentLoadingService"); - }), - resumeAgentFromPersistence: vi.fn(async () => { - throw new Error("Session should delegate persistence resume to AgentLoadingService"); - }), - reloadAgentSession: vi.fn(async () => { - throw new Error("Session should delegate refresh reload to AgentLoadingService"); - }), - hydrateTimelineFromProvider: vi.fn(async () => { - throw new Error("Session should not call hydrateTimelineFromProvider directly"); - }), - fetchTimeline: vi.fn(async () => ({ - rows: options?.timelineRows ?? [], - hasOlder: false, - hasNewer: false, - })), - recordUserMessage: vi.fn(), - waitForAgentRunStart: vi.fn(async () => undefined), - getTimeline: vi.fn(() => []), - }; - - const session = new Session({ - clientId: "test-client", - onMessage: (message) => emitted.push(message as any), - logger: logger as any, - downloadTokenStore: {} as any, - pushTokenStore: {} as any, - paseoHome: "/tmp/paseo-test", - agentManager: agentManager as any, - agentStorage: { - list: async () => (options?.storedRecord ? [options.storedRecord as any] : []), - get: async () => (options?.storedRecord as any) ?? null, - } as any, - projectRegistry: { - initialize: async () => {}, - existsOnDisk: async () => true, - list: async () => [], - get: async () => null, - upsert: async () => {}, - archive: async () => {}, - remove: async () => {}, - } as any, - workspaceRegistry: { - initialize: async () => {}, - existsOnDisk: async () => true, - list: async () => [], - get: async () => null, - upsert: async () => {}, - archive: async () => {}, - remove: async () => {}, - } as any, - createAgentMcpTransport: async () => { - throw new Error("not used"); - }, - stt: null, - tts: null, - terminalManager: null, - agentLoadingService: options?.agentLoadingService, - } as any) as any; - - return { session, emitted, agentManager }; -} - -describe("provider history compatibility ownership", () => { - test("fetch_agent_timeline_request delegates unloaded bootstrap through the compatibility seam", async () => { - const ensureAgentLoaded = vi.fn(async () => createCompatibilitySnapshot()); - const { session, emitted } = createSessionForOwnershipTests({ - storedRecord: createStoredAgentRecord(), - timelineRows: [ - { - seq: 1, - item: { type: "assistant_message", text: "rehydrated from provider history" }, - timestamp: new Date("2026-03-24T00:00:01.000Z"), - }, - ], - agentLoadingService: { - ensureAgentLoaded, - }, - }); - - session.buildAgentPayload = vi.fn(async () => ({ id: "agent-1" })); - - await session.handleMessage({ - type: "fetch_agent_timeline_request", - requestId: "req-fetch", - agentId: "agent-1", - }); - - expect(ensureAgentLoaded).toHaveBeenCalledWith({ agentId: "agent-1" }); - expect(emitted).toContainEqual({ - type: "fetch_agent_timeline_response", - payload: expect.objectContaining({ - requestId: "req-fetch", - agentId: "agent-1", - error: null, - entries: [ - expect.objectContaining({ - seq: 1, - }), - ], - }), - }); - }); - - test("fetch_agent_timeline_request emits legacy compatibility fields for older clients", async () => { - const ensureAgentLoaded = vi.fn(async () => createCompatibilitySnapshot()); - const { session, emitted } = createSessionForOwnershipTests({ - storedRecord: createStoredAgentRecord(), - timelineRows: [ - { - seq: 7, - item: { type: "assistant_message", text: "compat row" }, - timestamp: new Date("2026-03-24T00:00:07.000Z"), - }, - ], - agentLoadingService: { - ensureAgentLoaded, - }, - }); - - session.buildAgentPayload = vi.fn(async () => ({ id: "agent-1" })); - - await session.handleMessage({ - type: "fetch_agent_timeline_request", - requestId: "req-compat", - agentId: "agent-1", - direction: "tail", - }); - - expect(emitted).toContainEqual({ - type: "fetch_agent_timeline_response", - payload: expect.objectContaining({ - requestId: "req-compat", - agentId: "agent-1", - epoch: "compat:agent-1", - reset: false, - staleCursor: false, - gap: false, - projection: "projected", - startCursor: { seq: 7, epoch: "compat:agent-1" }, - endCursor: { seq: 7, epoch: "compat:agent-1" }, - window: { - minSeq: 7, - maxSeq: 7, - nextSeq: 8, - startSeq: 7, - endSeq: 7, - hasOlder: false, - hasNewer: false, - }, - entries: [ - expect.objectContaining({ - seq: 7, - seqStart: 7, - seqEnd: 7, - sourceSeqRanges: [{ startSeq: 7, endSeq: 7 }], - collapsed: [], - }), - ], - }), - }); - }); - - test("send_agent_message_request delegates unloaded bootstrap before recording and streaming", async () => { - const ensureAgentLoaded = vi.fn(async () => createCompatibilitySnapshot()); - const { session, agentManager, emitted } = createSessionForOwnershipTests({ - storedRecord: createStoredAgentRecord(), - agentLoadingService: { - ensureAgentLoaded, - }, - }); - - session.resolveAgentIdentifier = vi.fn(async () => ({ ok: true, agentId: "agent-1" })); - session.unarchiveAgentState = vi.fn(async () => true); - session.buildAgentPrompt = vi.fn((text: string) => text); - session.startAgentStream = vi.fn(() => ({ ok: true })); - - await session.handleMessage({ - type: "send_agent_message_request", - requestId: "req-send", - agentId: "agent-1", - text: "hello", - images: [], - messageId: "msg-1", - }); - - expect(ensureAgentLoaded).toHaveBeenCalledWith({ agentId: "agent-1" }); - expect(ensureAgentLoaded.mock.invocationCallOrder[0]).toBeLessThan( - agentManager.recordUserMessage.mock.invocationCallOrder[0], - ); - expect(agentManager.recordUserMessage).toHaveBeenCalledWith("agent-1", "hello", { - messageId: "msg-1", - emitState: false, - }); - expect(session.startAgentStream).toHaveBeenCalledWith("agent-1", "hello"); - expect(emitted).toContainEqual({ - type: "send_agent_message_response", - payload: { - requestId: "req-send", - agentId: "agent-1", - accepted: true, - error: null, - }, - }); - }); - - test("resume_agent_request delegates persistence bootstrap through the compatibility seam", async () => { - const resumeAgent = vi.fn(async () => createCompatibilitySnapshot()); - const { session, emitted } = createSessionForOwnershipTests({ - agentLoadingService: { - resumeAgent, - }, - }); - - session.unarchiveAgentByHandle = vi.fn(async () => undefined); - session.unarchiveAgentState = vi.fn(async () => true); - session.forwardAgentUpdate = vi.fn(async () => undefined); - session.getAgentPayloadById = vi.fn(async () => ({ id: "agent-1" })); - - await session.handleMessage({ - type: "resume_agent_request", - requestId: "req-resume", - handle: { - provider: "codex", - sessionId: "provider-session-1", - }, - overrides: { - model: "gpt-5.4", - }, - }); - - expect(resumeAgent).toHaveBeenCalledWith({ - handle: { - provider: "codex", - sessionId: "provider-session-1", - }, - overrides: { - model: "gpt-5.4", - }, - }); - expect(emitted).toContainEqual({ - type: "status", - payload: expect.objectContaining({ - status: "agent_resumed", - requestId: "req-resume", - agentId: "agent-1", - }), - }); - }); - - test("refresh_agent_request delegates loaded persisted refresh through the compatibility seam", async () => { - const refreshAgent = vi.fn(async () => - createCompatibilitySnapshot({ - persistence: { - provider: "codex", - sessionId: "provider-session-1", - }, - }), - ); - const { session, emitted } = createSessionForOwnershipTests({ - loadedAgent: createCompatibilitySnapshot(), - agentLoadingService: { - refreshAgent, - }, - }); - - session.unarchiveAgentState = vi.fn(async () => true); - session.interruptAgentIfRunning = vi.fn(async () => undefined); - session.forwardAgentUpdate = vi.fn(async () => undefined); - - await session.handleMessage({ - type: "refresh_agent_request", - requestId: "req-refresh-loaded", - agentId: "agent-1", - }); - - expect(session.interruptAgentIfRunning).toHaveBeenCalledWith("agent-1"); - expect(refreshAgent).toHaveBeenCalledWith({ agentId: "agent-1" }); - expect(emitted).toContainEqual({ - type: "status", - payload: { - status: "agent_refreshed", - requestId: "req-refresh-loaded", - agentId: "agent-1", - timelineSize: 0, - }, - }); - }); - - test("refresh_agent_request delegates unloaded persisted refresh through the compatibility seam", async () => { - const refreshAgent = vi.fn(async () => createCompatibilitySnapshot()); - const { session, emitted } = createSessionForOwnershipTests({ - storedRecord: createStoredAgentRecord(), - agentLoadingService: { - refreshAgent, - }, - }); - - session.unarchiveAgentState = vi.fn(async () => true); - session.interruptAgentIfRunning = vi.fn(async () => undefined); - session.forwardAgentUpdate = vi.fn(async () => undefined); - - await session.handleMessage({ - type: "refresh_agent_request", - requestId: "req-refresh-unloaded", - agentId: "agent-1", - }); - - expect(session.interruptAgentIfRunning).not.toHaveBeenCalled(); - expect(refreshAgent).toHaveBeenCalledWith({ agentId: "agent-1" }); - expect(emitted).toContainEqual({ - type: "status", - payload: { - status: "agent_refreshed", - requestId: "req-refresh-unloaded", - agentId: "agent-1", - timelineSize: 0, - }, - }); - }); -}); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index bad1d7b7c..fcc9ffd99 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -92,6 +92,11 @@ import { appendTimelineItemIfAgentKnown, emitLiveTimelineItemIfAgentKnown, } from "./agent/timeline-append.js"; +import { + projectTimelineRows, + selectTimelineWindowByProjectedLimit, + type TimelineProjectionMode, +} from "./agent/timeline-projection.js"; import { DEFAULT_STRUCTURED_GENERATION_PROVIDERS, StructuredAgentFallbackError, @@ -1143,6 +1148,7 @@ 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({ @@ -1705,7 +1711,7 @@ export class Session { break; case "cancel_agent_request": - await this.handleCancelAgentRequest(msg.agentId); + await this.handleCancelAgentRequest(msg.agentId, msg.requestId); break; case "restart_server_request": @@ -1886,7 +1892,7 @@ export class Session { break; case "clear_agent_attention": - await this.handleClearAgentAttention(msg.agentId); + await this.handleClearAgentAttention(msg.agentId, msg.requestId); break; case "client_heartbeat": @@ -3337,11 +3343,23 @@ export class Session { } } - private async handleCancelAgentRequest(agentId: string): Promise { + private async handleCancelAgentRequest(agentId: string, requestId?: string): Promise { this.sessionLogger.info({ agentId }, `Cancel request received for agent ${agentId}`); try { await this.interruptAgentIfRunning(agentId); + if (requestId) { + const agent = this.agentManager.getAgent(agentId); + const payload = agent ? await this.buildAgentPayload(agent) : null; + this.emit({ + type: "cancel_agent_response", + payload: { + requestId, + agentId, + agent: payload, + }, + }); + } } catch (error) { this.handleAgentRunError(agentId, error, "Failed to cancel running agent on request"); } @@ -4005,11 +4023,32 @@ export class Session { /** * Handle clearing agent attention flag */ - private async handleClearAgentAttention(agentId: string | string[]): Promise { + private async handleClearAgentAttention( + agentId: string | string[], + requestId?: string, + ): Promise { const agentIds = Array.isArray(agentId) ? agentId : [agentId]; try { await Promise.all(agentIds.map((id) => this.agentManager.clearAgentAttention(id))); + if (requestId) { + const agents = ( + await Promise.all( + agentIds.map(async (id) => { + const agent = this.agentManager.getAgent(id); + return agent ? this.buildAgentPayload(agent) : null; + }), + ) + ).filter((payload): payload is NonNullable => payload !== null); + this.emit({ + type: "clear_agent_attention_response", + payload: { + requestId, + agentId, + agents, + }, + }); + } } catch (error: any) { this.sessionLogger.error({ err: error, agentIds }, "Failed to clear agent attention"); // Don't throw - this is not critical @@ -6884,10 +6923,17 @@ 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; @@ -6895,46 +6941,86 @@ export class Session { try { const snapshot = await this.ensureAgentLoaded(msg.agentId); const agentPayload = await this.buildAgentPayload(snapshot); - const timeline = await this.agentManager.fetchTimeline(msg.agentId, { + + let timeline = this.agentManager.fetchTimeline(msg.agentId, { direction, cursor, - limit, + limit: + shouldLimitByProjectedWindow && typeof requestedLimit === "number" + ? Math.max(1, Math.floor(requestedLimit)) + : limit, }); - const firstRow = timeline.rows[0]; - const lastRow = timeline.rows[timeline.rows.length - 1]; - const timelineWindow = - timeline.window ?? - (() => { - const minSeq = firstRow?.seq ?? 0; - const maxSeq = lastRow?.seq ?? 0; - return { - minSeq, - maxSeq, - nextSeq: maxSeq > 0 ? maxSeq + 1 : 0, - }; - })(); - const epoch = `compat:${msg.agentId}`; - const startCursor = firstRow ? { seq: firstRow.seq, epoch } : null; - const endCursor = lastRow ? { seq: lastRow.seq, epoch } : null; - const window = { - minSeq: timelineWindow.minSeq, - maxSeq: timelineWindow.maxSeq, - nextSeq: timelineWindow.nextSeq, - startSeq: firstRow?.seq ?? null, - endSeq: lastRow?.seq ?? null, - hasOlder: timeline.hasOlder, - hasNewer: timeline.hasNewer, - }; - const entries = timeline.rows.map((row) => ({ - provider: snapshot.provider, - item: row.item, - timestamp: row.timestamp, - seq: row.seq, - seqStart: row.seq, - seqEnd: row.seq, - sourceSeqRanges: [{ startSeq: row.seq, endSeq: row.seq }], - collapsed: [], - })); + 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); + } this.emit({ type: "fetch_agent_timeline_response", @@ -6943,24 +7029,21 @@ export class Session { agentId: msg.agentId, agent: agentPayload, direction, - startSeq: firstRow?.seq ?? null, - endSeq: lastRow?.seq ?? null, - hasOlder: timeline.hasOlder, - hasNewer: timeline.hasNewer, - entries, - error: null, - epoch, - reset: false, - staleCursor: false, - gap: false, - window, + projection, + epoch: timeline.epoch, + reset: timeline.reset, + staleCursor: timeline.staleCursor, + gap: timeline.gap, + window: timeline.window, startCursor, endCursor, - projection: msg.projection ?? "projected", + hasOlder, + hasNewer, + entries, + error: null, }, }); } catch (error) { - const epoch = `compat:${msg.agentId}`; this.sessionLogger.error( { err: error, agentId: msg.agentId }, "Failed to handle fetch_agent_timeline_request", @@ -6972,28 +7055,18 @@ export class Session { agentId: msg.agentId, agent: null, direction, - startSeq: null, - endSeq: null, + projection, + epoch: "", + reset: false, + staleCursor: false, + gap: false, + window: { minSeq: 0, maxSeq: 0, nextSeq: 0 }, + startCursor: null, + endCursor: null, hasOlder: false, hasNewer: false, entries: [], error: error instanceof Error ? error.message : String(error), - epoch, - reset: false, - staleCursor: false, - gap: false, - window: { - minSeq: 0, - maxSeq: 0, - nextSeq: 0, - startSeq: null, - endSeq: null, - hasOlder: false, - hasNewer: false, - }, - startCursor: null, - endCursor: null, - projection: msg.projection ?? "projected", }, }); } diff --git a/packages/server/src/server/test-utils/fake-agent-client.ts b/packages/server/src/server/test-utils/fake-agent-client.ts index e79fe6b47..bfa99d532 100644 --- a/packages/server/src/server/test-utils/fake-agent-client.ts +++ b/packages/server/src/server/test-utils/fake-agent-client.ts @@ -935,9 +935,6 @@ export function createTestAgentClients(): Record { return { claude: new FakeAgentClient("claude"), codex: new FakeAgentClient("codex"), - gemini: new FakeAgentClient("gemini"), - amp: new FakeAgentClient("amp"), - aider: new FakeAgentClient("aider"), opencode: new FakeAgentClient("opencode"), }; } diff --git a/packages/server/src/shared/messages.stream-parsing.test.ts b/packages/server/src/shared/messages.stream-parsing.test.ts index b645d5d11..c5698c901 100644 --- a/packages/server/src/shared/messages.stream-parsing.test.ts +++ b/packages/server/src/shared/messages.stream-parsing.test.ts @@ -2,7 +2,6 @@ import { describe, expect, it } from "vitest"; import { AgentStreamMessageSchema, - FetchAgentTimelineRequestMessageSchema, FetchAgentTimelineResponseMessageSchema, SessionInboundMessageSchema, SessionOutboundMessageSchema, @@ -18,55 +17,16 @@ describe("shared messages stream parsing", () => { agentId: "agent_live", agent: null, direction: "tail", - startSeq: 1, - endSeq: 2, + 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 }, hasOlder: false, hasNewer: false, - entries: [ - { - provider: "codex", - item: { type: "assistant_message", text: "hello" }, - timestamp: "2026-02-08T20:10:00.000Z", - seq: 2, - }, - ], - error: null, - }, - }); - - expect(parsed.payload.entries).toHaveLength(1); - expect(parsed.payload.entries[0]?.item.type).toBe("assistant_message"); - }); - - it("parses legacy fetch timeline request baggage at the parser boundary", () => { - const parsed = FetchAgentTimelineRequestMessageSchema.parse({ - type: "fetch_agent_timeline_request", - agentId: "agent_live", - requestId: "req-legacy", - direction: "after", - cursor: { - seq: 12, - epoch: "legacy-epoch", - }, - projection: "canonical", - }); - - expect(parsed.cursor?.epoch).toBe("legacy-epoch"); - expect(parsed.projection).toBe("canonical"); - }); - - it("parses legacy fetch timeline response baggage at the parser boundary", () => { - const parsed = FetchAgentTimelineResponseMessageSchema.parse({ - type: "fetch_agent_timeline_response", - payload: { - requestId: "req-1", - agentId: "agent_live", - agent: null, - direction: "tail", - reset: false, - startCursor: { seq: 1, epoch: "legacy-epoch" }, - endCursor: { seq: 2, epoch: "legacy-epoch" }, - projection: "canonical", entries: [ { provider: "codex", @@ -82,10 +42,8 @@ describe("shared messages stream parsing", () => { }, }); - expect(parsed.payload.startSeq).toBe(1); - expect(parsed.payload.endSeq).toBe(2); - expect(parsed.payload.projection).toBe("canonical"); - expect(parsed.payload.entries[0]?.seq).toBe(2); + expect(parsed.payload.entries).toHaveLength(1); + expect(parsed.payload.entries[0]?.item.type).toBe("assistant_message"); }); it("parses explicit shutdown and restart lifecycle request payloads as distinct message types", () => { @@ -112,7 +70,6 @@ describe("shared messages stream parsing", () => { payload: { agentId: "agent_live", timestamp: "2026-02-08T20:10:00.000Z", - seq: 12, event: { type: "timeline", provider: "claude", @@ -140,27 +97,6 @@ describe("shared messages stream parsing", () => { } }); - it("parses legacy agent_stream baggage at the parser boundary", () => { - const parsed = AgentStreamMessageSchema.parse({ - 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.payload.epoch).toBe("legacy-epoch"); - }); - 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 8d72782fa..c61832748 100644 --- a/packages/server/src/shared/messages.ts +++ b/packages/server/src/shared/messages.ts @@ -908,6 +908,7 @@ export const RefreshAgentRequestMessageSchema = z.object({ export const CancelAgentRequestMessageSchema = z.object({ type: z.literal("cancel_agent_request"), agentId: z.string(), + requestId: z.string().optional(), }); export const RestartServerRequestMessageSchema = z.object({ @@ -922,9 +923,8 @@ export const ShutdownServerRequestMessageSchema = z.object({ }); export const AgentTimelineCursorSchema = z.object({ + epoch: z.string(), seq: z.number().int().nonnegative(), - // COMPAT(timeline): retain legacy cursor epoch for older clients. - epoch: z.string().optional(), }); export const FetchAgentTimelineRequestMessageSchema = z.object({ @@ -935,8 +935,8 @@ export const FetchAgentTimelineRequestMessageSchema = z.object({ cursor: AgentTimelineCursorSchema.optional(), // 0 means "all matching rows for this query window". limit: z.number().int().nonnegative().optional(), - // COMPAT(timeline): retain removed projection so older clients can still send it. - projection: z.enum(["canonical", "projected"]).optional(), + // Default should be projected for app timeline loading. + projection: z.enum(["projected", "canonical"]).optional(), }); export const SetAgentModeRequestMessageSchema = z.object({ @@ -1367,6 +1367,7 @@ export const FileDownloadTokenRequestSchema = z.object({ export const ClearAgentAttentionMessageSchema = z.object({ type: z.literal("clear_agent_attention"), agentId: z.union([z.string(), z.array(z.string())]), + requestId: z.string().optional(), }); export const ClientHeartbeatMessageSchema = z.object({ @@ -2036,9 +2037,8 @@ export const AgentStreamMessageSchema = z.object({ agentId: z.string(), event: AgentStreamEventPayloadSchema, timestamp: z.string(), - // Present only for committed timeline events. + // Present for timeline events. Maps 1:1 to canonical in-memory timeline rows. seq: z.number().int().nonnegative().optional(), - // COMPAT(timeline): retain removed epoch for older clients. epoch: z.string().optional(), }), }); @@ -2196,72 +2196,63 @@ export const FetchAgentResponseMessageSchema = z.object({ }), }); -const TimelineProjectionKindSchema = z.enum(["assistant_merge", "tool_lifecycle"]); - -const TimelineSeqRangeSchema = 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().optional(), - // COMPAT(timeline): retain legacy projected timeline fields. - seqStart: z.number().int().nonnegative().optional(), - seqEnd: z.number().int().nonnegative().optional(), - sourceSeqRanges: z.array(TimelineSeqRangeSchema).optional(), - collapsed: z.array(TimelineProjectionKindSchema).optional(), - }) - .transform((entry, ctx) => { - const seq = entry.seq ?? entry.seqEnd ?? entry.seqStart; - if (typeof seq !== "number") { - ctx.addIssue({ - code: z.ZodIssueCode.custom, - message: "Agent timeline entry must include seq or legacy seqStart/seqEnd", - }); - return z.NEVER; - } - - return { - ...entry, - seq, - }; - }); +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"]), - startSeq: z.number().int().nonnegative().nullable().optional(), - endSeq: z.number().int().nonnegative().nullable().optional(), - hasOlder: z.boolean().optional(), - hasNewer: z.boolean().optional(), - entries: z.array(AgentTimelineEntryPayloadSchema), - error: z.string().nullable(), - // COMPAT(timeline): retain legacy response baggage for older clients. - epoch: z.string().optional(), - reset: z.boolean().optional(), - staleCursor: z.boolean().optional(), - gap: z.unknown().optional(), - window: z.unknown().optional(), - startCursor: AgentTimelineCursorSchema.nullable().optional(), - endCursor: AgentTimelineCursorSchema.nullable().optional(), - projection: z.enum(["canonical", "projected"]).optional(), - }) - .transform((payload) => ({ - ...payload, - startSeq: payload.startSeq ?? payload.startCursor?.seq ?? null, - endSeq: payload.endSeq ?? payload.endCursor?.seq ?? null, - hasOlder: payload.hasOlder ?? false, - hasNewer: payload.hasNewer ?? false, - })), + 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 CancelAgentResponseMessageSchema = z.object({ + type: z.literal("cancel_agent_response"), + payload: z.object({ + requestId: z.string(), + agentId: z.string(), + agent: AgentSnapshotPayloadSchema.nullable(), + }), +}); + +export const ClearAgentAttentionResponseMessageSchema = z.object({ + type: z.literal("clear_agent_attention_response"), + payload: z.object({ + requestId: z.string(), + agentId: z.string().or(z.array(z.string())), + agents: z.array(AgentSnapshotPayloadSchema), + }), }); export const SendAgentMessageResponseMessageSchema = z.object({ @@ -2960,6 +2951,8 @@ export const SessionOutboundMessageSchema = z.discriminatedUnion("type", [ ArchiveWorkspaceResponseMessageSchema, FetchAgentResponseMessageSchema, FetchAgentTimelineResponseMessageSchema, + CancelAgentResponseMessageSchema, + ClearAgentAttentionResponseMessageSchema, SendAgentMessageResponseMessageSchema, SetVoiceModeResponseMessageSchema, GetDaemonConfigResponseMessageSchema, @@ -3085,6 +3078,7 @@ export type FetchAgentResponseMessage = z.infer; +export type CancelAgentResponseMessage = z.infer; export type SendAgentMessageResponseMessage = z.infer; export type SetVoiceModeResponseMessage = z.infer; export type SetAgentModeResponseMessage = z.infer; @@ -3256,6 +3250,9 @@ export type FileDownloadTokenResponse = z.infer; export type ShutdownServerRequestMessage = z.infer; export type ClearAgentAttentionMessage = z.infer; +export type ClearAgentAttentionResponseMessage = z.infer< + typeof ClearAgentAttentionResponseMessageSchema +>; export type ClientHeartbeatMessage = z.infer; export type ListCommandsRequest = z.infer; export type ListCommandsResponse = z.infer;