feat: add v0.1 runtime skeleton
This commit is contained in:
Codex
@@ -0,0 +1,246 @@
|
||||
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import { accessSync, constants as fsConstants } from "node:fs";
|
||||
import * as readline from "node:readline";
|
||||
import type { BackendEvent, BackendTurnResult, FailureKind, JsonRecord, JsonValue, TerminalStatus } from "../common/types.js";
|
||||
import { redactJson, redactText } from "../common/redaction.js";
|
||||
|
||||
export interface CodexStdioTurnOptions {
|
||||
prompt: string;
|
||||
cwd: string;
|
||||
model?: string;
|
||||
approvalPolicy: string;
|
||||
sandbox: string;
|
||||
timeoutMs: number;
|
||||
command?: string;
|
||||
args?: string[];
|
||||
env?: NodeJS.ProcessEnv;
|
||||
codexHome?: string;
|
||||
}
|
||||
|
||||
interface PendingRequest {
|
||||
resolve: (value: unknown) => void;
|
||||
reject: (error: Error) => void;
|
||||
}
|
||||
|
||||
export class CodexStdioClient {
|
||||
private readonly child: ChildProcessWithoutNullStreams;
|
||||
private readonly pending = new Map<number, PendingRequest>();
|
||||
private readonly stderrChunks: Buffer[] = [];
|
||||
private nextId = 1;
|
||||
private closed = false;
|
||||
readonly closedPromise: Promise<{ code: number | null; signal: string | null; stderrTail: string }>;
|
||||
private closeResolve!: (value: { code: number | null; signal: string | null; stderrTail: string }) => void;
|
||||
|
||||
constructor(options: { command?: string; args?: string[]; cwd: string; env?: NodeJS.ProcessEnv; onNotification: (message: JsonRecord) => void }) {
|
||||
this.closedPromise = new Promise((resolve) => { this.closeResolve = resolve; });
|
||||
this.child = spawn(options.command ?? "codex", options.args ?? ["app-server", "--listen", "stdio://"], {
|
||||
cwd: options.cwd,
|
||||
env: options.env ?? process.env,
|
||||
stdio: "pipe",
|
||||
});
|
||||
this.child.stderr.on("data", (chunk: Buffer) => {
|
||||
this.stderrChunks.push(chunk);
|
||||
while (Buffer.concat(this.stderrChunks).length > 64_000) this.stderrChunks.shift();
|
||||
});
|
||||
const rl = readline.createInterface({ input: this.child.stdout, crlfDelay: Infinity });
|
||||
void this.readLines(rl, options.onNotification);
|
||||
this.child.on("close", (code, signal) => this.handleClose(code, signal));
|
||||
this.child.on("error", (error) => this.handleClose(127, error.message));
|
||||
}
|
||||
|
||||
request(method: string, params: JsonRecord): Promise<unknown> {
|
||||
if (this.closed) return Promise.reject(new Error("codex app-server is closed"));
|
||||
const id = this.nextId++;
|
||||
const message = { id, method, params };
|
||||
return new Promise((resolve, reject) => {
|
||||
this.pending.set(id, { resolve, reject });
|
||||
this.child.stdin.write(`${JSON.stringify(message)}\n`);
|
||||
});
|
||||
}
|
||||
|
||||
notify(method: string, params: JsonRecord = {}): void {
|
||||
if (!this.closed) this.child.stdin.write(`${JSON.stringify({ method, params })}\n`);
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
if (this.closed) return;
|
||||
this.child.kill("SIGTERM");
|
||||
setTimeout(() => {
|
||||
if (!this.closed) this.child.kill("SIGKILL");
|
||||
}, 1500).unref?.();
|
||||
}
|
||||
|
||||
private async readLines(rl: readline.Interface, onNotification: (message: JsonRecord) => void): Promise<void> {
|
||||
try {
|
||||
for await (const line of rl) {
|
||||
const trimmed = String(line).trim();
|
||||
if (trimmed.length === 0) continue;
|
||||
const message = JSON.parse(trimmed) as JsonRecord;
|
||||
const id = typeof message.id === "number" ? message.id : null;
|
||||
const method = typeof message.method === "string" ? message.method : null;
|
||||
if (id !== null && method === null) {
|
||||
const pending = this.pending.get(id);
|
||||
if (!pending) continue;
|
||||
this.pending.delete(id);
|
||||
if (message.error !== undefined) pending.reject(new Error(JSON.stringify(redactJson(message.error))));
|
||||
else pending.resolve(message.result);
|
||||
continue;
|
||||
}
|
||||
if (id !== null && method !== null) {
|
||||
this.handleServerRequest(id, method);
|
||||
continue;
|
||||
}
|
||||
if (method !== null) onNotification(message);
|
||||
}
|
||||
} catch (error) {
|
||||
this.rejectAll(error instanceof Error ? error : new Error(String(error)));
|
||||
}
|
||||
}
|
||||
|
||||
private handleServerRequest(id: number, method: string): void {
|
||||
if (method === "item/commandExecution/requestApproval" || method === "item/fileChange/requestApproval") {
|
||||
this.child.stdin.write(`${JSON.stringify({ id, result: { decision: "decline" } })}\n`);
|
||||
return;
|
||||
}
|
||||
this.child.stdin.write(`${JSON.stringify({ id, error: { code: -32601, message: `Unsupported client-side request: ${method}` } })}\n`);
|
||||
}
|
||||
|
||||
private rejectAll(error: Error): void {
|
||||
for (const pending of this.pending.values()) pending.reject(error);
|
||||
this.pending.clear();
|
||||
}
|
||||
|
||||
private handleClose(code: number | null, signal: string | null): void {
|
||||
if (this.closed) return;
|
||||
this.closed = true;
|
||||
const stderrTail = redactText(Buffer.concat(this.stderrChunks).toString("utf8").slice(-8000));
|
||||
this.rejectAll(new Error(`codex app-server closed code=${code} signal=${signal}`));
|
||||
this.closeResolve({ code, signal, stderrTail });
|
||||
}
|
||||
}
|
||||
|
||||
export async function runCodexStdioTurn(options: CodexStdioTurnOptions): Promise<BackendTurnResult> {
|
||||
const secretFailure = codexHomeReadiness(options.codexHome ?? options.env?.CODEX_HOME ?? `${options.env?.HOME ?? process.env.HOME ?? ""}/.codex`);
|
||||
if (secretFailure) return secretFailure;
|
||||
const events: BackendEvent[] = [{ type: "backend_status", payload: { phase: "codex-app-server-starting", protocol: "codex-app-server-jsonrpc-stdio" } }];
|
||||
let assistantText = "";
|
||||
let threadId: string | undefined;
|
||||
let turnId: string | undefined;
|
||||
let terminal: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } | null = null;
|
||||
let terminalResolve!: () => void;
|
||||
const terminalPromise = new Promise<void>((resolve) => { terminalResolve = resolve; });
|
||||
const clientOptions: ConstructorParameters<typeof CodexStdioClient>[0] = {
|
||||
cwd: options.cwd,
|
||||
env: childEnv(options),
|
||||
onNotification: (message) => {
|
||||
const normalized = normalizeCodexNotification(message);
|
||||
if (normalized.threadId) threadId = normalized.threadId;
|
||||
if (normalized.turnId) turnId = normalized.turnId;
|
||||
if (normalized.assistantDelta) assistantText += normalized.assistantDelta;
|
||||
events.push(...normalized.events);
|
||||
if (normalized.terminal) {
|
||||
terminal = normalized.terminal;
|
||||
terminalResolve();
|
||||
}
|
||||
},
|
||||
};
|
||||
if (options.command) clientOptions.command = options.command;
|
||||
if (options.args) clientOptions.args = options.args;
|
||||
const client = new CodexStdioClient(clientOptions);
|
||||
const timeout = setTimeout(() => {
|
||||
if (!terminal) {
|
||||
terminal = { status: "failed", failureKind: "backend-timeout", message: `codex stdio turn timed out after ${options.timeoutMs}ms` };
|
||||
events.push({ type: "error", payload: terminal });
|
||||
client.stop();
|
||||
terminalResolve();
|
||||
}
|
||||
}, options.timeoutMs);
|
||||
try {
|
||||
await client.request("initialize", { clientInfo: { name: "agentrun", title: "AgentRun", version: "0.1.0" }, capabilities: { experimentalApi: true } });
|
||||
const threadResponse = asResponseRecord(await client.request("thread/start", { model: options.model ?? "default", cwd: options.cwd, approvalPolicy: options.approvalPolicy, sandbox: options.sandbox, serviceName: "agentrun" }));
|
||||
threadId = stringAt(asRecordAt(threadResponse, "thread"), "id") ?? threadId;
|
||||
const turnResponse = asResponseRecord(await client.request("turn/start", { threadId: threadId ?? "", input: [{ type: "text", text: options.prompt, text_elements: [] }], cwd: options.cwd, approvalPolicy: options.approvalPolicy, model: options.model ?? "default" }));
|
||||
turnId = stringAt(asRecordAt(turnResponse, "turn"), "id") ?? turnId;
|
||||
await Promise.race([terminalPromise, client.closedPromise]);
|
||||
if (!terminal) terminal = { status: "failed", failureKind: "backend-protocol-error", message: "codex app-server closed before turn/completed" };
|
||||
} catch (error) {
|
||||
terminal = { status: "failed", failureKind: "backend-protocol-error", message: error instanceof Error ? error.message : String(error) };
|
||||
events.push({ type: "error", payload: { failureKind: terminal.failureKind, message: terminal.message } });
|
||||
} finally {
|
||||
clearTimeout(timeout);
|
||||
client.stop();
|
||||
}
|
||||
if (assistantText.trim().length > 0) events.push({ type: "assistant_message", payload: { text: assistantText } });
|
||||
events.push({ type: "terminal_status", payload: { terminalStatus: terminal.status, failureKind: terminal.failureKind, message: terminal.message } });
|
||||
return { terminalStatus: terminal.status, failureKind: terminal.failureKind, failureMessage: terminal.message, events: events.map((event) => ({ ...event, payload: redactJson(event.payload) })), ...(threadId ? { threadId } : {}), ...(turnId ? { turnId } : {}) };
|
||||
}
|
||||
|
||||
function codexHomeReadiness(codexHome: string): BackendTurnResult | null {
|
||||
try {
|
||||
accessSync(`${codexHome}/auth.json`, fsConstants.R_OK);
|
||||
accessSync(`${codexHome}/config.toml`, fsConstants.R_OK);
|
||||
return null;
|
||||
} catch {
|
||||
return {
|
||||
terminalStatus: "blocked",
|
||||
failureKind: "secret-unavailable",
|
||||
failureMessage: "Codex auth.json or config.toml projection is not readable",
|
||||
events: [
|
||||
{ type: "error", payload: { failureKind: "secret-unavailable", credentialSource: { codexHome, valuesPrinted: false } } },
|
||||
{ type: "terminal_status", payload: { terminalStatus: "blocked", failureKind: "secret-unavailable" } },
|
||||
],
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeCodexNotification(message: JsonRecord): { events: BackendEvent[]; assistantDelta?: string; threadId?: string; turnId?: string; terminal?: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } } {
|
||||
const method = typeof message.method === "string" ? message.method : "unknown";
|
||||
const params = asRecordAt(message, "params");
|
||||
if (method === "thread/started") {
|
||||
const threadId = stringAt(asRecordAt(params, "thread"), "id");
|
||||
return { events: [{ type: "backend_status", payload: { phase: method, threadId } }], ...(threadId ? { threadId } : {}) };
|
||||
}
|
||||
if (method === "turn/started") {
|
||||
const turnId = stringAt(asRecordAt(params, "turn"), "id");
|
||||
return { events: [{ type: "backend_status", payload: { phase: method, turnId } }], ...(turnId ? { turnId } : {}) };
|
||||
}
|
||||
if (method === "item/agentMessage/delta") return { events: [], assistantDelta: typeof params.delta === "string" ? params.delta : "" };
|
||||
if (method === "item/commandExecution/outputDelta") return { events: [{ type: "command_output", payload: { stream: "stdout", text: typeof params.delta === "string" ? params.delta : "" } }] };
|
||||
if (method === "item/started" || method === "item/completed") return { events: [{ type: "tool_call", payload: { method, item: redactJson(asRecordAt(params, "item")) } }] };
|
||||
if (method === "error") return { events: [{ type: "error", payload: { failureKind: "backend-failed", error: redactJson(params.error ?? params) } }] };
|
||||
if (method === "turn/completed") {
|
||||
const turn = asRecordAt(params, "turn");
|
||||
const status = terminalStatusFromValue(turn.status);
|
||||
const error = asRecordAt(turn, "error");
|
||||
const messageText = typeof error.message === "string" ? error.message : null;
|
||||
return { events: [{ type: "backend_status", payload: { phase: method, terminalStatus: status } }], terminal: { status, failureKind: status === "completed" ? null : "backend-failed", message: messageText } };
|
||||
}
|
||||
return { events: [{ type: "backend_status", payload: { phase: method } }] };
|
||||
}
|
||||
|
||||
function terminalStatusFromValue(value: unknown): TerminalStatus {
|
||||
if (value === "completed") return "completed";
|
||||
if (value === "cancelled" || value === "canceled") return "cancelled";
|
||||
if (value === "blocked") return "blocked";
|
||||
return "failed";
|
||||
}
|
||||
|
||||
function childEnv(options: CodexStdioTurnOptions): NodeJS.ProcessEnv {
|
||||
const env: NodeJS.ProcessEnv = { ...process.env, ...options.env };
|
||||
const codexHome = options.codexHome ?? options.env?.CODEX_HOME;
|
||||
if (codexHome) env.CODEX_HOME = codexHome;
|
||||
return env;
|
||||
}
|
||||
|
||||
function asResponseRecord(value: unknown): JsonRecord {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as JsonRecord : {};
|
||||
}
|
||||
|
||||
function asRecordAt(value: JsonRecord, key: string): JsonRecord {
|
||||
const next = value[key];
|
||||
return typeof next === "object" && next !== null && !Array.isArray(next) ? next as JsonRecord : {};
|
||||
}
|
||||
|
||||
function stringAt(value: JsonRecord, key: string): string | null {
|
||||
return typeof value[key] === "string" ? value[key] : null;
|
||||
}
|
||||
Reference in New Issue
Block a user