diff --git a/examples/steering.ts b/examples/steering.ts index 3971b3e1..9a0b4b5a 100644 --- a/examples/steering.ts +++ b/examples/steering.ts @@ -8,18 +8,18 @@ * 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: + * The wire protocol has three moving parts: * * 1. The agent advertises support in its `initialize` response, at the - * top-level `_meta.steering.supported` (a sibling of `agentCapabilities`). + * top-level `_meta.steering` (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. + * - "startedNewTurn" the turn had already finished and the legacy + * detached fallback was used; + * - "promptRequired" the turn had already finished, but this request + * explicitly opted into Host-owned prompt delivery. * * This example launches the agent as a subprocess, starts a deliberately * long-running prompt, and — as soon as the agent begins streaming — injects a @@ -40,7 +40,13 @@ 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"; +import { + client as acpClient, + methods, + ndJsonStream, + type PromptRequest, + type PromptResponse, +} from "@agentclientprotocol/sdk"; /** The steering extension method, per the ACP steering wire protocol. */ const STEERING_METHOD = "_session/steering"; @@ -50,12 +56,22 @@ const STEERING_METHOD = "_session/steering"; type SteeringRequest = { sessionId: string; prompt: Array<{ type: "text"; text: string }>; + _meta?: { steering?: { idleBehavior?: "promptRequired" } }; }; -/** 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"; +/** Result of a `_session/steering` request. `injected` means the message joined + * the running turn; `promptRequired` means the turn had already settled, so the + * message was NOT consumed and must be (re)sent through a normal + * `session/prompt`. Both are successes. */ +type SteeringResponse = + | { outcome: "injected" } + | { outcome: "startedNewTurn" } + | { outcome: "promptRequired"; reason: "noRunningTurn" }; + +/** The existing steering capability advertised at the top-level `_meta.steering` + * of the `initialize` result. The idle behavior is selected per request. */ +type SteeringCapability = { + supported?: boolean; }; // The built agent entry. Run `npm run build` first so this exists. @@ -87,105 +103,119 @@ async function main() { }); 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}`); + 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); + + try { + 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 steering = (init._meta as { steering?: SteeringCapability } | null | undefined) + ?.steering; + if (steering?.supported !== true) { + throw new Error("agent does not advertise steering support"); + } + log("agent advertises steering support"); + + // 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 }], + // Opt into the host-owned idle fallback. Without this request metadata, + // the Adapter preserves its legacy `startedNewTurn` behavior. + _meta: { steering: { idleBehavior: "promptRequired" } }, + }; + const result = await agent.request(STEERING_METHOD, steerRequest); + log(`steer outcome: ${result.outcome}`); + + if (result.outcome === "promptRequired") { + // The target turn already settled, so steering did not consume the message. + // Start a normal session/prompt on the same session to deliver it — that + // request owns the continuation's updates and terminal response. + log(`steer fallback: ${result.reason}; starting a normal session/prompt`); + const continuationRequest: PromptRequest = { + sessionId: steerRequest.sessionId, + prompt: steerRequest.prompt, + }; + const continuation = await agent.request( + methods.agent.session.prompt, + continuationRequest, + ); + log(`continuation stopReason: ${continuation.stopReason}`); + } + + // 5. Await the original turn. With outcome "injected" the steer already + // reshaped the output above; the promptRequired branch owns its own turn. + const response = await turn; + log(`original turn stopReason: ${response.stopReason}`); + process.stdout.write("\n----- end of agent output -----\n"); + } finally { + connection.close(); + } + } finally { + child.kill(); } - - // 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); + process.exitCode = 1; }); diff --git a/src/acp-agent.ts b/src/acp-agent.ts index 7cd77f64..766ca778 100644 --- a/src/acp-agent.ts +++ b/src/acp-agent.ts @@ -208,6 +208,15 @@ const STEER_METHOD = "_session/steering"; * steering always uses `now` so the running turn adapts as soon as possible. */ const STEER_PRIORITY = "now" as const; +/** Request-level steering options. `promptRequired` is opt-in so existing Hosts + * keep the established idle fallback behavior. */ +type SteerMeta = { + [key: string]: unknown; + steering?: { + idleBehavior?: "promptRequired"; + }; +}; + /** 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 @@ -215,23 +224,16 @@ const STEER_PRIORITY = "now" as const; export type SteerRequest = { sessionId: string; prompt: PromptRequest["prompt"]; + _meta?: SteerMeta | null; }; -/** 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; -}; +/** Result of a {@link STEER_METHOD} request. The legacy `startedNewTurn` result + * remains the default idle behavior; `promptRequired` is returned only when the + * Host explicitly opts into the host-owned fallback in request `_meta`. */ +export type SteerResponse = + | { outcome: "injected" } + | { outcome: "startedNewTurn" } + | { outcome: "promptRequired"; reason: "noRunningTurn" }; /** Validate raw JSON-RPC params into a {@link SteerRequest}. Kept minimal — the * content blocks are handed to `promptToClaude`, which tolerates unknown block @@ -240,16 +242,26 @@ 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; + const { sessionId, prompt, _meta } = 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"); } + const steering = + _meta && typeof _meta === "object" ? (_meta as Record).steering : undefined; + const idleBehavior = + steering && typeof steering === "object" + ? (steering as Record).idleBehavior + : undefined; + if (idleBehavior !== undefined && idleBehavior !== "promptRequired") { + throw RequestError.invalidParams(undefined, "unsupported steering idleBehavior"); + } return { sessionId, prompt: prompt as PromptRequest["prompt"], + _meta: _meta as SteerMeta | null | undefined, }; } @@ -1477,10 +1489,9 @@ 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`. + // Top-level `_meta` (sibling of `agentCapabilities`), per the existing ACP + // steering extension contract: advertises the `_session/steering` request + // so clients know they may inject a follow-up into a running turn. _meta: { steering: { supported: true, @@ -1783,11 +1794,10 @@ 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}). + /** Steer the session per the ACP steering wire protocol: inject a follow-up + * message into the turn that is currently running. If that turn already + * settled, the established default starts a new detached turn; Hosts may opt + * into the host-owned `promptRequired` fallback through request `_meta`. * * When a turn is in flight this injects (returns `injected`): unlike * `prompt()`, it does NOT create a Turn or enqueue on `turnQueue`; it pushes @@ -1800,16 +1810,13 @@ export class ClaudeAcpAgent { * 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. */ + * When the session is idle, the opt-in path returns `promptRequired` WITHOUT + * calling `prompt()`, pushing SDK input, or mutating `turnQueue`: the content + * stays Host-owned so the Host can submit it through a standard + * `session/prompt`. Without the opt-in, the existing detached `prompt()` and + * `startedNewTurn` result are preserved for compatibility. */ 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"); @@ -1817,28 +1824,36 @@ export class ClaudeAcpAgent { 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, - }; - + // which is exactly the window in which steering is meaningful. This check + // and the active-path push below stay in one synchronous section so the + // turn cannot settle in the gap between deciding to inject and enqueueing. + const turnInFlight = (session.turnQueue ?? []).some((turn) => !turn.settled); 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. + const promptRequest: PromptRequest = { + sessionId, + prompt: params.prompt, + }; + if (params._meta?.steering?.idleBehavior === "promptRequired") { + // The opt-in path leaves the content untouched so the Host can retry via + // a normal session/prompt whose lifecycle owns the continuation result. + return { outcome: "promptRequired", reason: "noRunningTurn" }; + } + + // Preserve the established default for Hosts that do not opt in. This is + // intentionally detached for compatibility with the existing contract. this.prompt(promptRequest).catch((error) => { this.logger.error(`Session ${sessionId}: steered new turn failed: ${error}`); }); return { outcome: "startedNewTurn" }; } + const promptRequest: PromptRequest = { + sessionId, + prompt: params.prompt, + }; const userMessage = promptToClaude(promptRequest); userMessage.uuid = randomUUID(); // Deliver into the running turn rather than queuing behind it as a fresh diff --git a/src/tests/acp-agent.test.ts b/src/tests/acp-agent.test.ts index 95090e3d..42c82c40 100644 --- a/src/tests/acp-agent.test.ts +++ b/src/tests/acp-agent.test.ts @@ -43,6 +43,7 @@ import { runPromptWithCancellation, type AcpClient, type SDKMessageFilter, + type SteerRequest, type StreamedToolInputCache, } from "../acp-agent.js"; import { Pushable } from "../utils.js"; @@ -9691,6 +9692,29 @@ describe("turn steering (_session/steering)", () => { }; } + // A minimal SDK assistant message carrying a single text block. Used by the + // promptRequired retry test to model the continuation turn's streamed reply. + function createAssistantText(text: string) { + return { + type: "assistant" as const, + parent_tool_use_id: null, + uuid: randomUUID(), + session_id: "test-session", + message: { + role: "assistant", + model: "claude-sonnet-4-5", + stop_reason: "end_turn", + usage: { + input_tokens: 1, + output_tokens: 1, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + }, + content: [{ type: "text", text }], + }, + }; + } + const waitFor = async (cond: () => boolean) => { for (let i = 0; i < 200; i++) { if (cond()) return; @@ -9726,30 +9750,85 @@ describe("turn steering (_session/steering)", () => { ).rejects.toThrow(); }); - it("advertises steering support at _meta.steering.supported", async () => { + it("preserves the existing steering capability declaration", 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 }); + // Top-level _meta (sibling of agentCapabilities), per the existing steering + // extension contract. Idle behavior is selected per steering request. + expect((response._meta as any)?.steering).toEqual({ + supported: true, + }); }); - it("starts a new turn (outcome 'startedNewTurn') when no turn is in flight (race)", async () => { + it("preserves startedNewTurn by default when no turn is in flight", async () => { const agent = createMockAgent(); + const prompt = vi.spyOn(agent, "prompt").mockResolvedValue({ stopReason: "end_turn" }); + agent.sessions["test-session"] = mockSessionState({ + input: new Pushable(), + turnQueue: [], + }); + + const request: SteerRequest = { + sessionId: "test-session", + prompt: [{ type: "text", text: "late follow-up" }], + }; + + await expect(agent.steer(request)).resolves.toEqual({ outcome: "startedNewTurn" }); + expect(prompt).toHaveBeenCalledWith(request); + }); + + it("returns promptRequired without consuming an opted-in prompt", async () => { + const agent = createMockAgent(); + const input = new Pushable(); + const inputPush = vi.spyOn(input, "push"); + const prompt = vi.spyOn(agent, "prompt").mockResolvedValue({ stopReason: "end_turn" }); + // Idle session: turnQueue is empty, so the turn we meant to steer has (from + // the agent's view) already finished. The content must stay Host-owned. + agent.sessions["test-session"] = mockSessionState({ + input, + turnQueue: [], + }); + + const response = await agent.steer({ + sessionId: "test-session", + prompt: [{ type: "text", text: "late follow-up" }], + _meta: { steering: { idleBehavior: "promptRequired" } }, + }); + + expect(response).toEqual({ + outcome: "promptRequired", + reason: "noRunningTurn", + }); + // The idle branch must not start a detached turn, push SDK input, or mutate + // the queue — the Host retries the exact same content via session/prompt. + expect(prompt).not.toHaveBeenCalled(); + expect(inputPush).not.toHaveBeenCalled(); + expect(agent.sessions["test-session"].turnQueue).toEqual([]); + }); + + it("lets the host retry promptRequired content through session/prompt exactly once", async () => { + const updates: string[] = []; + const client = { + sessionUpdate: async (notification: any) => { + if (notification.update?.sessionUpdate === "agent_message_chunk") { + updates.push(notification.update.content?.text); + } + }, + } as unknown as AcpClient; + const agent = new ClaudeAcpAgent(client, { log: () => {}, error: () => {} }); 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 + const continuation = await iter.next(); + captured.push(continuation.value); + yield userEcho(continuation.value); + yield createAssistantText("continuation response"); yield createResultMessage(); yield { type: "system", subtype: "session_state_changed", state: "idle" }; } @@ -9757,19 +9836,31 @@ describe("turn steering (_session/steering)", () => { }, { turnQueue: [] }, ); - - const res = await agent.steer({ + const request: SteerRequest = { sessionId: "test-session", prompt: [{ type: "text", text: "late follow-up" }], - }); - expect(res.outcome).toBe("startedNewTurn"); + _meta: { steering: { idleBehavior: "promptRequired" } }, + }; - // A real turn was enqueued and drains like any normal prompt. - await waitFor(() => (agent.sessions["test-session"].turnQueue ?? []).length === 0); + // Steering an idle session leaves the content unconsumed... + await expect(agent.steer(request)).resolves.toEqual({ + outcome: "promptRequired", + reason: "noRunningTurn", + }); + await Promise.resolve(); + expect(captured).toHaveLength(0); + + // ...so the Host can submit the same content through a normal session/prompt, + // which owns the continuation's updates and terminal response. + await expect(agent.prompt(request)).resolves.toEqual( + expect.objectContaining({ stopReason: "end_turn" }), + ); 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"); + expect(updates).toContain("continuation response"); + expect(agent.sessions["test-session"].turnQueue).toHaveLength(0); + await agent.sessions["test-session"]?.consumer; }); it("injects (outcome 'injected') a priority:'now' message into the running turn without spawning a new turn", async () => {