feat(cloud): add native steering to agent adapters

Author: tatoalo

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";

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,
               );

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<string>;
   isLocalOnlyCommand: boolean;
   commandName?: string;
   /** Invoked once at activation, matching the pre-consumer broadcast timing. */

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();
   });
 

packages/agent/src/adapters/codex-app-server/app-server-client.ts

@@ -24,6 +24,17 @@ export interface AppServerRpc {
   close(): Promise<void>;
 }
 
+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);
     }

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<string, unknown>) {
         };
       }
       const response = responses[method];
+      if (response instanceof Error) {
+        throw response;
+      }
       return (
         typeof response === "function"
           ? await response(params)
@@ -1960,7 +1964,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: {
@@ -1972,12 +1981,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({
@@ -1991,6 +1999,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" } },