From 7e757468a191c32619119a4083b85ed2051653ed Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 1 Aug 2026 04:23:24 +0200 Subject: [PATCH] fix(timeline): gate prompt revisions by client capability --- docs/timeline-sync.md | 8 + packages/app/src/runtime/host-runtime.test.ts | 8 + packages/app/src/runtime/host-runtime.ts | 7 +- packages/protocol/src/client-capabilities.ts | 3 + .../src/messages.client-capabilities.test.ts | 19 ++ packages/protocol/src/messages.ts | 1 + .../src/server/agent/agent-manager.test.ts | 113 +++++-- .../server/src/server/agent/agent-manager.ts | 53 ++- .../agent/agent-timeline-store-types.ts | 3 + .../src/server/agent/agent-timeline-store.ts | 3 +- .../server/agent/submitted-prompt-delivery.ts | 63 ++++ packages/server/src/server/session.test.ts | 306 +++++++++++++++++- packages/server/src/server/session.ts | 170 +++++++--- 13 files changed, 661 insertions(+), 96 deletions(-) create mode 100644 packages/protocol/src/messages.client-capabilities.test.ts create mode 100644 packages/server/src/server/agent/submitted-prompt-delivery.ts diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index dc56a268f5..d9ae2dfcdf 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -178,6 +178,14 @@ provider accepts the turn. A correlated provider echo enriches that same sequenc `messageId`; it never appends a second source row. The row's identity and pagination position are therefore independent of provider notification timing. A late echo is streamed as a revision at the original sequence; clients apply it only to the already-covered row with the same client identity. + +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. Content matching is limited to the dated compatibility path for daemon timelines created before diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 74cdf85316..56463faa11 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -8,12 +8,14 @@ import type { import type { ConnectionOffer } from "@getpaseo/protocol/connection-offer"; import type { SessionOutboundMessage } from "@getpaseo/protocol/messages"; import type { AgentPermissionRequest } from "@getpaseo/protocol/agent-types"; +import { CLIENT_CAPS } from "@getpaseo/protocol/client-capabilities"; import type { HostConnection, HostProfile } from "@/types/host-connection"; import { useSessionStore, type Agent } from "@/stores/session-store"; 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, @@ -22,6 +24,12 @@ 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, + }); +}); + class FakeDaemonClient { private state: ConnectionState = { status: "idle" }; private listeners = new Set<(status: ConnectionState) => void>(); diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 6d05aa0e7f..5abe672a18 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -456,6 +456,11 @@ 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"; @@ -468,7 +473,7 @@ function createDefaultDeps(): HostRuntimeControllerDeps { } : undefined; const appCapabilities = { - [CLIENT_CAPS.selectiveAgentTimeline]: true, + ...APP_CLIENT_CAPABILITIES, ...browserAutomationCapabilities, }; diff --git a/packages/protocol/src/client-capabilities.ts b/packages/protocol/src/client-capabilities.ts index f9ccbeec99..b49ad5181d 100644 --- a/packages/protocol/src/client-capabilities.ts +++ b/packages/protocol/src/client-capabilities.ts @@ -20,6 +20,9 @@ export const CLIENT_CAPS = { providerSubagents: "provider_subagents", // COMPAT(projectUpdates): added in v0.1.109, remove gate after 2027-01-15. projectUpdates: "project_updates", + // COMPAT(canonicalSubmittedPromptRevisions): added in v0.2.6, remove gate after 2027-02-01. + // Capable clients reconcile same-sequence provider-identity revisions in canonical prompt rows. + canonicalSubmittedPromptRevisions: "canonical_submitted_prompt_revisions", browserHost: "browser_host", } as const; diff --git a/packages/protocol/src/messages.client-capabilities.test.ts b/packages/protocol/src/messages.client-capabilities.test.ts new file mode 100644 index 0000000000..c0e78466b2 --- /dev/null +++ b/packages/protocol/src/messages.client-capabilities.test.ts @@ -0,0 +1,19 @@ +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/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 7c51205fc6..53d99f6b83 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -5886,6 +5886,7 @@ export const WSHelloMessageSchema = z.object({ [CLIENT_CAPS.terminalReflowableSnapshot]: z.boolean().optional(), [CLIENT_CAPS.providerSubagents]: z.boolean().optional(), [CLIENT_CAPS.projectUpdates]: z.boolean().optional(), + [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: z.boolean().optional(), [CLIENT_CAPS.browserHost]: BrowserAutomationHostCapabilitySchema.optional(), }) .passthrough() diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index a768c67b7b..99e5173d4c 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -8076,23 +8076,21 @@ test("provider user_message is recorded from the live stream", async () => { class UnexpectedUserMessageSession extends TestAgentSession { override async startTurn(): Promise<{ turnId: string }> { const turnId = "turn-unexpected-1"; - setTimeout(() => { - this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); - // Provider yields user_message (e.g., system continuation) - this.pushEvent({ - type: "timeline", - provider: this.provider, - item: { type: "user_message", text: "continuation prompt" }, - turnId, - }); - this.pushEvent({ - type: "timeline", - provider: this.provider, - item: { type: "assistant_message", text: "continuation reply" }, - turnId, - }); - this.pushEvent({ type: "turn_completed", provider: this.provider, turnId }); - }, 0); + this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); + // Provider yields a staged user_message without client identity (e.g., system continuation). + this.pushEvent({ + type: "timeline", + provider: this.provider, + item: { type: "user_message", text: "continuation prompt" }, + turnId, + }); + this.pushEvent({ + type: "timeline", + provider: this.provider, + item: { type: "assistant_message", text: "continuation reply" }, + turnId, + }); + this.pushEvent({ type: "turn_completed", provider: this.provider, turnId }); return { turnId }; } } @@ -8130,6 +8128,9 @@ test("provider user_message is recorded from the live stream", async () => { // Provider's user_message should be recorded (no canonical to dedup against) expect(userMessages).toHaveLength(1); expect(userMessages[0].text).toBe("continuation prompt"); + expect( + manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 0 }).rows[0]?.delivery, + ).toBeUndefined(); }); test("authoritative timeline includes provider-emitted submitted user prompt", async () => { @@ -8138,13 +8139,20 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a const storage = new AgentStorage(storagePath, logger); class SubmittedUserMessageSession extends TestAgentSession { + override readonly capabilities = { + ...TEST_CAPABILITIES, + supportsRewindConversation: true, + }; + private turnCount = 0; + override async startTurn( prompt: AgentPromptInput, options?: AgentRunOptions, ): Promise<{ turnId: string }> { - const turnId = "turn-submitted-user-message"; + this.turnCount += 1; + const turnId = `turn-submitted-user-message-${this.turnCount}`; const text = typeof prompt === "string" ? prompt : ""; - setTimeout(() => { + const publishProviderIdentity = () => { this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); this.pushEvent({ type: "timeline", @@ -8153,12 +8161,22 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a item: { type: "user_message", text, - messageId: "provider-message-1", + messageId: `provider-message-${this.turnCount}`, clientMessageId: options?.clientMessageId, }, }); + if (this.turnCount === 1) { + this.pushEvent({ + type: "timeline", + provider: this.provider, + turnId, + item: { type: "assistant_message", text: "provider output after identity" }, + }); + } this.pushEvent({ type: "turn_completed", provider: this.provider, turnId }); - }, 0); + }; + if (this.turnCount === 1) setTimeout(publishProviderIdentity, 0); + else publishProviderIdentity(); return { turnId }; } } @@ -8209,6 +8227,7 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a type: event.event.type, item: event.event.item, seq: event.seq, + delivery: event.delivery, } : { type: event.event.type }, ), @@ -8222,6 +8241,7 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a clientMessageId: "msg-client-1", }, seq: 1, + delivery: "awaiting_provider_identity", }, { type: "timeline", @@ -8232,17 +8252,57 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a clientMessageId: "msg-client-1", }, seq: 1, + delivery: "awaiting_provider_identity", + }, + { + type: "timeline", + item: { type: "assistant_message", text: "provider output after identity" }, + seq: 2, + delivery: undefined, }, { type: "turn_completed" }, ]); const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows; - expect(timeline.map((row) => row.item)).toEqual([ + expect(timeline).toEqual([ { - type: "user_message", - text: "hello from composer", - messageId: "provider-message-1", - clientMessageId: "msg-client-1", + seq: 1, + timestamp: expect.any(String), + delivery: "awaiting_provider_identity", + item: { + 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" }, + }, + ]); + + 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([ + { + seq: 3, + delivery: undefined, + item: { + type: "user_message", + text: "staged identity", + messageId: "provider-message-2", + clientMessageId: "msg-client-2", + }, }, ]); } finally { @@ -8324,6 +8384,7 @@ test("authoritative timeline records a daemon-handled submitted prompt before it }, { type: "assistant_message", text: "Handled by the daemon" }, ]); + expect(timeline.every((row) => row.delivery === undefined)).toBe(true); } finally { await manager.flush().catch(() => undefined); await storage.flush().catch(() => undefined); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 5902e8e368..6698589944 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -61,6 +61,7 @@ import { AgentStreamCoalescer, } from "./agent-stream-coalescer.js"; import { limitAgentTimelineItemContent } from "./agent-timeline-content.js"; +import { AWAITING_PROVIDER_IDENTITY_DELIVERY } from "./submitted-prompt-delivery.js"; import { AgentRunState, type ForegroundTurnWaiter } from "./agent-run-state.js"; import { getAgentProviderDefinition } from "@getpaseo/protocol/provider-manifest"; import { invokeRewindCapability, type RewindMode } from "./rewind/rewind.js"; @@ -185,6 +186,7 @@ export type AgentManagerEvent = seq?: number; epoch?: string; timestamp?: string; + delivery?: AgentTimelineRow["delivery"]; }; export type AgentSubscriber = (event: AgentManagerEvent) => void; @@ -2070,12 +2072,14 @@ export class AgentManager { { type: "turn_started", provider: agent.provider, turnId }, { timestamp: turnStartedAt.toISOString() }, ); - const stagedSubmittedPromptEcho = pendingRun.stagedEvents.find( - (event): event is Extract => - event.type === "timeline" && - event.item.type === "user_message" && - event.item.clientMessageId === options?.clientMessageId, - ); + const stagedSubmittedPromptEcho = options?.clientMessageId + ? pendingRun.stagedEvents.find( + (event): event is Extract => + event.type === "timeline" && + event.item.type === "user_message" && + event.item.clientMessageId === options.clientMessageId, + ) + : undefined; if (options?.clientMessageId) { this.recordSubmittedPrompt( agent, @@ -3932,15 +3936,30 @@ export class AgentManager { } this.touchUpdatedAt(agent); agent.lastUserMessageAt = new Date(); - this.recordAndDispatchTimelineItem( + const item: AgentTimelineItem = { + type: "user_message", + text: submittedPromptText(prompt), + clientMessageId, + ...(messageId ? { messageId } : {}), + }; + const capabilities = agent.session.capabilities; + const delivery = + messageId === undefined && + (capabilities.supportsRewindConversation || + capabilities.supportsRewindFiles || + capabilities.supportsRewindBoth) + ? 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 }, { - type: "user_message", - text: submittedPromptText(prompt), - clientMessageId, - ...(messageId ? { messageId } : {}), + seq: row.seq, + epoch: this.timelineStore.getEpoch(agent.id), + timestamp: row.timestamp, + ...(row.delivery ? { delivery: row.delivery } : {}), }, - agent.provider, ); } @@ -3972,6 +3991,7 @@ export class AgentManager { seq: row.seq, epoch: this.timelineStore.getEpoch(agent.id), timestamp: row.timestamp, + ...(row.delivery ? { delivery: row.delivery } : {}), }, ); } @@ -4035,7 +4055,7 @@ export class AgentManager { private recordTimeline( agentId: string, item: AgentTimelineItem, - options?: { timestamp?: string }, + options?: { timestamp?: string; delivery?: AgentTimelineRow["delivery"] }, ): AgentTimelineRow { item = limitAgentTimelineItemContent(item); const row = this.timelineStore.append(agentId, item, options); @@ -4236,7 +4256,12 @@ export class AgentManager { private dispatchStream( agentId: string, event: AgentStreamEvent, - metadata?: { seq?: number; epoch?: string; timestamp?: string }, + metadata?: { + seq?: number; + epoch?: string; + timestamp?: string; + delivery?: AgentTimelineRow["delivery"]; + }, ): void { if (event.type === "timeline") { event = { 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 20c3d4d01e..8a189668b9 100644 --- a/packages/server/src/server/agent/agent-timeline-store-types.ts +++ b/packages/server/src/server/agent/agent-timeline-store-types.ts @@ -1,9 +1,12 @@ import type { AgentTimelineItem } from "./agent-sdk-types.js"; +export type AgentTimelineRowDelivery = "awaiting_provider_identity"; + export interface AgentTimelineRow { seq: number; timestamp: string; item: AgentTimelineItem; + readonly delivery?: AgentTimelineRowDelivery; } export interface AgentTimelineCursor { diff --git a/packages/server/src/server/agent/agent-timeline-store.ts b/packages/server/src/server/agent/agent-timeline-store.ts index d49341cfc6..2ddfea30e6 100644 --- a/packages/server/src/server/agent/agent-timeline-store.ts +++ b/packages/server/src/server/agent/agent-timeline-store.ts @@ -260,13 +260,14 @@ export class InMemoryAgentTimelineStore { append( agentId: string, item: AgentTimelineItem, - options?: { timestamp?: string }, + options?: { timestamp?: string; delivery?: AgentTimelineRow["delivery"] }, ): AgentTimelineRow { const state = this.requireState(agentId); const row: AgentTimelineRow = { seq: state.nextSeq, timestamp: options?.timestamp ?? new Date().toISOString(), item, + ...(options?.delivery ? { delivery: options.delivery } : {}), }; state.nextSeq += 1; state.rows.push(row); diff --git a/packages/server/src/server/agent/submitted-prompt-delivery.ts b/packages/server/src/server/agent/submitted-prompt-delivery.ts new file mode 100644 index 0000000000..ab5f5d0044 --- /dev/null +++ b/packages/server/src/server/agent/submitted-prompt-delivery.ts @@ -0,0 +1,63 @@ +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 { + return ( + input.supportsRevisions || + input.delivery !== AWAITING_PROVIDER_IDENTITY_DELIVERY || + input.item.type !== "user_message" || + input.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; + } + 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; + return { + ...input.timeline, + window: { + ...input.timeline.window, + maxSeq: input.visibleMaxSeq, + nextSeq: input.visibleMaxSeq + 1, + }, + hasNewer: input.timeline.hasNewer && selectedEndSeq < input.visibleMaxSeq, + rows, + }; +} diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index 5677750165..eef73462c2 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -22,7 +22,11 @@ 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 } from "./agent/agent-manager.js"; +import type { + AgentManagerEvent, + AgentTimelineFetchResult, + ManagedAgent, +} 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"; @@ -4880,6 +4884,306 @@ 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 () => {}; + }), + }, + }); + const legacySocket = {}; + const capableSocket = {}; + session.updateClientCapabilities(null, legacySocket); + session.updateClientCapabilities( + { [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true }, + capableSocket, + ); + + const listener = agentEventListeners[0]; + if (!listener) throw new Error("Agent event listener was not installed"); + listener({ + type: "agent_stream", + agentId: "agent-revisions", + event: { + type: "timeline", + provider: "codex", + item: { + type: "user_message", + text: "submitted prompt", + clientMessageId: "client-message-1", + }, + }, + seq: 5, + epoch: "epoch-1", + delivery: "awaiting_provider_identity", + }); + listener({ + type: "agent_stream", + agentId: "agent-revisions", + event: { + type: "timeline", + provider: "codex", + item: { + type: "user_message", + text: "submitted prompt", + clientMessageId: "client-message-1", + messageId: "provider-message-1", + }, + }, + seq: 5, + epoch: "epoch-1", + delivery: "awaiting_provider_identity", + }); + listener({ + type: "agent_stream", + agentId: "agent-revisions", + event: { + type: "timeline", + provider: "codex", + item: { + type: "user_message", + text: "staged identity", + clientMessageId: "client-message-2", + messageId: "provider-message-2", + }, + }, + seq: 6, + epoch: "epoch-1", + delivery: "awaiting_provider_identity", + }); + listener({ + type: "agent_stream", + agentId: "agent-revisions", + event: { + type: "timeline", + provider: "codex", + item: { + type: "user_message", + text: "provider message without revision contract", + }, + }, + seq: 7, + epoch: "epoch-1", + }); + + const timelineFor = (source: object) => + targetedMessages.flatMap(({ source: destination, message }) => + destination === source && + message.type === "agent_stream" && + message.payload.event.type === "timeline" + ? [ + { + seq: message.payload.seq, + clientMessageId: + message.payload.event.item.type === "user_message" + ? message.payload.event.item.clientMessageId + : undefined, + messageId: + message.payload.event.item.type === "user_message" + ? message.payload.event.item.messageId + : undefined, + }, + ] + : [], + ); + expect(timelineFor(capableSocket)).toEqual([ + { seq: 5, clientMessageId: "client-message-1", messageId: undefined }, + { seq: 5, clientMessageId: "client-message-1", messageId: "provider-message-1" }, + { seq: 6, clientMessageId: "client-message-2", messageId: "provider-message-2" }, + { seq: 7, clientMessageId: undefined, messageId: undefined }, + ]); + expect(timelineFor(legacySocket)).toEqual([ + { seq: 5, clientMessageId: "client-message-1", messageId: "provider-message-1" }, + { 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", + }, + }, + ], + }; + const session = createSessionForTest({ + targetedMessages, + agentManager: { + waitForAgentClose: vi.fn().mockResolvedValue(undefined), + getAgent: vi.fn(() => agent), + fetchTimeline: vi.fn(() => timeline), + }, + }); + 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", + }, + }, + ], + }; + 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", + }, + }, + ], + }, + }); +}); + test("sends project updates only to capable sockets in a retained session", () => { const messages: SessionOutboundMessage[] = []; const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = []; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 8c11ae35b3..7e1484a97e 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -76,6 +76,11 @@ import type { AgentTimelineFetchResult, ManagedAgent, } from "./agent/agent-manager.js"; +import { + findSubmittedPromptVisibleMaxSeq, + projectSubmittedPromptTimeline, + shouldDeliverSubmittedPromptRow, +} from "./agent/submitted-prompt-delivery.js"; import { createAgentCommand } from "./agent/create-agent/create.js"; import { resolveCreateAgentIntent, type CreateAgentIntent } from "./agent/create-agent/intent.js"; import { @@ -1063,6 +1068,16 @@ export class Session { serializedEvent: Extract["payload"]["event"], ): void { if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) { + if ( + serializedEvent.type === "timeline" && + !shouldDeliverSubmittedPromptRow({ + delivery: event.delivery, + item: serializedEvent.item, + supportsRevisions: this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions), + }) + ) { + return; + } if (this.usesSelectiveTimelineDelivery() && serializedEvent.type === "attention_required") { this.emit({ type: "agent_attention_required", @@ -1088,6 +1103,16 @@ export class Session { for (const [source, capabilities] of this.clientCapabilitiesBySource) { const supportsSelectiveDelivery = capabilities.has(CLIENT_CAPS.selectiveAgentTimeline); + if ( + serializedEvent.type === "timeline" && + !shouldDeliverSubmittedPromptRow({ + delivery: event.delivery, + item: serializedEvent.item, + supportsRevisions: capabilities.has(CLIENT_CAPS.canonicalSubmittedPromptRevisions), + }) + ) { + continue; + } if (supportsSelectiveDelivery && serializedEvent.type === "attention_required") { this.onMessageToSource(source, { type: "agent_attention_required", @@ -1881,7 +1906,7 @@ export class Session { ): Promise | undefined { switch (msg.type) { case "fetch_agent_timeline_request": - return this.handleFetchAgentTimelineRequest(msg); + return this.handleFetchAgentTimelineRequest(msg, source); case "agent.provider_subagents.list.request": return this.handleProviderSubagentListRequest(msg); case "agent.provider_subagents.timeline.get.request": @@ -6095,13 +6120,22 @@ export class Session { direction: AgentTimelineFetchDirection; cursor?: AgentTimelineCursor; pageLimit: number; + visibleMaxSeq?: number; }): AgentTimelineProjectionSelection { - const timeline = this.shouldUseFullTimelineForProjectedPage({ + const selectedTimeline = this.shouldUseFullTimelineForProjectedPage({ timeline: input.controlTimeline, pageLimit: input.pageLimit, }) ? 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, @@ -6127,6 +6161,7 @@ export class Session { direction: AgentTimelineFetchDirection; cursor?: AgentTimelineCursor; pageLimit: number; + visibleMaxSeq?: number; }): AgentTimelineProjectionSelection { if (input.projection === "canonical") { return this.selectCanonicalTimelineProjection({ timeline: input.controlTimeline }); @@ -6137,6 +6172,7 @@ export class Session { private async handleFetchAgentTimelineRequest( msg: Extract, + source?: object, ): Promise { const direction: AgentTimelineFetchDirection = msg.direction ?? (msg.cursor ? "after" : "tail"); const projection: TimelineProjectionMode = msg.projection ?? "projected"; @@ -6157,11 +6193,28 @@ export class Session { }); const agentPayload = await this.buildAgentPayload(snapshot); - const controlTimeline = this.agentManager.fetchTimeline(msg.agentId, { + const fetchedControlTimeline = this.agentManager.fetchTimeline(msg.agentId, { direction, cursor, limit: pageLimit, }); + 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, + ); + const controlTimeline = + visibleMaxSeq === undefined + ? fetchedControlTimeline + : projectSubmittedPromptTimeline({ + timeline: fetchedControlTimeline, + visibleMaxSeq, + ...(cursor ? { cursorSeq: cursor.seq } : {}), + }); const selectedTimeline = this.selectTimelineProjection({ agentId: msg.agentId, projection, @@ -6169,6 +6222,7 @@ export class Session { direction, ...(cursor ? { cursor } : {}), pageLimit, + ...(visibleMaxSeq !== undefined ? { visibleMaxSeq } : {}), }); const startCursor = selectedTimeline.startSeq !== null @@ -6179,63 +6233,73 @@ export class Session { ? { epoch: selectedTimeline.timeline.epoch, seq: selectedTimeline.endSeq } : null; - this.emit({ - type: "fetch_agent_timeline_response", - payload: { - requestId: msg.requestId, - agentId: msg.agentId, - agent: agentPayload, - direction, - projection, - epoch: selectedTimeline.timeline.epoch, - reset: controlTimeline.reset, - staleCursor: controlTimeline.staleCursor, - gap: controlTimeline.gap, - window: selectedTimeline.timeline.window, - startCursor, - endCursor, - hasOlder: selectedTimeline.hasOlder, - hasNewer: selectedTimeline.hasNewer, - entries: selectedTimeline.entries.map((entry) => ({ - provider: snapshot.provider, - item: entry.item, - timestamp: entry.timestamp, - seqStart: entry.seqStart, - seqEnd: entry.seqEnd, - sourceSeqRanges: entry.sourceSeqRanges, - collapsed: this.supports(CLIENT_CAPS.reasoningMergeEnum) - ? entry.collapsed - : entry.collapsed.filter((value) => value !== "reasoning_merge"), - })), - error: null, + this.emitForSource( + { + type: "fetch_agent_timeline_response", + payload: { + requestId: msg.requestId, + agentId: msg.agentId, + agent: agentPayload, + direction, + projection, + epoch: selectedTimeline.timeline.epoch, + reset: controlTimeline.reset, + staleCursor: controlTimeline.staleCursor, + gap: controlTimeline.gap, + window: selectedTimeline.timeline.window, + startCursor, + endCursor, + hasOlder: selectedTimeline.hasOlder, + hasNewer: selectedTimeline.hasNewer, + entries: selectedTimeline.entries.map((entry) => ({ + provider: snapshot.provider, + item: entry.item, + timestamp: entry.timestamp, + seqStart: entry.seqStart, + seqEnd: entry.seqEnd, + sourceSeqRanges: entry.sourceSeqRanges, + collapsed: ( + source + ? this.supportsForSource(CLIENT_CAPS.reasoningMergeEnum, source) + : this.supports(CLIENT_CAPS.reasoningMergeEnum) + ) + ? entry.collapsed + : entry.collapsed.filter((value) => value !== "reasoning_merge"), + })), + error: null, + }, }, - }); + source, + ); } catch (error) { this.sessionLogger.error( { err: error, agentId: msg.agentId }, "Failed to handle fetch_agent_timeline_request", ); - this.emit({ - type: "fetch_agent_timeline_response", - payload: { - requestId: msg.requestId, - agentId: msg.agentId, - agent: null, - direction, - projection, - epoch: "", - reset: false, - staleCursor: false, - gap: false, - window: { minSeq: 0, maxSeq: 0, nextSeq: 0 }, - startCursor: null, - endCursor: null, - hasOlder: false, - hasNewer: false, - entries: [], - error: error instanceof Error ? error.message : String(error), + this.emitForSource( + { + type: "fetch_agent_timeline_response", + payload: { + requestId: msg.requestId, + agentId: msg.agentId, + agent: null, + direction, + projection, + epoch: "", + reset: false, + staleCursor: false, + gap: false, + window: { minSeq: 0, maxSeq: 0, nextSeq: 0 }, + startCursor: null, + endCursor: null, + hasOlder: false, + hasNewer: false, + entries: [], + error: error instanceof Error ? error.message : String(error), + }, }, - }); + source, + ); } }