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.