fix(app): settle out-of-band submissions compatibly

This commit is contained in:
Mohamed Boudra
2026-07-27 19:18:06 +02:00
parent eec9fb3992
commit 73119d0f36
15 changed files with 108 additions and 33 deletions
@@ -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<string, unknown> };
payload?: Record<string, unknown>;
};
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;
},
@@ -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 });
}
});
});
+1 -1
View File
@@ -223,7 +223,7 @@ function createFakeStream(initialHead: Map<string, StreamItem[]> = 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) => {
+4 -4
View File
@@ -46,7 +46,7 @@ export interface ComposerSendClient {
images: Array<{ data: string; mimeType: string }>;
attachments: ReturnType<typeof splitComposerAttachmentsForSubmit>["attachments"];
},
) => Promise<void>;
) => Promise<void | { outOfBand?: boolean }>;
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;
@@ -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 }),
+11 -1
View File
@@ -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[];
@@ -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),
};
@@ -80,13 +80,16 @@ class FakeDaemonClient {
this.setConnectionState({ status: "disconnected", reason: "client_closed" });
}
async sendAgentMessage(...args: Parameters<DaemonClient["sendAgentMessage"]>): Promise<void> {
async sendAgentMessage(
...args: Parameters<DaemonClient["sendAgentMessage"]>
): ReturnType<DaemonClient["sendAgentMessage"]> {
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<void> {
+3 -1
View File
@@ -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<SessionStore>()(
});
},
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<SessionStore>()(
currentSubmissions,
clientMessageId,
session.agents.get(agentId)?.status === "running",
outOfBand,
);
if (submissions === currentSubmissions) return prev;
const messageSubmissions = new Map(session.messageSubmissions);
+7 -1
View File
@@ -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<void> {
): Promise<SendMessageResult> {
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<void> {
+3 -1
View File
@@ -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) {
+2
View File
@@ -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(),
}),
});
@@ -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 });
@@ -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));
+2
View File
@@ -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) {