diff --git a/packages/app/e2e/helpers/daemon-websocket-gate.ts b/packages/app/e2e/helpers/daemon-websocket-gate.ts index 379717c668..9d18c398d1 100644 --- a/packages/app/e2e/helpers/daemon-websocket-gate.ts +++ b/packages/app/e2e/helpers/daemon-websocket-gate.ts @@ -68,6 +68,24 @@ function stripAssistantMessageId( return JSON.stringify(envelope); } +function stripMessageSubmissionDisposition( + message: string | Buffer, + enabled: boolean, + messageType: unknown, +): string | Buffer { + if (!enabled || messageType !== "send_agent_message_response" || typeof message !== "string") { + return message; + } + const envelope = JSON.parse(message) as { + message?: { payload?: Record }; + payload?: Record; + }; + const payload = envelope.message?.payload ?? envelope.payload; + if (!payload) return message; + delete payload.outOfBand; + return JSON.stringify(envelope); +} + function forceTimelineReset(message: string | Buffer, enabled: boolean): string | Buffer { if (!enabled || typeof message !== "string") return message; const envelope = JSON.parse(message) as { @@ -119,6 +137,7 @@ export async function installDaemonWebSocketGate(page: Page) { let suppressAgentStream = false; let forceTimelineEpochReset = false; let stripAssistantMessageIds = false; + let stripSubmissionDisposition = false; let heldClientRequestType: string | null = null; let heldClientRequest: { server: WebSocketRoute; message: string | Buffer } | null = null; let resolveHeldClientRequest: (() => void) | null = null; @@ -192,6 +211,11 @@ export async function installDaemonWebSocketGate(page: Page) { stripAssistantMessageIds, serverMessage?.type, ); + outboundMessage = stripMessageSubmissionDisposition( + outboundMessage, + stripSubmissionDisposition, + serverMessage?.type, + ); const shouldForceTimelineReset = forceTimelineEpochReset && serverMessage?.type === "fetch_agent_timeline_response"; outboundMessage = forceTimelineReset(outboundMessage, shouldForceTimelineReset); @@ -360,6 +384,9 @@ export async function installDaemonWebSocketGate(page: Page) { setAssistantMessageIdsStripped(stripped: boolean): void { stripAssistantMessageIds = stripped; }, + setMessageSubmissionDispositionStripped(stripped: boolean): void { + stripSubmissionDisposition = 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 cbec6733c1..2df34398d8 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,6 +4,7 @@ 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); @@ -26,4 +27,26 @@ 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/composer/actions.test.ts b/packages/app/src/composer/actions.test.ts index e42937d386..8896650f00 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), + submissions: acceptMessageSubmission(current.submissions, clientMessageId, true, false), }); }, reject: (agentId, clientMessageId) => { diff --git a/packages/app/src/composer/actions.ts b/packages/app/src/composer/actions.ts index d622269aa2..08cd0d03e7 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) => void; + accept: (agentId: string, clientMessageId: string, outOfBand: boolean | undefined) => 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); - await input.client.sendAgentMessage(input.agentId, input.text, { + const result = await input.client.sendAgentMessage(input.agentId, input.text, { messageId: clientMessageId, images: imagesData ?? [], attachments: wirePayload.attachments, }); - input.submission.accept(input.agentId, clientMessageId); + input.submission.accept(input.agentId, clientMessageId, result?.outOfBand); } catch (error) { const outcome = input.submission.reject(input.agentId, clientMessageId); if (outcome === "accepted") return; diff --git a/packages/app/src/composer/submission/model.test.ts b/packages/app/src/composer/submission/model.test.ts index eee4bd4a18..cab44f8a3e 100644 --- a/packages/app/src/composer/submission/model.test.ts +++ b/packages/app/src/composer/submission/model.test.ts @@ -32,7 +32,7 @@ describe("message submission transactions", () => { { clientMessageId: "client-2", submittedAt }, ); - expect(acceptMessageSubmission(both, "client-1", true)).toEqual([ + expect(acceptMessageSubmission(both, "client-1", true, false)).toEqual([ { clientMessageId: "client-2", submittedAt, @@ -44,7 +44,7 @@ describe("message submission transactions", () => { it("bridges an accepted RPC until the correlated running state is observed", () => { const sending = beginMessageSubmission([], { clientMessageId: "client-1", submittedAt }); - const accepted = acceptMessageSubmission(sending, "client-1", false); + const accepted = acceptMessageSubmission(sending, "client-1", false, false); expect(getActiveMessageSubmissions(accepted)).toHaveLength(1); expect(accepted[0].rpcAccepted).toBe(true); @@ -53,11 +53,23 @@ describe("message submission transactions", () => { 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); + const accepted = acceptMessageSubmission(sending, "client-1", false, false); 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 2a46280571..9efc5e66a0 100644 --- a/packages/app/src/composer/submission/model.ts +++ b/packages/app/src/composer/submission/model.ts @@ -45,12 +45,22 @@ 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[]; - if (isAgentRunning || submissions[index].providerAcknowledged) { + // COMPAT(messageSubmissionDisposition): daemons before v0.2.3 omitted outOfBand. + // Their normal-send response follows the ordered running/canonical events, while an + // out-of-band response arrives with the agent still idle. Remove after 2027-01-27. + const legacyOutOfBand = outOfBand === undefined && !isAgentRunning; + if ( + outOfBand === true || + legacyOutOfBand || + isAgentRunning || + submissions[index].providerAcknowledged + ) { return submissions.filter((_, submissionIndex) => submissionIndex !== index); } if (submissions[index].rpcAccepted) return submissions as MessageSubmissionRecord[]; diff --git a/packages/app/src/composer/submission/writer.ts b/packages/app/src/composer/submission/writer.ts index 755eed1564..9b41c84e75 100644 --- a/packages/app/src/composer/submission/writer.ts +++ b/packages/app/src/composer/submission/writer.ts @@ -10,8 +10,10 @@ export function createMessageSubmissionWriter(serverId: string): MessageSubmissi return { begin: (agentId, message) => useSessionStore.getState().beginAgentMessageSubmission(serverId, agentId, message), - accept: (agentId, clientMessageId) => - useSessionStore.getState().acceptAgentMessageSubmission(serverId, agentId, clientMessageId), + accept: (agentId, clientMessageId, outOfBand) => + useSessionStore + .getState() + .acceptAgentMessageSubmission(serverId, agentId, clientMessageId, outOfBand), reject: (agentId, clientMessageId) => useSessionStore.getState().rejectAgentMessageSubmission(serverId, agentId, clientMessageId), }; diff --git a/packages/app/src/runtime/host-runtime.test.ts b/packages/app/src/runtime/host-runtime.test.ts index 64207753cc..6f73dd640f 100644 --- a/packages/app/src/runtime/host-runtime.test.ts +++ b/packages/app/src/runtime/host-runtime.test.ts @@ -80,13 +80,16 @@ class FakeDaemonClient { this.setConnectionState({ status: "disconnected", reason: "client_closed" }); } - async sendAgentMessage(...args: Parameters): Promise { + async sendAgentMessage( + ...args: Parameters + ): ReturnType { this.sentAgentMessages.push(args); for (const waiter of this.sentMessageWaiters) waiter(); const response = this.sendAgentMessageResponses.shift(); if (response) await response; const failure = this.sendAgentMessageFailures.shift(); if (failure) throw failure; + return {}; } async waitForSentMessages(count: number): Promise { diff --git a/packages/app/src/stores/session-store.ts b/packages/app/src/stores/session-store.ts index 76c121be0e..3a5504310e 100644 --- a/packages/app/src/stores/session-store.ts +++ b/packages/app/src/stores/session-store.ts @@ -486,6 +486,7 @@ interface SessionStoreActions { serverId: string, agentId: string, clientMessageId: string, + outOfBand: boolean | undefined, ) => void; rejectAgentMessageSubmission: ( serverId: string, @@ -1177,7 +1178,7 @@ export const useSessionStore = create()( }); }, - acceptAgentMessageSubmission: (serverId, agentId, clientMessageId) => { + acceptAgentMessageSubmission: (serverId, agentId, clientMessageId, outOfBand) => { set((prev) => { const session = prev.sessions[serverId]; if (!session) return prev; @@ -1186,6 +1187,7 @@ export const useSessionStore = create()( currentSubmissions, clientMessageId, session.agents.get(agentId)?.status === "running", + outOfBand, ); if (submissions === currentSubmissions) return prev; const messageSubmissions = new Map(session.messageSubmissions); diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index 8e67ea4b5a..bed840c0e9 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -324,6 +324,11 @@ 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"; @@ -2859,7 +2864,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({ @@ -2888,6 +2893,7 @@ 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/client/src/index.ts b/packages/client/src/index.ts index d7f21f1246..a0fdbd5e77 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -478,7 +478,9 @@ function createAgentHandleFactory(daemonClient: DaemonClient): AgentHandleFactor latest = result?.agent ?? null; return result; }, - send: (text, options) => daemonClient.sendAgentMessage(id, text, options), + send: async (text, options) => { + await daemonClient.sendAgentMessage(id, text, options); + }, archive: async () => { const result = await daemonClient.archiveAgent(id); if (latest) { diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 31f33685db..5efa8f6a7d 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -3708,6 +3708,8 @@ 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.ts b/packages/server/src/server/agent/agent-manager.ts index 7b1c7d4789..dc87d31969 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -1952,19 +1952,12 @@ export class AgentManager { * emitted by the handler flow through dispatchStream so they persist and * broadcast like normal timeline events. */ - tryRunOutOfBand( - agentId: string, - prompt: AgentPromptInput, - runOptions?: AgentRunOptions, - ): boolean { + tryRunOutOfBand(agentId: string, prompt: AgentPromptInput): boolean { const agent = this.requireSessionAgent(agentId); const handler = agent.session.tryHandleOutOfBand?.(prompt); if (!handler) { return false; } - if (typeof prompt !== "string") { - throw new Error("Out-of-band commands require a text prompt"); - } const dispatch = (event: AgentStreamEvent): void => { // Persist timeline items so they show up in fetchAgentTimeline; broadcast // for live subscribers. Other event types are broadcast only. @@ -1980,15 +1973,6 @@ export class AgentManager { } this.dispatchStream(agent.id, event, { timestamp: new Date().toISOString() }); }; - dispatch({ - type: "timeline", - provider: agent.provider, - item: { - type: "user_message", - text: prompt, - ...(runOptions?.clientMessageId ? { clientMessageId: runOptions.clientMessageId } : {}), - }, - }); void (async () => { try { await handler.run({ emit: dispatch }); diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index 53d3674162..22bd8cdc24 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, options?.runOptions)) { + if (agentManager.tryRunOutOfBand(agentId, prompt)) { return { outOfBand: true }; } const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId)); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 04f04fe6f2..364766c75f 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -6335,6 +6335,7 @@ export class Session { agentId, accepted: true, error: null, + outOfBand: true, }, }); return; @@ -6362,6 +6363,7 @@ export class Session { agentId, accepted: true, error: null, + outOfBand: false, }, }); } catch (error) {