From 64be90766bf5e74813c5b3c41384777fe5b063fd Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 1 Aug 2026 04:57:49 +0200 Subject: [PATCH] fix(timeline): finalize canonical prompt delivery Treat provider identity as a single stored-row transition so legacy clients receive one deliverable canonical prompt. Release identity-pending rows before later foreground output or the current terminal while preserving stale-turn isolation. --- docs/timeline-sync.md | 11 +- packages/app/src/runtime/host-runtime.test.ts | 55 ++--- packages/app/src/runtime/host-runtime.ts | 8 +- .../src/messages.client-capabilities.test.ts | 19 -- .../src/server/agent/agent-manager.test.ts | 200 ++++++++++------ .../server/src/server/agent/agent-manager.ts | 160 ++++++++----- .../server/agent/agent-timeline-store.test.ts | 19 +- .../src/server/agent/agent-timeline-store.ts | 20 +- .../server/agent/submitted-prompt-delivery.ts | 75 +++--- packages/server/src/server/session.test.ts | 225 ++++-------------- packages/server/src/server/session.ts | 75 +++--- .../server/src/server/wire-compat.test.ts | 142 +++++++++-- 12 files changed, 520 insertions(+), 489 deletions(-) delete mode 100644 packages/protocol/src/messages.client-capabilities.test.ts diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index d9ae2dfcdf..8d2867fe1e 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -181,13 +181,10 @@ original sequence; clients apply it only to the already-covered row with the sam Clients that understand these same-sequence updates advertise the `canonical_submitted_prompt_revisions` client capability. For older destinations, the daemon keeps -the canonical row globally but withholds a rewind-capable foreground prompt until provider identity -arrives, then delivers the enriched row once at its original sequence. Timeline fetches apply the -same destination projection: a trailing identity-pending suffix is temporarily outside the visible -window, while daemon-handled prompts and providers without the rewind identity contract remain -immediately visible. -An immediately interrupted turn remains -canonical even when the provider never emits its echo. +the canonical row globally but withholds an identity-awaiting foreground prompt until provider identity +arrives or the turn produces later output or a terminal. Finalization optionally adds provider +identity, removes the internal delivery marker, and redispatches the same stored sequence. Timeline +fetches hide only that trailing marked row; daemon-handled prompts are never marked. Content matching is limited to the dated compatibility path for daemon timelines created before that field existed. Canonical ingestion may match only an explicit unreconciled local candidate; the draft-create handoff is the one boundary that also permits the legacy canonical twin to have diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 56463faa11..9773e02e8e 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -15,7 +15,6 @@ import { normalizeAgentSnapshot } from "@/utils/agent-snapshots"; import { isAgentArchiving, setAgentArchiving } from "@/hooks/use-archive-agent"; import { queryClient } from "@/data/query-client"; import { - APP_CLIENT_CAPABILITIES, HostRuntimeController, HostRuntimeStore, readInitialDaemonConnectionHint, @@ -24,10 +23,17 @@ import { } from "./host-runtime"; import { ReplicaCache } from "./replica-cache"; -it("advertises canonical submitted prompt revision support", () => { - expect(APP_CLIENT_CAPABILITIES).toMatchObject({ - [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true, - }); +const capturedDefaultClientOptions = vi.hoisted(() => [] as Array>); + +vi.mock("@getpaseo/client/internal/daemon-client", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + DaemonClient: function DaemonClientDouble(options: Record) { + capturedDefaultClientOptions.push(options); + return new FakeDaemonClient(); + }, + }; }); class FakeDaemonClient { @@ -566,36 +572,25 @@ describe("HostRuntimeController", () => { expect(controller.getSnapshot().agentDirectoryStatus).toBe("initial_loading"); }); - it("passes resolved client id into created active clients", async () => { - const host = makeHost({ - connections: [ - { - id: "direct:lan:6767", - type: "directTcp", - endpoint: "lan:6767", - }, - ], - }); - const seenClientIds: string[] = []; - const fakeClient = new FakeDaemonClient(); - const controller = new HostRuntimeController({ - host, - deps: { - createClient: ({ clientId }) => { - seenClientIds.push(clientId); - return fakeClient as unknown as DaemonClient; - }, - connectToDaemon: async () => { - throw new Error("probe unavailable"); - }, - getClientId: async () => "cid_runtime_stable", + it("passes resolved client configuration into created active clients", async () => { + capturedDefaultClientOptions.length = 0; + const values = new Map([["@paseo:client-id-v1", "cid_runtime_stable"]]); + (globalThis as { window?: unknown }).window = { + localStorage: { + getItem: (key: string) => values.get(key) ?? null, + setItem: (key: string, value: string) => values.set(key, value), }, - }); + }; + const controller = new HostRuntimeController({ host: makeHost() }); await controller.activateConnection({ connectionId: "direct:lan:6767" }); - expect(seenClientIds).toEqual(["cid_runtime_stable"]); + expect(capturedDefaultClientOptions[0]).toMatchObject({ + clientId: "cid_runtime_stable", + capabilities: { [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true }, + }); expect(controller.getSnapshot().connectionStatus).toBe("online"); + await controller.stop(); }); it("keeps browser client lifecycle tied to the active host runtime client", async () => { diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 5abe672a18..11cb3e4829 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -456,11 +456,6 @@ function probeIntervalForConnection( return PROBE_MAX_BACKOFF_MS; } -export const APP_CLIENT_CAPABILITIES = { - [CLIENT_CAPS.selectiveAgentTimeline]: true, - [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true, -} as const; - function createDefaultDeps(): HostRuntimeControllerDeps { const browserHostAvailable = typeof getDesktopHost()?.browser?.executeAutomationCommand === "function"; @@ -473,7 +468,8 @@ function createDefaultDeps(): HostRuntimeControllerDeps { } : undefined; const appCapabilities = { - ...APP_CLIENT_CAPABILITIES, + [CLIENT_CAPS.selectiveAgentTimeline]: true, + [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true, ...browserAutomationCapabilities, }; diff --git a/packages/protocol/src/messages.client-capabilities.test.ts b/packages/protocol/src/messages.client-capabilities.test.ts deleted file mode 100644 index c0e78466b2..0000000000 --- a/packages/protocol/src/messages.client-capabilities.test.ts +++ /dev/null @@ -1,19 +0,0 @@ -import { describe, expect, it } from "vitest"; -import { CLIENT_CAPS } from "./client-capabilities.js"; -import { WSHelloMessageSchema } from "./messages.js"; - -describe("client capabilities", () => { - it("accepts canonical submitted prompt revision support in hello", () => { - const hello = WSHelloMessageSchema.parse({ - type: "hello", - clientId: "client-capability-test", - clientType: "mobile", - protocolVersion: 1, - capabilities: { - [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true, - }, - }); - - expect(hello.capabilities?.[CLIENT_CAPS.canonicalSubmittedPromptRevisions]).toBe(true); - }); -}); diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 99e5173d4c..53c3951bf8 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -4633,6 +4633,7 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the const interruptStarted = deferred(); const allowInterruptToFinish = deferred(); const allowSecondStartToResolve = deferred(); + const allowSecondTurnToComplete = deferred(); let capturedSession: StaleReplacementSession | null = null; class StaleReplacementSession extends TestAgentSession { @@ -4651,8 +4652,9 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the secondStartEntered.resolve(); await allowSecondStartToResolve.promise; - setTimeout(() => { + setTimeout(async () => { this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); + await allowSecondTurnToComplete.promise; this.pushEvent({ type: "turn_completed", provider: this.provider, turnId }); }, 0); return { turnId }; @@ -4709,7 +4711,9 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the { agentId: snapshot.id, replayState: false }, ); - const firstRun = manager.streamAgent(snapshot.id, "first run"); + const firstRun = manager.streamAgent(snapshot.id, "first run", { + clientMessageId: "first-client-message", + }); const firstRunDrain = (async () => { for await (const _event of firstRun) { // Drain events so lifecycle updates are applied. @@ -4720,7 +4724,9 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the const replaceUpdatesStart = stateUpdates.length; const beforeReplaceUpdatedAt = manager.getAgent(snapshot.id)?.updatedAt.getTime() ?? 0; - const secondRunPromise = manager.replaceAgentRun(snapshot.id, "replacement run"); + const secondRunPromise = manager.replaceAgentRun(snapshot.id, "replacement run", { + clientMessageId: "replacement-client-message", + }); await interruptStarted.promise; const replacementUpdates = stateUpdates.slice(replaceUpdatesStart); @@ -4766,6 +4772,18 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the allowSecondStartToResolve.resolve(); await replacementStart; + capturedSession!.pushEvent({ type: "turn_completed", provider: "codex", turnId: "turn-1" }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect( + manager + .fetchTimeline(snapshot.id, { direction: "tail", limit: 0 }) + .rows.find( + (row) => + row.item.type === "user_message" && + row.item.clientMessageId === "replacement-client-message", + )?.delivery, + ).toBe("awaiting_provider_identity"); + allowSecondTurnToComplete.resolve(); await firstRunDrain; await secondRunDrain; unsubscribe(); @@ -8154,17 +8172,19 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a const text = typeof prompt === "string" ? prompt : ""; const publishProviderIdentity = () => { this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); - this.pushEvent({ - type: "timeline", - provider: this.provider, - turnId, - item: { - type: "user_message", - text, - messageId: `provider-message-${this.turnCount}`, - clientMessageId: options?.clientMessageId, - }, - }); + if (this.turnCount !== 3 && this.turnCount !== 5) { + this.pushEvent({ + type: "timeline", + provider: this.provider, + turnId, + item: { + type: "user_message", + text, + ...(this.turnCount === 4 ? {} : { messageId: `provider-message-${this.turnCount}` }), + clientMessageId: options?.clientMessageId, + }, + }); + } if (this.turnCount === 1) { this.pushEvent({ type: "timeline", @@ -8172,11 +8192,18 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a turnId, item: { type: "assistant_message", text: "provider output after identity" }, }); + } else if (this.turnCount === 3) { + this.pushEvent({ + type: "timeline", + provider: this.provider, + turnId, + item: { type: "assistant_message", text: "provider output without echo" }, + }); } this.pushEvent({ type: "turn_completed", provider: this.provider, turnId }); }; - if (this.turnCount === 1) setTimeout(publishProviderIdentity, 0); - else publishProviderIdentity(); + if (this.turnCount === 2) publishProviderIdentity(); + else setTimeout(publishProviderIdentity, 0); return { turnId }; } } @@ -8195,6 +8222,26 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a }); const events: AgentManagerEvent[] = []; manager.subscribe((event) => events.push(event), { replayState: false }); + const streamEvents = () => + events.flatMap((event) => { + if (event.type !== "agent_stream") return []; + if (event.event.type !== "timeline") return [{ type: event.event.type }]; + const item = event.event.item; + return [ + { + type: item.type, + seq: event.seq, + delivery: event.delivery, + ...(item.type === "user_message" + ? { + text: item.text, + clientMessageId: item.clientMessageId, + messageId: item.messageId, + } + : {}), + }, + ]; + }); try { const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { @@ -8216,95 +8263,91 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a { clientMessageId: "msg-client-1" }, ); - const liveEvents = events.filter( - (event): event is Extract => - event.type === "agent_stream" && event.agentId === snapshot.id, - ); - expect( - liveEvents.map((event) => - event.event.type === "timeline" - ? { - type: event.event.type, - item: event.event.item, - seq: event.seq, - delivery: event.delivery, - } - : { type: event.event.type }, - ), - ).toEqual([ + expect(streamEvents()).toEqual([ { type: "turn_started" }, { - type: "timeline", - item: { - type: "user_message", - text: "hello from composer", - clientMessageId: "msg-client-1", - }, + type: "user_message", seq: 1, delivery: "awaiting_provider_identity", + text: "hello from composer", + clientMessageId: "msg-client-1", + messageId: undefined, }, { - type: "timeline", - item: { - type: "user_message", - text: "hello from composer", - messageId: "provider-message-1", - clientMessageId: "msg-client-1", - }, + type: "user_message", seq: 1, - delivery: "awaiting_provider_identity", - }, - { - type: "timeline", - item: { type: "assistant_message", text: "provider output after identity" }, - seq: 2, delivery: undefined, + text: "hello from composer", + clientMessageId: "msg-client-1", + messageId: "provider-message-1", }, + { type: "assistant_message", seq: 2, delivery: undefined }, { type: "turn_completed" }, ]); const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows; - expect(timeline).toEqual([ - { - seq: 1, - timestamp: expect.any(String), - delivery: "awaiting_provider_identity", - item: { + expect(timeline.map((row) => [row.seq, row.delivery, row.item])).toEqual([ + [ + 1, + undefined, + { type: "user_message", text: "hello from composer", messageId: "provider-message-1", clientMessageId: "msg-client-1", }, - }, - { - seq: 2, - timestamp: expect.any(String), - item: { type: "assistant_message", text: "provider output after identity" }, - }, + ], + [2, undefined, { type: "assistant_message", text: "provider output after identity" }], ]); events.length = 0; await manager.runAgent(snapshot.id, "staged identity", { clientMessageId: "msg-client-2", }); - expect( - events.flatMap((event) => - event.type === "agent_stream" && event.event.type === "timeline" - ? [{ seq: event.seq, delivery: event.delivery, item: event.event.item }] - : [], - ), - ).toEqual([ + expect(streamEvents().filter((event) => event.type === "user_message")).toEqual([ { + type: "user_message", seq: 3, delivery: undefined, - item: { - type: "user_message", - text: "staged identity", - messageId: "provider-message-2", - clientMessageId: "msg-client-2", - }, + text: "staged identity", + messageId: "provider-message-2", + clientMessageId: "msg-client-2", }, ]); + + events.length = 0; + await manager.runAgent(snapshot.id, "output without echo", { + clientMessageId: "msg-client-3", + }); + expect(streamEvents()).toEqual([ + { type: "turn_started" }, + expect.objectContaining({ + type: "user_message", + seq: 4, + delivery: "awaiting_provider_identity", + }), + expect.objectContaining({ type: "user_message", seq: 4, delivery: undefined }), + { type: "assistant_message", seq: 5, delivery: undefined }, + { type: "turn_completed" }, + ]); + + for (const [prompt, clientMessageId, seq] of [ + ["echo without identity", "msg-client-4", 6], + ["terminal without echo", "msg-client-5", 7], + ] as const) { + events.length = 0; + await manager.runAgent(snapshot.id, prompt, { clientMessageId }); + expect(streamEvents()).toEqual([ + { type: "turn_started" }, + expect.objectContaining({ + type: "user_message", + seq, + delivery: "awaiting_provider_identity", + }), + expect.objectContaining({ type: "user_message", seq, delivery: undefined }), + { type: "turn_completed" }, + ]); + } } finally { await manager.flush().catch(() => undefined); await storage.flush().catch(() => undefined); @@ -8319,6 +8362,11 @@ test("authoritative timeline records a daemon-handled submitted prompt before it const commandCompleted = deferred(); class DaemonHandledPromptSession extends TestAgentSession { + override readonly capabilities = { + ...TEST_CAPABILITIES, + supportsRewindConversation: true, + }; + override tryHandleOutOfBand(prompt: AgentPromptInput) { if (prompt !== "/handled") return null; return { diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 6698589944..f068757ece 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -2081,14 +2081,13 @@ export class AgentManager { ) : undefined; if (options?.clientMessageId) { - this.recordSubmittedPrompt( - agent, - prompt, - options.clientMessageId, - stagedSubmittedPromptEcho?.item.type === "user_message" - ? stagedSubmittedPromptEcho.item.messageId - : undefined, - ); + this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, { + awaitProviderIdentity: true, + messageId: + stagedSubmittedPromptEcho?.item.type === "user_message" + ? stagedSubmittedPromptEcho.item.messageId + : undefined, + }); } for (const stagedEvent of pendingRun.stagedEvents.splice(0)) { const isAcceptedTurnStart = @@ -3432,6 +3431,13 @@ export class AgentManager { return false; } + this.finalizeSubmittedPromptBeforeTimeline( + agent, + event, + options?.fromHistory === true, + isForegroundEvent, + ); + // Only update timestamp for live events, not history replay if (!options?.fromHistory) { this.touchUpdatedAt(agent); @@ -3442,12 +3448,13 @@ export class AgentManager { this.agentStreamCoalescer.flushFor(agent.id); } - let terminalDisposition: ActiveTurnTerminalDisposition = "untracked"; - if (isTurnTerminalEvent(event)) { - terminalDisposition = options?.fromHistory - ? "stale" - : this.applyActiveTurnTerminal(agent, eventTurnId); - } + const terminalDisposition = this.applyStreamTerminalDisposition( + agent, + event, + eventTurnId, + options?.fromHistory === true, + isForegroundEvent, + ); const flags: StreamEventFlags = { shouldDispatchEvent: true, shouldNotifyWaiters: true }; @@ -3501,6 +3508,21 @@ export class AgentManager { ); } + private applyStreamTerminalDisposition( + agent: ActiveManagedAgent, + event: AgentStreamEvent, + turnId: string | undefined, + fromHistory: boolean, + isForegroundEvent: boolean, + ): ActiveTurnTerminalDisposition { + if (!isTurnTerminalEvent(event)) return "untracked"; + const disposition = fromHistory ? "stale" : this.applyActiveTurnTerminal(agent, turnId); + if (isForegroundEvent && disposition !== "stale") { + this.finalizeSubmittedPrompt(agent, event); + } + return disposition; + } + private traceCoalescerBuffered( agent: ActiveManagedAgent, event: AgentStreamEvent, @@ -3896,8 +3918,9 @@ export class AgentManager { item: AgentTimelineItem, provider: AgentProvider, turnId?: string, + delivery?: AgentTimelineRow["delivery"], ): AgentStreamEvent { - const row = this.recordTimeline(agentId, item); + const row = this.recordTimeline(agentId, item, delivery ? { delivery } : undefined); const event: AgentStreamEvent = { type: "timeline", item, @@ -3908,6 +3931,7 @@ export class AgentManager { seq: row.seq, epoch: this.timelineStore.getEpoch(agentId), timestamp: row.timestamp, + ...(row.delivery ? { delivery: row.delivery } : {}), }); if ( @@ -3929,7 +3953,7 @@ export class AgentManager { agent: ActiveManagedAgent, prompt: AgentPromptInput, clientMessageId: string, - messageId?: string, + options?: { awaitProviderIdentity: boolean; messageId: string | undefined }, ): void { if (this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId)) { return; @@ -3940,27 +3964,13 @@ export class AgentManager { type: "user_message", text: submittedPromptText(prompt), clientMessageId, - ...(messageId ? { messageId } : {}), + ...(options?.messageId ? { messageId: options.messageId } : {}), }; - const capabilities = agent.session.capabilities; const delivery = - messageId === undefined && - (capabilities.supportsRewindConversation || - capabilities.supportsRewindFiles || - capabilities.supportsRewindBoth) + options?.awaitProviderIdentity === true && options.messageId === undefined ? AWAITING_PROVIDER_IDENTITY_DELIVERY : undefined; - const row = this.recordTimeline(agent.id, item, delivery ? { delivery } : undefined); - this.dispatchStream( - agent.id, - { type: "timeline", item, provider: agent.provider }, - { - seq: row.seq, - epoch: this.timelineStore.getEpoch(agent.id), - timestamp: row.timestamp, - ...(row.delivery ? { delivery: row.delivery } : {}), - }, - ); + this.recordAndDispatchTimelineItem(agent.id, item, agent.provider, undefined, delivery); } private reconcileSubmittedPromptEcho( @@ -3971,31 +3981,69 @@ export class AgentManager { const { clientMessageId, messageId } = item; if (!clientMessageId) return null; const existing = this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId); - if ( - !existing || - existing.item.type !== "user_message" || - !messageId || - existing.item.messageId === messageId - ) { - return existing; + if (!existing || existing.item.type !== "user_message") return null; + this.finalizeSubmittedPrompt(agent, event, clientMessageId, messageId); + return existing; + } + + private finalizeSubmittedPromptBeforeTimeline( + agent: ActiveManagedAgent, + event: AgentStreamEvent, + fromHistory: boolean, + isForegroundEvent: boolean, + ): void { + if (fromHistory || !isForegroundEvent || event.type !== "timeline") return; + const item = event.item; + if (item.type === "user_message") { + if (isSystemInjectedEnvelope(item.text)) return; + if ( + item.clientMessageId && + this.timelineStore.getSubmittedUserMessage(agent.id, item.clientMessageId) + ) { + return; + } } - const row = this.timelineStore.enrichSubmittedUserMessage(agent.id, clientMessageId, { - messageId, - }); - if (row) { - this.enqueueDurableTimelineUpdate(agent.id, row); - this.dispatchStream( - agent.id, - { ...event, item: row.item }, - { - seq: row.seq, - epoch: this.timelineStore.getEpoch(agent.id), - timestamp: row.timestamp, - ...(row.delivery ? { delivery: row.delivery } : {}), - }, + this.finalizeSubmittedPrompt(agent, event); + } + + private finalizeSubmittedPrompt( + agent: ActiveManagedAgent, + event: AgentStreamEvent, + correlatedClientMessageId?: string, + messageId?: string, + ): void { + const pending = this.timelineStore + .getRows(agent.id) + .findLast( + (row) => + row.delivery === AWAITING_PROVIDER_IDENTITY_DELIVERY && row.item.type === "user_message", ); - } - return row; + const clientMessageId = + correlatedClientMessageId ?? + (pending?.item.type === "user_message" ? pending.item.clientMessageId : undefined); + if (!clientMessageId) return; + const row = this.timelineStore.finalizeSubmittedUserMessage( + agent.id, + clientMessageId, + messageId ? { messageId } : undefined, + ); + if (!row) return; + this.enqueueDurableTimelineUpdate(agent.id, row); + const turnId = getAgentStreamEventTurnId(event); + this.dispatchStream( + agent.id, + { + type: "timeline", + item: row.item, + provider: event.provider, + ...(turnId ? { turnId } : {}), + }, + { + seq: row.seq, + epoch: this.timelineStore.getEpoch(agent.id), + timestamp: row.timestamp, + }, + ); } private async appendSystemErrorTimelineMessage( diff --git a/packages/server/src/server/agent/agent-timeline-store.test.ts b/packages/server/src/server/agent/agent-timeline-store.test.ts index b6afa939b2..8668cf3b23 100644 --- a/packages/server/src/server/agent/agent-timeline-store.test.ts +++ b/packages/server/src/server/agent/agent-timeline-store.test.ts @@ -2,16 +2,20 @@ import { describe, expect, it } from "vitest"; import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js"; describe("InMemoryAgentTimelineStore", () => { - it("enriches a submitted prompt without allocating a second sequence", () => { + it("finalizes a submitted prompt without allocating a second sequence", () => { const store = new InMemoryAgentTimelineStore(); store.initialize("agent-1", { epoch: "epoch-1" }); - store.append("agent-1", { - type: "user_message", - text: "Accepted prompt", - clientMessageId: "client-message-1", - }); + store.append( + "agent-1", + { + type: "user_message", + text: "Accepted prompt", + clientMessageId: "client-message-1", + }, + { delivery: "awaiting_provider_identity" }, + ); - const updated = store.enrichSubmittedUserMessage("agent-1", "client-message-1", { + const updated = store.finalizeSubmittedUserMessage("agent-1", "client-message-1", { messageId: "provider-message-1", }); @@ -24,6 +28,7 @@ describe("InMemoryAgentTimelineStore", () => { messageId: "provider-message-1", }, }); + expect(updated?.delivery).toBeUndefined(); expect(store.getRows("agent-1")).toEqual([updated]); expect(store.fetch("agent-1", { direction: "tail", limit: 1 }).window).toEqual({ minSeq: 1, diff --git a/packages/server/src/server/agent/agent-timeline-store.ts b/packages/server/src/server/agent/agent-timeline-store.ts index 2ddfea30e6..93938803ff 100644 --- a/packages/server/src/server/agent/agent-timeline-store.ts +++ b/packages/server/src/server/agent/agent-timeline-store.ts @@ -176,20 +176,28 @@ export class InMemoryAgentTimelineStore { return row ? cloneRow(row) : null; } - enrichSubmittedUserMessage( + finalizeSubmittedUserMessage( agentId: string, clientMessageId: string, - enrichment: { messageId: string }, + identity?: { messageId: string }, ): AgentTimelineRow | null { const state = this.requireState(agentId); - const row = state.rows.find( + const index = state.rows.findIndex( (candidate) => candidate.item.type === "user_message" && candidate.item.clientMessageId === clientMessageId, ); - if (!row || row.item.type !== "user_message") return null; - row.item = { ...row.item, messageId: enrichment.messageId }; - return cloneRow(row); + const row = state.rows[index]; + if (!row || row.item.type !== "user_message" || row.delivery !== "awaiting_provider_identity") { + return null; + } + const finalized: AgentTimelineRow = { + seq: row.seq, + timestamp: row.timestamp, + item: identity ? { ...row.item, messageId: identity.messageId } : row.item, + }; + state.rows[index] = finalized; + return cloneRow(finalized); } getEpoch(agentId: string): string { diff --git a/packages/server/src/server/agent/submitted-prompt-delivery.ts b/packages/server/src/server/agent/submitted-prompt-delivery.ts index ab5f5d0044..1372f91645 100644 --- a/packages/server/src/server/agent/submitted-prompt-delivery.ts +++ b/packages/server/src/server/agent/submitted-prompt-delivery.ts @@ -1,63 +1,50 @@ import type { AgentTimelineItem } from "./agent-sdk-types.js"; import type { AgentTimelineFetchResult, - AgentTimelineRow, AgentTimelineRowDelivery, } from "./agent-timeline-store-types.js"; export const AWAITING_PROVIDER_IDENTITY_DELIVERY = "awaiting_provider_identity" as const; -export function shouldDeliverSubmittedPromptRow(input: { - delivery: AgentTimelineRowDelivery | undefined; - item: AgentTimelineItem; - supportsRevisions: boolean; -}): boolean { +export function shouldDeliverSubmittedPromptRow( + delivery: AgentTimelineRowDelivery | undefined, + item: AgentTimelineItem, + supportsRevisions: boolean, +): boolean { return ( - input.supportsRevisions || - input.delivery !== AWAITING_PROVIDER_IDENTITY_DELIVERY || - input.item.type !== "user_message" || - input.item.messageId !== undefined + supportsRevisions || + delivery !== AWAITING_PROVIDER_IDENTITY_DELIVERY || + item.type !== "user_message" || + item.messageId !== undefined ); } -export function findSubmittedPromptVisibleMaxSeq( - rows: readonly AgentTimelineRow[], - maxSeq: number, -): number { - let visibleMaxSeq = maxSeq; - for (let index = rows.length - 1; index >= 0; index -= 1) { - const row = rows[index]; - if ( - !row || - row.seq !== visibleMaxSeq || - shouldDeliverSubmittedPromptRow({ - delivery: row.delivery, - item: row.item, - supportsRevisions: false, - }) - ) { - break; - } - visibleMaxSeq -= 1; +export function projectSubmittedPromptTimeline( + timeline: AgentTimelineFetchResult, + fullTimeline = timeline, + cursorSeq = 0, +): AgentTimelineFetchResult { + const pending = fullTimeline.rows.at(-1); + if ( + !pending || + pending.seq !== fullTimeline.window.maxSeq || + shouldDeliverSubmittedPromptRow(pending.delivery, pending.item, false) + ) { + return timeline; } - return visibleMaxSeq; -} - -export function projectSubmittedPromptTimeline(input: { - timeline: AgentTimelineFetchResult; - visibleMaxSeq: number; - cursorSeq?: number; -}): AgentTimelineFetchResult { - const rows = input.timeline.rows.filter((row) => row.seq <= input.visibleMaxSeq); - const selectedEndSeq = rows.at(-1)?.seq ?? input.cursorSeq ?? 0; + const visibleMaxSeq = pending.seq - 1; + const rows = timeline.rows.filter((row) => row.seq <= visibleMaxSeq); + const selectedEndSeq = rows.at(-1)?.seq ?? cursorSeq; return { - ...input.timeline, + ...timeline, window: { - ...input.timeline.window, - maxSeq: input.visibleMaxSeq, - nextSeq: input.visibleMaxSeq + 1, + ...timeline.window, + minSeq: + visibleMaxSeq === 0 || timeline.window.minSeq > visibleMaxSeq ? 0 : timeline.window.minSeq, + maxSeq: visibleMaxSeq, + nextSeq: visibleMaxSeq + 1, }, - hasNewer: input.timeline.hasNewer && selectedEndSeq < input.visibleMaxSeq, + hasNewer: timeline.hasNewer && selectedEndSeq < visibleMaxSeq, rows, }; } diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index eef73462c2..71c0928268 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -22,11 +22,7 @@ import { isSessionRpcAllowed, Session } from "./session.js"; import { DownloadTokenStore } from "./file-download/token-store.js"; import { StructuredAgentFallbackError } from "./agent/agent-response-loop.js"; import type { StoredAgentRecord } from "./agent/agent-storage.js"; -import type { - AgentManagerEvent, - AgentTimelineFetchResult, - ManagedAgent, -} from "./agent/agent-manager.js"; +import type { AgentManagerEvent } from "./agent/agent-manager.js"; import type { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js"; import { createPersistedProjectRecord } from "./workspace-registry.js"; import { deriveProjectKey } from "./project-key.js"; @@ -4887,11 +4883,9 @@ test("keeps selective delivery scoped per socket when a retained session also ha test("projects canonical submitted prompt revisions per retained socket capability", async () => { const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = []; const agentEventListeners: Array<(event: AgentManagerEvent) => void> = []; - const rewind = vi.fn().mockResolvedValue(undefined); const session = createSessionForTest({ targetedMessages, agentManager: { - rewind, subscribe: vi.fn((listener: (event: AgentManagerEvent) => void) => { agentEventListeners.push(listener); return () => {}; @@ -4939,7 +4933,6 @@ test("projects canonical submitted prompt revisions per retained socket capabili }, seq: 5, epoch: "epoch-1", - delivery: "awaiting_provider_identity", }); listener({ type: "agent_stream", @@ -4956,7 +4949,6 @@ test("projects canonical submitted prompt revisions per retained socket capabili }, seq: 6, epoch: "epoch-1", - delivery: "awaiting_provider_identity", }); listener({ type: "agent_stream", @@ -5004,184 +4996,61 @@ test("projects canonical submitted prompt revisions per retained socket capabili { seq: 6, clientMessageId: "client-message-2", messageId: "provider-message-2" }, { seq: 7, clientMessageId: undefined, messageId: undefined }, ]); - - await session.handleMessage( - { - type: "agent.rewind.request", - requestId: "rewind-provider-identity", - agentId: "agent-revisions", - messageId: "provider-message-1", - mode: "conversation", - }, - legacySocket, - ); - expect(rewind).toHaveBeenCalledWith("agent-revisions", "provider-message-1", "conversation"); }); -test("projects a coherent legacy fetch window around pending provider identity", async () => { - const source = {}; - const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = []; - const timestamp = "2026-08-01T00:00:00.000Z"; - const agent = { - id: "agent-fetch-revisions", - provider: "codex", - cwd: "/tmp/agent-fetch-revisions", - capabilities: { - supportsStreaming: true, - supportsSessionPersistence: true, - supportsDynamicModes: false, - supportsMcpServers: false, - supportsReasoningStream: false, - supportsToolInvocations: false, - supportsRewindConversation: true, - }, - config: { provider: "codex", cwd: "/tmp/agent-fetch-revisions" }, - runtimeInfo: { provider: "codex", sessionId: "session-fetch-revisions" }, - createdAt: new Date(timestamp), - updatedAt: new Date(timestamp), - availableModes: [], - currentModeId: null, - pendingPermissions: new Map(), - persistence: null, - historyPrimed: true, - lastUserMessageAt: new Date(timestamp), - activeTurnId: null, - activeTurnStartedAt: null, - lastUsage: undefined, - attention: { requiresAttention: false }, - labels: {}, - lifecycle: "idle", - activeForegroundTurnId: null, - } as unknown as ManagedAgent; - let timeline: AgentTimelineFetchResult = { - epoch: "epoch-fetch-revisions", - direction: "tail", - reset: false, - staleCursor: false, - gap: false, - window: { minSeq: 1, maxSeq: 5, nextSeq: 6 }, - hasOlder: true, - hasNewer: false, - rows: [ - { - seq: 4, - timestamp, - item: { type: "assistant_message", text: "before submitted prompt" }, - }, - { - seq: 5, - timestamp, - delivery: "awaiting_provider_identity", - item: { - type: "user_message", - text: "submitted prompt", - clientMessageId: "client-message-fetch", - }, - }, - ], - }; +test.each([ + [null, ["provider-message-1"]], + [{ [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true }, [undefined, "provider-message-1"]], +])("uses the negotiated fallback capability for prompt revisions", (capabilities, expectedIds) => { + const messages: SessionOutboundMessage[] = []; + let listener: ((event: AgentManagerEvent) => void) | undefined; const session = createSessionForTest({ - targetedMessages, + messages, agentManager: { - waitForAgentClose: vi.fn().mockResolvedValue(undefined), - getAgent: vi.fn(() => agent), - fetchTimeline: vi.fn(() => timeline), + subscribe: vi.fn((next: (event: AgentManagerEvent) => void) => { + listener = next; + return () => {}; + }), }, }); - session.updateClientCapabilities(null, source); - - await session.handleMessage( - { - type: "fetch_agent_timeline_request", - requestId: "fetch-before-provider-identity", - agentId: agent.id, - direction: "tail", - projection: "canonical", - limit: 20, - }, - source, - ); - - const before = targetedMessages.find( - ({ message }) => - message.type === "fetch_agent_timeline_response" && - message.payload.requestId === "fetch-before-provider-identity", - )?.message; - expect(before).toMatchObject({ - type: "fetch_agent_timeline_response", - payload: { - requestId: "fetch-before-provider-identity", - error: null, - window: { minSeq: 1, maxSeq: 4, nextSeq: 5 }, - endCursor: { epoch: "epoch-fetch-revisions", seq: 4 }, - hasNewer: false, - entries: [ - { - seqStart: 4, - seqEnd: 4, - item: { type: "assistant_message", text: "before submitted prompt" }, - }, - ], - }, - }); - - timeline = { - ...timeline, - rows: [ - timeline.rows[0]!, - { - ...timeline.rows[1]!, - item: { - type: "user_message", - text: "submitted prompt", - clientMessageId: "client-message-fetch", - messageId: "provider-message-fetch", - }, - }, - ], + session.updateClientCapabilities(capabilities); + const item = { + type: "user_message" as const, + text: "submitted prompt", + clientMessageId: "client-message-1", }; - targetedMessages.length = 0; - await session.handleMessage( - { - type: "fetch_agent_timeline_request", - requestId: "fetch-after-provider-identity", - agentId: agent.id, - direction: "tail", - projection: "canonical", - limit: 20, - }, - source, - ); - - const after = targetedMessages.find( - ({ message }) => - message.type === "fetch_agent_timeline_response" && - message.payload.requestId === "fetch-after-provider-identity", - )?.message; - expect(after).toMatchObject({ - type: "fetch_agent_timeline_response", - payload: { - requestId: "fetch-after-provider-identity", - error: null, - window: { minSeq: 1, maxSeq: 5, nextSeq: 6 }, - endCursor: { epoch: "epoch-fetch-revisions", seq: 5 }, - hasNewer: false, - entries: [ - { seqStart: 4, seqEnd: 4 }, - { - seqStart: 5, - seqEnd: 5, - item: { - type: "user_message", - text: "submitted prompt", - clientMessageId: "client-message-fetch", - messageId: "provider-message-fetch", - }, - }, - ], - }, + listener?.({ + type: "agent_stream", + agentId: "agent-revisions", + event: { type: "timeline", provider: "codex", item }, + seq: 1, + epoch: "epoch-1", + delivery: "awaiting_provider_identity", }); + listener?.({ + type: "agent_stream", + agentId: "agent-revisions", + event: { + type: "timeline", + provider: "codex", + item: { ...item, messageId: "provider-message-1" }, + }, + seq: 1, + epoch: "epoch-1", + }); + + expect( + messages.flatMap((message) => + message.type === "agent_stream" && message.payload.event.type === "timeline" + ? [ + message.payload.event.item.type === "user_message" + ? message.payload.event.item.messageId + : undefined, + ] + : [], + ), + ).toEqual(expectedIds); }); test("sends project updates only to capable sockets in a retained session", () => { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 7e1484a97e..13a6026db6 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -77,7 +77,6 @@ import type { ManagedAgent, } from "./agent/agent-manager.js"; import { - findSubmittedPromptVisibleMaxSeq, projectSubmittedPromptTimeline, shouldDeliverSubmittedPromptRow, } from "./agent/submitted-prompt-delivery.js"; @@ -1070,11 +1069,11 @@ export class Session { if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) { if ( serializedEvent.type === "timeline" && - !shouldDeliverSubmittedPromptRow({ - delivery: event.delivery, - item: serializedEvent.item, - supportsRevisions: this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions), - }) + !shouldDeliverSubmittedPromptRow( + event.delivery, + serializedEvent.item, + this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions), + ) ) { return; } @@ -1105,11 +1104,11 @@ export class Session { const supportsSelectiveDelivery = capabilities.has(CLIENT_CAPS.selectiveAgentTimeline); if ( serializedEvent.type === "timeline" && - !shouldDeliverSubmittedPromptRow({ - delivery: event.delivery, - item: serializedEvent.item, - supportsRevisions: capabilities.has(CLIENT_CAPS.canonicalSubmittedPromptRevisions), - }) + !shouldDeliverSubmittedPromptRow( + event.delivery, + serializedEvent.item, + capabilities.has(CLIENT_CAPS.canonicalSubmittedPromptRevisions), + ) ) { continue; } @@ -6120,36 +6119,30 @@ export class Session { direction: AgentTimelineFetchDirection; cursor?: AgentTimelineCursor; pageLimit: number; - visibleMaxSeq?: number; + fullTimeline?: AgentTimelineFetchResult; }): AgentTimelineProjectionSelection { const selectedTimeline = this.shouldUseFullTimelineForProjectedPage({ timeline: input.controlTimeline, pageLimit: input.pageLimit, }) - ? this.agentManager.fetchTimeline(input.agentId, { direction: "tail", limit: 0 }) + ? (input.fullTimeline ?? + this.agentManager.fetchTimeline(input.agentId, { direction: "tail", limit: 0 })) : input.controlTimeline; - const timeline = - input.visibleMaxSeq === undefined - ? selectedTimeline - : projectSubmittedPromptTimeline({ - timeline: selectedTimeline, - visibleMaxSeq: input.visibleMaxSeq, - ...(input.cursor ? { cursorSeq: input.cursor.seq } : {}), - }); const page = selectProjectedTimelinePage({ - rows: timeline.rows, - bounds: timeline.window, + rows: selectedTimeline.rows, + bounds: selectedTimeline.window, direction: input.controlTimeline.reset ? "tail" : input.direction, ...(input.cursor ? { cursorSeq: input.cursor.seq } : {}), limit: input.pageLimit, }); return { - timeline, + timeline: selectedTimeline, entries: page.entries, startSeq: page.startSeq, endSeq: page.endSeq, - hasOlder: page.hasOlder || (page.startSeq !== null && page.startSeq > timeline.window.minSeq), + hasOlder: + page.hasOlder || (page.startSeq !== null && page.startSeq > selectedTimeline.window.minSeq), hasNewer: page.hasNewer, }; } @@ -6161,7 +6154,7 @@ export class Session { direction: AgentTimelineFetchDirection; cursor?: AgentTimelineCursor; pageLimit: number; - visibleMaxSeq?: number; + fullTimeline?: AgentTimelineFetchResult; }): AgentTimelineProjectionSelection { if (input.projection === "canonical") { return this.selectCanonicalTimelineProjection({ timeline: input.controlTimeline }); @@ -6201,20 +6194,24 @@ export class Session { const supportsSubmittedPromptRevisions = source ? this.supportsForSource(CLIENT_CAPS.canonicalSubmittedPromptRevisions, source) : this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions); - const visibleMaxSeq = supportsSubmittedPromptRevisions - ? undefined - : findSubmittedPromptVisibleMaxSeq( - this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 0 }).rows, - fetchedControlTimeline.window.maxSeq, - ); + let fetchedFullTimeline: AgentTimelineFetchResult | undefined; + if (!supportsSubmittedPromptRevisions) { + fetchedFullTimeline = + direction === "tail" && pageLimit === 0 + ? fetchedControlTimeline + : this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 0 }); + } const controlTimeline = - visibleMaxSeq === undefined + fetchedFullTimeline === undefined ? fetchedControlTimeline - : projectSubmittedPromptTimeline({ - timeline: fetchedControlTimeline, - visibleMaxSeq, - ...(cursor ? { cursorSeq: cursor.seq } : {}), - }); + : projectSubmittedPromptTimeline( + fetchedControlTimeline, + fetchedFullTimeline, + cursor?.seq, + ); + const fullTimeline = fetchedFullTimeline + ? projectSubmittedPromptTimeline(fetchedFullTimeline) + : undefined; const selectedTimeline = this.selectTimelineProjection({ agentId: msg.agentId, projection, @@ -6222,7 +6219,7 @@ export class Session { direction, ...(cursor ? { cursor } : {}), pageLimit, - ...(visibleMaxSeq !== undefined ? { visibleMaxSeq } : {}), + ...(fullTimeline ? { fullTimeline } : {}), }); const startCursor = selectedTimeline.startSeq !== null diff --git a/packages/server/src/server/wire-compat.test.ts b/packages/server/src/server/wire-compat.test.ts index 6bc09fdc30..daa1061acb 100644 --- a/packages/server/src/server/wire-compat.test.ts +++ b/packages/server/src/server/wire-compat.test.ts @@ -16,6 +16,8 @@ import { import { Session, type SessionOptions } from "./session.js"; import { createProviderSnapshotManagerStub } from "./test-utils/session-stubs.js"; import type { AgentTimelineRow } from "./agent/agent-manager.js"; +import { InMemoryAgentTimelineStore } from "./agent/agent-timeline-store.js"; +import type { AgentTimelineFetchOptions } from "./agent/agent-timeline-store-types.js"; import { handleCreatePaseoWorktreeRequest } from "./worktree-session.js"; import { createPersistedProjectRecord } from "./workspace-registry.js"; @@ -87,7 +89,15 @@ interface SessionInternals { } class InMemoryAgentManager { - constructor(private readonly rows: AgentTimelineRow[]) {} + private readonly timeline = new InMemoryAgentTimelineStore(); + + constructor(rows: AgentTimelineRow[]) { + this.timeline.initialize("agent-1", { + epoch: "epoch-1", + rows, + nextSeq: (rows.at(-1)?.seq ?? 0) + 1, + }); + } getAgent() { return { @@ -133,17 +143,8 @@ class InMemoryAgentManager { }; } - fetchTimeline() { - return { - epoch: "epoch-1", - reset: false, - staleCursor: false, - gap: false, - window: { minSeq: 1, maxSeq: 3, nextSeq: 4 }, - rows: this.rows, - hasOlder: false, - hasNewer: false, - }; + fetchTimeline(_agentId: string, options?: AgentTimelineFetchOptions) { + return this.timeline.fetch("agent-1", options); } listAgents() { @@ -218,6 +219,7 @@ class InMemoryWorktreeWorkflow { function createSessionForWireCompatTest(options?: { clientCapabilities?: Record | null; messages?: SessionOutboundMessage[]; + rows?: AgentTimelineRow[]; }): Session { const messages = options?.messages ?? []; const rows: AgentTimelineRow[] = [ @@ -247,7 +249,9 @@ function createSessionForWireCompatTest(options?: { downloadTokenStore: {} as SessionOptions["downloadTokenStore"], pushTokenStore: {} as SessionOptions["pushTokenStore"], paseoHome: "/tmp/paseo-home", - agentManager: new InMemoryAgentManager(rows) as unknown as SessionOptions["agentManager"], + agentManager: new InMemoryAgentManager( + options?.rows ?? rows, + ) as unknown as SessionOptions["agentManager"], agentStorage: new EmptyAgentStorage() as unknown as SessionOptions["agentStorage"], projectRegistry: new EmptyProjectRegistry() as unknown as SessionOptions["projectRegistry"], workspaceRegistry: @@ -308,11 +312,19 @@ function createSessionForWireCompatTest(options?: { return session; } -async function emitTimelineResponse( - clientCapabilities?: Record | null, -): Promise> { +async function emitTimelineResponse(options?: { + clientCapabilities?: Record | null; + rows?: AgentTimelineRow[]; + request?: Partial< + Extract, { type: "fetch_agent_timeline_request" }> + >; +}): Promise> { const messages: SessionOutboundMessage[] = []; - const session = createSessionForWireCompatTest({ clientCapabilities, messages }); + const session = createSessionForWireCompatTest({ + clientCapabilities: options?.clientCapabilities, + rows: options?.rows, + messages, + }); const internals = session as unknown as SessionInternals; await internals.handleFetchAgentTimelineRequest({ @@ -320,6 +332,7 @@ async function emitTimelineResponse( requestId: "req-timeline", agentId: "agent-1", projection: "projected", + ...options?.request, }); const response = messages[0]; @@ -377,7 +390,7 @@ describe("wire compatibility", () => { ]); }); - test("hello parses with and without the project update capability", () => { + test("hello parses with and without current client capabilities", () => { const legacy = WSHelloMessageSchema.parse({ type: "hello", clientId: "legacy-client", @@ -389,7 +402,10 @@ describe("wire compatibility", () => { clientId: "capable-client", clientType: "mobile", protocolVersion: 1, - capabilities: { [CLIENT_CAPS.projectUpdates]: true }, + capabilities: { + [CLIENT_CAPS.projectUpdates]: true, + [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true, + }, }); expect([legacy, capable]).toEqual([ @@ -404,7 +420,10 @@ describe("wire compatibility", () => { clientId: "capable-client", clientType: "mobile", protocolVersion: 1, - capabilities: { project_updates: true }, + capabilities: { + project_updates: true, + canonical_submitted_prompt_revisions: true, + }, }, ]); }); @@ -463,13 +482,94 @@ describe("wire compatibility", () => { test("preserves reasoning_merge for clients that declare the capability", async () => { const response = await emitTimelineResponse({ - [CLIENT_CAPS.reasoningMergeEnum]: true, + clientCapabilities: { [CLIENT_CAPS.reasoningMergeEnum]: true }, }); const currentParsed = FetchAgentTimelineResponseMessageSchema.parse(response); expect(currentParsed.payload.entries[0]?.collapsed).toContain("reasoning_merge"); }); + test("projects identity-pending prompt rows coherently for legacy fetches", async () => { + const timestamp = "2026-08-01T00:00:00.000Z"; + const prompt = (options?: { delivery?: AgentTimelineRow["delivery"]; messageId?: string }) => ({ + seq: 1, + timestamp, + ...(options?.delivery ? { delivery: options.delivery } : {}), + item: { + type: "user_message" as const, + text: "hello", + clientMessageId: "client-message-1", + ...(options?.messageId ? { messageId: options.messageId } : {}), + }, + }); + + const pending = await emitTimelineResponse({ + rows: [prompt({ delivery: "awaiting_provider_identity" })], + request: { projection: "canonical" }, + }); + const pageShape = (response: typeof pending) => [ + [ + response.payload.window.minSeq, + response.payload.window.maxSeq, + response.payload.window.nextSeq, + ], + response.payload.startCursor?.seq ?? null, + response.payload.endCursor?.seq ?? null, + response.payload.hasOlder, + response.payload.hasNewer, + response.payload.entries.map((entry) => entry.seqStart), + ]; + expect(pageShape(pending)).toEqual([[0, 0, 1], null, null, false, false, []]); + + for (const request of [ + { projection: "canonical" as const }, + { + projection: "canonical" as const, + direction: "after" as const, + cursor: { epoch: "epoch-1", seq: 0 }, + }, + { projection: "projected" as const }, + ]) { + const finalized = await emitTimelineResponse({ + rows: [prompt({ messageId: "provider-message-1" })], + request, + }); + expect(pageShape(finalized)).toEqual([[1, 1, 2], 1, 1, false, false, [1]]); + expect(finalized.payload.entries[0]?.item).toMatchObject({ + type: "user_message", + messageId: "provider-message-1", + }); + } + + const daemonHandled = await emitTimelineResponse({ + rows: [prompt()], + request: { projection: "canonical" }, + }); + expect(daemonHandled.payload.entries).toHaveLength(1); + + const pageRows: AgentTimelineRow[] = [1, 2, 3].map((seq) => ({ + seq, + timestamp, + item: { + type: "user_message", + text: `prompt ${seq}`, + clientMessageId: `client-message-${seq}`, + }, + })); + for (const projection of ["canonical", "projected"] as const) { + const page = await emitTimelineResponse({ + rows: pageRows, + request: { + projection, + direction: "before", + cursor: { epoch: "epoch-1", seq: 3 }, + limit: 1, + }, + }); + expect(pageShape(page)).toEqual([[1, 3, 4], 2, 2, true, true, [2]]); + } + }); + test("sub_agent tool-call payload still parses against the v0.1.65-beta.3 schema", () => { const parsed = LegacySubAgentToolCallSchema.parse({ type: "tool_call",