Steering implementation (#885)

.gitignore

@@ -3,3 +3,4 @@ dist
 .DS_Store
 .claude/*.local.*
 CLAUDE.local.md
+.idea/

examples/steering.ts

@@ -0,0 +1,191 @@
+/**
+ * Example: mid-turn steering over ACP with the `_session/steering` extension.
+ *
+ * "Steering" lets a client deliver a follow-up message to a turn that is still
+ * running, instead of waiting for it to finish and sending a fresh
+ * `session/prompt`. This is what powers "the user typed something while the
+ * agent was still working" — the new message joins the in-flight turn so the
+ * agent can adapt immediately (it shines in multi-step / tool-using turns,
+ * where the message slots in between tool calls).
+ *
+ * The wire protocol (see ../steering_protocol.md) has three moving parts:
+ *
+ *   1. The agent advertises support in its `initialize` response, at the
+ *      top-level `_meta.steering.supported` (a sibling of `agentCapabilities`).
+ *   2. The client calls the `_session/steering` request with `{ sessionId,
+ *      prompt }` while a turn is running.
+ *   3. The agent replies with an `outcome`:
+ *        - "injected"       the message joined the running turn;
+ *        - "startedNewTurn" the turn had already finished (an unavoidable race),
+ *                           so the message began a fresh turn instead.
+ *      Both are success outcomes — the message is never dropped and the race is
+ *      never surfaced as an error.
+ *
+ * This example launches the agent as a subprocess, starts a deliberately
+ * long-running prompt, and — as soon as the agent begins streaming — injects a
+ * steering message and prints the outcome. All agent output is streamed to
+ * stdout so you can watch the turn change course.
+ *
+ * Run (build the agent first so `dist/index.js` exists):
+ *
+ *   npm run build
+ *   node examples/steering.ts
+ *
+ * (Node < 22.18 needs `node --experimental-strip-types examples/steering.ts`.)
+ *
+ * Override the prompts with the PROMPT / STEER env vars. Requires the agent to
+ * be authenticated, since it talks to the real model.
+ */
+import { spawn } from "node:child_process";
+import { Readable, Writable } from "node:stream";
+import { fileURLToPath } from "node:url";
+import path from "node:path";
+import { client as acpClient, methods, ndJsonStream } from "@agentclientprotocol/sdk";
+
+/** The steering extension method, per the ACP steering wire protocol. */
+const STEERING_METHOD = "_session/steering";
+
+/** Params for a `_session/steering` request — the same shape as the relevant
+ *  subset of a `session/prompt`. */
+type SteeringRequest = {
+  sessionId: string;
+  prompt: Array<{ type: "text"; text: string }>;
+};
+
+/** Result of a `_session/steering` request. Both values are successes: they
+ *  tell the client where the message landed, not whether it succeeded. */
+type SteeringResponse = {
+  outcome: "injected" | "startedNewTurn";
+};
+
+// The built agent entry. Run `npm run build` first so this exists.
+const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "..");
+const AGENT_ENTRY = process.env.AGENT_ENTRY ?? path.join(repoRoot, "dist", "index.js");
+const CWD = process.env.CWD ?? process.cwd();
+
+// A deliberately long-running first prompt, and the follow-up injected while it
+// is still streaming. Override either via env vars to experiment.
+const PROMPT =
+  process.env.PROMPT ??
+  "Count slowly from 1 to 30, one number per line, with a short sentence of " +
+    "commentary after each. Do not stop early.";
+const STEER =
+  process.env.STEER ?? "Actually stop counting and instead reply with exactly one line: STEERED-OK";
+
+function log(msg: string) {
+  process.stderr.write(`\x1b[2m[client]\x1b[0m ${msg}\n`);
+}
+
+const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+async function main() {
+  // An ACP client launches the agent as a subprocess and speaks JSON-RPC over
+  // its stdin/stdout. stderr is inherited so the agent's own logs stay visible.
+  const child = spawn(process.execPath, [AGENT_ENTRY], {
+    stdio: ["pipe", "pipe", "inherit"],
+    env: process.env,
+  });
+  child.on("error", (err) => {
+    log(`failed to spawn agent (${AGENT_ENTRY}): ${err}`);
+    process.exit(1);
+  });
+
+  const stream = ndJsonStream(
+    Writable.toWeb(child.stdin!) as WritableStream<Uint8Array>,
+    Readable.toWeb(child.stdout!) as unknown as ReadableStream<Uint8Array>,
+  );
+
+  // Resolves the first time the agent streams assistant text — our signal that
+  // the turn is genuinely underway and therefore steerable.
+  let signalFirstOutput = () => {};
+  const firstOutput = new Promise<void>((resolve) => (signalFirstOutput = resolve));
+
+  const connection = acpClient({ name: "steering-example" })
+    .onNotification(methods.client.session.update, (ctx) => {
+      const update = ctx.params.update;
+      if (update.sessionUpdate === "agent_message_chunk" && update.content.type === "text") {
+        process.stdout.write(update.content.text);
+        signalFirstOutput();
+      }
+    })
+    // Auto-approve permission prompts so the turn is never blocked on us.
+    .onRequest(methods.client.session.requestPermission, (ctx) => {
+      const options = ctx.params.options;
+      const option = options.find((o) => o.kind === "allow_once") ?? options[0];
+      return { outcome: { outcome: "selected", optionId: option.optionId } };
+    })
+    // Minimal file-system stubs; the example prompts don't touch files.
+    .onRequest(methods.client.fs.readTextFile, () => ({ content: "" }))
+    .onRequest(methods.client.fs.writeTextFile, () => ({}))
+    .connect(stream);
+
+  const agent = connection.agent;
+
+  // 1. Initialize and confirm the agent advertises steering. Per the wire
+  //    protocol the capability lives at the TOP-LEVEL `_meta` of the initialize
+  //    result — a sibling of `agentCapabilities`, not nested inside it.
+  const init = await agent.request(methods.agent.initialize, {
+    protocolVersion: 1,
+    clientCapabilities: { fs: { readTextFile: true, writeTextFile: true } },
+  });
+  const meta = init._meta as { steering?: { supported?: boolean } } | null | undefined;
+  const steeringSupported = meta?.steering?.supported === true;
+  log(`agent advertises steering: ${steeringSupported}`);
+  if (!steeringSupported) {
+    log("agent does not advertise steering; the steering request may be rejected.");
+  }
+
+  // 2. Open a session.
+  const { sessionId } = await agent.request(methods.agent.session.new, {
+    cwd: CWD,
+    mcpServers: [],
+  });
+  log(`session: ${sessionId}`);
+
+  // 3. Start a long turn, but DON'T await it yet — we need it in flight so we
+  //    can steer it. Its output streams through the notification handler above.
+  log(`prompt: ${PROMPT}`);
+  process.stdout.write("\n----- agent output -----\n");
+  const turn = agent.request(methods.agent.session.prompt, {
+    sessionId,
+    prompt: [{ type: "text", text: PROMPT }],
+  });
+
+  // 4. Once the turn is producing output, inject the follow-up. Wait for the
+  //    first streamed chunk (with a fallback) plus a beat, so the steer clearly
+  //    lands mid-turn.
+  await Promise.race([firstOutput, delay(5000)]);
+  await delay(1000);
+
+  process.stdout.write("\n");
+  log(`steer: ${STEER}`);
+  const steerRequest: SteeringRequest = {
+    sessionId,
+    prompt: [{ type: "text", text: STEER }],
+  };
+  try {
+    const result = await agent.request<SteeringResponse>(STEERING_METHOD, steerRequest);
+    log(`steer outcome: ${result.outcome}`);
+  } catch (err) {
+    log(`steer rejected: ${err}`);
+  }
+
+  // 5. Await the turn. With outcome "injected" the steer already reshaped the
+  //    output above; with "startedNewTurn" the follow-up runs as a fresh turn
+  //    that may still be streaming, so linger briefly to capture it.
+  const response = await turn.catch((err: unknown) => {
+    log(`turn error: ${err}`);
+    return undefined;
+  });
+  if (response) log(`turn stopReason: ${response.stopReason}`);
+  await delay(3000);
+  process.stdout.write("\n----- end of agent output -----\n");
+
+  connection.close();
+  child.kill();
+}
+
+main().catch((err) => {
+  log(`fatal: ${err?.stack ?? err}`);
+  process.exit(1);
+});

src/acp-agent.ts

@@ -192,6 +192,66 @@ const TURN_NO_RESULT_MESSAGE =
   "The turn ended without a result: the agent went idle while this prompt was still in flight " +
   "(e.g. the model stream dropped mid-turn). Any partial output may be incomplete; please retry.";
 
+/** Custom (extension) request method a client uses to steer the turn that is
+ *  currently running: the message is injected into the in-flight turn rather
+ *  than queued as a separate `session/prompt`. Named `_session/steering` per the
+ *  agreed ACP steering wire protocol; advertised to clients via the top-level
+ *  `InitializeResponse._meta.steering.supported`. */
+const STEER_METHOD = "_session/steering";
+
+/** How urgently the SDK delivers a steered message relative to the running
+ *  turn — an internal Claude implementation detail, not part of the wire
+ *  contract. `now` pre-empts the current generation and handles the message
+ *  immediately (interrupting a single-shot response, or slotting in between a
+ *  multi-step turn's tool calls). Maps to `SDKUserMessage.priority`; injected
+ *  steering always uses `now` so the running turn adapts as soon as possible. */
+const STEER_PRIORITY = "now" as const;
+
+/** Params of a {@link STEER_METHOD} request. Shaped like the relevant subset of
+ *  a `PromptRequest` so the same `promptToClaude` conversion applies. Delivery
+ *  priority is deliberately NOT exposed here — it's an internal detail the agent
+ *  chooses (see {@link STEER_PRIORITY}). */
+export type SteerRequest = {
+  sessionId: string;
+  prompt: PromptRequest["prompt"];
+};
+
+/** Where a steering message was accepted, per the wire protocol's two
+ *  successful outcomes:
+ *   - `injected`: a turn was still running and the message was applied to it;
+ *   - `startedNewTurn`: the turn we meant to steer had already finished (an
+ *     unavoidable race), so the message began a fresh turn instead of being
+ *     dropped.
+ *  Both are success results — never a JSON-RPC error — and tell the client
+ *  where the message landed. */
+type SteerOutcome = "injected" | "startedNewTurn";
+
+/** Result of a {@link STEER_METHOD} request: the single required `outcome`
+ *  field the client reads to learn where its steering message was accepted. */
+export type SteerResponse = {
+  outcome: SteerOutcome;
+};
+
+/** Validate raw JSON-RPC params into a {@link SteerRequest}. Kept minimal — the
+ *  content blocks are handed to `promptToClaude`, which tolerates unknown block
+ *  types — but `sessionId` and a non-empty `prompt` array are required. */
+function parseSteerRequest(params: unknown): SteerRequest {
+  if (!params || typeof params !== "object") {
+    throw RequestError.invalidParams(undefined, "steer params must be an object");
+  }
+  const { sessionId, prompt } = params as Record<string, unknown>;
+  if (typeof sessionId !== "string" || sessionId.length === 0) {
+    throw RequestError.invalidParams(undefined, "steer params require a non-empty sessionId");
+  }
+  if (!Array.isArray(prompt) || prompt.length === 0) {
+    throw RequestError.invalidParams(undefined, "steer params require a non-empty prompt array");
+  }
+  return {
+    sessionId,
+    prompt: prompt as PromptRequest["prompt"],
+  };
+}
+
 /** Internal model-selection state. Mirrors the shape the ACP SDK exposed as
  *  `SessionModelState` before model selection moved entirely into
  *  `SessionConfigOption` (category "model"). Retained internally to track the
@@ -1368,6 +1428,15 @@ export class ClaudeAcpAgent {
         ...terminalAuthMethods,
         ...(supportsGatewayAuth ? [gatewayAuthMethod, gatewayBedrockAuthMethod] : []),
       ],
+      // Top-level `_meta` (sibling of `agentCapabilities`), per the ACP steering
+      // wire protocol: advertises the `_session/steering` extension request so
+      // clients know they may inject a follow-up into the running turn (see
+      // STEER_METHOD) instead of queuing it as a separate `session/prompt`.
+      _meta: {
+        steering: {
+          supported: true,
+        },
+      },
     };
   }
 
@@ -1665,6 +1734,71 @@ export class ClaudeAcpAgent {
     return response;
   }
 
+  /** Steer the session per the ACP steering wire protocol: apply a follow-up
+   *  message to the turn that is currently running, or — if that turn already
+   *  finished — start a fresh turn with it. Never drops the message and never
+   *  returns a JSON-RPC error for the "arrived too late" race; both paths are
+   *  success outcomes (see {@link SteerOutcome}).
+   *
+   *  When a turn is in flight this injects (returns `injected`): unlike
+   *  `prompt()`, it does NOT create a Turn or enqueue on `turnQueue`; it pushes
+   *  an `SDKUserMessage` onto the same streaming input, which the SDK routes
+   *  into the in-flight turn. The injected message's echo carries a uuid that
+   *  matches no queued turn, so the consumer drops it as an unrelated replay
+   *  without promoting/settling anything. It is delivered at {@link
+   *  STEER_PRIORITY} (`now`) so it pre-empts the current generation (interrupting
+   *  a single-shot response, or slotting in between a multi-step turn's tool
+   *  calls). The steered message's own output streams via `session/update`, not
+   *  this response.
+   *
+   *  When the session is idle (no unsettled turn — the turn we meant to steer
+   *  raced ahead and finished), this starts a normal new turn with the message
+   *  and returns `startedNewTurn`. That turn is fire-and-forget from the steer
+   *  request's view: its `PromptResponse` and output flow through the usual
+   *  `prompt()`/`session/update` path, so we return the outcome immediately
+   *  rather than awaiting turn completion. */
+  async steer(params: SteerRequest): Promise<SteerResponse> {
+    const sessionId = params.sessionId;
+    const prompt = params.prompt;
+
+    const session = this.sessions[sessionId];
+    if (!session) {
+      throw new Error("Session not found");
+    }
+    if (session.queryClosed) {
+      throw RequestError.internalError(undefined, SESSION_ENDED_MESSAGE);
+    }
+    // "A turn is running" = the queue holds an unsettled turn. This covers both
+    // the activated turn and one just submitted but not yet echoed/activated,
+    // which is exactly the window in which steering is meaningful.
+    const turnInFlight = (session.turnQueue ?? []).some((t) => !t.settled);
+    const promptRequest: PromptRequest = {
+      sessionId: sessionId,
+      prompt: prompt,
+    };
+
+    if (!turnInFlight) {
+      // Race: the turn we meant to steer already finished. Per the protocol the
+      // message must not be dropped nor surfaced as an error — start a fresh
+      // turn with it. Don't await: the new turn streams via session/update and
+      // its PromptResponse is consumed by the normal prompt() path; we only owe
+      // the client the outcome. `.catch` keeps the detached promise from
+      // becoming an unhandled rejection.
+      this.prompt(promptRequest).catch((error) => {
+        this.logger.error(`Session ${sessionId}: steered new turn failed: ${error}`);
+      });
+      return { outcome: "startedNewTurn" };
+    }
+
+    const userMessage = promptToClaude(promptRequest);
+    userMessage.uuid = randomUUID();
+    // Deliver into the running turn rather than queuing behind it as a fresh
+    // prompt would.
+    userMessage.priority = STEER_PRIORITY;
+    session.input.push(userMessage);
+    return { outcome: "injected" };
+  }
+
   /** Lazily start the per-session consumer that drains the SDK query stream for
    *  the session's whole life. Idempotent: only the first `prompt()` starts it. */
   private ensureConsumer(session: Session, sessionId: string): void {
@@ -7242,6 +7376,9 @@ export function runAcp() {
       runPromptWithCancellation(agent, ctx.params, ctx.signal),
     )
     .onNotification(methods.agent.session.cancel, (ctx) => agent.cancel(ctx.params))
+    .onRequest<SteerRequest, SteerResponse>(STEER_METHOD, { parse: parseSteerRequest }, (ctx) =>
+      agent.steer(ctx.params),
+    )
     .connect(stream);
 
   agent = new ClaudeAcpAgent(new ClientConnection(connection.client));

src/tests/acp-agent.test.ts

@@ -9458,6 +9458,204 @@ describe("deferred settlement for live background subagents (issues #864/#866)",
   });
 });
 
+describe("turn steering (_session/steering)", () => {
+  function createMockAgent() {
+    const mockClient = {
+      sessionUpdate: async () => {},
+    } as unknown as AcpClient;
+    return new ClaudeAcpAgent(mockClient, { log: () => {}, error: () => {} });
+  }
+
+  function createResultMessage() {
+    return {
+      type: "result" as const,
+      subtype: "success" as const,
+      stop_reason: "end_turn",
+      is_error: false,
+      result: "",
+      errors: [],
+      duration_ms: 0,
+      duration_api_ms: 0,
+      num_turns: 1,
+      total_cost_usd: 0,
+      usage: {
+        input_tokens: 10,
+        output_tokens: 5,
+        cache_read_input_tokens: 0,
+        cache_creation_input_tokens: 0,
+      },
+      modelUsage: {},
+      permission_denials: [],
+      uuid: randomUUID(),
+      session_id: "test-session",
+    };
+  }
+
+  const waitFor = async (cond: () => boolean) => {
+    for (let i = 0; i < 200; i++) {
+      if (cond()) return;
+      await new Promise((r) => setTimeout(r, 0));
+    }
+    throw new Error("waitFor timed out");
+  };
+
+  it("rejects when the session is unknown", async () => {
+    const agent = createMockAgent();
+    await expect(
+      agent.steer({ sessionId: "missing", prompt: [{ type: "text", text: "hi" }] }),
+    ).rejects.toThrow("Session not found");
+  });
+
+  it("rejects when the query stream has already closed", async () => {
+    const agent = createMockAgent();
+    agent.sessions["test-session"] = mockSessionState({
+      input: new Pushable<any>(),
+      queryClosed: true,
+      turnQueue: [
+        {
+          promptUuid: "x",
+          isLocalOnlyCommand: false,
+          settled: false,
+          resolve: () => {},
+          reject: () => {},
+        },
+      ],
+    });
+    await expect(
+      agent.steer({ sessionId: "test-session", prompt: [{ type: "text", text: "hi" }] }),
+    ).rejects.toThrow();
+  });
+
+  it("advertises steering support at _meta.steering.supported", async () => {
+    const agent = createMockAgent();
+    const response = await agent.initialize({
+      protocolVersion: 1,
+      clientCapabilities: {},
+    });
+    // Top-level _meta (sibling of agentCapabilities), per the wire protocol.
+    expect((response._meta as any)?.steering).toEqual({ supported: true });
+  });
+
+  it("starts a new turn (outcome 'startedNewTurn') when no turn is in flight (race)", async () => {
+    const agent = createMockAgent();
+    const captured: any[] = [];
+    // Idle session: turnQueue starts empty, so the turn we meant to steer has
+    // (from the agent's view) already finished. Steering must not error — it
+    // starts a fresh turn with the message.
+    injectGeneratorSession(
+      agent,
+      (input) => {
+        async function* messageGenerator() {
+          const iter = input[Symbol.asyncIterator]();
+          const u1 = await iter.next();
+          captured.push(u1.value);
+          yield userEcho(u1.value); // activates the new turn
+          yield createResultMessage();
+          yield { type: "system", subtype: "session_state_changed", state: "idle" };
+        }
+        return messageGenerator();
+      },
+      { turnQueue: [] },
+    );
+
+    const res = await agent.steer({
+      sessionId: "test-session",
+      prompt: [{ type: "text", text: "late follow-up" }],
+    });
+    expect(res.outcome).toBe("startedNewTurn");
+
+    // A real turn was enqueued and drains like any normal prompt.
+    await waitFor(() => (agent.sessions["test-session"].turnQueue ?? []).length === 0);
+    expect(captured).toHaveLength(1);
+    // It went through the normal prompt() path — no steering delivery priority.
+    expect(captured[0].priority).toBeUndefined();
+    expect(JSON.stringify(captured[0].message.content)).toContain("late follow-up");
+  });
+
+  it("injects (outcome 'injected') a priority:'now' message into the running turn without spawning a new turn", async () => {
+    const agent = createMockAgent();
+    const captured: any[] = [];
+    injectGeneratorSession(agent, (input) => {
+      async function* messageGenerator() {
+        const iter = input[Symbol.asyncIterator]();
+        const u1 = await iter.next();
+        yield userEcho(u1.value); // turn becomes active
+        // The steered message is pushed next; capture it, replay its echo
+        // (which matches no queued turn and must be dropped), then finish.
+        const steered = await iter.next();
+        captured.push(steered.value);
+        yield userEcho(steered.value); // unrelated replay — must NOT settle a turn
+        yield createResultMessage();
+        yield { type: "system", subtype: "session_state_changed", state: "idle" };
+      }
+      return messageGenerator();
+    });
+
+    const turn = agent.prompt({
+      sessionId: "test-session",
+      prompt: [{ type: "text", text: "start" }],
+    });
+    await waitFor(() => !!agent.sessions["test-session"]?.activeTurn);
+
+    const steerRes = await agent.steer({
+      sessionId: "test-session",
+      prompt: [{ type: "text", text: "also handle X" }],
+    });
+
+    // Steering into a live turn reports the 'injected' outcome.
+    expect(steerRes.outcome).toBe("injected");
+
+    // The original turn resolves as a single, normal end_turn — the injected
+    // message steered it rather than producing a second PromptResponse.
+    await expect(turn).resolves.toEqual(expect.objectContaining({ stopReason: "end_turn" }));
+    expect(agent.sessions["test-session"].turnQueue).toHaveLength(0);
+
+    // The pushed message carried priority:'now' and a fresh uuid (matching no
+    // queued turn, so its echo is dropped rather than settling anything).
+    expect(captured).toHaveLength(1);
+    const injected = captured[0];
+    expect(injected.priority).toBe("now");
+    expect(typeof injected.uuid).toBe("string");
+    expect(JSON.stringify(injected.message.content)).toContain("also handle X");
+  });
+
+  it("always injects at 'now' priority, ignoring any client-supplied priority", async () => {
+    const agent = createMockAgent();
+    const captured: any[] = [];
+    injectGeneratorSession(agent, (input) => {
+      async function* messageGenerator() {
+        const iter = input[Symbol.asyncIterator]();
+        const u1 = await iter.next();
+        yield userEcho(u1.value);
+        const steered = await iter.next();
+        captured.push(steered.value);
+        yield createResultMessage();
+        yield { type: "system", subtype: "session_state_changed", state: "idle" };
+      }
+      return messageGenerator();
+    });
+
+    const turn = agent.prompt({
+      sessionId: "test-session",
+      prompt: [{ type: "text", text: "start" }],
+    });
+    await waitFor(() => !!agent.sessions["test-session"]?.activeTurn);
+
+    // Delivery priority is an internal detail, not part of the wire contract.
+    // Even if a client sneaks a `priority` field into the params, it's ignored
+    // and the message is always delivered at 'now'.
+    const res = await agent.steer({
+      sessionId: "test-session",
+      prompt: [{ type: "text", text: "queue this after" }],
+      priority: "next",
+    } as any);
+    await turn;
+
+    expect(res.outcome).toBe("injected");
+    expect(captured[0]?.priority).toBe("now");
+  });
+});
+
 describe("session/cancel wedge recovery (issue #680)", () => {
   function createMockAgent() {
     const mockClient = {