From 20a542126e77e5ba079e919ae303c49f820d1106 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 1 Aug 2026 11:46:13 +0200 Subject: [PATCH] refactor(timeline): mint submitted prompt identity at acceptance Canonical prompt rows now keep one immutable wire identity while provider identity remains internal for rewind translation. This removes the unreleased same-sequence revision and legacy projection contract. --- packages/app/src/runtime/host-runtime.ts | 1 - packages/protocol/src/client-capabilities.ts | 3 - packages/protocol/src/messages.ts | 1 - .../src/server/agent/agent-manager.test.ts | 180 ++++++------------ .../server/src/server/agent/agent-manager.ts | 122 +++--------- .../agent/agent-timeline-store-types.ts | 4 +- .../server/agent/agent-timeline-store.test.ts | 35 ---- .../src/server/agent/agent-timeline-store.ts | 20 +- .../server/agent/submitted-prompt-delivery.ts | 69 ------- packages/server/src/server/session.test.ts | 173 ----------------- packages/server/src/server/session.ts | 54 +----- .../server/src/server/wire-compat.test.ts | 114 ----------- 12 files changed, 98 insertions(+), 678 deletions(-) delete mode 100644 packages/server/src/server/agent/submitted-prompt-delivery.ts diff --git a/packages/app/src/runtime/host-runtime.ts b/packages/app/src/runtime/host-runtime.ts index 11cb3e4829..6d05aa0e7f 100644 --- a/packages/app/src/runtime/host-runtime.ts +++ b/packages/app/src/runtime/host-runtime.ts @@ -469,7 +469,6 @@ function createDefaultDeps(): HostRuntimeControllerDeps { : undefined; const appCapabilities = { [CLIENT_CAPS.selectiveAgentTimeline]: true, - [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true, ...browserAutomationCapabilities, }; diff --git a/packages/protocol/src/client-capabilities.ts b/packages/protocol/src/client-capabilities.ts index b49ad5181d..f9ccbeec99 100644 --- a/packages/protocol/src/client-capabilities.ts +++ b/packages/protocol/src/client-capabilities.ts @@ -20,9 +20,6 @@ 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.ts b/packages/protocol/src/messages.ts index 53d99f6b83..7c51205fc6 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -5886,7 +5886,6 @@ 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 53c3951bf8..b03b9fd6b9 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -4781,8 +4781,8 @@ test("replaceAgentRun stays running when a stale old terminal arrives before the (row) => row.item.type === "user_message" && row.item.clientMessageId === "replacement-client-message", - )?.delivery, - ).toBe("awaiting_provider_identity"); + )?.item, + ).toMatchObject({ messageId: "replacement-client-message" }); allowSecondTurnToComplete.resolve(); await firstRunDrain; await secondRunDrain; @@ -8146,12 +8146,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 () => { +test("canonical submitted prompt keeps wire identity while rewind resolves provider identity", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-submitted-prompt-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -8159,63 +8156,57 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a class SubmittedUserMessageSession extends TestAgentSession { override readonly capabilities = { ...TEST_CAPABILITIES, - supportsRewindConversation: true, + supportsRewindFiles: true, }; - private turnCount = 0; + readonly rewindMessageIds: string[] = []; override async startTurn( prompt: AgentPromptInput, options?: AgentRunOptions, ): Promise<{ turnId: string }> { - this.turnCount += 1; - const turnId = `turn-submitted-user-message-${this.turnCount}`; + const turnId = "turn-submitted-user-message"; const text = typeof prompt === "string" ? prompt : ""; - const publishProviderIdentity = () => { + setTimeout(() => { this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); - 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", - provider: this.provider, - 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: "timeline", + provider: this.provider, + turnId, + item: { type: "assistant_message", text: "output before provider echo" }, + }); + this.pushEvent({ + type: "timeline", + provider: this.provider, + turnId, + item: { + type: "user_message", + text, + messageId: "provider-message-1", + clientMessageId: options?.clientMessageId, + }, + }); this.pushEvent({ type: "turn_completed", provider: this.provider, turnId }); - }; - if (this.turnCount === 2) publishProviderIdentity(); - else setTimeout(publishProviderIdentity, 0); + }, 0); return { turnId }; } + + override async revertFiles({ messageId }: { messageId: string }): Promise { + this.rewindMessageIds.push(messageId); + } } class SubmittedUserMessageClient extends TestAgentClient { + session: SubmittedUserMessageSession | null = null; + override async createSession(config: AgentSessionConfig): Promise { - return new SubmittedUserMessageSession(config); + this.session = new SubmittedUserMessageSession(config); + return this.session; } } + const client = new SubmittedUserMessageClient(); const manager = new AgentManager({ - clients: { codex: new SubmittedUserMessageClient() }, + clients: { codex: client }, registry: storage, logger, idFactory: () => "00000000-0000-4000-8000-000000000402", @@ -8231,7 +8222,6 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a { type: item.type, seq: event.seq, - delivery: event.delivery, ...(item.type === "user_message" ? { text: item.text, @@ -8248,106 +8238,47 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a workspaceId: undefined, }); - await manager.runAgent( - snapshot.id, - [ - { - type: "text", - mimeType: "text/plain", - contextKind: "chat_history", - title: "Chat history", - text: "hidden fork history", - }, - { type: "text", text: "hello from composer" }, - ], - { clientMessageId: "msg-client-1" }, - ); + await manager.runAgent(snapshot.id, "hello from composer", { clientMessageId: "msg-client-1" }); expect(streamEvents()).toEqual([ { type: "turn_started" }, { type: "user_message", seq: 1, - delivery: "awaiting_provider_identity", text: "hello from composer", clientMessageId: "msg-client-1", - messageId: undefined, + messageId: "msg-client-1", }, - { - type: "user_message", - seq: 1, - delivery: undefined, - text: "hello from composer", - clientMessageId: "msg-client-1", - messageId: "provider-message-1", - }, - { type: "assistant_message", seq: 2, delivery: undefined }, + { type: "assistant_message", seq: 2 }, { type: "turn_completed" }, ]); const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows; - expect(timeline.map((row) => [row.seq, row.delivery, row.item])).toEqual([ - [ - 1, - undefined, - { + expect(timeline).toEqual([ + { + seq: 1, + timestamp: expect.any(String), + providerMessageId: "provider-message-1", + item: { type: "user_message", text: "hello from composer", - messageId: "provider-message-1", + messageId: "msg-client-1", clientMessageId: "msg-client-1", }, - ], - [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(streamEvents().filter((event) => event.type === "user_message")).toEqual([ + }, { - type: "user_message", - seq: 3, - delivery: undefined, - text: "staged identity", - messageId: "provider-message-2", - clientMessageId: "msg-client-2", + seq: 2, + timestamp: expect.any(String), + item: { type: "assistant_message", text: "output before provider echo" }, }, ]); - 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" }, + await manager.rewind(snapshot.id, "msg-client-1", "files"); + await manager.rewind(snapshot.id, "provider-native-message", "files"); + expect(client.session?.rewindMessageIds).toEqual([ + "provider-message-1", + "provider-native-message", ]); - - 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); @@ -8419,6 +8350,7 @@ test("authoritative timeline records a daemon-handled submitted prompt before it type: "user_message", text: "/handled", clientMessageId: "msg-client-daemon-handled", + messageId: "msg-client-daemon-handled", }, { type: "assistant_message", text: "Handled by the daemon" }, ]); @@ -8429,10 +8361,10 @@ test("authoritative timeline records a daemon-handled submitted prompt before it type: "user_message", text: "/handled", clientMessageId: "msg-client-daemon-handled", + messageId: "msg-client-daemon-handled", }, { 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 ea28126d36..a8dce41f1c 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -61,7 +61,6 @@ 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"; @@ -186,7 +185,6 @@ export type AgentManagerEvent = seq?: number; epoch?: string; timestamp?: string; - delivery?: AgentTimelineRow["delivery"]; }; export type AgentSubscriber = (event: AgentManagerEvent) => void; @@ -2082,8 +2080,7 @@ export class AgentManager { : undefined; if (options?.clientMessageId) { this.recordSubmittedPrompt(agent, prompt, options.clientMessageId, { - awaitProviderIdentity: true, - messageId: + providerMessageId: stagedSubmittedPromptEcho?.item.type === "user_message" ? stagedSubmittedPromptEcho.item.messageId : undefined, @@ -2493,11 +2490,23 @@ export class AgentManager { const lock = this.runs.createPendingRun(agentId); try { + const submittedRow = this.timelineStore + .getRows(agentId) + .find( + (row) => + row.item.type === "user_message" && + row.item.messageId === messageId && + row.item.clientMessageId === messageId, + ); + if (submittedRow && !submittedRow.providerMessageId) { + throw new Error("Cannot rewind before the provider acknowledges the submitted prompt"); + } + const providerMessageId = submittedRow?.providerMessageId ?? messageId; this.logger.info( { agentId, provider: agent.provider, messageId, mode }, "agent.rewind.start", ); - await invokeRewindCapability(agent.session, { messageId, mode }); + await invokeRewindCapability(agent.session, { messageId: providerMessageId, mode }); if (mode !== "files") { await this.hydrateTimelineFromProvider(agentId, { force: true, broadcast: true }); } @@ -3433,13 +3442,6 @@ 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); @@ -3457,9 +3459,6 @@ export class AgentManager { eventTurnId, options?.fromHistory === true, ); - if (isForegroundEvent && terminalDisposition !== "stale") { - this.finalizeSubmittedPrompt(agent, event); - } } const flags: StreamEventFlags = { shouldDispatchEvent: true, shouldNotifyWaiters: true }; @@ -3680,7 +3679,7 @@ export class AgentManager { if ( event.item.type === "user_message" && event.item.clientMessageId && - this.reconcileSubmittedPromptEcho(agent, event, event.item) + this.reconcileSubmittedPromptEcho(agent, event.item) ) { flags.shouldDispatchEvent = false; flags.shouldNotifyWaiters = false; @@ -3911,9 +3910,9 @@ export class AgentManager { item: AgentTimelineItem, provider: AgentProvider, turnId?: string, - delivery?: AgentTimelineRow["delivery"], + options?: { providerMessageId?: string }, ): AgentStreamEvent { - const row = this.recordTimeline(agentId, item, delivery ? { delivery } : undefined); + const row = this.recordTimeline(agentId, item, options); const event: AgentStreamEvent = { type: "timeline", item, @@ -3924,7 +3923,6 @@ export class AgentManager { seq: row.seq, epoch: this.timelineStore.getEpoch(agentId), timestamp: row.timestamp, - ...(row.delivery ? { delivery: row.delivery } : {}), }); if ( @@ -3946,7 +3944,7 @@ export class AgentManager { agent: ActiveManagedAgent, prompt: AgentPromptInput, clientMessageId: string, - options?: { awaitProviderIdentity: boolean; messageId: string | undefined }, + options?: { providerMessageId?: string }, ): void { if (this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId)) { return; @@ -3957,91 +3955,30 @@ export class AgentManager { type: "user_message", text: submittedPromptText(prompt), clientMessageId, - ...(options?.messageId ? { messageId: options.messageId } : {}), + messageId: clientMessageId, }; - const delivery = - options?.awaitProviderIdentity === true && options.messageId === undefined - ? AWAITING_PROVIDER_IDENTITY_DELIVERY - : undefined; - this.recordAndDispatchTimelineItem(agent.id, item, agent.provider, undefined, delivery); + this.recordAndDispatchTimelineItem(agent.id, item, agent.provider, undefined, options); } private reconcileSubmittedPromptEcho( agent: ActiveManagedAgent, - event: Extract, item: Extract, ): AgentTimelineRow | null { const { clientMessageId, messageId } = item; if (!clientMessageId) return null; const existing = this.timelineStore.getSubmittedUserMessage(agent.id, clientMessageId); if (!existing || existing.item.type !== "user_message") return null; - this.finalizeSubmittedPrompt(agent, event, clientMessageId, messageId); + if (messageId) { + const enriched = this.timelineStore.enrichSubmittedUserMessage( + agent.id, + clientMessageId, + messageId, + ); + if (enriched) this.enqueueDurableTimelineUpdate(agent.id, enriched); + } 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; - } - } - this.finalizeSubmittedPrompt(agent, event); - } - - private finalizeSubmittedPrompt( - agent: ActiveManagedAgent, - event: AgentStreamEvent, - correlatedClientMessageId?: string, - messageId?: string, - ): void { - let clientMessageId = correlatedClientMessageId; - if (!clientMessageId) { - const pending = this.timelineStore - .getRows(agent.id) - .findLast( - (row) => - row.delivery === AWAITING_PROVIDER_IDENTITY_DELIVERY && - row.item.type === "user_message", - ); - clientMessageId = - 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( agent: ActiveManagedAgent, provider: AgentProvider, @@ -4099,7 +4036,7 @@ export class AgentManager { private recordTimeline( agentId: string, item: AgentTimelineItem, - options?: { timestamp?: string; delivery?: AgentTimelineRow["delivery"] }, + options?: { timestamp?: string; providerMessageId?: string }, ): AgentTimelineRow { item = limitAgentTimelineItemContent(item); const row = this.timelineStore.append(agentId, item, options); @@ -4304,7 +4241,6 @@ export class AgentManager { seq?: number; epoch?: string; timestamp?: string; - delivery?: AgentTimelineRow["delivery"]; }, ): void { if (event.type === "timeline") { 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 8a189668b9..7440f9b146 100644 --- a/packages/server/src/server/agent/agent-timeline-store-types.ts +++ b/packages/server/src/server/agent/agent-timeline-store-types.ts @@ -1,12 +1,10 @@ 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; + readonly providerMessageId?: string; } export interface AgentTimelineCursor { 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 8668cf3b23..b033a856e4 100644 --- a/packages/server/src/server/agent/agent-timeline-store.test.ts +++ b/packages/server/src/server/agent/agent-timeline-store.test.ts @@ -2,41 +2,6 @@ import { describe, expect, it } from "vitest"; import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js"; describe("InMemoryAgentTimelineStore", () => { - 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", - }, - { delivery: "awaiting_provider_identity" }, - ); - - const updated = store.finalizeSubmittedUserMessage("agent-1", "client-message-1", { - messageId: "provider-message-1", - }); - - expect(updated).toMatchObject({ - seq: 1, - item: { - type: "user_message", - text: "Accepted prompt", - clientMessageId: "client-message-1", - 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, - maxSeq: 1, - nextSeq: 2, - }); - }); - it("returns a bounded reset window when an after cursor is behind retained history", () => { const store = new InMemoryAgentTimelineStore(); store.initialize("agent-1", { diff --git a/packages/server/src/server/agent/agent-timeline-store.ts b/packages/server/src/server/agent/agent-timeline-store.ts index 93938803ff..2500bc19a6 100644 --- a/packages/server/src/server/agent/agent-timeline-store.ts +++ b/packages/server/src/server/agent/agent-timeline-store.ts @@ -176,10 +176,10 @@ export class InMemoryAgentTimelineStore { return row ? cloneRow(row) : null; } - finalizeSubmittedUserMessage( + enrichSubmittedUserMessage( agentId: string, clientMessageId: string, - identity?: { messageId: string }, + providerMessageId: string, ): AgentTimelineRow | null { const state = this.requireState(agentId); const index = state.rows.findIndex( @@ -188,16 +188,12 @@ export class InMemoryAgentTimelineStore { candidate.item.clientMessageId === clientMessageId, ); const row = state.rows[index]; - if (!row || row.item.type !== "user_message" || row.delivery !== "awaiting_provider_identity") { + if (!row || row.item.type !== "user_message") { 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); + const enriched: AgentTimelineRow = { ...row, providerMessageId }; + state.rows[index] = enriched; + return cloneRow(enriched); } getEpoch(agentId: string): string { @@ -268,14 +264,14 @@ export class InMemoryAgentTimelineStore { append( agentId: string, item: AgentTimelineItem, - options?: { timestamp?: string; delivery?: AgentTimelineRow["delivery"] }, + options?: { timestamp?: string; providerMessageId?: string }, ): AgentTimelineRow { const state = this.requireState(agentId); const row: AgentTimelineRow = { seq: state.nextSeq, timestamp: options?.timestamp ?? new Date().toISOString(), item, - ...(options?.delivery ? { delivery: options.delivery } : {}), + ...(options?.providerMessageId ? { providerMessageId: options.providerMessageId } : {}), }; 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 deleted file mode 100644 index 693e0c3678..0000000000 --- a/packages/server/src/server/agent/submitted-prompt-delivery.ts +++ /dev/null @@ -1,69 +0,0 @@ -import type { AgentTimelineItem } from "./agent-sdk-types.js"; -import type { - AgentTimelineFetchOptions, - AgentTimelineFetchResult, - AgentTimelineRowDelivery, -} from "./agent-timeline-store-types.js"; -import { InMemoryAgentTimelineStore } from "./agent-timeline-store.js"; - -export const AWAITING_PROVIDER_IDENTITY_DELIVERY = "awaiting_provider_identity" as const; - -export function shouldDeliverSubmittedPromptRow( - delivery: AgentTimelineRowDelivery | undefined, - item: AgentTimelineItem, - supportsRevisions: boolean, -): boolean { - return ( - supportsRevisions || - delivery !== AWAITING_PROVIDER_IDENTITY_DELIVERY || - item.type !== "user_message" || - item.messageId !== undefined - ); -} - -export function projectSubmittedPromptTimeline( - timeline: AgentTimelineFetchResult, -): AgentTimelineFetchResult { - const pending = timeline.rows.at(-1); - if ( - !pending || - pending.seq !== timeline.window.maxSeq || - shouldDeliverSubmittedPromptRow(pending.delivery, pending.item, false) - ) { - return timeline; - } - const rows = timeline.rows.slice(0, -1); - const minSeq = rows[0]?.seq ?? 0; - const maxSeq = rows.at(-1)?.seq ?? 0; - return { - ...timeline, - window: { minSeq, maxSeq, nextSeq: maxSeq + 1 }, - hasOlder: false, - hasNewer: false, - rows, - }; -} - -export function selectSubmittedPromptTimelinePage( - fullTimeline: AgentTimelineFetchResult, - controlTimeline: AgentTimelineFetchResult, - options: AgentTimelineFetchOptions, -): AgentTimelineFetchResult { - const timeline = new InMemoryAgentTimelineStore(); - timeline.initialize("compatibility-projection", { - epoch: fullTimeline.epoch, - rows: fullTimeline.rows, - nextSeq: fullTimeline.window.nextSeq, - }); - const selected = timeline.fetch("compatibility-projection", { - ...options, - ...(controlTimeline.reset ? { direction: "tail", cursor: undefined } : {}), - }); - return { - ...selected, - direction: controlTimeline.direction, - reset: controlTimeline.reset, - staleCursor: controlTimeline.staleCursor, - gap: controlTimeline.gap, - }; -} diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index 71c0928268..5677750165 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -4880,179 +4880,6 @@ 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 session = createSessionForTest({ - targetedMessages, - agentManager: { - 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", - }); - 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", - }); - 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 }, - ]); -}); - -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({ - messages, - agentManager: { - subscribe: vi.fn((next: (event: AgentManagerEvent) => void) => { - listener = next; - return () => {}; - }), - }, - }); - session.updateClientCapabilities(capabilities); - const item = { - type: "user_message" as const, - text: "submitted prompt", - clientMessageId: "client-message-1", - }; - - 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", () => { 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 129d7037de..0e2e6e2dcc 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -76,11 +76,6 @@ import type { AgentTimelineFetchResult, ManagedAgent, } from "./agent/agent-manager.js"; -import { - projectSubmittedPromptTimeline, - selectSubmittedPromptTimelinePage, - 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 { @@ -1064,16 +1059,6 @@ export class Session { serializedEvent: Extract["payload"]["event"], ): void { if (this.clientCapabilitiesBySource.size === 0 || !this.onMessageToSource) { - if ( - serializedEvent.type === "timeline" && - !shouldDeliverSubmittedPromptRow( - event.delivery, - serializedEvent.item, - this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions), - ) - ) { - return; - } if (this.usesSelectiveTimelineDelivery() && serializedEvent.type === "attention_required") { this.emit({ type: "agent_attention_required", @@ -1099,16 +1084,6 @@ export class Session { for (const [source, capabilities] of this.clientCapabilitiesBySource) { const supportsSelectiveDelivery = capabilities.has(CLIENT_CAPS.selectiveAgentTimeline); - if ( - serializedEvent.type === "timeline" && - !shouldDeliverSubmittedPromptRow( - event.delivery, - serializedEvent.item, - capabilities.has(CLIENT_CAPS.canonicalSubmittedPromptRevisions), - ) - ) { - continue; - } if (supportsSelectiveDelivery && serializedEvent.type === "attention_required") { this.onMessageToSource(source, { type: "agent_attention_required", @@ -6182,34 +6157,13 @@ export class Session { cursor, limit: pageLimit, }); - const supportsSubmittedPromptRevisions = source - ? this.supportsForSource(CLIENT_CAPS.canonicalSubmittedPromptRevisions, source) - : this.supports(CLIENT_CAPS.canonicalSubmittedPromptRevisions); - let fetchedFullTimeline: AgentTimelineFetchResult | undefined; - if (!supportsSubmittedPromptRevisions) { - fetchedFullTimeline = - direction === "tail" && pageLimit === 0 - ? fetchedControlTimeline - : this.agentManager.fetchTimeline(msg.agentId, { direction: "tail", limit: 0 }); - } - const fullTimeline = fetchedFullTimeline - ? projectSubmittedPromptTimeline(fetchedFullTimeline) - : undefined; - const controlTimeline = fullTimeline - ? selectSubmittedPromptTimelinePage(fullTimeline, fetchedControlTimeline, { - direction, - cursor, - limit: pageLimit, - }) - : fetchedControlTimeline; const selectedTimeline = this.selectTimelineProjection({ agentId: msg.agentId, projection, - controlTimeline, + controlTimeline: fetchedControlTimeline, direction, ...(cursor ? { cursor } : {}), pageLimit, - ...(fullTimeline ? { fullTimeline } : {}), }); const startCursor = selectedTimeline.startSeq !== null @@ -6230,9 +6184,9 @@ export class Session { direction, projection, epoch: selectedTimeline.timeline.epoch, - reset: controlTimeline.reset, - staleCursor: controlTimeline.staleCursor, - gap: controlTimeline.gap, + reset: fetchedControlTimeline.reset, + staleCursor: fetchedControlTimeline.staleCursor, + gap: fetchedControlTimeline.gap, window: selectedTimeline.timeline.window, startCursor, endCursor, diff --git a/packages/server/src/server/wire-compat.test.ts b/packages/server/src/server/wire-compat.test.ts index ce5b4d952c..78af3adecc 100644 --- a/packages/server/src/server/wire-compat.test.ts +++ b/packages/server/src/server/wire-compat.test.ts @@ -404,7 +404,6 @@ describe("wire compatibility", () => { protocolVersion: 1, capabilities: { [CLIENT_CAPS.projectUpdates]: true, - [CLIENT_CAPS.canonicalSubmittedPromptRevisions]: true, }, }); @@ -422,7 +421,6 @@ describe("wire compatibility", () => { protocolVersion: 1, capabilities: { project_updates: true, - canonical_submitted_prompt_revisions: true, }, }, ]); @@ -489,118 +487,6 @@ describe("wire compatibility", () => { 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 = ( - seq: number, - options?: { delivery?: AgentTimelineRow["delivery"]; messageId?: string }, - ) => ({ - seq, - timestamp, - ...(options?.delivery ? { delivery: options.delivery } : {}), - item: { - type: "user_message" as const, - text: `prompt ${seq}`, - clientMessageId: `client-message-${seq}`, - ...(options?.messageId ? { messageId: options.messageId } : {}), - }, - }); - - const pending = await emitTimelineResponse({ - rows: [prompt(1, { 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(1, { 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(1)], - request: { projection: "canonical" }, - }); - expect(daemonHandled.payload.entries).toHaveLength(1); - - const rowsWithPendingSuffix = [ - prompt(1), - prompt(2, { delivery: "awaiting_provider_identity" }), - ]; - for (const projection of ["canonical", "projected"] as const) { - for (const request of [ - { direction: "tail" as const, limit: 1 }, - { direction: "after" as const, cursor: { epoch: "epoch-1", seq: 0 }, limit: 1 }, - { direction: "after" as const, cursor: { epoch: "epoch-1", seq: 0 }, limit: 0 }, - { direction: "before" as const, cursor: { epoch: "epoch-1", seq: 2 }, limit: 1 }, - { direction: "before" as const, cursor: { epoch: "epoch-1", seq: 2 }, limit: 0 }, - ]) { - const page = await emitTimelineResponse({ - rows: rowsWithPendingSuffix, - request: { projection, ...request }, - }); - expect(pageShape(page)).toEqual([[1, 1, 2], 1, 1, false, false, [1]]); - } - } - - for (const { rows, cursor, flags } of [ - { - rows: rowsWithPendingSuffix, - cursor: { epoch: "stale-epoch", seq: 0 }, - flags: [true, true, false], - }, - { - rows: [prompt(5), prompt(6, { delivery: "awaiting_provider_identity" })], - cursor: { epoch: "epoch-1", seq: 1 }, - flags: [true, false, true], - }, - ] as const) { - const page = await emitTimelineResponse({ - rows, - request: { projection: "canonical", direction: "after", cursor, limit: 1 }, - }); - expect([page.payload.reset, page.payload.staleCursor, page.payload.gap]).toEqual(flags); - expect(page.payload.entries.map((entry) => entry.seqStart)).toEqual([rows[0].seq]); - } - - const finalizedSuffix = await emitTimelineResponse({ - rows: [prompt(1), prompt(2, { messageId: "provider-message-2" })], - request: { projection: "canonical", direction: "tail", limit: 1 }, - }); - expect(finalizedSuffix.payload.entries.map((entry) => entry.seqStart)).toEqual([2]); - expect(finalizedSuffix.payload.entries[0]?.item).toMatchObject({ - type: "user_message", - messageId: "provider-message-2", - }); - }); - test("sub_agent tool-call payload still parses against the v0.1.65-beta.3 schema", () => { const parsed = LegacySubAgentToolCallSchema.parse({ type: "tool_call",