diff --git a/src/local-agent-adapters.ts b/src/local-agent-adapters.ts index f5845086..f2bc358d 100644 --- a/src/local-agent-adapters.ts +++ b/src/local-agent-adapters.ts @@ -9,7 +9,7 @@ import { } from "./local-agent-codex.js"; import { removeDevspaceNodeModulesBinFromPath } from "./local-agent-path.js"; import { - CodexCliLocalAgentRuntime, + CodexAppServerLocalAgentRuntime, type LocalAgentRunInput, type LocalAgentRunResult, } from "./local-agent-runtime.js"; @@ -59,7 +59,7 @@ class CodexLocalAgentAdapter implements LocalAgentAdapter { "codex provider is not available: codex executable not found. Install codex or set CODEX_COMMAND.", ); } - const runtime = new CodexCliLocalAgentRuntime({ + const runtime = new CodexAppServerLocalAgentRuntime({ command: resolved.executable, env, version: resolved.version, diff --git a/src/local-agent-runtime.test.ts b/src/local-agent-runtime.test.ts index 785fa15b..13d75b14 100644 --- a/src/local-agent-runtime.test.ts +++ b/src/local-agent-runtime.test.ts @@ -1,100 +1,91 @@ import assert from "node:assert/strict"; import { - codexCliArguments, - codexCliError, - CodexCliLocalAgentRuntime, - createCodexCliLocalAgentRuntime, - parseCodexCliLines, - type CodexCliInvocation, - type CodexCliRunner, + codexAppServerError, + codexAppServerThreadParams, + codexAppServerTurnParams, + CodexAppServerLocalAgentRuntime, + createCodexAppServerLocalAgentRuntime, + parseCodexAppServerEvents, + runCodexAppServerTurn, + sandboxModeFor, + type CodexAppServerInvocation, + type CodexAppServerRunner, + type CodexAppServerWire, } from "./local-agent-runtime.js"; -const startedArgs = (writeMode: "read_only" | "allowed" | "full_access" | undefined = undefined) => - codexCliArguments({ +const startedThreadParams = (writeMode: "read_only" | "allowed" | "full_access" | undefined = undefined) => + codexAppServerThreadParams({ prompt: "inspect only", workspace: "/tmp/project", writeMode, }); -assert.deepEqual(startedArgs(), [ - "exec", - "--experimental-json", - "--config", - 'approval_policy="never"', - "--sandbox", - "read-only", - "--cd", - "/tmp/project", -]); +assert.deepEqual(startedThreadParams().params, { + cwd: "/tmp/project", + approvalPolicy: "never", + sandbox: "read-only", +}); -assert.deepEqual(codexCliArguments({ +assert.deepEqual(codexAppServerThreadParams({ prompt: "make change", workspace: "/tmp/project", writeMode: "allowed", model: "gpt-5.4", - thinking: "high", -}), [ - "exec", - "--experimental-json", - "--model", - "gpt-5.4", - "--config", - 'model_reasoning_effort="high"', - "--config", - 'approval_policy="never"', - "--sandbox", - "workspace-write", - "--cd", - "/tmp/project", -]); - -assert.deepEqual(codexCliArguments({ - prompt: "make change", - workspace: "/tmp/project", - writeMode: "full_access", - thinking: "max", -}), [ - "exec", - "--experimental-json", - "--config", - 'model_reasoning_effort="max"', - "--config", - 'approval_policy="never"', - "--sandbox", - "danger-full-access", - "--cd", - "/tmp/project", -]); +}), { + method: "thread/start", + params: { + cwd: "/tmp/project", + approvalPolicy: "never", + sandbox: "workspace-write", + model: "gpt-5.4", + }, +}); -assert.deepEqual(codexCliArguments({ +assert.deepEqual(codexAppServerThreadParams({ prompt: "continue", workspace: "/tmp/project", providerSessionId: "existing-thread", writeMode: "full_access", -}), [ - "exec", - "--experimental-json", - "--config", - 'approval_policy="never"', - "--sandbox", - "danger-full-access", - "--cd", - "/tmp/project", - "resume", - "existing-thread", -]); +}), { + method: "thread/resume", + params: { + threadId: "existing-thread", + cwd: "/tmp/project", + approvalPolicy: "never", + sandbox: "danger-full-access", + }, +}); + +assert.deepEqual(codexAppServerTurnParams({ + prompt: "make change", + workspace: "/tmp/project", + writeMode: "allowed", + model: "gpt-5.4", + thinking: "high", +}, "thread-9"), { + threadId: "thread-9", + input: [{ type: "text", text: "make change" }], + approvalPolicy: "never", + sandboxPolicy: { type: "workspaceWrite" }, + model: "gpt-5.4", + effort: "high", +}); + +assert.equal(sandboxModeFor("read_only"), "read-only"); +assert.equal(sandboxModeFor("allowed"), "workspace-write"); +assert.equal(sandboxModeFor("full_access"), "danger-full-access"); -const invocations: CodexCliInvocation[] = []; -const runner: CodexCliRunner = async (invocation) => { +const invocations: CodexAppServerInvocation[] = []; +const runner: CodexAppServerRunner = async (invocation) => { invocations.push(invocation); return { threadId: "new-thread", - finalResponse: `response:${invocation.prompt}`, - items: [{ type: "agent_message", text: `response:${invocation.prompt}` }], + finalResponse: `response:${invocation.input.prompt}`, + items: [{ type: "agent_message", text: `response:${invocation.input.prompt}` }], }; }; -const runtime = new CodexCliLocalAgentRuntime({ +const runtime = new CodexAppServerLocalAgentRuntime({ command: "/usr/local/bin/codex", env: { PATH: "/usr/local/bin" }, runner, @@ -108,16 +99,6 @@ assert.equal(readOnly.provider, "codex"); assert.equal(readOnly.providerSessionId, "new-thread"); assert.equal(readOnly.finalResponse, "response:inspect only"); assert.deepEqual(readOnly.items, [{ type: "agent_message", text: "response:inspect only" }]); -assert.deepEqual(invocations.at(-1)?.args, [ - "exec", - "--experimental-json", - "--config", - 'approval_policy="never"', - "--sandbox", - "read-only", - "--cd", - "/tmp/project", -]); const resumed = await runtime.run({ prompt: "continue", @@ -129,55 +110,271 @@ const resumed = await runtime.run({ }); assert.equal(resumed.providerSessionId, "new-thread"); assert.equal(resumed.finalResponse, "response:continue"); -assert.deepEqual(invocations.at(-1)?.args, [ - "exec", - "--experimental-json", - "--model", - "gpt-5.4", - "--config", - 'model_reasoning_effort="high"', - "--config", - 'approval_policy="never"', - "--sandbox", - "workspace-write", - "--cd", - "/tmp/project", - "resume", - "existing-thread", -]); +assert.deepEqual(invocations.at(-1)?.input, { + prompt: "continue", + workspace: "/tmp/project", + providerSessionId: "existing-thread", + writeMode: "allowed", + model: "gpt-5.4", + thinking: "high", +}); -const created = createCodexCliLocalAgentRuntime({ +const created = createCodexAppServerLocalAgentRuntime({ command: "/usr/local/bin/codex", env: process.env, runner, }); assert.equal(created.provider, "codex"); -const parsed = parseCodexCliLines([ - JSON.stringify({ type: "thread.started", thread_id: "thread-9" }), - JSON.stringify({ - type: "item.completed", - item: { id: "item_1", type: "tool_output", result: "ls" }, - }), - JSON.stringify({ - type: "item.completed", - item: { id: "item_2", type: "agent_message", text: "Final answer." }, - }), +const parsed = parseCodexAppServerEvents([ + { + method: "thread/started", + params: { thread: { id: "thread-9" } }, + }, + { + method: "item/completed", + params: { item: { id: "item_1", type: "tool_call", tool_name: "shell" } }, + }, + { + method: "item/completed", + params: { item: { id: "item_2", type: "agentMessage", text: "Final answer.", phase: "final_answer" } }, + }, ]); assert.equal(parsed.threadId, "thread-9"); assert.equal(parsed.finalResponse, "Final answer."); assert.equal(parsed.items.length, 2); +assert.equal(parsed.failure, undefined); -const failed = parseCodexCliLines([ - JSON.stringify({ type: "turn.started" }), - JSON.stringify({ - type: "turn.failed", - error: { message: "model rejected: cli too old" }, - }), +const failed = parseCodexAppServerEvents([ + { + method: "turn/completed", + params: { + turn: { + status: "failed", + error: { message: "model rejected: cli too old" }, + items: [], + }, + }, + }, ]); assert.equal(failed.failure, "model rejected: cli too old"); -const error = codexCliError("codex turn failed: boom", "0.147.0", "raw stderr"); +const error = codexAppServerError("codex turn failed: boom", "0.147.0", "raw stderr"); assert.match(error.message, /codex turn failed: boom/); assert.match(error.message, /codex version: 0\.147\.0/); -assert.match(error.message, /raw stderr/); \ No newline at end of file +assert.match(error.message, /raw stderr/); + +// A scripted in-memory app-server peer. Client requests are answered in FIFO +// order from `script`; each reply may also carry app-server notifications to +// emit (streamed item events, turn completed). +class MockCodexAppServer { + readonly outgoing: Array> = []; + private readonly lineHandlers: Array<(line: string) => void> = []; + script: Array<{ + result: unknown; + notifications?: Array<{ method: string; params: unknown }>; + }> = []; + + wire(): CodexAppServerWire { + return { + writeLine: (line) => this.writeLine(line), + onLine: (handler) => { + this.lineHandlers.push(handler); + }, + onExit: () => {}, + endStdin: () => {}, + }; + } + + writeLine(line: string): void { + const message = JSON.parse(line) as Record; + this.outgoing.push(message); + if (!("id" in message) || !("method" in message)) return; + const reply = this.script.shift(); + if (!reply) return; + const id = message.id; + this.emit(JSON.stringify({ id, result: reply.result })); + for (const notification of reply.notifications ?? []) { + this.emit(JSON.stringify({ method: notification.method, params: notification.params })); + } + } + + private emit(line: string): void { + for (const handler of this.lineHandlers) handler(line); + } +} + +// Drive the full protocol offline: initialize -> thread/start -> turn/start on +// a scripted peer, collecting the final agent message from the turn payload. +{ + const peer = new MockCodexAppServer(); + peer.script = [ + { result: { userAgent: "probe/0.147.0" } }, + { + result: { thread: { id: "thread-77" } }, + notifications: [{ method: "thread/started", params: { thread: { id: "thread-77" } } }], + }, + { + result: { turn: { id: "turn-1", status: "inProgress" } }, + notifications: [ + { + method: "turn/started", + params: { threadId: "thread-77", turn: { id: "turn-1", status: "inProgress" } }, + }, + { + method: "item/completed", + params: { + item: { id: "item_2", type: "agentMessage", text: "Mock final.", phase: "final_answer" }, + }, + }, + { + method: "turn/completed", + params: { + threadId: "thread-77", + turn: { + id: "turn-1", + status: "completed", + items: [ + { id: "item_2", type: "agentMessage", text: "Mock final.", phase: "final_answer" }, + ], + }, + }, + }, + ], + }, + ]; + const turn = await runCodexAppServerTurn(peer.wire(), { + prompt: "hello", + workspace: "/tmp/project", + model: "gpt-5.4", + thinking: "high", + }); + assert.equal(turn.threadId, "thread-77"); + assert.equal(turn.finalResponse, "Mock final."); + assert.deepEqual(turn.items, [ + { id: "item_2", type: "agentMessage", text: "Mock final.", phase: "final_answer" }, + ]); + assert.deepEqual(peer.outgoing.map((message) => message.method), [ + "initialize", + "initialized", + "thread/start", + "turn/start", + ]); + const turnRequest = peer.outgoing[3].params as Record; + assert.equal(turnRequest.threadId, "thread-77"); + assert.equal(turnRequest.effort, "high"); +} + +// A provider-session id resumes the thread instead of starting a new one. +{ + const peer = new MockCodexAppServer(); + peer.script = [ + { result: { userAgent: "probe/0.147.0" } }, + { + result: { thread: { id: "thread-77" } }, + notifications: [{ method: "thread/resumed", params: { thread: { id: "thread-77" } } }], + }, + { + result: { turn: { id: "turn-1", status: "inProgress" } }, + notifications: [ + { + method: "turn/completed", + params: { + turn: { + id: "turn-1", + status: "completed", + items: [ + { id: "item_2", type: "agentMessage", text: "Same thread.", phase: "final_answer" }, + ], + }, + }, + }, + ], + }, + ]; + const turn = await runCodexAppServerTurn(peer.wire(), { + prompt: "continue", + workspace: "/tmp/project", + providerSessionId: "thread-77", + }); + assert.equal(peer.outgoing[2].method, "thread/resume"); + const resumeParams = peer.outgoing[2].params as Record; + assert.equal(resumeParams.threadId, "thread-77"); + assert.equal(turn.finalResponse, "Same thread."); +} + +// A failed turn surfaces the error from the terminal payload. +{ + const peer = new MockCodexAppServer(); + peer.script = [ + { result: { userAgent: "probe/0.147.0" } }, + { + result: { thread: { id: "thread-77" } }, + notifications: [{ method: "thread/started", params: { thread: { id: "thread-77" } } }], + }, + { + result: { turn: { id: "turn-1", status: "inProgress" } }, + notifications: [ + { + method: "turn/completed", + params: { + threadId: "thread-77", + turn: { + id: "turn-1", + status: "failed", + error: { message: "model rejected: cli too old" }, + items: [], + }, + }, + }, + ], + }, + ]; + await assert.rejects( + runCodexAppServerTurn(peer.wire(), { prompt: "hello", workspace: "/tmp/project" }), + /codex turn failed: model rejected: cli too old/, + ); +} + +// A provider-gate failure arrives as a JSON-encoded error; the human-readable +// inner message is peeled out for the session error row. +{ + const peer = new MockCodexAppServer(); + peer.script = [ + { result: { userAgent: "probe/0.147.0" } }, + { + result: { thread: { id: "thread-77" } }, + notifications: [{ method: "thread/started", params: { thread: { id: "thread-77" } } }], + }, + { + result: { turn: { id: "turn-1", status: "inProgress" } }, + notifications: [ + { + method: "turn/completed", + params: { + threadId: "thread-77", + turn: { + id: "turn-1", + status: "failed", + error: { + message: JSON.stringify({ + type: "error", + status: 400, + error: { + type: "invalid_request_error", + message: "The 'bogus-model' model is not supported.", + }, + }), + }, + items: [], + }, + }, + }, + ], + }, + ]; + await assert.rejects( + runCodexAppServerTurn(peer.wire(), { prompt: "hello", workspace: "/tmp/project" }), + /codex turn failed: The 'bogus-model' model is not supported\./, + ); +} \ No newline at end of file diff --git a/src/local-agent-runtime.ts b/src/local-agent-runtime.ts index 34411cfb..3cdcfed0 100644 --- a/src/local-agent-runtime.ts +++ b/src/local-agent-runtime.ts @@ -24,88 +24,134 @@ export interface LocalAgentRuntime { run(input: LocalAgentRunInput): Promise; } -export interface CodexCliInvocation { +// The Codex harness speaks the `codex app-server` JSON-RPC protocol over +// newline-delimited stdio: the client drives a thread (`thread/start` or +// `thread/resume`), runs one turn (`turn/start`), and observes streamed +// notifications until `turn/completed` reports the final item list. +export interface CodexAppServerInvocation { readonly command: string; - readonly args: string[]; readonly env: NodeJS.ProcessEnv; - readonly prompt: string; + readonly input: LocalAgentRunInput; } -export interface CodexCliTurn { +export interface CodexAppServerTurn { readonly threadId: string | null; readonly finalResponse: string; readonly items: unknown[]; } -export type CodexCliRunner = ( - invocation: CodexCliInvocation, -) => Promise; +export type CodexAppServerRunner = ( + invocation: CodexAppServerInvocation, +) => Promise; -export interface CodexCliRuntimeOptions { +export interface CodexAppServerRuntimeOptions { readonly command: string; readonly env: NodeJS.ProcessEnv; readonly version?: string; - readonly runner?: CodexCliRunner; -} - -interface ParsedCodexCliLines { - threadId: string | null; - finalResponse: string; - items: unknown[]; - failure: unknown; + readonly runner?: CodexAppServerRunner; } const CODEX_APPROVAL_POLICY = "never"; -export function codexCliArguments(input: LocalAgentRunInput): string[] { - const args = ["exec", "--experimental-json"]; - if (input.model) { - args.push("--model", input.model); - } - if (input.thinking) { - args.push("--config", `model_reasoning_effort="${input.thinking}"`); - } - args.push("--config", `approval_policy="${CODEX_APPROVAL_POLICY}"`); - args.push("--sandbox", sandboxModeFor(input.writeMode)); - args.push("--cd", input.workspace); +// Build the thread-opening request: a fresh `thread/start` unless the host +// supplied a provider session id, which resumes that thread in place. +export function codexAppServerThreadParams(input: LocalAgentRunInput): { + method: "thread/start" | "thread/resume"; + params: Record; +} { + const sandbox = sandboxModeFor(input.writeMode); if (input.providerSessionId) { - args.push("resume", input.providerSessionId); + return { + method: "thread/resume", + params: { + threadId: input.providerSessionId, + cwd: input.workspace, + approvalPolicy: CODEX_APPROVAL_POLICY, + sandbox, + ...(input.model ? { model: input.model } : {}), + }, + }; } - return args; + return { + method: "thread/start", + params: { + cwd: input.workspace, + approvalPolicy: CODEX_APPROVAL_POLICY, + sandbox, + ...(input.model ? { model: input.model } : {}), + }, + }; } -// Mirror the SDK's event handling, minus the parts DevSpace does not use: -// `thread.started` supplies the thread id, `item.completed` yields the final -// agent message and the item log, and a `turn.failed` event aborts the run. -export function parseCodexCliLines(lines: string[]): ParsedCodexCliLines { +// Build the per-turn request against the opened thread. Reasoning effort is a +// free-form string passed through verbatim, matching the CLI's +// `model_reasoning_effort` override. +export function codexAppServerTurnParams( + input: LocalAgentRunInput, + threadId: string, +): Record { + return { + threadId, + input: [{ type: "text", text: input.prompt }], + approvalPolicy: CODEX_APPROVAL_POLICY, + sandboxPolicy: turnSandboxPolicyFor(input.writeMode), + ...(input.model ? { model: input.model } : {}), + ...(input.thinking ? { effort: input.thinking } : {}), + }; +} + +export interface ParsedCodexAppServerEvents { + threadId: string | null; + finalResponse: string; + items: unknown[]; + failure: unknown; +} + +// Reduce the streamed notifications and the terminal `turn/completed` payload +// into the bounded-run result: the thread id, the final agent message, the +// item log, and any turn failure. `turn/completed` carries the authoritative +// item list for the turn; per-item notifications provide the thread id and +// the streaming fallback when that list is empty. +export function parseCodexAppServerEvents( + events: Array<{ method: string; params?: unknown }>, +): ParsedCodexAppServerEvents { let threadId: string | null = null; let finalResponse = ""; - const items: unknown[] = []; + let items: unknown[] = []; let failure: unknown; - for (const rawLine of lines) { - const line = rawLine.trim(); - if (!line) continue; - let event: Record; - try { - event = JSON.parse(line) as Record; - } catch { - throw new Error(`Failed to parse codex CLI output line: ${line.slice(0, 200)}`); - } - switch (event.type) { - case "thread.started": { - if (typeof event.thread_id === "string") threadId = event.thread_id; + for (const event of events) { + switch (event.method) { + case "thread/started": { + const thread = asRecord(event.params)?.thread as Record | undefined; + if (thread && typeof thread.id === "string") threadId = thread.id; break; } - case "item.completed": { - const record = event.item as Record | undefined; - if (record) items.push(record); - if (record?.type === "agent_message" && typeof record.text === "string") { - finalResponse = record.text; + case "item/completed": { + const item = asRecord(asRecord(event.params)?.item); + if (item) items.push(item); + if (item?.type === "agentMessage" && typeof item.text === "string") { + finalResponse = item.text; } break; } - case "turn.failed": { - failure = failureMessage(event.error); + case "turn/completed": { + const params = asRecord(event.params); + const turn = asRecord(params?.turn); + if (turn?.status === "failed") { + failure = failureMessage(asRecord(turn.error)?.message); + } + if (Array.isArray(turn?.items) && turn.items.length > 0) { + items = turn.items; + } else if (items.length === 0) { + items = []; + } + for (const item of items) { + if (finalResponse) break; + const record = asRecord(item); + if (record?.type === "agentMessage" && typeof record.text === "string") { + finalResponse = record.text; + } + } break; } } @@ -113,20 +159,19 @@ export function parseCodexCliLines(lines: string[]): ParsedCodexCliLines { return { threadId, finalResponse, items, failure }; } -export class CodexCliLocalAgentRuntime implements LocalAgentRuntime { +export class CodexAppServerLocalAgentRuntime implements LocalAgentRuntime { readonly provider = "codex" as const; - private readonly runner: CodexCliRunner; + private readonly runner: CodexAppServerRunner; - constructor(private readonly options: CodexCliRuntimeOptions) { - this.runner = options.runner ?? createCodexCliSpawnRunner({ version: options.version }); + constructor(private readonly options: CodexAppServerRuntimeOptions) { + this.runner = options.runner ?? createCodexAppServerSpawnRunner({ version: options.version }); } async run(input: LocalAgentRunInput): Promise { const turn = await this.runner({ command: this.options.command, - args: codexCliArguments(input), env: this.options.env, - prompt: input.prompt, + input, }); return { provider: this.provider, @@ -137,98 +182,307 @@ export class CodexCliLocalAgentRuntime implements LocalAgentRuntime { } } -export function createCodexCliLocalAgentRuntime( - options: CodexCliRuntimeOptions, -): CodexCliLocalAgentRuntime { - return new CodexCliLocalAgentRuntime(options); +export function createCodexAppServerLocalAgentRuntime( + options: CodexAppServerRuntimeOptions, +): CodexAppServerLocalAgentRuntime { + return new CodexAppServerLocalAgentRuntime(options); } -export function createCodexCliSpawnRunner(options: { version?: string } = {}): CodexCliRunner { +interface CodexAppServerRpc { + readonly events: Array<{ method: string; params?: unknown }>; + request(method: string, params?: unknown): Promise; + notify(method: string, params?: unknown): void; + waitForNotification( + predicate: (event: { method: string; params?: unknown }) => boolean, + ): Promise<{ method: string; params?: unknown }>; +} + +// The app-server wire is the transport boundary for the JSON-RPC protocol. +// The default spawn runner builds a wire over a live child; tests inject an +// in-memory peer to exercise the protocol offline. +export interface CodexAppServerWire { + writeLine(line: string): void; + onLine(handler: (line: string) => void): void; + onExit(handler: (code: number | null, signal: NodeJS.Signals | null) => void): void; + endStdin(): void; +} + +// Drive the host `codex app-server` over a wire and eventually through one +// bounded turn. The wire is injected so the protocol can be exercised offline +// against an in-memory peer (or a live child for the default spawn runner). +export async function runCodexAppServerTurn( + wire: CodexAppServerWire, + input: LocalAgentRunInput, +): Promise { + const rpc = attachCodexAppServerRpc(wire); + await initializeCodexAppServer(rpc); + const threadId = await openCodexThread(rpc, input); + await rpc.request("turn/start", codexAppServerTurnParams(input, threadId)); + await waitForTurnCompleted(rpc); + const parsed = parseCodexAppServerEvents([...rpc.events]); + if (parsed.failure) { + throw new Error(`codex turn failed: ${String(parsed.failure)}`); + } + return { + threadId: parsed.threadId ?? threadId, + finalResponse: parsed.finalResponse, + items: parsed.items, + }; +} + +export function createCodexAppServerSpawnRunner(options: { + version?: string; +} = {}): CodexAppServerRunner { const version = options.version; return async (invocation) => { - const { command, args, env, prompt } = invocation; - const child = spawn(command, args, { + const { command, env, input } = invocation; + const child = spawn(command, ["app-server"], { env, windowsHide: true, }); let spawnError: Error | undefined; let stderr = ""; - const exitPromise = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>( - (resolve) => { - child.once("exit", (code, signal) => resolve({ code, signal })); - }, - ); child.once("error", (error) => { spawnError = error; }); - if (!child.stdin) { - child.kill(); - throw codexCliError("codex CLI did not expose stdin", version); - } - child.stdin.write(prompt); - child.stdin.end(); if (child.stderr) { child.stderr.on("data", (chunk: Buffer) => { stderr += chunk.toString("utf8"); }); } - const output = child.stdout; - if (!output) { - child.kill(); - throw codexCliError("codex CLI did not expose stdout", version); - } - const lines: string[] = []; - const reader = createInterface({ - input: output, - crlfDelay: Infinity, - }); + const wire: CodexAppServerWire = { + onLine: (handler) => { + if (!child.stdout) { + throw new Error("codex app-server did not expose stdout."); + } + createInterface({ input: child.stdout, crlfDelay: Infinity }).on("line", handler); + }, + onExit: (handler) => { + child.once("exit", (code, signal) => handler(code, signal)); + }, + endStdin: () => { + try { + if (child.stdin && !child.stdin.destroyed) child.stdin.end(); + } catch { + // The process may already be gone. + } + }, + writeLine: (line) => { + if (!child.stdin) { + throw new Error("codex app-server did not expose stdin."); + } + child.stdin.write(`${line}\n`); + }, + }; try { - for await (const line of reader) { - lines.push(line); + const turn = await runCodexAppServerTurn(wire, input); + return turn; + } catch (error) { + if (spawnError) { + throw codexAppServerError( + `Failed to start codex app-server: ${spawnError.message}`, + version, + stderr, + ); + } + if (error instanceof CodexAppServerError) { + throw error; } + throw codexAppServerError(errorMessage(error), version, stderr); } finally { - reader.close(); + await terminateWire(wire); + } + }; +} + +async function initializeCodexAppServer(rpc: CodexAppServerRpc): Promise { + await rpc.request("initialize", { + clientInfo: { + name: "devspace", + title: "DevSpace", + version: "1.0.0", + }, + capabilities: { + experimentalApi: true, + }, + }); + rpc.notify("initialized"); +} + +async function openCodexThread( + rpc: CodexAppServerRpc, + input: LocalAgentRunInput, +): Promise { + const params = codexAppServerThreadParams(input); + const response = asRecord(await rpc.request(params.method, params.params)); + const thread = asRecord(response?.thread); + const threadId = thread && typeof thread.id === "string" ? thread.id : undefined; + if (!threadId) { + throw new Error(`codex app-server did not return a thread id for ${params.method}.`); + } + return threadId; +} + +function attachCodexAppServerRpc( + wire: CodexAppServerWire, +): CodexAppServerRpc { + let nextRequestId = 1; + const pending = new Map void; + reject: (error: Error) => void; + }>(); + const waiters: Array<{ + predicate: (event: { method: string; params?: unknown }) => boolean; + resolve: (event: { method: string; params?: unknown }) => void; + reject: (error: Error) => void; + }> = []; + const events: Array<{ method: string; params?: unknown }> = []; + let fatalError: Error | undefined; + + const dispatch = (event: { method: string; params?: unknown }): void => { + events.push(event); + for (const waiter of waiters) { + if (waiter.predicate(event)) { + waiter.resolve(event); + waiters.splice(waiters.indexOf(waiter), 1); + return; + } + } + }; + + const handleLine = (line: string): void => { + const trimmed = line.trim(); + if (!trimmed) return; + let message: Record; + try { + message = JSON.parse(trimmed) as Record; + } catch { + return; + } + const method = typeof message.method === "string" ? message.method : undefined; + const id = typeof message.id === "string" ? message.id : undefined; + if (id !== undefined && method === undefined) { + const pendingEntry = id ? pending.get(id) : undefined; + if (!pendingEntry) return; + pending.delete(id); + if (message.error !== undefined) { + pendingEntry.reject(new Error(protocolErrorText(message.error))); + return; + } + pendingEntry.resolve(message.result); + return; } - if (spawnError) { - throw codexCliError( - `Failed to start codex CLI: ${spawnError.message}`, - version, - stderr, - ); + if (id !== undefined && method !== undefined) { + // The app-server can request approvals or user input mid-turn. DevSpace + // runs bounded non-interactive turns, so decline anything that asks. + wire.writeLine(JSON.stringify({ + id, + error: { code: -32601, message: `unsupported app-server request: ${method}` }, + })); + return; } + if (method !== undefined) { + dispatch({ method, params: message.params }); + } + }; - const parsed = parseCodexCliLines(lines); - if (parsed.failure) { - throw codexCliError(`codex turn failed: ${String(parsed.failure)}`, version, stderr); + wire.onLine(handleLine); + wire.onExit((code, signal) => { + if (fatalError) return; + fatalError = new Error( + `codex app-server exited with ${signal ? `signal ${signal}` : `code ${code ?? 1}`} before the turn completed`, + ); + for (const entry of pending.values()) { + entry.reject(fatalError); } - const { code, signal } = await exitPromise; - if (code !== 0 || signal) { - throw codexCliError( - `codex CLI exited with ${signal ? `signal ${signal}` : `code ${code ?? 1}`}`, - version, - stderr, - ); + for (const waiter of waiters) { + waiter.reject(fatalError); } - return { - threadId: parsed.threadId, - finalResponse: parsed.finalResponse, - items: parsed.items, - }; + pending.clear(); + waiters.length = 0; + }); + + return { + events, + request(method, params) { + if (fatalError) return Promise.reject(fatalError); + const id = String(nextRequestId); + nextRequestId += 1; + return new Promise((resolve, reject) => { + pending.set(id, { resolve, reject }); + wire.writeLine(JSON.stringify({ + id, + method, + ...(params !== undefined ? { params } : {}), + })); + }); + }, + notify(method, params) { + wire.writeLine(JSON.stringify({ + method, + ...(params !== undefined ? { params } : {}), + })); + }, + waitForNotification(predicate) { + if (fatalError) return Promise.reject(fatalError); + const existing = events.find(predicate); + if (existing) return Promise.resolve(existing); + return new Promise((resolve, reject) => { + waiters.push({ predicate, resolve, reject }); + }); + }, }; } +function waitForTurnCompleted( + rpc: CodexAppServerRpc, +): Promise<{ method: string; params?: unknown }> { + return rpc.waitForNotification((event) => { + if (event.method !== "turn/completed") return false; + const turn = asRecord(asRecord(event.params)?.turn); + return turn?.status !== undefined; + }); +} + +function protocolErrorText(error: unknown): string { + const record = asRecord(error); + if (!record) return String(error); + const message = record.message; + const code = record.code; + if (typeof message === "string") { + return typeof code === "string" || typeof code === "number" + ? `codex app-server ${code}: ${message}` + : message; + } + return String(error); +} + +async function terminateWire(wire: CodexAppServerWire): Promise { + try { + wire.endStdin(); + } catch { + // The process may already be gone. + } +} + // Surface the CLI version and raw stderr in the exception so the session error // row lets the host reason about model gates without forensics. -export function codexCliError(message: string, version?: string, stderr?: string): Error { +export class CodexAppServerError extends Error {} + +export function codexAppServerError( + message: string, + version?: string, + stderr?: string, +): CodexAppServerError { const details = [ message, version ? `codex version: ${version}` : undefined, stderr && stderr.trim() ? `stderr:\n${stderr.trim()}` : undefined, ].filter(Boolean).join("\n"); - return new Error(details); + return new CodexAppServerError(details); } -function sandboxModeFor(writeMode: LocalAgentWriteMode | undefined): string { +export function sandboxModeFor(writeMode: LocalAgentWriteMode | undefined): string { switch (writeMode) { case "allowed": return "workspace-write"; @@ -240,11 +494,66 @@ function sandboxModeFor(writeMode: LocalAgentWriteMode | undefined): string { } } +function turnSandboxPolicyFor( + writeMode: LocalAgentWriteMode | undefined, +): { type: "readOnly" | "workspaceWrite" | "dangerFullAccess" } { + switch (writeMode) { + case "allowed": + return { type: "workspaceWrite" }; + case "full_access": + return { type: "dangerFullAccess" }; + case "read_only": + case undefined: + return { type: "readOnly" }; + } +} + function failureMessage(error: unknown): unknown { if (error && typeof error === "object") { const message = (error as Record).message; - if (typeof message === "string" && message.trim()) return message; + if (typeof message === "string" && message.trim()) { + return unwrapErrorMessage(message); + } } - if (typeof error === "string" && error.trim()) return error; + if (typeof error === "string" && error.trim()) return unwrapErrorMessage(error); return error; -} \ No newline at end of file +} + +// The CLI can surface `turn.error.message` as a JSON-encoded error payload +// (e.g. provider HTTP errors); peel down to the innermost human-readable +// message so the session error row stays legible. +function unwrapErrorMessage(message: string): string { + if (!message.startsWith("{") && !message.startsWith("[")) return message; + for (let depth = 0; depth < 3; depth += 1) { + let parsed: unknown; + try { + parsed = JSON.parse(message) as unknown; + } catch { + return message; + } + const current = asRecord(parsed); + if (!current) return message; + const inner = current.message ?? current.error; + if (typeof inner === "string" && inner.trim()) return inner; + const innerRecord = asRecord(inner); + if (typeof innerRecord?.message === "string" && innerRecord.message.trim()) { + message = innerRecord.message; + continue; + } + if (typeof innerRecord?.error === "string" && innerRecord.error.trim()) { + message = innerRecord.error; + continue; + } + return message; + } + return message; +} + +function asRecord(value: unknown): Record | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + return value as Record; +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +}