From bea9503c4268288268926190823f5ac927706aec Mon Sep 17 00:00:00 2001 From: "mark.tkachenko" Date: Mon, 13 Jul 2026 13:02:23 +0200 Subject: [PATCH 1/2] Steering implementation --- examples/steering.ts | 444 ++++++++++++++++++ examples/tsconfig.json | 13 + package-lock.json | 112 ++--- package.json | 6 +- src/AcpExtensions.ts | 27 +- src/CodexAcpClient.ts | 9 + src/CodexAcpServer.ts | 159 ++++++- src/CodexAppServerClient.ts | 6 + .../CodexACPAgent/initialize.test.ts | 5 + .../CodexACPAgent/steer-events.test.ts | 222 +++++++++ src/index.ts | 11 +- tsconfig.json | 2 +- 12 files changed, 957 insertions(+), 59 deletions(-) create mode 100644 examples/steering.ts create mode 100644 examples/tsconfig.json create mode 100644 src/__tests__/CodexACPAgent/steer-events.test.ts diff --git a/examples/steering.ts b/examples/steering.ts new file mode 100644 index 00000000..f9f34e46 --- /dev/null +++ b/examples/steering.ts @@ -0,0 +1,444 @@ +#!/usr/bin/env tsx + +/** + * Steering demo — mid-turn edition. + * + * The plain `steering.ts` example steers a single-message answer ("count to + * 30"). There the injected prompt can only take effect *after* that message is + * finished: a turn made of one model step has no earlier boundary for Codex to + * inject at, so `turn/steer` appends the message and the model reads it on its + * next step — which is the end. + * + * This example instead gives Codex a genuinely multi-step task: a "treasure + * hunt" where each clue file only reveals the *name of the next clue*. Because + * the reads are sequential and dependent, the agent cannot batch them or read + * ahead — the turn contains several model steps. A steering message injected + * part-way through is therefore picked up *between* steps and visibly changes + * what the agent does next (it stops the hunt early). + * + * Run it with: + * node --import tsx examples/steering.ts + * # or: npm run example:steering:multistep + * + * Auth: uses your existing Codex login in ~/.codex. If you instead export + * CODEX_API_KEY / OPENAI_API_KEY it will authenticate with that. + * + * Knobs (env): + * STEERING_EXAMPLE_MODEL model id (default gpt-5.6-sol) + * STEER_AFTER_TOOL_CALLS inject the steer after N clue reads (default 2) + * NO_COLOR disable ANSI colors + */ + +import * as acp from "@agentclientprotocol/sdk"; +import {type ChildProcess, spawn} from "node:child_process"; +import {fileURLToPath} from "node:url"; +import {mkdtemp, rm, writeFile} from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import {Readable, Writable} from "node:stream"; + +const STEERING_METHOD = "_session/steering"; +const EXAMPLE_TIMEOUT_MS = 60_000; +const DEFAULT_EXAMPLE_MODEL = "gpt-5.6-sol"; +const exampleModel = process.env["STEERING_EXAMPLE_MODEL"] ?? DEFAULT_EXAMPLE_MODEL; + +const parsedSteerAfter = Number(process.env["STEER_AFTER_TOOL_CALLS"] ?? "2"); +const STEER_AFTER_TOOL_CALLS = Number.isFinite(parsedSteerAfter) && parsedSteerAfter > 0 + ? Math.floor(parsedSteerAfter) + : 2; + +const repositoryRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), ".."); + +// One note per clue file. The hunt is CLUE_NOTES.length steps long. +const CLUE_NOTES = ["compass", "lantern", "map", "brass key", "torch", "rope", "chart", "chest"]; + +const initialPrompt = [ + "You're solving a treasure hunt inside this folder.", + "Start by reading the file `clue-1.txt`.", + "Each clue holds one note to remember and tells you the exact filename of the next clue.", + "Follow the trail, reading EXACTLY ONE clue per step — use a separate read for each file,", + "and do NOT list the folder or read several files at once.", + "Keep going until a clue tells you to STOP, then reply with the full ordered list of notes you collected.", +].join(" "); + +const steeringPrompt = [ + "Change of plan — stop the treasure hunt immediately.", + "Do not open any more clues.", + "Just tell me the notes you've collected so far and which clue number you stopped on.", +].join(" "); + +type SteeringRequest = { + sessionId: acp.SessionId; + prompt: acp.ContentBlock[]; +}; + +type SteeringResponse = { + outcome: "injected" | "startedNewTurn"; +}; + +type ThreadStatusType = "active" | "idle" | "systemError"; +type StateListener = () => void; + +let trackedSessionId: acp.SessionId | null = null; +const toolCallsSeen = new Set(); +let finishedTransitions = 0; +let lastChannel: string | null = null; +const stateListeners = new Set(); + +// --------------------------------------------------------------------------- +// Tiny ANSI helpers (no dependencies). Honors NO_COLOR and non-TTY output. +// --------------------------------------------------------------------------- +const useColor = Boolean(process.stdout.isTTY) && !process.env["NO_COLOR"]; +const paint = (code: string) => (text: string): string => (useColor ? `\x1b[${code}m${text}\x1b[0m` : text); +const c = { + bold: paint("1"), + dim: paint("2"), + red: paint("31"), + green: paint("32"), + yellow: paint("33"), + magenta: paint("35"), + cyan: paint("36"), +}; + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function createAgentEnvironment(): NodeJS.ProcessEnv { + const configString = process.env["CODEX_CONFIG"]; + let config: Record = {}; + if (configString) { + const parsedConfig: unknown = JSON.parse(configString); + if (!isRecord(parsedConfig)) { + throw new Error("CODEX_CONFIG must contain a JSON object"); + } + config = parsedConfig; + } + return { + ...process.env, + CODEX_CONFIG: JSON.stringify({ + ...config, + model: exampleModel, + }), + }; +} + +function supportsSteering(response: acp.InitializeResponse): boolean { + const steering = response._meta?.["steering"]; + return isRecord(steering) && steering["supported"] === true; +} + +function readThreadStatus(update: acp.SessionUpdate): ThreadStatusType | undefined { + if (update.sessionUpdate !== "session_info_update") { + return undefined; + } + const codex = update._meta?.["codex"]; + if (!isRecord(codex)) { + return undefined; + } + const threadStatus = codex["threadStatus"]; + if (!isRecord(threadStatus)) { + return undefined; + } + const type = threadStatus["type"]; + return type === "active" || type === "idle" || type === "systemError" ? type : undefined; +} + +function notifyStateListeners(): void { + for (const listener of stateListeners) { + listener(); + } +} + +// --------------------------------------------------------------------------- +// Streaming output: group consecutive chunks of the same kind under a header +// so thinking / agent text / tool calls stay visually separated. +// --------------------------------------------------------------------------- +function writeChannel(channel: string, label: string, text: string): void { + if (lastChannel !== channel) { + process.stdout.write(`\n${label}\n`); + lastChannel = channel; + } + process.stdout.write(text); +} + +function writeEvent(line: string): void { + process.stdout.write(`\n${line}\n`); + lastChannel = null; +} + +function recordSessionUpdate(params: acp.SessionNotification): void { + if (params.sessionId !== trackedSessionId) { + return; + } + + const update = params.update; + switch (update.sessionUpdate) { + case "agent_message_chunk": + if (update.content.type === "text") { + writeChannel("message", c.bold(c.cyan("🤖 agent")), update.content.text); + } + break; + case "agent_thought_chunk": + if (update.content.type === "text") { + writeChannel("thought", c.dim("💭 thinking"), c.dim(update.content.text)); + } + break; + case "tool_call": { + const isNew = !toolCallsSeen.has(update.toolCallId); + toolCallsSeen.add(update.toolCallId); + writeEvent(c.yellow(`🔧 tool call #${toolCallsSeen.size}: ${update.title} [${update.status}]`)); + if (isNew) { + notifyStateListeners(); + } + break; + } + case "tool_call_update": + if (update.status) { + writeEvent(c.dim(` ↳ ${update.toolCallId} [${update.status}]`)); + } + break; + } + + const threadStatus = readThreadStatus(update); + if (threadStatus === "idle" || threadStatus === "systemError") { + finishedTransitions += 1; + notifyStateListeners(); + } +} + +async function waitForState( + predicate: () => boolean, + description: string, + timeoutMs = EXAMPLE_TIMEOUT_MS, +): Promise { + if (predicate()) { + return; + } + + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + stateListeners.delete(checkState); + reject(new Error(`Timed out waiting for ${description}`)); + }, timeoutMs); + const checkState = (): void => { + if (!predicate()) { + return; + } + clearTimeout(timeout); + stateListeners.delete(checkState); + resolve(); + }; + stateListeners.add(checkState); + }); +} + +async function createTreasureHunt(): Promise<{workspaceDir: string; clueCount: number}> { + const workspaceDir = await mkdtemp(path.join(os.tmpdir(), "codex-steering-")); + const clueCount = CLUE_NOTES.length; + for (let index = 0; index < clueCount; index += 1) { + const step = index + 1; + const note = CLUE_NOTES[index]; + const isLast = step === clueCount; + const nextInstruction = isLast + ? "This is the final clue. STOP here — do not open any more files. Report every note you collected, in order." + : `When ready, read the next clue in the file named "clue-${step + 1}.txt".`; + const body = + `Treasure hunt — clue ${step} of ${clueCount}\n\n` + + `Note to remember #${step}: ${note}\n\n` + + `${nextInstruction}\n`; + await writeFile(path.join(workspaceDir, `clue-${step}.txt`), body, "utf8"); + } + return {workspaceDir, clueCount}; +} + +function printHeader(workspaceDir: string, clueCount: number): void { + const line = "─".repeat(66); + console.log(c.bold(`\n${line}`)); + console.log(c.bold(" Codex ACP — mid-turn steering demo")); + console.log(line); + console.log(` model : ${c.cyan(exampleModel)}`); + console.log(` workspace : ${c.dim(workspaceDir)}`); + console.log(` clue files : ${clueCount} (clue-1.txt … clue-${clueCount}.txt)`); + console.log(` steer after : ${STEER_AFTER_TOOL_CALLS} tool call(s)`); + console.log(line); + console.log(c.dim(" Task: follow the treasure-hunt chain, one clue at a time.")); + console.log(c.dim(" Mid-turn we inject a steering message telling it to stop early.")); + console.log(`${line}\n`); +} + +function printBanner(text: string): void { + const line = "═".repeat(66); + process.stdout.write(`\n${c.magenta(line)}\n${c.magenta(c.bold(` ${text}`))}\n${c.magenta(line)}\n`); + lastChannel = null; +} + +function printSummary(clueCount: number, cluesAtSteer: number, stopReason: string, steered: boolean): void { + const line = "─".repeat(66); + const stoppedEarly = toolCallsSeen.size < clueCount; + console.log(`\n\n${c.bold(line)}`); + console.log(c.bold(" Summary")); + console.log(line); + console.log(` tool calls total : ${toolCallsSeen.size} of up to ${clueCount} clues`); + console.log(` steered after : ${steered ? `${cluesAtSteer} clue(s)` : "not steered"}`); + console.log(` stop reason : ${stopReason}`); + console.log(line); + if (!steered) { + console.log(c.yellow(" • The turn finished before we could steer. Lower STEER_AFTER_TOOL_CALLS")); + console.log(c.yellow(" or use a slower model to catch the turn while it is still running.")); + } else if (stoppedEarly) { + console.log(c.green(" ✔ The agent stopped BEFORE reading every clue — the steering message")); + console.log(c.green(" was picked up mid-turn and changed its course.")); + } else { + console.log(c.yellow(" • The agent read every clue. Steering still applied, but the turn was")); + console.log(c.yellow(" short — try a longer chain (add CLUE_NOTES) or steer earlier.")); + } + console.log(`${line}\n`); +} + +async function stopAgent(agentProcess: ChildProcess): Promise { + if (agentProcess.stdin && !agentProcess.stdin.destroyed && !agentProcess.stdin.writableEnded) { + agentProcess.stdin.end(); + } + if (agentProcess.exitCode !== null || agentProcess.signalCode !== null) { + return; + } + + await new Promise((resolve) => { + const timeout = setTimeout(resolve, 2_000); + agentProcess.once("exit", () => { + clearTimeout(timeout); + resolve(); + }); + }); + if (agentProcess.exitCode === null && agentProcess.signalCode === null) { + agentProcess.kill(); + } +} + +async function main(): Promise { + const {workspaceDir, clueCount} = await createTreasureHunt(); + const npmCommand = process.platform === "win32" ? "npm.cmd" : "npm"; + const agentProcess = spawn(npmCommand, ["run", "--silent", "start"], { + cwd: repositoryRoot, + env: createAgentEnvironment(), + stdio: ["pipe", "pipe", "inherit"], + }); + if (!agentProcess.stdin || !agentProcess.stdout) { + throw new Error("Failed to open stdio pipes for the ACP agent"); + } + const stream = acp.ndJsonStream( + Writable.toWeb(agentProcess.stdin), + Readable.toWeb(agentProcess.stdout) as ReadableStream, + ); + + try { + await acp.client({name: "steering-multistep-example"}) + .onRequest(acp.methods.client.session.requestPermission, (ctx) => { + // A real client would prompt the user here. To keep the demo + // hands-free we auto-approve each read once. + const {toolCall, options} = ctx.params; + const allow = options.find((option) => option.kind === "allow_once") ?? options[0]; + if (!allow) { + return {outcome: {outcome: "cancelled"}}; + } + writeEvent(c.green(` ✔ auto-approving: ${toolCall.title ?? toolCall.toolCallId} → "${allow.name}"`)); + return {outcome: {outcome: "selected", optionId: allow.optionId}}; + }) + .onNotification(acp.methods.client.session.update, (ctx) => { + recordSessionUpdate(ctx.params); + }) + .connectWith(stream, async (agent) => { + const initializeResponse = await agent.request(acp.methods.agent.initialize, { + protocolVersion: acp.PROTOCOL_VERSION, + clientInfo: { + name: "steering-multistep-example", + version: "1.0.0", + }, + }); + if (!supportsSteering(initializeResponse)) { + throw new Error("The agent did not advertise steering support"); + } + + const apiKey = process.env["CODEX_API_KEY"] ?? process.env["OPENAI_API_KEY"]; + if (apiKey && initializeResponse.authMethods?.some((method) => method.id === "api-key")) { + await agent.request(acp.methods.agent.authenticate, { + methodId: "api-key", + _meta: { + "api-key": {apiKey}, + }, + }); + } + + const session = await agent.request(acp.methods.agent.session.new, { + cwd: workspaceDir, + mcpServers: [], + }); + trackedSessionId = session.sessionId; + + printHeader(workspaceDir, clueCount); + process.stdout.write(c.dim(`📤 prompt → ${initialPrompt}\n`)); + + let promptDone = false; + const promptPromise = agent.request(acp.methods.agent.session.prompt, { + sessionId: trackedSessionId, + prompt: [{type: "text", text: initialPrompt}], + }).finally(() => { + promptDone = true; + notifyStateListeners(); + }); + promptPromise.catch(() => {}); + + // Let the agent work through a couple of clues, then steer mid-turn. + await Promise.race([ + waitForState( + () => toolCallsSeen.size >= STEER_AFTER_TOOL_CALLS || promptDone, + `the agent to open ${STEER_AFTER_TOOL_CALLS} clue(s)`, + ).catch(() => {}), + promptPromise.then(() => undefined, () => undefined), + ]); + + const cluesAtSteer = toolCallsSeen.size; + const turnAlreadyFinished = promptDone || finishedTransitions > 0; + let steered = false; + + if (turnAlreadyFinished) { + writeEvent(c.red("⚠ The turn finished before we could steer — skipping the steering step.")); + } else { + steered = true; + printBanner(`Injecting steering message after ${cluesAtSteer} clue(s)`); + process.stdout.write(`${c.magenta(`✋ steer → ${steeringPrompt}`)}\n`); + lastChannel = null; + + const steeringResponse = await agent.request(STEERING_METHOD, { + sessionId: trackedSessionId, + prompt: [{type: "text", text: steeringPrompt}], + }); + if (steeringResponse.outcome !== "injected" && steeringResponse.outcome !== "startedNewTurn") { + throw new Error(`Unexpected steering response: ${JSON.stringify(steeringResponse)}`); + } + writeEvent(c.magenta(c.bold(` outcome: ${steeringResponse.outcome}`))); + if (steeringResponse.outcome === "injected") { + writeEvent(c.dim(" → injected into the running turn; the agent picks it up at its next step.")); + } else { + writeEvent(c.dim(" → the turn had already ended, so this started a fresh turn.")); + } + } + + const promptResponse = await promptPromise; + printSummary(clueCount, cluesAtSteer, promptResponse.stopReason, steered); + + await agent.request(acp.methods.agent.session.close, { + sessionId: trackedSessionId, + }); + }); + } finally { + await stopAgent(agentProcess); + await rm(workspaceDir, {recursive: true, force: true}); + } +} + +main().catch((error: unknown) => { + console.error("Steering multistep example failed:", error); + process.exitCode = 1; +}); diff --git a/examples/tsconfig.json b/examples/tsconfig.json new file mode 100644 index 00000000..b71b3eec --- /dev/null +++ b/examples/tsconfig.json @@ -0,0 +1,13 @@ +{ + "extends": "../tsconfig.json", + "compilerOptions": { + "rootDir": ".", + "declaration": false, + "declarationMap": false, + "sourceMap": false + }, + "include": [ + "steering.ts" + ], + "exclude": [] +} diff --git a/package-lock.json b/package-lock.json index bf8967a7..a9ba4301 100644 --- a/package-lock.json +++ b/package-lock.json @@ -10,7 +10,7 @@ "license": "Apache-2.0", "dependencies": { "@agentclientprotocol/sdk": "^1.2.1", - "@openai/codex": "^0.144.4", + "@openai/codex": "^0.144.5", "diff": "^9.0.0", "open": "^11.0.0", "vscode-jsonrpc": "^9.0.1", @@ -37,29 +37,6 @@ "zod": "^3.25.0 || ^4.0.0" } }, - "node_modules/@emnapi/core": { - "version": "1.11.1", - "resolved": "https://registry.npmjs.org/@emnapi/core/-/core-1.11.1.tgz", - "integrity": "sha512-RSvbQmHzdKzNsLYa/wHrbc3KN4sYLKAdPZxqiM2HATqv/SBk2/ENSHpvXGaLOMcsAyz0poEGqkmmKYG3OWiJEQ==", - "dev": true, - "license": "MIT", - "optional": true, - "dependencies": { - "@emnapi/wasi-threads": "1.2.2", - "tslib": "^2.4.0" - } - }, - "node_modules/@emnapi/runtime": { - "version": "1.11.1", - "resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.11.1.tgz", - "integrity": "sha512-vgj7R3y3Wgx24IQaGPA/R6YFXLHVMOZ0uVEyIQPaWs+rd1AzfEMXlAC22FYwO1XkKR6NPsq7mUandH8oIRdZFw==", - "dev": true, - "license": "MIT", - "optional": true, - "dependencies": { - "tslib": "^2.4.0" - } - }, "node_modules/@emnapi/wasi-threads": { "version": "1.2.2", "resolved": "https://registry.npmjs.org/@emnapi/wasi-threads/-/wasi-threads-1.2.2.tgz", @@ -929,9 +906,9 @@ } }, "node_modules/@openai/codex": { - "version": "0.144.4", - "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.4.tgz", - "integrity": "sha512-DTHzYatlKq9dw55E0/HsbK4tRCEKabuJ10ybbqpsG8gVv/kvwEdg3Z4OI3cvLXKa21xkIa4lkGlZoO/HmqmFFw==", + "version": "0.144.5", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.5.tgz", + "integrity": "sha512-jjB+K+OMv572mKhS+2QuLxWXDJNdpwbPenf+V+8bdq7wg4Scqt3cn6WEekD8wPqDVZqck0HSX17K9rD9kbDJQA==", "license": "Apache-2.0", "bin": { "codex": "bin/codex.js" @@ -940,19 +917,19 @@ "node": ">=16" }, "optionalDependencies": { - "@openai/codex-darwin-arm64": "npm:@openai/codex@0.144.4-darwin-arm64", - "@openai/codex-darwin-x64": "npm:@openai/codex@0.144.4-darwin-x64", - "@openai/codex-linux-arm64": "npm:@openai/codex@0.144.4-linux-arm64", - "@openai/codex-linux-x64": "npm:@openai/codex@0.144.4-linux-x64", - "@openai/codex-win32-arm64": "npm:@openai/codex@0.144.4-win32-arm64", - "@openai/codex-win32-x64": "npm:@openai/codex@0.144.4-win32-x64" + "@openai/codex-darwin-arm64": "npm:@openai/codex@0.144.5-darwin-arm64", + "@openai/codex-darwin-x64": "npm:@openai/codex@0.144.5-darwin-x64", + "@openai/codex-linux-arm64": "npm:@openai/codex@0.144.5-linux-arm64", + "@openai/codex-linux-x64": "npm:@openai/codex@0.144.5-linux-x64", + "@openai/codex-win32-arm64": "npm:@openai/codex@0.144.5-win32-arm64", + "@openai/codex-win32-x64": "npm:@openai/codex@0.144.5-win32-x64" } }, "node_modules/@openai/codex-darwin-arm64": { "name": "@openai/codex", - "version": "0.144.4-darwin-arm64", - "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.4-darwin-arm64.tgz", - "integrity": "sha512-6J3g498cM2oA7vYIJhpuGJlnIi/M5JdYmjB5BZ1Of5HQ0ziIlplFSvH801oVy9J5TQFp642ODzOu/ZEokDUXsg==", + "version": "0.144.5-darwin-arm64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.5-darwin-arm64.tgz", + "integrity": "sha512-zcT6NfBCqLFt+BReNSETTZW6v6PdbH0dzNtm9j7l7mDGqwPbKZDGJdnpkBao2389I0ZacyIKgSZoI0vez1d4Dw==", "cpu": [ "arm64" ], @@ -967,9 +944,9 @@ }, "node_modules/@openai/codex-darwin-x64": { "name": "@openai/codex", - "version": "0.144.4-darwin-x64", - "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.4-darwin-x64.tgz", - "integrity": "sha512-k1HC8gdbAy+VmMbekYkhM+r+QE2Xfgd67n1VSp94tjz7aXVKoalHcDkdKNM/uUQ8o2tvbiwhHSUftJF8Sm9/Lw==", + "version": "0.144.5-darwin-x64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.5-darwin-x64.tgz", + "integrity": "sha512-//Mo0m1MwaoT6psu5xsmofXpKx4/0irIkeq10xJvk59+886EG355ibjA+ZmlRcKhE3bLjsKD7p81nTbAdRL/bw==", "cpu": [ "x64" ], @@ -984,9 +961,9 @@ }, "node_modules/@openai/codex-linux-arm64": { "name": "@openai/codex", - "version": "0.144.4-linux-arm64", - "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.4-linux-arm64.tgz", - "integrity": "sha512-OlKx65579OwIzech9Tt3OUH9+hFZfFrCBP1hL2MudnMIoNr1+cFZjB5YIj5MWMRoBD+K5W3wdBIpQSH855b5Sg==", + "version": "0.144.5-linux-arm64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.5-linux-arm64.tgz", + "integrity": "sha512-zAHggxVwR2TBxKmybXY7ZMiB0G8DMonY2YPdwNNjwXcf+LOIqNGgswwNCDMbP/HEe6r8j+R9ZX/yYoo8f+n/RQ==", "cpu": [ "arm64" ], @@ -1001,9 +978,9 @@ }, "node_modules/@openai/codex-linux-x64": { "name": "@openai/codex", - "version": "0.144.4-linux-x64", - "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.4-linux-x64.tgz", - "integrity": "sha512-2jxrmV6+/7eBNdg5uhhmOEPFu2o28eYY/ClLzWhSBHH8uo3f2KA1z9JQcVtwlbToW03nEPlEzYNYfCF1UBqsVQ==", + "version": "0.144.5-linux-x64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.5-linux-x64.tgz", + "integrity": "sha512-FalLJlBQGFdK8Gc3kj9sa/ekNdgkHhUawLaKkvy5CtB18JaP2YxtTP/Pe1pD2iBiq8mMUliRnafpF6AdBdQMbg==", "cpu": [ "x64" ], @@ -1018,9 +995,9 @@ }, "node_modules/@openai/codex-win32-arm64": { "name": "@openai/codex", - "version": "0.144.4-win32-arm64", - "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.4-win32-arm64.tgz", - "integrity": "sha512-CCgfI1smFhHZTIpTuBwDJwBr/AR40RTqaFxbBWVabu0RMeYDteRuPiDfdTlktf3C43Y1q10VZXhVGYtCokDg2w==", + "version": "0.144.5-win32-arm64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.5-win32-arm64.tgz", + "integrity": "sha512-0Pj7iqjEOEvPQPO3kFfCy9vGX4BTu76ChFFZHr2eNNIfVc3FOENAv/X98u4L+iIUtDOK9DbqmfUudW3DPapshg==", "cpu": [ "arm64" ], @@ -1035,9 +1012,9 @@ }, "node_modules/@openai/codex-win32-x64": { "name": "@openai/codex", - "version": "0.144.4-win32-x64", - "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.4-win32-x64.tgz", - "integrity": "sha512-iL1ky0ERgdQJOKzom/Ms1fhpwkSmpsA9eVrzAqURFlYGS8z7JqwEgm33+nLGCsY7y25d8Xs/LJ91Oiqz3yXcUg==", + "version": "0.144.5-win32-x64", + "resolved": "https://registry.npmjs.org/@openai/codex/-/codex-0.144.5-win32-x64.tgz", + "integrity": "sha512-DnsSTlnnzleTxvLwIGnBitKInscxn2I7qASqosS8Fv+qysBygd+ZiBn/SQsRCgQ28PAlsNzmd3Gf3ZTecolAmg==", "cpu": [ "x64" ], @@ -1283,6 +1260,29 @@ "node": "^20.19.0 || >=22.12.0" } }, + "node_modules/@rolldown/binding-wasm32-wasi/node_modules/@emnapi/core": { + "version": "1.11.1", + "resolved": "https://registry.npmjs.org/@emnapi/core/-/core-1.11.1.tgz", + "integrity": "sha512-RSvbQmHzdKzNsLYa/wHrbc3KN4sYLKAdPZxqiM2HATqv/SBk2/ENSHpvXGaLOMcsAyz0poEGqkmmKYG3OWiJEQ==", + "dev": true, + "license": "MIT", + "optional": true, + "dependencies": { + "@emnapi/wasi-threads": "1.2.2", + "tslib": "^2.4.0" + } + }, + "node_modules/@rolldown/binding-wasm32-wasi/node_modules/@emnapi/runtime": { + "version": "1.11.1", + "resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.11.1.tgz", + "integrity": "sha512-vgj7R3y3Wgx24IQaGPA/R6YFXLHVMOZ0uVEyIQPaWs+rd1AzfEMXlAC22FYwO1XkKR6NPsq7mUandH8oIRdZFw==", + "dev": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, "node_modules/@rolldown/binding-win32-arm64-msvc": { "version": "1.1.5", "resolved": "https://registry.npmjs.org/@rolldown/binding-win32-arm64-msvc/-/binding-win32-arm64-msvc-1.1.5.tgz", @@ -1373,6 +1373,7 @@ "integrity": "sha512-nxAkRSVkN1Y0JC1W8ky/fTfkGsMmcrRsbx+3XoZE+rMOX71kLYTV7fLXpqud1GpbpP5TuffXFqfX7fH2GgZREw==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "undici-types": "~8.3.0" } @@ -2013,6 +2014,7 @@ "integrity": "sha512-IuL+Elrou2ZvCFHs18/CIzy2Nzvo25nZ1/D2eIZlz7c+QUayAcYoiM2BthCjs+EBHVpjYjcuLDAiCWgeIX3X1Q==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "accepts": "~1.3.8", "array-flatten": "1.1.1", @@ -2263,6 +2265,7 @@ "integrity": "sha512-YwUvVpSF7m1yOblFPrU3Hbo8XhPheBoiyfGuII6z19LnOr6JpDnyyp7LFNrfV56wS8tpvtBFGRISHN02pDdLOA==", "dev": true, "license": "MIT", + "peer": true, "engines": { "node": ">=16.9.0" } @@ -2965,6 +2968,7 @@ "integrity": "sha512-RvwwcruNjI1ncT5xRakeyS9Lf8lcItv34KD+aif+VH9kduAyfYBipGh12274xtenIPZ119/R9BdTBa8gAwSh0A==", "dev": true, "license": "MIT", + "peer": true, "engines": { "node": ">=12" }, @@ -3264,8 +3268,7 @@ "resolved": "https://registry.npmjs.org/scheduler/-/scheduler-0.27.0.tgz", "integrity": "sha512-eNv+WrVbKu1f3vbYJT/xtiF5syA5HPIMtf9IgY/nKg0sWqzAUEvqY/xm7OcZc/qafLx/iO9FgOmeSAp4v5ti/Q==", "dev": true, - "license": "MIT", - "peer": true + "license": "MIT" }, "node_modules/send": { "version": "0.19.2", @@ -3530,6 +3533,7 @@ "integrity": "sha512-eUdUIaCr963q2h5u3+QwvYp0+eqPvn+egeqZUm0hwERCqqx1E3kK5ehbGCvqSE5MQAULr67ww0cA3jKc3YkM1w==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "esbuild": "~0.28.0" }, @@ -3614,6 +3618,7 @@ "integrity": "sha512-bTT9PsdWO+MQMNG9ZXIP/qM9wGh37DFxTV/sPq9cFpHr3w4jkgef032PkAL9jAqhk3Nz8NQw3O8n6/xFkqO4QQ==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "lightningcss": "^1.32.0", "picomatch": "^4.0.5", @@ -3846,6 +3851,7 @@ "resolved": "https://registry.npmjs.org/zod/-/zod-4.4.3.tgz", "integrity": "sha512-ytENFjIJFl2UwYglde2jchW2Hwm4GJFLDiSXWdTrJQBIN9Fcyp7n4DhxJEiWNAJMV1/BqWfW/kkg71UDcHJyTQ==", "license": "MIT", + "peer": true, "funding": { "url": "https://github.com/sponsors/colinhacks" } diff --git a/package.json b/package.json index 0c107d36..0c16f19b 100644 --- a/package.json +++ b/package.json @@ -33,11 +33,13 @@ "package:win-x64": "cd dist/bin && zip codex-acp-x64-windows.zip codex-acp-x64-windows.exe", "package:win-arm64": "cd dist/bin && zip codex-acp-arm64-windows.zip codex-acp-arm64-windows.exe", "start": "node --import tsx src/index.ts", + "example:steering": "node --import tsx examples/steering.ts", + "example:steering:multistep": "node --import tsx examples/steering.ts", "generate-types": "./node_modules/.bin/codex app-server generate-ts --out src/app-server", "test": "vitest run", "test:e2e": "npm run build && RUN_E2E_TESTS=true vitest run src/__tests__/CodexACPAgent/e2e", "test:watch": "vitest", - "typecheck": "tsc --noEmit", + "typecheck": "tsc --noEmit && tsc --noEmit -p examples/tsconfig.json", "codex-test": "tsx .claude/skills/run-codex/scripts/run-codex-test.ts" }, "homepage": "https://github.com/agentclientprotocol/codex-acp#readme", @@ -62,7 +64,7 @@ }, "dependencies": { "@agentclientprotocol/sdk": "^1.2.1", - "@openai/codex": "^0.144.4", + "@openai/codex": "^0.144.5", "diff": "^9.0.0", "open": "^11.0.0", "vscode-jsonrpc": "^9.0.1", diff --git a/src/AcpExtensions.ts b/src/AcpExtensions.ts index 5a10ea5e..a0b7afa3 100644 --- a/src/AcpExtensions.ts +++ b/src/AcpExtensions.ts @@ -1,5 +1,6 @@ import type { ClientContext, + ContentBlock, LoadSessionResponse, NewSessionResponse, ResumeSessionResponse, @@ -7,6 +8,7 @@ import type { } from "@agentclientprotocol/sdk"; export const LEGACY_SET_SESSION_MODEL_METHOD = "session/set_model"; +export const SESSION_STEERING_METHOD = "_session/steering"; export const GOAL_CONTROL_METHOD = "_codex/session/goal_control"; export type LegacySessionModel = { @@ -43,13 +45,15 @@ export type ExtMethodRequest = AuthenticationStatusRequest | AuthenticationLogoutRequest | LegacySetSessionModelExtRequest + | SessionSteeringExtRequest | GoalControlExtRequest export function isExtMethodRequest(request: { method: string, params: Record }): request is ExtMethodRequest { return request.method === "authentication/status" || request.method === "authentication/logout" || request.method === LEGACY_SET_SESSION_MODEL_METHOD - || request.method === GOAL_CONTROL_METHOD; + || request.method === GOAL_CONTROL_METHOD + || request.method === SESSION_STEERING_METHOD; } export type AuthenticationStatusRequest = { method: "authentication/status", params: {} } @@ -79,3 +83,24 @@ export async function legacySetSessionModel( ): Promise { return await connection.request(LEGACY_SET_SESSION_MODEL_METHOD, params); } + +export type SessionSteerRequest = { + sessionId: SessionId; + prompt: ContentBlock[]; +} + +export type SessionSteeringResponse = { + outcome: "injected" | "startedNewTurn"; +} + +export type SessionSteeringExtRequest = { + method: typeof SESSION_STEERING_METHOD; + params: SessionSteerRequest; +} + +export async function steerSessionWithFallback( + connection: Pick, + params: SessionSteerRequest, +): Promise { + return await connection.request(SESSION_STEERING_METHOD, params); +} diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index b7664d36..67950100 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -37,6 +37,7 @@ import type { ThreadGoalStatus, ThreadSourceKind, TurnCompletedNotification, + TurnSteerResponse, UserInput, } from "./app-server/v2"; import packageJson from "../package.json"; @@ -833,6 +834,14 @@ export class CodexAcpClient { }); } + async steerTurn(params: { threadId: string, turnId: string, prompt: acp.ContentBlock[] }): Promise { + return await this.codexClient.turnSteer({ + threadId: params.threadId, + expectedTurnId: params.turnId, + input: buildPromptItems(params.prompt), + }); + } + async fetchAvailableModels(): Promise { const models: Model[] = []; let cursor: string | null = null; diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 24a883e7..1fe54602 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -46,9 +46,12 @@ import { type LegacySessionModelState, type LegacySetSessionModelRequest, type LegacySetSessionModelResponse, + type SessionSteerRequest, + type SessionSteeringResponse, GOAL_CONTROL_METHOD, isExtMethodRequest, LEGACY_SET_SESSION_MODEL_METHOD, + SESSION_STEERING_METHOD, } from "./AcpExtensions"; import { createCollabAgentToolCallUpdate, @@ -163,6 +166,7 @@ export class CodexAcpServer { private readonly pendingMcpStartupSessions: Map; private readonly pendingTurnStarts: Map; private readonly activePrompts: Map; + private readonly steeringRequests: Map>; private readonly closingSessions: Map; private readonly sessionGenerations: Map; private readonly sessionOpenGenerations: Map; @@ -178,6 +182,7 @@ export class CodexAcpServer { this.pendingMcpStartupSessions = new Map(); this.pendingTurnStarts = new Map(); this.activePrompts = new Map(); + this.steeringRequests = new Map(); this.closingSessions = new Map(); this.sessionGenerations = new Map(); this.sessionOpenGenerations = new Map(); @@ -238,6 +243,11 @@ export class CodexAcpServer { } }, authMethods: getCodexAuthMethods(_params.clientCapabilities), + _meta: { + steering: { + supported: true, + }, + }, }; } @@ -255,6 +265,8 @@ export class CodexAcpServer { } case LEGACY_SET_SESSION_MODEL_METHOD: return await this.unstable_setSessionModel(this.parseLegacySetSessionModelParams(methodRequest.params)); + case SESSION_STEERING_METHOD: + return await this.steerSessionWithFallback(this.parseSessionSteerParams(methodRequest.params)); case GOAL_CONTROL_METHOD: { const sessionState = this.sessions.get(methodRequest.params.sessionId); if (!sessionState) { @@ -861,6 +873,145 @@ export class CodexAcpServer { }; } + async steerSessionWithFallback(params: SessionSteerRequest): Promise { + const previousRequest = this.steeringRequests.get(params.sessionId) ?? Promise.resolve(); + let releaseRequest: () => void = () => {}; + const requestCompleted = new Promise((resolve) => { + releaseRequest = resolve; + }); + const requestQueue = previousRequest.then(() => requestCompleted); + this.steeringRequests.set(params.sessionId, requestQueue); + + await previousRequest; + try { + return await this.performSteeringRequest(params); + } finally { + releaseRequest(); + if (this.steeringRequests.get(params.sessionId) === requestQueue) { + this.steeringRequests.delete(params.sessionId); + } + } + } + + private async performSteeringRequest(params: SessionSteerRequest): Promise { + logger.log("Steering session requested", { + sessionId: params.sessionId, + prompt: params.prompt, + }); + const sessionState = this.getSessionState(params.sessionId); + + if (!sessionState.supportedInputModalities.includes("image") && params.prompt.some(b => b.type === "image")) { + throw RequestError.invalidRequest("The current model does not support image input"); + } + + const activePrompt = this.activePrompts.get(params.sessionId); + const turnId = await this.getSteerableTurnId(sessionState); + if (!turnId) { + return await this.startNewTurnFromSteering(params, activePrompt); + } + + try { + await this.runWithProcessCheck(() => this.codexAcpClient.steerTurn({ + threadId: params.sessionId, + turnId, + prompt: params.prompt, + })); + } catch (err) { + await this.codexAcpClient.waitForSessionNotifications(params.sessionId); + if (sessionState.currentTurnId === turnId && !this.isNoActiveTurnToSteerError(err)) { + throw err; + } + return await this.startNewTurnFromSteering(params, activePrompt); + } + + logger.log("Steering session injected", { + sessionId: params.sessionId, + turnId, + }); + return {outcome: "injected"}; + } + + private async startNewTurnFromSteering( + params: SessionSteerRequest, + previousPrompt: ActivePrompt | undefined, + ): Promise { + await previousPrompt?.completion; + if (this.sessionIsClosing(params.sessionId)) { + throw RequestError.invalidRequest(`Session ${params.sessionId} is closing`); + } + + let resolveTurnStarted: () => void = () => {}; + const turnStarted = new Promise((resolve) => { + resolveTurnStarted = resolve; + }); + const promptResult = this.prompt(params, undefined, resolveTurnStarted).then( + (response) => ({status: "completed" as const, response}), + (error: unknown) => ({status: "failed" as const, error}), + ); + const acceptance = await Promise.race([ + turnStarted.then(() => ({status: "started" as const})), + promptResult, + ]); + + if (acceptance.status === "failed") { + throw acceptance.error; + } + if (acceptance.status === "completed" && acceptance.response.stopReason === "cancelled") { + throw RequestError.invalidRequest(`Session ${params.sessionId} was cancelled before the steering turn started`); + } + + void promptResult.then((result) => { + if (result.status === "failed") { + logger.error(`Steering-started prompt for session ${params.sessionId} failed`, result.error); + } + }); + logger.log("Steering session started a new turn", {sessionId: params.sessionId}); + return {outcome: "startedNewTurn"}; + } + + private isNoActiveTurnToSteerError(error: unknown): boolean { + const messages = error instanceof Error ? [error.message] : []; + if (typeof error === "object" && error !== null && "data" in error) { + const data = (error as {data?: unknown}).data; + if (typeof data === "string") { + messages.push(data); + } else if (typeof data === "object" && data !== null && "details" in data) { + const details = (data as {details?: unknown}).details; + if (typeof details === "string") { + messages.push(details); + } + } + } + return messages.some(message => message.toLowerCase().includes("no active turn to steer")); + } + + private async getSteerableTurnId(sessionState: SessionState): Promise { + if (this.sessionIsClosing(sessionState.sessionId)) { + return null; + } + if (sessionState.currentTurnId) { + return sessionState.currentTurnId; + } + + const pendingTurnStart = this.pendingTurnStarts.get(sessionState.sessionId); + if (!pendingTurnStart) { + return null; + } + return await pendingTurnStart.promise; + } + + private parseSessionSteerParams(params: Record): SessionSteerRequest { + const sessionId = params["sessionId"]; + const prompt = params["prompt"]; + if (typeof sessionId !== "string" || !Array.isArray(prompt)) { + throw RequestError.invalidParams(); + } + return { + sessionId: sessionId, + prompt: prompt as acp.ContentBlock[], + }; + } + private createSessionConfigOptions(sessionState: SessionState): Array { const currentModelId = ModelId.fromString(sessionState.currentModelId); const configOptions = [ @@ -1609,7 +1760,11 @@ export class CodexAcpServer { return turnId; } - async prompt(params: acp.PromptRequest, signal?: AbortSignal): Promise { + async prompt( + params: acp.PromptRequest, + signal?: AbortSignal, + onTurnStarted?: () => void, + ): Promise { logger.log("Prompt received", { sessionId: params.sessionId, prompt: params.prompt, @@ -1662,6 +1817,7 @@ export class CodexAcpServer { } sessionState.currentTurnId = turnId; pendingTurnStart?.resolve(turnId); + onTurnStarted?.(); }, setConfigOption: async (configId, value) => { await this.applySessionConfigOption(sessionState, { @@ -1751,6 +1907,7 @@ export class CodexAcpServer { } sessionState.currentTurnId = turnId; pendingTurnStart?.resolve(turnId); + onTurnStarted?.(); }, () => this.promptShouldStop(params.sessionId, activePrompt), )); diff --git a/src/CodexAppServerClient.ts b/src/CodexAppServerClient.ts index aa88155d..eb26c83f 100644 --- a/src/CodexAppServerClient.ts +++ b/src/CodexAppServerClient.ts @@ -61,6 +61,8 @@ import type { TurnInterruptResponse, TurnStartParams, TurnStartResponse, + TurnSteerParams, + TurnSteerResponse, CommandExecutionRequestApprovalParams, CommandExecutionRequestApprovalResponse, FileChangeRequestApprovalParams, @@ -504,6 +506,10 @@ export class CodexAppServerClient { return await this.sendRequest({ method: "turn/interrupt", params: params }); } + async turnSteer(params: TurnSteerParams): Promise { + return await this.sendRequest({ method: "turn/steer", params: params }); + } + async reviewStart(params: ReviewStartParams): Promise { return await this.sendRequest({ method: "review/start", params: params }); } diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 3180835b..9d6dc2b8 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -61,6 +61,11 @@ describe('CodexACPAgent - initialize', () => { }, }, authMethods: getCodexAuthMethods(), + _meta: { + steering: { + supported: true, + }, + }, }); }); diff --git a/src/__tests__/CodexACPAgent/steer-events.test.ts b/src/__tests__/CodexACPAgent/steer-events.test.ts new file mode 100644 index 00000000..d0d0606f --- /dev/null +++ b/src/__tests__/CodexACPAgent/steer-events.test.ts @@ -0,0 +1,222 @@ +import {beforeEach, describe, expect, it, vi} from 'vitest'; +import type * as acp from "@agentclientprotocol/sdk"; +import {RequestError} from "@agentclientprotocol/sdk"; +import {createCodexMockTestFixture, createTestSessionState} from "../acp-test-utils"; +import type {SessionState} from "../../CodexAcpServer"; +import type {TurnCompletedNotification} from "../../app-server/v2"; +import {SESSION_STEERING_METHOD} from "../../AcpExtensions"; + +function createTurn(id: string, status: "inProgress" | "completed" | "interrupted") { + return { + id, + items: [], + itemsView: "notLoaded" as const, + status, + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }; +} + +function deferred(): {promise: Promise, resolve: (value: T) => void} { + let resolve: (value: T) => void = () => {}; + const promise = new Promise((innerResolve) => { + resolve = innerResolve; + }); + return {promise, resolve}; +} + +/** + * Drives a prompt to the point where a turn is active (in progress) and paused + * on turn completion, so a steer can be injected mid-turn. + */ +function startActiveTurn(sessionOverrides?: Partial) { + const mockFixture = createCodexMockTestFixture(); + const sessionState = createTestSessionState(sessionOverrides); + vi.spyOn(mockFixture.getCodexAppServerClient(), "turnStart").mockResolvedValue({ + turn: createTurn("turn-id", "inProgress"), + }); + const turnCompleted = deferred(); + vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted") + .mockReturnValue(turnCompleted.promise); + vi.spyOn(mockFixture.getCodexAcpAgent(), "getSessionState").mockReturnValue(sessionState); + return {mockFixture, sessionState, turnCompleted}; +} + +describe('_session/steering', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('reports injected when the input joins the active turn', async () => { + const {mockFixture, sessionState, turnCompleted} = startActiveTurn(); + const turnSteerSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "turnSteer") + .mockResolvedValue({turnId: "turn-id"}); + + const promptPromise = mockFixture.getCodexAcpAgent().prompt({ + sessionId: "session-id", + prompt: [{type: "text", text: "long running prompt"}], + }); + await vi.waitFor(() => { + expect(sessionState.currentTurnId).toBe("turn-id"); + }); + + await expect(mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, { + sessionId: "session-id", + prompt: [{type: "text", text: "also keep backward compatibility"}], + })).resolves.toEqual({outcome: "injected"}); + + expect(turnSteerSpy).toHaveBeenCalledWith({ + threadId: "session-id", + expectedTurnId: "turn-id", + input: [{type: "text", text: "also keep backward compatibility", text_elements: []}], + }); + + turnCompleted.resolve({ + threadId: "session-id", + turn: createTurn("turn-id", "completed"), + }); + await expect(promptPromise).resolves.toMatchObject({stopReason: "end_turn"}); + }); + + it('starts a new turn when no turn is active', async () => { + const mockFixture = createCodexMockTestFixture(); + const sessionState = createTestSessionState(); + vi.spyOn(mockFixture.getCodexAcpAgent(), "getSessionState").mockReturnValue(sessionState); + const turnStartSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "turnStart") + .mockResolvedValue({turn: createTurn("new-turn-id", "inProgress")}); + const turnCompleted = deferred(); + vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted") + .mockReturnValue(turnCompleted.promise); + + await expect(mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, { + sessionId: "session-id", + prompt: [{type: "text", text: "too late for the previous turn"}], + })).resolves.toEqual({outcome: "startedNewTurn"}); + + expect(turnStartSpy).toHaveBeenCalledWith(expect.objectContaining({ + threadId: "session-id", + input: [{type: "text", text: "too late for the previous turn", text_elements: []}], + })); + + turnCompleted.resolve({ + threadId: "session-id", + turn: createTurn("new-turn-id", "completed"), + }); + await vi.waitFor(() => { + expect(sessionState.currentTurnId).toBeNull(); + }); + }); + + it('starts a new turn when Codex reports that the tracked turn is no longer active', async () => { + const {mockFixture, sessionState, turnCompleted} = startActiveTurn(); + const nextTurnCompleted = deferred(); + vi.spyOn(mockFixture.getCodexAppServerClient(), "turnStart") + .mockResolvedValueOnce({turn: createTurn("turn-id", "inProgress")}) + .mockResolvedValueOnce({turn: createTurn("new-turn-id", "inProgress")}); + vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted") + .mockReturnValueOnce(turnCompleted.promise) + .mockReturnValueOnce(nextTurnCompleted.promise); + vi.spyOn(mockFixture.getCodexAppServerClient(), "turnSteer").mockImplementation(async () => { + turnCompleted.resolve({ + threadId: "session-id", + turn: createTurn("turn-id", "completed"), + }); + throw Object.assign(new Error("Internal error"), { + data: {details: "no active turn to steer"}, + }); + }); + + const promptPromise = mockFixture.getCodexAcpAgent().prompt({ + sessionId: "session-id", + prompt: [{type: "text", text: "long running prompt"}], + }); + await vi.waitFor(() => { + expect(sessionState.currentTurnId).toBe("turn-id"); + }); + + await expect(mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, { + sessionId: "session-id", + prompt: [{type: "text", text: "racing follow-up"}], + })).resolves.toEqual({outcome: "startedNewTurn"}); + await expect(promptPromise).resolves.toMatchObject({stopReason: "end_turn"}); + expect(sessionState.currentTurnId).toBe("new-turn-id"); + + nextTurnCompleted.resolve({ + threadId: "session-id", + turn: createTurn("new-turn-id", "completed"), + }); + await vi.waitFor(() => { + expect(sessionState.currentTurnId).toBeNull(); + }); + }); + + it('serializes concurrent late steering requests without dropping either prompt', async () => { + const mockFixture = createCodexMockTestFixture(); + const sessionState = createTestSessionState(); + vi.spyOn(mockFixture.getCodexAcpAgent(), "getSessionState").mockReturnValue(sessionState); + vi.spyOn(mockFixture.getCodexAppServerClient(), "turnStart") + .mockResolvedValue({turn: createTurn("new-turn-id", "inProgress")}); + const turnCompleted = deferred(); + vi.spyOn(mockFixture.getCodexAppServerClient(), "awaitTurnCompleted") + .mockReturnValue(turnCompleted.promise); + const turnSteerSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "turnSteer") + .mockResolvedValue({turnId: "new-turn-id"}); + + const firstRequest = mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, { + sessionId: "session-id", + prompt: [{type: "text", text: "first late follow-up"}], + }); + const secondRequest = mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, { + sessionId: "session-id", + prompt: [{type: "text", text: "second late follow-up"}], + }); + + await expect(Promise.all([firstRequest, secondRequest])).resolves.toEqual([ + {outcome: "startedNewTurn"}, + {outcome: "injected"}, + ]); + expect(turnSteerSpy).toHaveBeenCalledWith({ + threadId: "session-id", + expectedTurnId: "new-turn-id", + input: [{type: "text", text: "second late follow-up", text_elements: []}], + }); + + turnCompleted.resolve({ + threadId: "session-id", + turn: createTurn("new-turn-id", "completed"), + }); + await vi.waitFor(() => { + expect(sessionState.currentTurnId).toBeNull(); + }); + }); + + it('rejects malformed steer params', async () => { + const mockFixture = createCodexMockTestFixture(); + + await expect(mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, { + sessionId: "session-id", + })).rejects.toThrow(RequestError); + }); + + it('rejects image input when the model does not support it', async () => { + const {mockFixture} = startActiveTurn({supportedInputModalities: ["text"]}); + const turnSteerSpy = vi.spyOn(mockFixture.getCodexAppServerClient(), "turnSteer"); + + const image: acp.ContentBlock = { + type: "image", + mimeType: "image/png", + data: "abc123", + }; + + const error = await mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, { + sessionId: "session-id", + prompt: [image], + }).catch((err: unknown) => err); + + expect(error).toBeInstanceOf(RequestError); + expect((error as RequestError).data).toContain("does not support image input"); + expect(turnSteerSpy).not.toHaveBeenCalled(); + }); +}); diff --git a/src/index.ts b/src/index.ts index b2edd2c9..014801ff 100644 --- a/src/index.ts +++ b/src/index.ts @@ -12,7 +12,10 @@ import packageJson from "../package.json"; import {logger} from "./Logger"; import {runLoginCommand} from "./login"; import {runCodexCli} from "./CodexCli"; -import {GOAL_CONTROL_METHOD, LEGACY_SET_SESSION_MODEL_METHOD} from "./AcpExtensions"; +import { + GOAL_CONTROL_METHOD, LEGACY_SET_SESSION_MODEL_METHOD, + SESSION_STEERING_METHOD, +} from "./AcpExtensions"; const emptyExtensionParamsParser = z.preprocess( (params) => params ?? {}, @@ -24,6 +27,11 @@ const legacySetSessionModelParamsParser = z.object({ modelId: z.string(), }).passthrough(); +const sessionSteerParamsParser = z.object({ + sessionId: z.string(), + prompt: z.array(z.any()), +}).passthrough(); + const goalControlParamsParser = z.object({ sessionId: z.string(), action: z.enum(["pause", "clear"]), @@ -137,6 +145,7 @@ function startAcpServer() { .onRequest("authentication/status", emptyExtensionParamsParser, (ctx) => getAgent().extMethod("authentication/status", ctx.params)) .onRequest("authentication/logout", emptyExtensionParamsParser, (ctx) => getAgent().extMethod("authentication/logout", ctx.params)) .onRequest(LEGACY_SET_SESSION_MODEL_METHOD, legacySetSessionModelParamsParser, (ctx) => getAgent().extMethod(LEGACY_SET_SESSION_MODEL_METHOD, ctx.params)) + .onRequest(SESSION_STEERING_METHOD, sessionSteerParamsParser, (ctx) => getAgent().extMethod(SESSION_STEERING_METHOD, ctx.params)) .onRequest(GOAL_CONTROL_METHOD, goalControlParamsParser, (ctx) => getAgent().extMethod(GOAL_CONTROL_METHOD, ctx.params)) .connect(acpJsonStream); } diff --git a/tsconfig.json b/tsconfig.json index 39a4fd81..423a1d8f 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -27,5 +27,5 @@ "noUncheckedSideEffectImports": true, "skipLibCheck": true, }, - "exclude": [".claude"] + "exclude": [".claude", "examples"] } From ea6a9a55463c15cd77c923fecef87cf6fc098a0d Mon Sep 17 00:00:00 2001 From: "mark.tkachenko" Date: Thu, 16 Jul 2026 18:16:55 +0200 Subject: [PATCH 2/2] npm install fix --- package-lock.json | 56 +++++++++++++++++++++-------------------------- 1 file changed, 25 insertions(+), 31 deletions(-) diff --git a/package-lock.json b/package-lock.json index a9ba4301..061220ac 100644 --- a/package-lock.json +++ b/package-lock.json @@ -37,6 +37,29 @@ "zod": "^3.25.0 || ^4.0.0" } }, + "node_modules/@emnapi/core": { + "version": "1.11.1", + "resolved": "https://registry.npmjs.org/@emnapi/core/-/core-1.11.1.tgz", + "integrity": "sha512-RSvbQmHzdKzNsLYa/wHrbc3KN4sYLKAdPZxqiM2HATqv/SBk2/ENSHpvXGaLOMcsAyz0poEGqkmmKYG3OWiJEQ==", + "dev": true, + "license": "MIT", + "optional": true, + "dependencies": { + "@emnapi/wasi-threads": "1.2.2", + "tslib": "^2.4.0" + } + }, + "node_modules/@emnapi/runtime": { + "version": "1.11.1", + "resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.11.1.tgz", + "integrity": "sha512-vgj7R3y3Wgx24IQaGPA/R6YFXLHVMOZ0uVEyIQPaWs+rd1AzfEMXlAC22FYwO1XkKR6NPsq7mUandH8oIRdZFw==", + "dev": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, "node_modules/@emnapi/wasi-threads": { "version": "1.2.2", "resolved": "https://registry.npmjs.org/@emnapi/wasi-threads/-/wasi-threads-1.2.2.tgz", @@ -1260,29 +1283,6 @@ "node": "^20.19.0 || >=22.12.0" } }, - "node_modules/@rolldown/binding-wasm32-wasi/node_modules/@emnapi/core": { - "version": "1.11.1", - "resolved": "https://registry.npmjs.org/@emnapi/core/-/core-1.11.1.tgz", - "integrity": "sha512-RSvbQmHzdKzNsLYa/wHrbc3KN4sYLKAdPZxqiM2HATqv/SBk2/ENSHpvXGaLOMcsAyz0poEGqkmmKYG3OWiJEQ==", - "dev": true, - "license": "MIT", - "optional": true, - "dependencies": { - "@emnapi/wasi-threads": "1.2.2", - "tslib": "^2.4.0" - } - }, - "node_modules/@rolldown/binding-wasm32-wasi/node_modules/@emnapi/runtime": { - "version": "1.11.1", - "resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.11.1.tgz", - "integrity": "sha512-vgj7R3y3Wgx24IQaGPA/R6YFXLHVMOZ0uVEyIQPaWs+rd1AzfEMXlAC22FYwO1XkKR6NPsq7mUandH8oIRdZFw==", - "dev": true, - "license": "MIT", - "optional": true, - "dependencies": { - "tslib": "^2.4.0" - } - }, "node_modules/@rolldown/binding-win32-arm64-msvc": { "version": "1.1.5", "resolved": "https://registry.npmjs.org/@rolldown/binding-win32-arm64-msvc/-/binding-win32-arm64-msvc-1.1.5.tgz", @@ -1373,7 +1373,6 @@ "integrity": "sha512-nxAkRSVkN1Y0JC1W8ky/fTfkGsMmcrRsbx+3XoZE+rMOX71kLYTV7fLXpqud1GpbpP5TuffXFqfX7fH2GgZREw==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "undici-types": "~8.3.0" } @@ -2014,7 +2013,6 @@ "integrity": "sha512-IuL+Elrou2ZvCFHs18/CIzy2Nzvo25nZ1/D2eIZlz7c+QUayAcYoiM2BthCjs+EBHVpjYjcuLDAiCWgeIX3X1Q==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "accepts": "~1.3.8", "array-flatten": "1.1.1", @@ -2265,7 +2263,6 @@ "integrity": "sha512-YwUvVpSF7m1yOblFPrU3Hbo8XhPheBoiyfGuII6z19LnOr6JpDnyyp7LFNrfV56wS8tpvtBFGRISHN02pDdLOA==", "dev": true, "license": "MIT", - "peer": true, "engines": { "node": ">=16.9.0" } @@ -2968,7 +2965,6 @@ "integrity": "sha512-RvwwcruNjI1ncT5xRakeyS9Lf8lcItv34KD+aif+VH9kduAyfYBipGh12274xtenIPZ119/R9BdTBa8gAwSh0A==", "dev": true, "license": "MIT", - "peer": true, "engines": { "node": ">=12" }, @@ -3268,7 +3264,8 @@ "resolved": "https://registry.npmjs.org/scheduler/-/scheduler-0.27.0.tgz", "integrity": "sha512-eNv+WrVbKu1f3vbYJT/xtiF5syA5HPIMtf9IgY/nKg0sWqzAUEvqY/xm7OcZc/qafLx/iO9FgOmeSAp4v5ti/Q==", "dev": true, - "license": "MIT" + "license": "MIT", + "peer": true }, "node_modules/send": { "version": "0.19.2", @@ -3533,7 +3530,6 @@ "integrity": "sha512-eUdUIaCr963q2h5u3+QwvYp0+eqPvn+egeqZUm0hwERCqqx1E3kK5ehbGCvqSE5MQAULr67ww0cA3jKc3YkM1w==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "esbuild": "~0.28.0" }, @@ -3618,7 +3614,6 @@ "integrity": "sha512-bTT9PsdWO+MQMNG9ZXIP/qM9wGh37DFxTV/sPq9cFpHr3w4jkgef032PkAL9jAqhk3Nz8NQw3O8n6/xFkqO4QQ==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "lightningcss": "^1.32.0", "picomatch": "^4.0.5", @@ -3851,7 +3846,6 @@ "resolved": "https://registry.npmjs.org/zod/-/zod-4.4.3.tgz", "integrity": "sha512-ytENFjIJFl2UwYglde2jchW2Hwm4GJFLDiSXWdTrJQBIN9Fcyp7n4DhxJEiWNAJMV1/BqWfW/kkg71UDcHJyTQ==", "license": "MIT", - "peer": true, "funding": { "url": "https://github.com/sponsors/colinhacks" }