diff --git a/packages/agent/src/adapters/claude/claude-agent.streamed-text.test.ts b/packages/agent/src/adapters/claude/claude-agent.streamed-text.test.ts index 96775fa040..4bd3d8e13f 100644 --- a/packages/agent/src/adapters/claude/claude-agent.streamed-text.test.ts +++ b/packages/agent/src/adapters/claude/claude-agent.streamed-text.test.ts @@ -243,6 +243,64 @@ describe("ClaudeAcpAgent.prompt — streamed assistant text wiring", () => { ]); }); + it("keeps the original turn open until a pending steer is consumed", async () => { + const { agent, client } = makeAgent(); + const sessionId = "s-steer-ordering"; + const { query, input } = installFakeSession(agent, sessionId); + + const promptPromise = agent.prompt({ + sessionId, + prompt: [{ type: "text", text: "use orange" }], + }); + let promptSettled = false; + void promptPromise.then(() => { + promptSettled = true; + }); + await tick(); + await echoUserMessage(query, input); + + const steerResult = await agent.prompt({ + sessionId, + prompt: [{ type: "text", text: "use green instead" }], + _meta: { steer: true }, + }); + expect(steerResult._meta).toEqual({ steer: true }); + + await send(query, assistantMessage(sessionId, "msg_orange", "ORANGE")); + await send(query, resultSuccess(sessionId)); + expect(promptSettled).toBe(false); + + await echoUserMessage(query, input); + await send(query, assistantMessage(sessionId, "msg_green", "GREEN")); + await send(query, resultSuccess(sessionId)); + + await expect(promptPromise).resolves.toMatchObject({ + stopReason: "end_turn", + }); + expect(messageChunkTexts(client.sessionUpdate.mock.calls)).toEqual([ + "ORANGE", + "GREEN", + ]); + }); + + it("declines an explicit steer after the active turn has ended", async () => { + const { agent } = makeAgent(); + const sessionId = "s-expired-steer"; + installFakeSession(agent, sessionId); + + await expect( + agent.prompt({ + sessionId, + prompt: [{ type: "text", text: "too late" }], + _meta: { steer: true }, + }), + ).resolves.toMatchObject({ _meta: { steer: false } }); + + const session = (agent as unknown as { session: { turnQueue: unknown[] } }) + .session; + expect(session.turnQueue).toHaveLength(0); + }); + it("reconnects a disconnected signed-commit server before the turn", async () => { const { agent } = makeAgent(); const sessionId = "s-heal"; diff --git a/packages/agent/src/adapters/claude/claude-agent.ts b/packages/agent/src/adapters/claude/claude-agent.ts index 402d8dbdd4..983567d8a5 100644 --- a/packages/agent/src/adapters/claude/claude-agent.ts +++ b/packages/agent/src/adapters/claude/claude-agent.ts @@ -482,12 +482,20 @@ export class ClaudeAcpAgent extends BaseAcpAgent { const hasInFlightTurns = this.session.activeTurn !== null || this.session.turnQueue.length > 0; - if (hasInFlightTurns && isSteerMeta(params._meta)) { + const isSteer = isSteerMeta(params._meta); + if (hasInFlightTurns && isSteer) { // Fold into the running turn (promptToClaude tagged it priority:"next"); // the benign end_turn is ignored by clients, which key off _meta.steer. + const owner = + this.session.activeTurn ?? + this.session.turnQueue.find((turn) => !turn.settled); + owner?.pendingSteerUuids.add(promptUuid); this.session.input.push(userMessage); await this.broadcastUserMessage(params); - return { stopReason: "end_turn" }; + return { stopReason: "end_turn", _meta: { steer: true } }; + } + if (isSteer) { + return { stopReason: "end_turn", _meta: { steer: false } }; } if (!hasInFlightTurns && !isLocalOnlyCommand) { @@ -507,6 +515,7 @@ export class ClaudeAcpAgent extends BaseAcpAgent { const turn: Turn = { promptUuid, + pendingSteerUuids: new Set(), isLocalOnlyCommand, commandName: commandMatch?.[1], broadcast: () => this.broadcastUserMessage(params), @@ -1059,6 +1068,21 @@ export class ClaudeAcpAgent extends BaseAcpAgent { }, ); + if ( + !isTaskNotification && + session.activeTurn && + session.activeTurn.pendingSteerUuids.size > 0 + ) { + this.logger.debug( + "Deferring turn completion until pending steers are consumed", + { + sessionId, + pendingSteers: session.activeTurn.pendingSteerUuids.size, + }, + ); + break; + } + if ( (message as { stop_reason?: string }).stop_reason === "refusal" ) { @@ -1198,6 +1222,9 @@ export class ClaudeAcpAgent extends BaseAcpAgent { // active one first), then drops from the feed. Runs before the // cancelled guard so a turn enqueued after a cancel still starts. if (message.type === "user" && "uuid" in message && message.uuid) { + if (session.activeTurn?.pendingSteerUuids.delete(message.uuid)) { + break; + } const queued = session.turnQueue.find( (t) => t.promptUuid === message.uuid && !t.settled, ); diff --git a/packages/agent/src/adapters/claude/types.ts b/packages/agent/src/adapters/claude/types.ts index 80078dfa8f..37d46ac0d6 100644 --- a/packages/agent/src/adapters/claude/types.ts +++ b/packages/agent/src/adapters/claude/types.ts @@ -42,6 +42,7 @@ export type BackgroundTerminal = /** One in-flight `prompt()` call, settled by the session's consumer. */ export type Turn = { promptUuid: string; + pendingSteerUuids: Set; isLocalOnlyCommand: boolean; commandName?: string; /** Invoked once at activation, matching the pre-consumer broadcast timing. */ diff --git a/packages/agent/src/adapters/codex-app-server/app-server-client.test.ts b/packages/agent/src/adapters/codex-app-server/app-server-client.test.ts index 9501f019ce..a46cc96091 100644 --- a/packages/agent/src/adapters/codex-app-server/app-server-client.test.ts +++ b/packages/agent/src/adapters/codex-app-server/app-server-client.test.ts @@ -4,7 +4,7 @@ import { createBidirectionalStreams, type StreamPair, } from "../../utils/streams"; -import { AppServerClient } from "./app-server-client"; +import { AppServerClient, AppServerRequestError } from "./app-server-client"; interface RpcMessage { id?: number | string; @@ -86,7 +86,12 @@ describe("AppServerClient", () => { error: { code: -32001, message: "Server overloaded; retry later." }, }); - await expect(pending).rejects.toThrow("Server overloaded; retry later."); + const error = await pending.catch((requestError: unknown) => requestError); + expect(error).toBeInstanceOf(AppServerRequestError); + expect(error).toMatchObject({ + code: -32001, + message: "Server overloaded; retry later.", + }); await client.close(); }); diff --git a/packages/agent/src/adapters/codex-app-server/app-server-client.ts b/packages/agent/src/adapters/codex-app-server/app-server-client.ts index ff32050f5c..7d911173e2 100644 --- a/packages/agent/src/adapters/codex-app-server/app-server-client.ts +++ b/packages/agent/src/adapters/codex-app-server/app-server-client.ts @@ -24,6 +24,17 @@ export interface AppServerRpc { close(): Promise; } +export class AppServerRequestError extends Error { + constructor( + readonly code: number, + message: string, + readonly data?: unknown, + ) { + super(message); + this.name = "AppServerRequestError"; + } +} + /** * Bidirectional newline-delimited JSON-RPC client for the native Codex `app-server` subprocess. * Transport-agnostic via a {@link StreamPair} so tests can drive it over in-memory streams. @@ -173,7 +184,13 @@ export class AppServerClient implements AppServerRpc { } this.pending.delete(message.id); if (message.error) { - call.reject(new Error(message.error.message)); + call.reject( + new AppServerRequestError( + message.error.code, + message.error.message, + message.error.data, + ), + ); } else { call.resolve(message.result); } diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts index 9761794839..7062d98f70 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.test.ts @@ -10,6 +10,7 @@ import type { AppServerClientHandlers, AppServerRpc, } from "./app-server-client"; +import { AppServerRequestError } from "./app-server-client"; import { CodexAppServerAgent } from "./codex-app-server-agent"; import { sandboxPolicyFor } from "./session-config"; @@ -48,6 +49,9 @@ function makeStubRpc(responses: Record) { }; } const response = responses[method]; + if (response instanceof Error) { + throw response; + } return ( typeof response === "function" ? await response(params) @@ -2133,7 +2137,12 @@ describe("CodexAppServerAgent", () => { prompt: [{ type: "text", text: "more context" }], } as unknown as PromptRequest); - // The single turn/completed resolves both the original and the folded prompt. + await expect(second).resolves.toMatchObject({ + stopReason: "end_turn", + _meta: { steer: true }, + }); + + // The original prompt remains the sole owner of turn completion. stub.emit("thread/tokenUsage/updated", { tokenUsage: { last: { @@ -2145,12 +2154,11 @@ describe("CodexAppServerAgent", () => { }, }); stub.emit("turn/completed", { turn: { status: "completed" } }); - const [firstResult, secondResult] = await Promise.all([first, second]); + const firstResult = await first; expect(firstResult).toMatchObject({ stopReason: "end_turn", usage: { totalTokens: 45 }, }); - expect(secondResult).toEqual({ stopReason: "end_turn" }); const steer = stub.requests.find((r) => r.method === "turn/steer"); expect(steer?.params).toMatchObject({ @@ -2164,6 +2172,118 @@ describe("CodexAppServerAgent", () => { ); }); + it("rejects a failed turn/steer without echoing or acknowledging it", async () => { + const stub = makeStubRpc({ + "thread/start": { thread: { id: "t" } }, + "turn/start": { turn: { id: "turn_1" } }, + "turn/steer": new Error("steer transport failed"), + }); + const { client, sessionUpdates } = makeFakeClient(); + const agent = new CodexAppServerAgent(client, { + processOptions: { binaryPath: "/x/codex" }, + rpcFactory: stub.factory, + }); + + await agent.newSession({ cwd: "/r" } as unknown as NewSessionRequest); + const first = agent.prompt({ + sessionId: "t", + prompt: [{ type: "text", text: "one" }], + } as unknown as PromptRequest); + stub.emit("turn/started", { threadId: "t", turn: { id: "turn_1" } }); + + await expect( + agent.prompt({ + sessionId: "t", + prompt: [{ type: "text", text: "lost steer" }], + _meta: { steer: true }, + } as unknown as PromptRequest), + ).rejects.toThrow("steer transport failed"); + expect(sessionUpdates).not.toContainEqual( + expect.objectContaining({ + update: expect.objectContaining({ + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "lost steer" }, + }), + }), + ); + + stub.emit("turn/completed", { turn: { status: "completed" } }); + await first; + }); + + it("declines a stale turn/steer so the caller can queue it normally", async () => { + const stub = makeStubRpc({ + "thread/start": { thread: { id: "t" } }, + "turn/start": { turn: { id: "turn_1" } }, + "turn/steer": new AppServerRequestError( + -32600, + "expected active turn id `turn_1` but found `turn_2`", + ), + }); + const { client, sessionUpdates } = makeFakeClient(); + const agent = new CodexAppServerAgent(client, { + processOptions: { binaryPath: "/x/codex" }, + rpcFactory: stub.factory, + }); + + await agent.newSession({ cwd: "/r" } as unknown as NewSessionRequest); + const first = agent.prompt({ + sessionId: "t", + prompt: [{ type: "text", text: "one" }], + } as unknown as PromptRequest); + stub.emit("turn/started", { threadId: "t", turn: { id: "turn_1" } }); + + await expect( + agent.prompt({ + sessionId: "t", + prompt: [{ type: "text", text: "queue me" }], + _meta: { steer: true }, + } as unknown as PromptRequest), + ).resolves.toMatchObject({ _meta: { steer: false } }); + expect(sessionUpdates).not.toContainEqual( + expect.objectContaining({ + update: expect.objectContaining({ + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "queue me" }, + }), + }), + ); + + stub.emit("turn/completed", { turn: { status: "completed" } }); + await first; + }); + + it("declines an explicit steer after the active turn has ended", async () => { + const stub = makeStubRpc({ + "thread/start": { thread: { id: "t" } }, + }); + const { client, sessionUpdates } = makeFakeClient(); + const agent = new CodexAppServerAgent(client, { + processOptions: { binaryPath: "/x/codex" }, + rpcFactory: stub.factory, + }); + + await agent.newSession({ cwd: "/r" } as unknown as NewSessionRequest); + await expect( + agent.prompt({ + sessionId: "t", + prompt: [{ type: "text", text: "too late" }], + _meta: { steer: true }, + } as unknown as PromptRequest), + ).resolves.toMatchObject({ _meta: { steer: false } }); + expect( + stub.requests.filter((request) => request.method === "turn/start"), + ).toHaveLength(0); + expect(sessionUpdates).not.toContainEqual( + expect.objectContaining({ + update: expect.objectContaining({ + sessionUpdate: "user_message_chunk", + content: { type: "text", text: "too late" }, + }), + }), + ); + }); + it("refreshes the live turnId from each turn/steer response", async () => { const stub = makeStubRpc({ "thread/start": { thread: { id: "t" } }, diff --git a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts index 823a23c6e3..5029f2127a 100644 --- a/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts +++ b/packages/agent/src/adapters/codex-app-server/codex-app-server-agent.ts @@ -50,6 +50,7 @@ import { resolveSpokenNarration } from "../session-meta"; import { AppServerClient, type AppServerClientHandlers, + AppServerRequestError, type AppServerRpc, } from "./app-server-client"; import { handleServerRequest } from "./approvals"; @@ -88,6 +89,16 @@ import { parseStructuredOutput } from "./structured-output"; import { TurnController } from "./turn-controller"; import { UsageTracker } from "./usage-tracker"; +function isStaleTurnSteerError(error: unknown): boolean { + if (!(error instanceof AppServerRequestError) || error.code !== -32600) { + return false; + } + return ( + error.message === "no active turn to steer" || + /^expected active turn id `.*` but found `.*`$/.test(error.message) + ); +} + type AppServerSessionMeta = { // The host sends either a plain string or the Claude-style `{ append }` form. systemPrompt?: string | { append?: string }; @@ -753,32 +764,43 @@ export class CodexAppServerAgent extends BaseAcpAgent { if (dropped > 0) { this.logger.warn("Dropped non-text/non-image prompt blocks", { dropped }); } - // Echo the user prompt (codex emits none), for fresh turns and steering alike. - this.broadcastUserInput(params.prompt); - if (this.turns.isRunning) { // A turn is already running: fold the message in via turn/steer (precondition: the // active turnId). Refresh from the response's rotated turnId so a later steer/interrupt // still targets the live turn (no turn/started is re-emitted for a steer). - const steerRes = await this.rpc - .request<{ turnId?: string }>(APP_SERVER_METHODS.TURN_STEER, { - threadId: this.threadId, - input, - expectedTurnId: this.turns.activeTurnId, - }) - .catch((err) => { - this.logger.warn("turn/steer failed", err); - return undefined; - }); + let steerRes: { turnId?: string }; + try { + steerRes = await this.rpc.request<{ turnId?: string }>( + APP_SERVER_METHODS.TURN_STEER, + { + threadId: this.threadId, + input, + expectedTurnId: this.turns.activeTurnId, + }, + ); + } catch (error) { + if ( + (params._meta as { steer?: unknown } | undefined)?.steer === true && + isStaleTurnSteerError(error) + ) { + return { stopReason: "end_turn", _meta: { steer: false } }; + } + throw error; + } this.turns.onSteered(steerRes?.turnId); - const response = await this.turns.awaitCompletion(); - return { stopReason: response.stopReason }; + this.broadcastUserInput(params.prompt); + return { stopReason: "end_turn", _meta: { steer: true } }; + } + if ((params._meta as { steer?: unknown } | undefined)?.steer === true) { + return { stopReason: "end_turn", _meta: { steer: false } }; } if (this.turns.isPending) { // A turn is pending but has no turnId yet, so we can't steer; fail fast. throw new Error("prompt() called while a turn is already in progress"); } + // Codex does not echo user input, so emit it only once delivery can proceed. + this.broadcastUserInput(params.prompt); const response = await this.runTurn(input); return this.maybeOfferPlanImplementation(response); } diff --git a/packages/agent/src/adapters/codex-app-server/turn-controller.ts b/packages/agent/src/adapters/codex-app-server/turn-controller.ts index e6f6483a9a..90e7bcb21b 100644 --- a/packages/agent/src/adapters/codex-app-server/turn-controller.ts +++ b/packages/agent/src/adapters/codex-app-server/turn-controller.ts @@ -48,11 +48,6 @@ export class TurnController { if (typeof id === "string") this.turnId = id; } - /** Await the in-flight turn's completion (the steer path reuses the original). */ - awaitCompletion(): Promise { - return this.completion ?? Promise.resolve({ stopReason: "end_turn" }); - } - /** Atomically claim the pending turn (clears the slot + turnId synchronously), or undefined if already claimed. */ claim(): PendingTurn | undefined { const pending = this.pending;