From a1ff4773e57984ff5f2812abf2f0b9bfb947d92f Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 31 Jul 2026 00:43:22 +0200 Subject: [PATCH] fix(timeline): make submitted prompts canonical Submission settlement previously depended on status updates that could arrive after the optimistic row. Persist every accepted prompt in the canonical timeline and keep the footer active until RPC and canonical acknowledgement converge. --- docs/timeline-sync.md | 24 +++- .../app/e2e/agent-consecutive-turns.spec.ts | 6 - .../app/e2e/agent-message-submission.spec.ts | 94 +++++++++++++++ .../app/e2e/helpers/daemon-websocket-gate.ts | 24 ++-- .../out-of-band-command.codex.real.spec.ts | 23 ---- packages/app/src/agent-stream/view.tsx | 6 +- packages/app/src/composer/actions.test.ts | 2 +- packages/app/src/composer/actions.ts | 8 +- .../app/src/composer/draft/create-flow.ts | 8 +- .../app/src/composer/submission/model.test.ts | 38 +++--- packages/app/src/composer/submission/model.ts | 20 +--- .../app/src/composer/submission/writer.ts | 74 ++++++++++-- packages/app/src/runtime/host-runtime.test.ts | 7 +- packages/app/src/stores/session-store.test.ts | 52 +++++++++ packages/app/src/stores/session-store.ts | 108 +++++++----------- .../timeline/session-stream-reducers.test.ts | 20 +++- packages/app/src/types/stream.ts | 5 +- packages/client/src/daemon-client.ts | 8 +- packages/protocol/src/messages.ts | 4 +- .../src/server/agent/agent-manager.test.ts | 63 +++++++++- .../server/src/server/agent/agent-manager.ts | 17 ++- .../server/src/server/agent/agent-prompt.ts | 2 +- .../server/agent/providers/acp-agent.test.ts | 9 ++ .../agent/providers/claude/agent.test.ts | 16 +++ .../providers/codex-app-server-agent.test.ts | 7 ++ .../providers/mock-load-test-agent.test.ts | 26 +++++ .../agent/providers/mock-load-test-agent.ts | 74 ++++++------ .../server/agent/providers/omp/agent.test.ts | 1 + .../providers/omp/test-utils/omp-harness.ts | 4 + .../agent/providers/opencode-agent.test.ts | 23 +++- .../server/agent/providers/pi/agent.test.ts | 6 + packages/server/src/server/session.ts | 2 - .../server/src/server/websocket-server.ts | 2 + 33 files changed, 551 insertions(+), 232 deletions(-) diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index 4a967fe118..a56fa3cc83 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -126,11 +126,25 @@ back a prompt already observed canonically; it does not own whether the local pr visible. The daemon's accepted response already waits for the correlated run start, but its response and the -directory update reach client state separately. An accepted transaction remains active until the -directory observes that run or canonical ingestion acknowledges the prompt, bridging those ordered -authorities without inspecting timeline snapshots. Either signal clears only an RPC-accepted -transaction, regardless of which arrived first; it cannot settle a fresh send. -Overlapping sends settle independently rather than collapsing to one newest pending message. +canonical submitted row reach client state independently and may arrive in either order. The +transaction therefore records RPC acceptance and provider acknowledgement as separate phases. RPC +acceptance marks the record accepted unless provider acknowledgement already arrived; canonical +acknowledgement marks the record acknowledged unless RPC acceptance already arrived. Only the second +authority removes the transaction. Directory status never settles a submission. Overlapping sends +settle independently rather than collapsing to one newest pending message. + +Daemons advertising `server_info.features.canonicalSubmittedPrompts` guarantee that every accepted +prompt carrying a client message id is recorded and streamed as a canonical `user_message` with that +same id. This includes daemon-handled commands that do not allocate a foreground turn; their submitted +row is recorded before handler output. The app creates two-phase submission records only for hosts with +this capability. Older hosts keep the shipped untracked optimistic-row behavior and roll that row back +on RPC rejection. + +Turn liveness is the union of an unsettled submission and an open stream turn. `turn_started` opens the +turn; ordered `turn_completed`, `turn_failed`, and `turn_canceled` events close it. A running directory +transition may seed or raise an open turn during hydration or after a selective stream subscription +lapses, but directory status never closes one or settles a submission. Disconnect and replica removal +are the other destructive close boundaries. Canonical submitted user rows carry the provider's `messageId` and Paseo's optional `clientMessageId`. The user-message producer reconciles them by `clientMessageId`, adds provider diff --git a/packages/app/e2e/agent-consecutive-turns.spec.ts b/packages/app/e2e/agent-consecutive-turns.spec.ts index ff2f3cdef2..f35b349c9b 100644 --- a/packages/app/e2e/agent-consecutive-turns.spec.ts +++ b/packages/app/e2e/agent-consecutive-turns.spec.ts @@ -533,8 +533,6 @@ async function recordDelayedRunningTransition( const prompt = "Second prompt keeps streaming."; gate.holdNextAgentUpdate(agent.agentId, "running"); - gate.setMessageSubmissionDispositionStripped(true); - gate.setAgentStreamEventSuppressed("turn_started", true); gate.setAgentStreamItemSuppressed("assistant_message", true); await recordTurnFrames(page, prompt); await submitMessage(page, prompt); @@ -553,14 +551,10 @@ async function recordDelayedRunningTransition( released = true; await waitForRecordedFrames(page, "turn-running", 3); const frames = await stopTurnFrameRecording(page); - gate.setMessageSubmissionDispositionStripped(false); - gate.setAgentStreamEventSuppressed("turn_started", false); gate.setAgentStreamItemSuppressed("assistant_message", false); return frames; } finally { if (!released) gate.releaseHeldServerMessage(); - gate.setMessageSubmissionDispositionStripped(false); - gate.setAgentStreamEventSuppressed("turn_started", false); gate.setAgentStreamItemSuppressed("assistant_message", false); } } diff --git a/packages/app/e2e/agent-message-submission.spec.ts b/packages/app/e2e/agent-message-submission.spec.ts index 9a07262bb9..0026448eaf 100644 --- a/packages/app/e2e/agent-message-submission.spec.ts +++ b/packages/app/e2e/agent-message-submission.spec.ts @@ -11,6 +11,7 @@ import { fillComposerDraft, sendDraftToQueue, startRunningMockAgent, + submitMessage, } from "./helpers/composer"; import { openAgentRoute, seedMockAgentWorkspace } from "./helpers/mock-agent"; import { seedWorkspace } from "./helpers/seed-client"; @@ -517,6 +518,89 @@ async function expectCanonicalOrderWinsAcrossOverlappingClients( } } +async function expectDaemonHandledSubmissionSurvivesReload( + page: Page, + testInfo: { workerIndex: number }, +): Promise { + const agent = await seedMockAgentWorkspace({ + repoPrefix: `daemon-handled-submission-${testInfo.workerIndex}-`, + title: "Daemon-handled submission", + }); + const prompt = "/mock handled-command"; + try { + await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId }); + await expectComposerVisible(page); + await expectAgentIdle(page); + await submitMessage(page, prompt); + const command = page.getByTestId("user-message").filter({ hasText: prompt }); + await expect(command).toHaveAttribute("aria-busy", "false"); + await expect(page.getByText("Mock command handled", { exact: true })).toBeVisible(); + await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); + + await page.reload(); + await expectComposerVisible(page); + await expect(command).toBeVisible(); + await expect(command).toHaveAttribute("aria-busy", "false"); + await expect(page.getByText("Mock command handled", { exact: true })).toBeVisible(); + await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); + } finally { + await agent.cleanup(); + } +} + +async function expectOldHostSubmissionBehavior( + page: Page, + testInfo: { workerIndex: number }, +): Promise { + const gate = await installDaemonWebSocketGate(page); + gate.setCanonicalSubmittedPromptsStripped(true); + const agent = await seedMockAgentWorkspace({ + repoPrefix: `old-host-submission-${testInfo.workerIndex}-`, + title: "Old host submission", + }); + const prompt = "/mock handled-command"; + try { + await openAgentRoute(page, { workspaceId: agent.workspaceId, agentId: agent.agentId }); + await expectComposerVisible(page); + await expectAgentIdle(page); + gate.holdNextClientRequest("send_agent_message_request"); + const composer = page.getByRole("textbox", { name: "Message agent..." }).first(); + await composer.fill(prompt); + await composer.press("Enter"); + const nextFrame = await composer.evaluate( + (_, submittedPrompt) => + new Promise<{ rowPresent: boolean; ariaBusy: string | null; workingPresent: boolean }>( + (resolve) => { + requestAnimationFrame(() => { + const row = Array.from( + document.querySelectorAll('[data-testid="user-message"]'), + ).find((candidate) => candidate.textContent?.includes(submittedPrompt)); + resolve({ + rowPresent: Boolean(row), + ariaBusy: row?.getAttribute("aria-busy") ?? null, + workingPresent: Boolean( + document.querySelector('[data-testid="turn-working-indicator"]'), + ), + }); + }); + }, + ), + prompt, + ); + expect(nextFrame).toEqual({ rowPresent: true, ariaBusy: "false", workingPresent: false }); + + await gate.waitForHeldClientRequest(); + gate.releaseHeldClientRequest(); + const command = page.getByTestId("user-message").filter({ hasText: prompt }); + await expect(command).toHaveAttribute("aria-busy", "false"); + await expect(page.getByText("Mock command handled", { exact: true })).toBeVisible(); + await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); + } finally { + gate.restore(); + await agent.cleanup(); + } +} + async function expectRenderedBefore(first: Locator, second: Locator): Promise { const secondElement = await second.elementHandle(); if (!secondElement) throw new Error("Expected the second timeline item to be rendered"); @@ -711,4 +795,14 @@ test.describe("Agent message submission", () => { }, testInfo) => { await expectCanonicalOrderWinsAcrossOverlappingClients(page, testInfo); }); + + test("keeps a daemon-handled submitted row after authoritative reload", async ({ + page, + }, testInfo) => { + await expectDaemonHandledSubmissionSurvivesReload(page, testInfo); + }); + + test("uses untracked optimistic rows for an old host", async ({ page }, testInfo) => { + await expectOldHostSubmissionBehavior(page, testInfo); + }); }); diff --git a/packages/app/e2e/helpers/daemon-websocket-gate.ts b/packages/app/e2e/helpers/daemon-websocket-gate.ts index 783228e9c3..0ab8144f3b 100644 --- a/packages/app/e2e/helpers/daemon-websocket-gate.ts +++ b/packages/app/e2e/helpers/daemon-websocket-gate.ts @@ -73,21 +73,19 @@ function stripAssistantMessageId( return JSON.stringify(envelope); } -function stripMessageSubmissionDisposition( +function stripCanonicalSubmittedPrompts( message: string | Buffer, enabled: boolean, messageType: unknown, ): string | Buffer { - if (!enabled || messageType !== "send_agent_message_response" || typeof message !== "string") { - return message; - } + if (!enabled || messageType !== "status" || typeof message !== "string") return message; const envelope = JSON.parse(message) as { - message?: { payload?: Record }; - payload?: Record; + message?: { payload?: { status?: unknown; features?: Record } }; + payload?: { status?: unknown; features?: Record }; }; const payload = envelope.message?.payload ?? envelope.payload; - if (!payload) return message; - delete payload.outOfBand; + if (payload?.status !== "server_info" || !payload.features) return message; + delete payload.features.canonicalSubmittedPrompts; return JSON.stringify(envelope); } @@ -164,7 +162,7 @@ export async function installDaemonWebSocketGate(page: Page) { let suppressAgentStream = false; let forceTimelineEpochReset = false; let stripAssistantMessageIds = false; - let stripSubmissionDisposition = false; + let stripCanonicalSubmittedPromptsFeature = false; let heldClientRequestType: string | null = null; let heldClientRequest: { server: WebSocketRoute; message: string | Buffer } | null = null; let resolveHeldClientRequest: (() => void) | null = null; @@ -240,9 +238,9 @@ export async function installDaemonWebSocketGate(page: Page) { stripAssistantMessageIds, serverMessage?.type, ); - outboundMessage = stripMessageSubmissionDisposition( + outboundMessage = stripCanonicalSubmittedPrompts( outboundMessage, - stripSubmissionDisposition, + stripCanonicalSubmittedPromptsFeature, serverMessage?.type, ); const shouldForceTimelineReset = @@ -428,8 +426,8 @@ export async function installDaemonWebSocketGate(page: Page) { setAssistantMessageIdsStripped(stripped: boolean): void { stripAssistantMessageIds = stripped; }, - setMessageSubmissionDispositionStripped(stripped: boolean): void { - stripSubmissionDisposition = stripped; + setCanonicalSubmittedPromptsStripped(stripped: boolean): void { + stripCanonicalSubmittedPromptsFeature = stripped; }, setAgentStreamSuppressed(suppressed: boolean): void { suppressAgentStream = suppressed; diff --git a/packages/app/e2e/out-of-band-command.codex.real.spec.ts b/packages/app/e2e/out-of-band-command.codex.real.spec.ts index 2df34398d8..cbec6733c1 100644 --- a/packages/app/e2e/out-of-band-command.codex.real.spec.ts +++ b/packages/app/e2e/out-of-band-command.codex.real.spec.ts @@ -4,7 +4,6 @@ import path from "node:path"; import { expect, test } from "./fixtures"; import { submitMessage } from "./helpers/composer"; import { cleanupRewindFlow, launchAgent, type AgentHandle } from "./helpers/rewind-flow"; -import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate"; test.describe("Codex out-of-band commands", () => { test.setTimeout(300_000); @@ -27,26 +26,4 @@ test.describe("Codex out-of-band commands", () => { await cleanupRewindFlow({ handle, cwd }); } }); - - test("settles the submitted row when an older daemon omits submission disposition", async ({ - page, - }) => { - const gate = await installDaemonWebSocketGate(page); - gate.setMessageSubmissionDispositionStripped(true); - const cwd = realpathSync(mkdtempSync(path.join(tmpdir(), "paseo-codex-command-compat-"))); - let handle: AgentHandle | undefined; - - try { - handle = await launchAgent({ page, provider: "codex", cwd, mode: "full-access" }); - await submitMessage(page, "/goal clear"); - const command = page.getByTestId("user-message").filter({ hasText: "/goal clear" }); - - await expect(command).toBeVisible(); - await expect(command).toHaveAttribute("aria-busy", "false", { timeout: 30_000 }); - await expect(page.getByTestId("turn-working-indicator")).toHaveCount(0); - } finally { - gate.restore(); - await cleanupRewindFlow({ handle, cwd }); - } - }); }); diff --git a/packages/app/src/agent-stream/view.tsx b/packages/app/src/agent-stream/view.tsx index 30d71e4bb0..9deb5f7d2d 100644 --- a/packages/app/src/agent-stream/view.tsx +++ b/packages/app/src/agent-stream/view.tsx @@ -565,11 +565,7 @@ const AgentStreamViewComponent = forwardRef 0, - hasActiveTurn, - ].some(Boolean); + const isTurnActive = pendingMessageSubmissions.length > 0 || hasActiveTurn; // Keep retained history outside the 48ms live-head flush path. const preparedToolCallHistory = useMemo( () => prepareToolCallHistory(toolCallDetailLevel, effectiveStreamItems), diff --git a/packages/app/src/composer/actions.test.ts b/packages/app/src/composer/actions.test.ts index 8896650f00..1597b82025 100644 --- a/packages/app/src/composer/actions.test.ts +++ b/packages/app/src/composer/actions.test.ts @@ -223,7 +223,7 @@ function createFakeStream(initialHead: Map = new Map()): F const current = readSubmission(fake, agentId); writeSubmission(fake, agentId, { ...current, - submissions: acceptMessageSubmission(current.submissions, clientMessageId, true, false), + submissions: acceptMessageSubmission(current.submissions, clientMessageId), }); }, reject: (agentId, clientMessageId) => { diff --git a/packages/app/src/composer/actions.ts b/packages/app/src/composer/actions.ts index 08cd0d03e7..d622269aa2 100644 --- a/packages/app/src/composer/actions.ts +++ b/packages/app/src/composer/actions.ts @@ -46,7 +46,7 @@ export interface ComposerSendClient { images: Array<{ data: string; mimeType: string }>; attachments: ReturnType["attachments"]; }, - ) => Promise; + ) => Promise; uploadFile: (input: { fileName: string; mimeType: string; bytes: Uint8Array }) => Promise<{ requestId: string; file: { @@ -67,7 +67,7 @@ export interface ComposerCancelClient { export interface MessageSubmissionWriter { begin: (agentId: string, message: UserMessageItem) => void; - accept: (agentId: string, clientMessageId: string, outOfBand: boolean | undefined) => void; + accept: (agentId: string, clientMessageId: string) => void; reject: (agentId: string, clientMessageId: string) => MessageSubmissionRejectionOutcome; } @@ -183,12 +183,12 @@ export async function dispatchComposerAgentMessage( input.submission.begin(input.agentId, userMessage); try { const imagesData = await input.encodeImages(wirePayload.images); - const result = await input.client.sendAgentMessage(input.agentId, input.text, { + await input.client.sendAgentMessage(input.agentId, input.text, { messageId: clientMessageId, images: imagesData ?? [], attachments: wirePayload.attachments, }); - input.submission.accept(input.agentId, clientMessageId, result?.outOfBand); + input.submission.accept(input.agentId, clientMessageId); } catch (error) { const outcome = input.submission.reject(input.agentId, clientMessageId); if (outcome === "accepted") return; diff --git a/packages/app/src/composer/draft/create-flow.ts b/packages/app/src/composer/draft/create-flow.ts index ea3638da6c..88046e7e60 100644 --- a/packages/app/src/composer/draft/create-flow.ts +++ b/packages/app/src/composer/draft/create-flow.ts @@ -6,6 +6,7 @@ import { splitComposerAttachmentsForSubmit, } from "@/composer/attachments/submit"; import { useCreateFlowStore } from "@/stores/create-flow-store"; +import { handoffCreatedAgentMessageSubmission } from "@/composer/submission/writer"; import { useSessionStore } from "@/stores/session-store"; import { createUserMessage, @@ -127,10 +128,6 @@ export function useDraftAgentCreateFlow({ const updatePendingAgentId = useCreateFlowStore((state) => state.updateAgentId); const markPendingCreateLifecycle = useCreateFlowStore((state) => state.markLifecycle); const clearPendingCreateAttempt = useCreateFlowStore((state) => state.clear); - const handoffCreatedAgentUserMessage = useSessionStore( - (state) => state.handoffCreatedAgentUserMessage, - ); - const formErrorMessage = machine.tag === "draft" ? machine.errorMessage : ""; const isSubmitting = machine.tag === "creating"; @@ -202,7 +199,7 @@ export function useDraftAgentCreateFlow({ if (createResult.agentId) { updatePendingAgentId({ draftId, agentId: createResult.agentId }); - handoffCreatedAgentUserMessage( + handoffCreatedAgentMessageSubmission( pendingServerId, createResult.agentId, createUserMessage({ @@ -228,7 +225,6 @@ export function useDraftAgentCreateFlow({ } }, [ - handoffCreatedAgentUserMessage, clearPendingCreateAttempt, createRequest, draftId, diff --git a/packages/app/src/composer/submission/model.test.ts b/packages/app/src/composer/submission/model.test.ts index cab44f8a3e..ba4a083a89 100644 --- a/packages/app/src/composer/submission/model.test.ts +++ b/packages/app/src/composer/submission/model.test.ts @@ -4,7 +4,6 @@ import { beginMessageSubmission, getActiveMessageSubmissions, getSendingClientMessageIds, - observeAcceptedMessageSubmissionsRunning, observeMessageSubmissionCanonical, rejectMessageSubmission, } from "./model"; @@ -26,13 +25,14 @@ describe("message submission transactions", () => { expect(getSendingClientMessageIds(both)).toEqual(["client-1", "client-2"]); }); - it("removes only the RPC-accepted transaction", () => { + it("removes only the provider-acknowledged transaction on RPC acceptance", () => { const both = beginMessageSubmission( beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }), { clientMessageId: "client-2", submittedAt }, ); + const acknowledged = observeMessageSubmissionCanonical(both, ["client-1"]); - expect(acceptMessageSubmission(both, "client-1", true, false)).toEqual([ + expect(acceptMessageSubmission(acknowledged, "client-1")).toEqual([ { clientMessageId: "client-2", submittedAt, @@ -42,34 +42,34 @@ describe("message submission transactions", () => { ]); }); - it("bridges an accepted RPC until the correlated running state is observed", () => { + it("keeps an accepted RPC active until canonical acknowledgement", () => { const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); - const accepted = acceptMessageSubmission(sending, "client-1", false, false); + const accepted = acceptMessageSubmission(sending, "client-1"); expect(getActiveMessageSubmissions(accepted)).toHaveLength(1); expect(accepted[0].rpcAccepted).toBe(true); - expect(observeAcceptedMessageSubmissionsRunning(accepted)).toEqual([]); + }); + + it("does not settle RPC acceptance from directory running status", () => { + const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); + + expect(acceptMessageSubmission(sending, "client-1")).toEqual([ + { + clientMessageId: "client-1", + submittedAt, + rpcAccepted: true, + providerAcknowledged: false, + }, + ]); }); it("settles an accepted RPC when provider acknowledgement arrives after running was missed", () => { const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); - const accepted = acceptMessageSubmission(sending, "client-1", false, false); + const accepted = acceptMessageSubmission(sending, "client-1"); expect(observeMessageSubmissionCanonical(accepted, ["client-1"])).toEqual([]); }); - it("settles an explicitly out-of-band acceptance without lifecycle inference", () => { - const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); - - expect(acceptMessageSubmission(sending, "client-1", false, true)).toEqual([]); - }); - - it("settles an idle acceptance from a daemon without submission disposition", () => { - const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); - - expect(acceptMessageSubmission(sending, "client-1", false, undefined)).toEqual([]); - }); - it("records provider acknowledgement without settling another transaction", () => { const both = beginMessageSubmission( beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }), diff --git a/packages/app/src/composer/submission/model.ts b/packages/app/src/composer/submission/model.ts index 250e3a581f..db8bafc1ea 100644 --- a/packages/app/src/composer/submission/model.ts +++ b/packages/app/src/composer/submission/model.ts @@ -44,23 +44,12 @@ export function beginMessageSubmission( export function acceptMessageSubmission( submissions: readonly MessageSubmissionRecord[], clientMessageId: string, - isAgentRunning: boolean, - outOfBand: boolean | undefined, ): MessageSubmissionRecord[] { const index = submissions.findIndex( (submission) => submission.clientMessageId === clientMessageId, ); if (index < 0) return submissions as MessageSubmissionRecord[]; - // COMPAT(messageSubmissionDisposition): daemons before v0.2.3 omitted outOfBand. - // An idle response cannot distinguish a normal send that has not reported running yet - // from an out-of-band send; the separate active-turn lease bridges that gap. Remove after 2027-01-27. - const legacyOutOfBand = outOfBand === undefined && !isAgentRunning; - if ( - outOfBand === true || - legacyOutOfBand || - isAgentRunning || - submissions[index].providerAcknowledged - ) { + if (submissions[index].providerAcknowledged) { return submissions.filter((_, submissionIndex) => submissionIndex !== index); } if (submissions[index].rpcAccepted) return submissions as MessageSubmissionRecord[]; @@ -69,13 +58,6 @@ export function acceptMessageSubmission( return next; } -export function observeAcceptedMessageSubmissionsRunning( - submissions: readonly MessageSubmissionRecord[], -): MessageSubmissionRecord[] { - const next = submissions.filter((submission) => !submission.rpcAccepted); - return next.length === submissions.length ? (submissions as MessageSubmissionRecord[]) : next; -} - export function observeMessageSubmissionCanonical( submissions: readonly MessageSubmissionRecord[], clientMessageIds: readonly string[], diff --git a/packages/app/src/composer/submission/writer.ts b/packages/app/src/composer/submission/writer.ts index 9b41c84e75..b9211cedef 100644 --- a/packages/app/src/composer/submission/writer.ts +++ b/packages/app/src/composer/submission/writer.ts @@ -1,20 +1,80 @@ import type { MessageSubmissionWriter } from "@/composer/actions"; import { useSessionStore } from "@/stores/session-store"; +import { + appendSubmittedUserMessage, + removeSubmittedUserMessage, + type UserMessageItem, +} from "@/types/stream"; + +function appendUntrackedSubmission(serverId: string, agentId: string, message: UserMessageItem) { + const session = useSessionStore.getState().sessions[serverId]; + if (!session) return; + const tail = session.agentStreamTail.get(agentId) ?? []; + const head = session.agentStreamHead.get(agentId) ?? []; + useSessionStore + .getState() + .setAgentStreamState(serverId, agentId, appendSubmittedUserMessage({ tail, head, message })); +} + +function removeUntrackedSubmission( + serverId: string, + agentId: string, + clientMessageId: string, +): void { + const session = useSessionStore.getState().sessions[serverId]; + if (!session) return; + const tail = session.agentStreamTail.get(agentId) ?? []; + const head = session.agentStreamHead.get(agentId) ?? []; + useSessionStore + .getState() + .setAgentStreamState( + serverId, + agentId, + removeSubmittedUserMessage({ tail, head, clientMessageId }), + ); +} + +function createUntrackedMessageSubmissionWriter(serverId: string): MessageSubmissionWriter { + return { + begin: (agentId, message) => appendUntrackedSubmission(serverId, agentId, message), + accept: () => undefined, + reject: (agentId, clientMessageId) => { + removeUntrackedSubmission(serverId, agentId, clientMessageId); + return "rejected"; + }, + }; +} /** - * Binds the submission lifecycle to a host session. Every path that sends a message to an - * agent — composer send, queued send-now, automatic queue drain — goes through this so a - * submitted row and its pending state are always created together. + * Binds message submission presentation to a host session. Capable hosts use the tracked + * two-phase lifecycle; older hosts preserve the untracked optimistic-row behavior. */ export function createMessageSubmissionWriter(serverId: string): MessageSubmissionWriter { + const supportsCanonicalSubmittedPrompts = + useSessionStore.getState().sessions[serverId]?.serverInfo?.features + ?.canonicalSubmittedPrompts === true; + if (!supportsCanonicalSubmittedPrompts) { + return createUntrackedMessageSubmissionWriter(serverId); + } return { begin: (agentId, message) => useSessionStore.getState().beginAgentMessageSubmission(serverId, agentId, message), - accept: (agentId, clientMessageId, outOfBand) => - useSessionStore - .getState() - .acceptAgentMessageSubmission(serverId, agentId, clientMessageId, outOfBand), + accept: (agentId, clientMessageId) => + useSessionStore.getState().acceptAgentMessageSubmission(serverId, agentId, clientMessageId), reject: (agentId, clientMessageId) => useSessionStore.getState().rejectAgentMessageSubmission(serverId, agentId, clientMessageId), }; } + +export function handoffCreatedAgentMessageSubmission( + serverId: string, + agentId: string, + message: UserMessageItem, +): boolean { + const supportsCanonicalSubmittedPrompts = + useSessionStore.getState().sessions[serverId]?.serverInfo?.features + ?.canonicalSubmittedPrompts === true; + return useSessionStore + .getState() + .handoffCreatedAgentUserMessage(serverId, agentId, message, supportsCanonicalSubmittedPrompts); +} diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 6f73dd640f..7507dd17d2 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -89,7 +89,6 @@ class FakeDaemonClient { if (response) await response; const failure = this.sendAgentMessageFailures.shift(); if (failure) throw failure; - return {}; } async waitForSentMessages(count: number): Promise { @@ -2280,6 +2279,12 @@ describe("HostRuntimeStore", () => { }); const sessionStore = useSessionStore.getState(); sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1); + sessionStore.updateSessionServerInfo(host.serverId, { + serverId: host.serverId, + hostname: null, + version: null, + features: { canonicalSubmittedPrompts: true }, + }); sessionStore.setQueuedMessages( host.serverId, new Map([ diff --git a/packages/app/src/stores/session-store.test.ts b/packages/app/src/stores/session-store.test.ts index 369097d267..68da911630 100644 --- a/packages/app/src/stores/session-store.test.ts +++ b/packages/app/src/stores/session-store.test.ts @@ -7,6 +7,7 @@ import { normalizeWorkspaceDescriptor, selectAgentTimelineState, useSessionStore, + type Agent, type WorkspaceDescriptor, } from "./session-store"; import type { StreamItem } from "../types/stream"; @@ -41,6 +42,37 @@ function initializeTestSession(): void { useSessionStore.getState().initializeSession("test-server", null as unknown as DaemonClient); } +function createAgent(status: Agent["status"]): Agent { + const timestamp = new Date("2026-07-30T10:00:00.000Z"); + return { + serverId: "test-server", + id: "agent-1", + provider: "codex", + status, + createdAt: timestamp, + updatedAt: timestamp, + lastUserMessageAt: null, + lastActivityAt: timestamp, + capabilities: { + supportsStreaming: true, + supportsSessionPersistence: true, + supportsDynamicModes: true, + supportsMcpServers: true, + supportsReasoningStream: true, + supportsToolInvocations: true, + }, + currentModeId: null, + availableModes: [], + pendingPermissions: [], + persistence: null, + title: "Agent", + cwd: "/repo", + model: null, + parentAgentId: null, + labels: {}, + }; +} + function getTestSessionReferences() { const state = useSessionStore.getState(); const session = state.sessions["test-server"]; @@ -101,6 +133,26 @@ describe("agent timeline state", () => { selectAgentTimelineState(useSessionStore.getState().sessions["test-server"], "agent-1"), ).toEqual({ status: "synced", items: [], range: null, older: "none" }); }); + + it("lets a later running directory transition raise but never lower an open turn", () => { + initializeTestSession(); + const store = useSessionStore.getState(); + store.setAgents("test-server", new Map([["agent-1", createAgent("idle")]])); + store.applyAgentTimelineResponseState("test-server", "agent-1", { + items: [], + head: [], + range: null, + older: "none", + synchronized: true, + acknowledgedClientMessageIds: [], + }); + + store.setAgents("test-server", new Map([["agent-1", createAgent("running")]])); + expect(store.getSession("test-server")?.activeAgentTurns.has("agent-1")).toBe(true); + + store.setAgents("test-server", new Map([["agent-1", createAgent("idle")]])); + expect(store.getSession("test-server")?.activeAgentTurns.has("agent-1")).toBe(true); + }); }); describe("normalizeWorkspaceDescriptor", () => { diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 97bbf4726c..760e703c79 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -7,6 +7,7 @@ import type { AgentDirectoryEntry } from "@/types/agent-directory"; import { appendSubmittedUserMessage, handoffCreatedAgentUserMessageToStream, + isUnreconciledLocalUserMessage, removeSubmittedUserMessage, type StreamItem, type UserMessageItem, @@ -14,7 +15,6 @@ import { import { acceptMessageSubmission, beginMessageSubmission, - observeAcceptedMessageSubmissionsRunning, observeMessageSubmissionCanonical, rejectMessageSubmission, type MessageSubmissionRecord, @@ -513,7 +513,6 @@ interface SessionStoreActions { serverId: string, agentId: string, clientMessageId: string, - outOfBand: boolean | undefined, ) => void; rejectAgentMessageSubmission: ( serverId: string, @@ -524,6 +523,7 @@ interface SessionStoreActions { serverId: string, agentId: string, message: UserMessageItem, + trackSubmission: boolean, ) => boolean; clearAgentStreamHead: (serverId: string, agentId: string) => void; setAgentTimelineCursor: ( @@ -638,27 +638,6 @@ type SessionStore = SessionStoreState & SessionStoreActions; const agentLastActivityCoalescer = createAgentLastActivityCoalescer(); -function applyRunningAgentsToAcceptedSubmissions(input: { - previousAgents: Map; - nextAgents: Map; - submissions: Map; -}): Map { - let nextSubmissions = input.submissions; - for (const [agentId, submissions] of input.submissions) { - const previousAgent = input.previousAgents.get(agentId); - const nextAgent = input.nextAgents.get(agentId); - if (!nextAgent || previousAgent?.status === "running" || nextAgent.status !== "running") { - continue; - } - const remaining = observeAcceptedMessageSubmissionsRunning(submissions); - if (remaining === submissions) continue; - if (nextSubmissions === input.submissions) nextSubmissions = new Map(input.submissions); - if (remaining.length > 0) nextSubmissions.set(agentId, remaining); - else nextSubmissions.delete(agentId); - } - return nextSubmissions; -} - function updateActiveAgentTurn( activeAgentTurns: ReadonlySet, agentId: string, @@ -1222,7 +1201,6 @@ export const useSessionStore = create()( stream.head === currentHead ? session.agentStreamHead : new Map(session.agentStreamHead).set(agentId, stream.head), - activeAgentTurns: updateActiveAgentTurn(session.activeAgentTurns, agentId, true), messageSubmissions, }, }, @@ -1230,30 +1208,13 @@ export const useSessionStore = create()( }); }, - acceptAgentMessageSubmission: (serverId, agentId, clientMessageId, outOfBand) => { + acceptAgentMessageSubmission: (serverId, agentId, clientMessageId) => { set((prev) => { const session = prev.sessions[serverId]; if (!session) return prev; const currentSubmissions = session.messageSubmissions.get(agentId) ?? []; - const isAgentRunning = session.agents.get(agentId)?.status === "running"; - const submissions = acceptMessageSubmission( - currentSubmissions, - clientMessageId, - isAgentRunning, - outOfBand, - ); - let activeAgentTurns = session.activeAgentTurns; - if ( - submissions !== currentSubmissions && - !isAgentRunning && - outOfBand === true && - submissions.length === 0 - ) { - activeAgentTurns = updateActiveAgentTurn(activeAgentTurns, agentId, false); - } - if (submissions === currentSubmissions && activeAgentTurns === session.activeAgentTurns) { - return prev; - } + const submissions = acceptMessageSubmission(currentSubmissions, clientMessageId); + if (submissions === currentSubmissions) return prev; const messageSubmissions = new Map(session.messageSubmissions); if (submissions.length > 0) { messageSubmissions.set(agentId, submissions); @@ -1264,7 +1225,7 @@ export const useSessionStore = create()( ...prev, sessions: { ...prev.sessions, - [serverId]: { ...session, activeAgentTurns, messageSubmissions }, + [serverId]: { ...session, messageSubmissions }, }, }; }); @@ -1295,12 +1256,6 @@ export const useSessionStore = create()( } else { messageSubmissions.delete(agentId); } - const activeAgentTurns = - outcome === "rejected" && - result.submissions.length === 0 && - session.agents.get(agentId)?.status !== "running" - ? updateActiveAgentTurn(session.activeAgentTurns, agentId, false) - : session.activeAgentTurns; return { ...prev, sessions: { @@ -1315,7 +1270,6 @@ export const useSessionStore = create()( stream.head === currentHead ? session.agentStreamHead : new Map(session.agentStreamHead).set(agentId, stream.head), - activeAgentTurns, messageSubmissions, }, }, @@ -1324,7 +1278,7 @@ export const useSessionStore = create()( return outcome; }, - handoffCreatedAgentUserMessage: (serverId, agentId, message) => { + handoffCreatedAgentUserMessage: (serverId, agentId, message, trackSubmission) => { let didHandoff = false; set((prev) => { const session = prev.sessions[serverId]; @@ -1349,6 +1303,28 @@ export const useSessionStore = create()( const nextHead = result.changedHead ? new Map(session.agentStreamHead).set(agentId, result.head) : session.agentStreamHead; + const clientMessageId = message.clientMessageId; + let messageSubmissions = session.messageSubmissions; + if ( + trackSubmission && + clientMessageId && + [...result.tail, ...result.head].some( + (item) => + item.kind === "user_message" && + item.clientMessageId === clientMessageId && + isUnreconciledLocalUserMessage(item), + ) + ) { + const currentSubmissions = session.messageSubmissions.get(agentId) ?? []; + const accepted = acceptMessageSubmission( + beginMessageSubmission(currentSubmissions, { + clientMessageId, + submittedAt: message.timestamp, + }), + clientMessageId, + ); + messageSubmissions = new Map(session.messageSubmissions).set(agentId, accepted); + } didHandoff = true; return { @@ -1359,7 +1335,7 @@ export const useSessionStore = create()( ...session, agentStreamTail: nextTail, agentStreamHead: nextHead, - activeAgentTurns: updateActiveAgentTurn(session.activeAgentTurns, agentId, true), + messageSubmissions, }, }, }; @@ -1613,12 +1589,14 @@ export const useSessionStore = create()( return prev; } const nextAgents = typeof agents === "function" ? agents(session.agents) : agents; - const messageSubmissions = applyRunningAgentsToAcceptedSubmissions({ - previousAgents: session.agents, - nextAgents, - submissions: session.messageSubmissions, - }); - if (session.agents === nextAgents && session.messageSubmissions === messageSubmissions) { + let activeAgentTurns = session.activeAgentTurns; + for (const [agentId, agent] of nextAgents) { + const previousAgent = session.agents.get(agentId); + if (previousAgent?.status !== "running" && agent.status === "running") { + activeAgentTurns = updateActiveAgentTurn(activeAgentTurns, agentId, true); + } + } + if (session.agents === nextAgents && activeAgentTurns === session.activeAgentTurns) { return prev; } return { @@ -1628,11 +1606,11 @@ export const useSessionStore = create()( [serverId]: { ...session, agents: nextAgents, - messageSubmissions, - workspaceAgentActivity: - nextAgents === session.agents - ? session.workspaceAgentActivity - : buildWorkspaceAgentActivityIndex(nextAgents, session.workspaceAgentActivity), + activeAgentTurns, + workspaceAgentActivity: buildWorkspaceAgentActivityIndex( + nextAgents, + session.workspaceAgentActivity, + ), }, }, }; diff --git a/packages/app/src/timeline/session-stream-reducers.test.ts b/packages/app/src/timeline/session-stream-reducers.test.ts index 6e8efbd82d..6503f62c6f 100644 --- a/packages/app/src/timeline/session-stream-reducers.test.ts +++ b/packages/app/src/timeline/session-stream-reducers.test.ts @@ -149,10 +149,22 @@ describe("deriveAgentStreamTurnActivity", () => { } }); - it("leaves activity unchanged when a batch has no lifecycle event", () => { - expect( - deriveAgentStreamTurnActivity([makeStreamReducerEvent(makeAssistantTimelineEvent("hi"), 1)]), - ).toBeUndefined(); + it("keeps an open turn without stream lifecycle and closes it on an ordered terminal event", () => { + let hasOpenTurn = true; + hasOpenTurn = + deriveAgentStreamTurnActivity([ + makeStreamReducerEvent(makeAssistantTimelineEvent("still working"), 1), + ]) ?? hasOpenTurn; + expect(hasOpenTurn).toBe(true); + + hasOpenTurn = + deriveAgentStreamTurnActivity([ + makeStreamReducerEvent( + { type: "turn_completed", provider: "claude" } as AgentStreamEventPayload, + 2, + ), + ]) ?? hasOpenTurn; + expect(hasOpenTurn).toBe(false); }); }); diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index b764890973..e9f2ef7082 100644 --- a/packages/app/src/types/stream.ts +++ b/packages/app/src/types/stream.ts @@ -161,7 +161,10 @@ export function removeSubmittedUserMessage(input: { }): { tail: StreamItem[]; head: StreamItem[] } { const remove = (items: StreamItem[]) => { const next = items.filter( - (item) => item.kind !== "user_message" || item.clientMessageId !== input.clientMessageId, + (item) => + item.kind !== "user_message" || + item.clientMessageId !== input.clientMessageId || + !isUnreconciledLocalUserMessage(item), ); return next.length === items.length ? items : next; }; diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index bed840c0e9..8e67ea4b5a 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -324,11 +324,6 @@ export interface SendMessageOptions { attachments?: SendAgentMessageRequest["attachments"]; } -export interface SendMessageResult { - /** Undefined when connected to a daemon predating message submission disposition. */ - outOfBand?: boolean; -} - export interface AgentAttentionRequiredNotification { agentId: string; reason: "finished" | "error" | "permission"; @@ -2864,7 +2859,7 @@ export class DaemonClient { agentId: string, text: string, options?: SendMessageOptions, - ): Promise { + ): Promise { const requestId = this.createRequestId(); const messageId = options?.messageId ?? crypto.randomUUID(); const message = SessionInboundMessageSchema.parse({ @@ -2893,7 +2888,6 @@ export class DaemonClient { if (!payload.accepted) { throw new Error(payload.error ?? "sendAgentMessage rejected"); } - return payload.outOfBand === undefined ? {} : { outOfBand: payload.outOfBand }; } async sendMessage(agentId: string, text: string, options?: SendMessageOptions): Promise { diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index dd9772e5c8..6624d58262 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -2828,6 +2828,8 @@ export const ServerInfoStatusPayloadSchema = z forgeProviders: z.boolean().optional(), // COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12. selectiveAgentTimeline: z.boolean().optional(), + // COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30. + canonicalSubmittedPrompts: z.boolean().optional(), // COMPAT(stableProjectIdentity): added in v0.1.109, remove gate after 2027-01-15. stableProjectIdentity: z.boolean().optional(), // COMPAT(workspaceScriptManagement): added in v0.1.105, remove gate after 2027-01-10. @@ -3710,8 +3712,6 @@ export const SendAgentMessageResponseMessageSchema = z.object({ agentId: z.string(), accepted: z.boolean(), error: z.string().nullable(), - // COMPAT(messageSubmissionDisposition): added in v0.2.3, remove optional parsing after 2027-01-27. - outOfBand: z.boolean().optional(), }), }); diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 1ff1b19337..d4cc21ea01 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -16,7 +16,7 @@ import { import { AgentStorage } from "./agent-storage.js"; import { toAgentPayload } from "./agent-projections.js"; import { PARENT_AGENT_ID_LABEL } from "@getpaseo/protocol/agent-labels"; -import { formatSystemNotificationPrompt } from "./agent-prompt.js"; +import { formatSystemNotificationPrompt, startAgentRun } from "./agent-prompt.js"; import { ensureAgentLoaded, ensureUnarchivedAgentLoaded } from "./agent-loading.js"; import type { StoredAgentRecord } from "./agent-storage.js"; import type { @@ -8001,6 +8001,67 @@ test("authoritative timeline includes provider-emitted submitted user prompt", a } }); +test("authoritative timeline records a daemon-handled submitted prompt before its output", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-daemon-handled-prompt-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const commandCompleted = deferred(); + + class DaemonHandledPromptSession extends TestAgentSession { + override tryHandleOutOfBand(prompt: AgentPromptInput) { + if (prompt !== "/handled") return null; + return { + run: async ({ emit }: { emit: (event: AgentStreamEvent) => void }) => { + emit({ + type: "timeline", + provider: this.provider, + item: { type: "assistant_message", text: "Handled by the daemon" }, + }); + commandCompleted.resolve(); + }, + }; + } + } + + class DaemonHandledPromptClient extends TestAgentClient { + override async createSession(config: AgentSessionConfig): Promise { + return new DaemonHandledPromptSession(config); + } + } + + const manager = new AgentManager({ + clients: { codex: new DaemonHandledPromptClient() }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000403", + }); + + try { + const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + + await startAgentRun(manager, snapshot.id, "/handled", logger, { + runOptions: { clientMessageId: "msg-client-daemon-handled" }, + }); + await commandCompleted.promise; + + const timeline = manager.fetchTimeline(snapshot.id, { direction: "tail", limit: 20 }).rows; + expect(timeline.map((row) => row.item)).toEqual([ + { + type: "user_message", + text: "/handled", + clientMessageId: "msg-client-daemon-handled", + }, + { type: "assistant_message", text: "Handled by the daemon" }, + ]); + } finally { + await manager.flush().catch(() => undefined); + await storage.flush().catch(() => undefined); + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("replaceAgentRun succeeds when foreground turn terminal event is never delivered", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-stale-fg-")); const storagePath = join(workdir, "agents"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index e6c4e68a88..b734624470 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -1868,12 +1868,27 @@ export class AgentManager { * emitted by the handler flow through dispatchStream so they persist and * broadcast like normal timeline events. */ - tryRunOutOfBand(agentId: string, prompt: AgentPromptInput): boolean { + tryRunOutOfBand(agentId: string, prompt: AgentPromptInput, options?: AgentRunOptions): boolean { const agent = this.requireSessionAgent(agentId); const handler = agent.session.tryHandleOutOfBand?.(prompt); if (!handler) { return false; } + if (options?.clientMessageId) { + if (typeof prompt !== "string") { + throw new Error("Daemon-handled submitted prompts must be text"); + } + this.touchUpdatedAt(agent); + this.recordAndDispatchTimelineItem( + agent.id, + { + type: "user_message", + text: prompt, + clientMessageId: options.clientMessageId, + }, + agent.provider, + ); + } const dispatch = (event: AgentStreamEvent): void => { // Persist timeline items so they show up in fetchAgentTimeline; broadcast // for live subscribers. Other event types are broadcast only. diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index 22bd8cdc24..53d3674162 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -41,7 +41,7 @@ export async function startAgentRun( // Out-of-band commands (e.g. /goal pause) must run WITHOUT canceling an // in-flight turn — replaceAgentRun would interrupt the running turn. The // intercept lives at this layer so it covers every prompt entrypoint. - if (agentManager.tryRunOutOfBand(agentId, prompt)) { + if (agentManager.tryRunOutOfBand(agentId, prompt, options?.runOptions)) { return { outOfBand: true }; } const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId)); diff --git a/packages/server/src/server/agent/providers/acp-agent.test.ts b/packages/server/src/server/agent/providers/acp-agent.test.ts index 49adf40a2c..2f5f919ebf 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -2345,6 +2345,15 @@ describe("ACPAgentSession", () => { messageId: "msg-client-1", prompt: [{ type: "text", text: "hello" }], }); + expect( + events + .filter( + (event) => + event.type === "turn_started" || + (event.type === "timeline" && event.item.type === "user_message"), + ) + .map((event) => event.type), + ).toEqual(["turn_started", "timeline"]); expect( events.filter((event) => event.type === "timeline" && event.item.type === "user_message"), ).toEqual([ diff --git a/packages/server/src/server/agent/providers/claude/agent.test.ts b/packages/server/src/server/agent/providers/claude/agent.test.ts index e4e8805928..4bcb0eeadc 100644 --- a/packages/server/src/server/agent/providers/claude/agent.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.test.ts @@ -1507,6 +1507,22 @@ describe("ClaudeAgentSession context window usage", () => { }; } + test("emits turn_started before the submitted user message", async () => { + const session = await createSessionForTurns([[]]); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + await session.startTurn("turn", { clientMessageId: "client-message-1" }); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(events.slice(0, 2).map((event) => event.type)).toEqual(["turn_started", "timeline"]); + expect(events[1]).toMatchObject({ + type: "timeline", + item: { type: "user_message", clientMessageId: "client-message-1" }, + }); + await session.close(); + }); + test("passes persistSession through to the Claude SDK query options", async () => { const createResultTurn = (sessionId: string) => [ { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 57235d3773..ddc96bf0b0 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -1119,7 +1119,13 @@ describe("Codex app-server provider", () => { async () => appServer.child, ); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); await session.startTurn("remember this", { clientMessageId: "client-message" }); + asInternals(session).handleNotification("turn/started", { + threadId: "thread-1", + turn: { id: "turn-1" }, + }); const userMessage = waitForNextTimelineItem(session, "user_message"); emitCodexUserMessage(appServer, { id: "codex-message", text: "remember this" }); @@ -1130,6 +1136,7 @@ describe("Codex app-server provider", () => { clientMessageId: "client-message", }, }); + expect(events.slice(0, 2).map((event) => event.type)).toEqual(["turn_started", "timeline"]); appServer.completeTurn(); await session.close(); }); diff --git a/packages/server/src/server/agent/providers/mock-load-test-agent.test.ts b/packages/server/src/server/agent/providers/mock-load-test-agent.test.ts index c32d9a25dd..c4377cb1a9 100644 --- a/packages/server/src/server/agent/providers/mock-load-test-agent.test.ts +++ b/packages/server/src/server/agent/providers/mock-load-test-agent.test.ts @@ -230,6 +230,32 @@ describe("MockLoadTestAgentClient", () => { }); }); + test("emits turn_started before the submitted user message", async () => { + vi.useFakeTimers(); + const client = new MockLoadTestAgentClient(); + const session = await client.createSession({ + provider: "mock", + cwd: process.cwd(), + model: "ten-second-stream", + }); + const events: AgentStreamEvent[] = []; + const unsubscribe = session.subscribe((event) => events.push(event)); + + await session.startTurn("Order the submitted prompt.", { + clientMessageId: "client-message-1", + }); + await vi.advanceTimersByTimeAsync(0); + unsubscribe(); + + expect( + events + .slice(0, 2) + .map((event) => + event.type === "timeline" ? `${event.type}:${event.item.type}` : event.type, + ), + ).toEqual(["turn_started", "timeline:user_message"]); + }); + test("emits the free-write question scenario selected by prompt", async () => { vi.useFakeTimers(); const client = new MockLoadTestAgentClient(); diff --git a/packages/server/src/server/agent/providers/mock-load-test-agent.ts b/packages/server/src/server/agent/providers/mock-load-test-agent.ts index 790e450620..fd2086efe2 100644 --- a/packages/server/src/server/agent/providers/mock-load-test-agent.ts +++ b/packages/server/src/server/agent/providers/mock-load-test-agent.ts @@ -33,6 +33,7 @@ import { getAgentProviderDefinition } from "@getpaseo/protocol/provider-manifest export const MOCK_LOAD_TEST_PROVIDER_ID = "mock"; export const MOCK_LOAD_TEST_DEFAULT_MODEL_ID = "five-minute-stream"; +export const MOCK_LOAD_TEST_HANDLED_COMMAND = "/mock handled-command"; const MOCK_LOAD_TEST_MODE_ID = "load-test"; const MOCK_LOAD_TEST_DURATION_MS = 5 * 60 * 1000; const MOCK_LOAD_TEST_INTERVAL_MS = 40; @@ -692,6 +693,7 @@ export class MockLoadTestAgentSession implements AgentSession { if (this.activeTurn?.turnId !== turnId) { return; } + this.emitTurnStarted(turn); this.emit({ type: "timeline", provider: this.provider, @@ -710,6 +712,21 @@ export class MockLoadTestAgentSession implements AgentSession { return { turnId }; } + tryHandleOutOfBand( + prompt: AgentPromptInput, + ): { run(ctx: { emit: (event: AgentStreamEvent) => void }): Promise } | null { + if (prompt !== MOCK_LOAD_TEST_HANDLED_COMMAND) return null; + return { + run: async ({ emit }) => { + emit({ + type: "timeline", + provider: this.provider, + item: { type: "assistant_message", text: "Mock command handled" }, + }); + }, + }; + } + subscribe(callback: (event: AgentStreamEvent) => void): () => void { this.listeners.add(callback); return () => { @@ -842,6 +859,18 @@ export class MockLoadTestAgentSession implements AgentSession { turn.timer.unref?.(); } + private emitTurnStarted(turn: ActiveTurn): void { + if (turn.turnStarted) { + return; + } + turn.turnStarted = true; + this.emit({ + type: "turn_started", + provider: this.provider, + turnId: turn.turnId, + }); + } + private failConfiguredRewind(): void { if (this.rewindError) { throw new Error(this.rewindError); @@ -864,11 +893,7 @@ export class MockLoadTestAgentSession implements AgentSession { return; } this.clearTurnTimer(turn); - this.emit({ - type: "turn_started", - provider: this.provider, - turnId: turn.turnId, - }); + this.emitTurnStarted(turn); this.activeTurn = null; this.emit({ type: "turn_failed", @@ -923,11 +948,7 @@ export class MockLoadTestAgentSession implements AgentSession { } this.clearTurnTimer(turn); - this.emit({ - type: "turn_started", - provider: this.provider, - turnId: turn.turnId, - }); + this.emitTurnStarted(turn); this.emitTimeline(turn.turnId, { type: "assistant_message", @@ -960,11 +981,7 @@ export class MockLoadTestAgentSession implements AgentSession { } this.clearTurnTimer(turn); - this.emit({ - type: "turn_started", - provider: this.provider, - turnId: turn.turnId, - }); + this.emitTurnStarted(turn); const request: AgentPermissionRequest = { id: `mock-plan-${turn.turnId}`, @@ -1015,11 +1032,7 @@ export class MockLoadTestAgentSession implements AgentSession { } this.clearTurnTimer(turn); - this.emit({ - type: "turn_started", - provider: this.provider, - turnId: turn.turnId, - }); + this.emitTurnStarted(turn); const request: AgentPermissionRequest = { id: `mock-questions-${turn.turnId}`, @@ -1050,11 +1063,7 @@ export class MockLoadTestAgentSession implements AgentSession { } this.clearTurnTimer(turn); - this.emit({ - type: "turn_started", - provider: this.provider, - turnId: turn.turnId, - }); + this.emitTurnStarted(turn); for (let index = 0; index < stress.count; index += 1) { this.emitTimeline( @@ -1103,11 +1112,7 @@ export class MockLoadTestAgentSession implements AgentSession { } this.clearTurnTimer(turn); - this.emit({ - type: "turn_started", - provider: this.provider, - turnId: turn.turnId, - }); + this.emitTurnStarted(turn); const payload = buildRepeatedPayload(largePayload.bytes, largePayload.kind); if (largePayload.kind === "diff") { @@ -1179,14 +1184,7 @@ export class MockLoadTestAgentSession implements AgentSession { } this.clearTurnTimer(turn); - if (!turn.turnStarted) { - turn.turnStarted = true; - this.emit({ - type: "turn_started", - provider: this.provider, - turnId: turn.turnId, - }); - } + this.emitTurnStarted(turn); const elapsedMs = Date.now() - turn.startedAt; if (elapsedMs >= turn.durationMs) { diff --git a/packages/server/src/server/agent/providers/omp/agent.test.ts b/packages/server/src/server/agent/providers/omp/agent.test.ts index 4e01109bea..d197714475 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -147,6 +147,7 @@ describe("OMP agent client and session", () => { { type: "user_message", text: "hello OMP", messageId: "user-1" }, { type: "assistant_message", text: "hello from OMP", messageId: "omp-assistant-1" }, ]); + expect(omp.eventTypes().slice(0, 2)).toEqual(["turn_started", "timeline"]); expect(omp.completedTurnCount()).toBe(1); }); diff --git a/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts b/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts index 00fed1ac89..556bbd3227 100644 --- a/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts +++ b/packages/server/src/server/agent/providers/omp/test-utils/omp-harness.ts @@ -391,6 +391,10 @@ export class OmpHarness { return this.events.flatMap((event) => (event.type === "timeline" ? [event.item] : [])); } + eventTypes(): AgentStreamEvent["type"][] { + return this.events.map((event) => event.type); + } + async history(): Promise { const items: AgentTimelineItem[] = []; for await (const event of this.requireSession().streamHistory()) { diff --git a/packages/server/src/server/agent/providers/opencode-agent.test.ts b/packages/server/src/server/agent/providers/opencode-agent.test.ts index b9af8193c7..6d454c2b32 100644 --- a/packages/server/src/server/agent/providers/opencode-agent.test.ts +++ b/packages/server/src/server/agent/providers/opencode-agent.test.ts @@ -1168,6 +1168,19 @@ describe("OpenCode adapter startTurn error handling", () => { properties: {}, }, }, + { + directory: "/tmp/test", + payload: { + type: "message.updated", + properties: { + info: { + id: "msg_user", + sessionID: "ses_unit_test", + role: "user", + }, + }, + }, + }, { directory: "/tmp/test", payload: { @@ -1229,17 +1242,25 @@ describe("OpenCode adapter startTurn error handling", () => { createTestLogger(), ); - const turn = await collectTurnEvents(streamSession(session, "hello")); + const turn = await collectTurnEvents( + streamSession(session, "hello", { clientMessageId: "client-message-1" }), + ); expect(turn.events.map((event) => event.type)).toEqual([ "turn_started", "timeline", + "timeline", "turn_completed", ]); + expect(turn.events[1]).toMatchObject({ + type: "timeline", + item: { type: "user_message", clientMessageId: "client-message-1" }, + }); expect(turn.events.map((event) => ("turnId" in event ? event.turnId : undefined))).toEqual([ "opencode-turn-0", "opencode-turn-0", "opencode-turn-0", + "opencode-turn-0", ]); }); diff --git a/packages/server/src/server/agent/providers/pi/agent.test.ts b/packages/server/src/server/agent/providers/pi/agent.test.ts index ee918354f6..c3fad836d9 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -181,6 +181,10 @@ class SessionEvents { }); } + eventTypes(): AgentStreamEvent["type"][] { + return this.events.map((event) => event.type); + } + turnCompletedEvents() { return this.events.filter( (event): event is Extract => @@ -608,6 +612,7 @@ describe("PiRpcAgentSession", () => { const fakeSession = pi.latestSession(); await session.startTurn("hello"); + fakeSession.emit({ type: "turn_start" }); fakeSession.finishSubmittedUserMessage({ id: "entry-user-1", parentId: null, @@ -619,6 +624,7 @@ describe("PiRpcAgentSession", () => { expect(events.timelineItems()).toEqual([ { type: "user_message", text: "hello", messageId: "entry-user-1" }, ]); + expect(events.eventTypes().slice(0, 2)).toEqual(["turn_started", "timeline"]); }); test("uses the Pi entry attached to a submitted prompt after resuming old history", async () => { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 51059637a4..2682979fe1 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -6345,7 +6345,6 @@ export class Session { agentId, accepted: true, error: null, - outOfBand: true, }, }); return; @@ -6373,7 +6372,6 @@ export class Session { agentId, accepted: true, error: null, - outOfBand: false, }, }); } catch (error) { diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 9501fe8639..d41ad55bef 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1574,6 +1574,8 @@ export class VoiceAssistantWebSocketServer { forgeProviders: true, // COMPAT(selectiveAgentTimeline): added in v0.1.106, remove after 2027-01-12. selectiveAgentTimeline: true, + // COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30. + canonicalSubmittedPrompts: true, // COMPAT(stableProjectIdentity): added in v0.1.109, remove gate after 2027-01-15. stableProjectIdentity: true, // COMPAT(workspaceScriptManagement): added in v0.1.105, remove gate after 2027-01-10.