From 980ac8dc56e671cbcdb607171f290dfcc024c696 Mon Sep 17 00:00:00 2001 From: Mark Tkachenko Date: Tue, 21 Jul 2026 18:53:56 +0200 Subject: [PATCH] Steering implementation (#885) --- .gitignore | 1 + examples/steering.ts | 191 ++++++++++++++++++++++++++++++++++ src/acp-agent.ts | 137 +++++++++++++++++++++++++ src/tests/acp-agent.test.ts | 198 ++++++++++++++++++++++++++++++++++++ 4 files changed, 527 insertions(+) create mode 100644 examples/steering.ts diff --git a/.gitignore b/.gitignore index c4b8d1cd..ff121fe6 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ dist .DS_Store .claude/*.local.* CLAUDE.local.md +.idea/ \ No newline at end of file diff --git a/examples/steering.ts b/examples/steering.ts new file mode 100644 index 00000000..3971b3e1 --- /dev/null +++ b/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, + Readable.toWeb(child.stdout!) as unknown as ReadableStream, + ); + + // 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((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(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); +}); diff --git a/src/acp-agent.ts b/src/acp-agent.ts index 49db45d9..8aa840da 100644 --- a/src/acp-agent.ts +++ b/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; + 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 { + 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(STEER_METHOD, { parse: parseSteerRequest }, (ctx) => + agent.steer(ctx.params), + ) .connect(stream); agent = new ClaudeAcpAgent(new ClientConnection(connection.client)); diff --git a/src/tests/acp-agent.test.ts b/src/tests/acp-agent.test.ts index 8b3fc937..6c3b9e22 100644 --- a/src/tests/acp-agent.test.ts +++ b/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(), + 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 = {