diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 03d5de9ab..e80d0381b 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -38,7 +38,6 @@ import * as fsSync from "node:fs"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import readline from "node:readline"; import { z } from "zod"; import { renderPromptAttachmentAsText } from "../prompt-attachments.js"; import { curateAgentActivity } from "../activity-curator.js"; @@ -53,10 +52,10 @@ import { type ProviderRuntimeSettings, } from "../provider-launch-config.js"; import { findExecutable, isCommandAvailable } from "../../../utils/executable.js"; -import { terminateWithTreeKill } from "../../../utils/tree-kill.js"; import { spawnProcess } from "../../../utils/spawn.js"; import { extractCodexTerminalSessionId, nonEmptyString } from "./tool-call-mapper-utils.js"; import { buildCodexFeatures, codexModelSupportsFastMode } from "./codex-feature-definitions.js"; +import { CodexAppServerClient } from "./codex/app-server-transport.js"; import { renderProviderImageOutputAsAssistantMarkdown, type ProviderImageOutput, @@ -83,11 +82,8 @@ function isRecord(value: unknown): value is Record { return value != null && typeof value === "object" && !Array.isArray(value); } -const DEFAULT_TIMEOUT_MS = 14 * 24 * 60 * 60 * 1000; const TURN_START_TIMEOUT_MS = 90 * 1000; const INTERRUPT_TIMEOUT_MS = 2_000; -const APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 2_000; -const APP_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS = 1_000; const CODEX_PROVIDER = "codex" as const; const CODEX_IMAGE_ATTACHMENT_DIR = "paseo-attachments"; const ASSISTANT_MESSAGE_BOUNDARY_MARKDOWN = "\n\n---\n\n"; @@ -627,53 +623,11 @@ function toCodexMcpConfig(config: McpServerConfig): CodexMcpServerConfig { } } } -interface JsonRpcRequest { - id: number; - method: string; - params?: unknown; -} - -interface JsonRpcResponse { - id: number; - result?: unknown; - error?: { code?: number; message: string }; -} - -interface JsonRpcNotification { - method: string; - params?: unknown; -} - -function isJsonRpcResponse(msg: unknown): msg is JsonRpcResponse { - if (!isRecord(msg)) return false; - if (typeof msg.id !== "number") return false; - return msg.result !== undefined || !!msg.error; -} - -function isJsonRpcRequest(msg: unknown): msg is JsonRpcRequest { - if (!isRecord(msg)) return false; - return typeof msg.id === "number" && typeof msg.method === "string"; -} - -function isJsonRpcNotification(msg: unknown): msg is JsonRpcNotification { - if (!isRecord(msg)) return false; - return typeof msg.method === "string" && typeof msg.id !== "number"; -} function toObjectRecord(value: unknown): Record | undefined { return isRecord(value) ? value : undefined; } -interface PendingRequest { - resolve: (value: unknown) => void; - reject: (error: Error) => void; - timer: ReturnType; -} - -type RequestHandler = (params: unknown) => unknown; - -type NotificationHandler = (method: string, params: unknown) => void; - // Codex app-server API response types interface CodexReasoningEffortEntry { reasoningEffort?: string; @@ -727,171 +681,6 @@ function filterCodexThreadsByCwd( return threads.filter((thread) => typeof thread.cwd === "string" && thread.cwd === cwd); } -class CodexAppServerClient { - private readonly rl: readline.Interface; - private readonly pending = new Map(); - private readonly requestHandlers = new Map(); - private notificationHandler: NotificationHandler | null = null; - private nextId = 1; - private disposed = false; - private stderrBuffer = ""; - - constructor( - private readonly child: ChildProcessWithoutNullStreams, - private readonly logger: Logger, - ) { - this.rl = readline.createInterface({ input: child.stdout }); - this.rl.on("line", (line) => this.handleLine(line)); - - child.stderr.on("data", (chunk) => { - this.stderrBuffer += chunk.toString(); - if (this.stderrBuffer.length > 8192) { - this.stderrBuffer = this.stderrBuffer.slice(-8192); - } - }); - - child.on("error", (err) => { - this.logger.error({ err }, "Codex app-server child process error"); - for (const pending of this.pending.values()) { - clearTimeout(pending.timer); - pending.reject(err); - } - this.pending.clear(); - this.disposed = true; - }); - - child.on("exit", (code, signal) => { - const message = - code === 0 && !signal - ? "Codex app-server exited" - : `Codex app-server exited with code ${code ?? "null"} and signal ${signal ?? "null"}`; - const error = new Error(`${message}\n${this.stderrBuffer}`.trim()); - for (const pending of this.pending.values()) { - clearTimeout(pending.timer); - pending.reject(error); - } - this.pending.clear(); - this.disposed = true; - }); - } - - setNotificationHandler(handler: NotificationHandler): void { - this.notificationHandler = handler; - } - - setRequestHandler(method: string, handler: RequestHandler): void { - this.requestHandlers.set(method, handler); - } - - request(method: string, params?: unknown, timeoutMs = DEFAULT_TIMEOUT_MS): Promise { - if (this.disposed) { - return Promise.reject(new Error("Codex app-server client is closed")); - } - const id = this.nextId++; - const payload: JsonRpcRequest = { id, method, params }; - const serialized = JSON.stringify(payload); - this.child.stdin.write(`${serialized}\n`); - return new Promise((resolve, reject) => { - const timer = setTimeout(() => { - this.pending.delete(id); - reject(new Error(`Codex app-server request timed out for ${method}`)); - }, timeoutMs); - this.pending.set(id, { resolve, reject, timer }); - }); - } - - notify(method: string, params?: unknown): void { - if (this.disposed) { - return; - } - const payload: JsonRpcNotification = { method, params }; - this.child.stdin.write(`${JSON.stringify(payload)}\n`); - } - - private writeJsonRpcResponse(response: JsonRpcResponse): void { - if (this.disposed || this.child.stdin.destroyed || !this.child.stdin.writable) { - return; - } - try { - this.child.stdin.write(`${JSON.stringify(response)}\n`); - } catch (error) { - this.logger.debug({ error }, "Failed to write Codex app-server JSON-RPC response"); - } - } - - async dispose(): Promise { - if (this.disposed) return; - this.disposed = true; - this.rl.close(); - try { - this.child.stdin.end(); - } catch { - // ignore - } - const result = await terminateWithTreeKill(this.child, { - gracefulTimeoutMs: APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS, - forceTimeoutMs: APP_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS, - onForceSignal: () => { - this.logger.warn( - { timeoutMs: APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS }, - "Codex app-server did not exit after SIGTERM; sending SIGKILL", - ); - }, - }); - if (result === "kill-timeout") { - this.logger.warn( - { timeoutMs: APP_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS }, - "Codex app-server did not report exit after SIGKILL", - ); - } - } - - private async handleLine(line: string): Promise { - if (!line.trim()) return; - const raw: unknown = JSON.parse(line); - if (!isRecord(raw)) { - this.logger.warn({ line }, "Parsed JSON is not an object"); - return; - } - - if (isJsonRpcResponse(raw)) { - const id = raw.id; - if (raw.result !== undefined || raw.error) { - const pending = this.pending.get(id); - if (!pending) return; - clearTimeout(pending.timer); - this.pending.delete(id); - if (raw.error) { - pending.reject(new Error(raw.error.message ?? "Unknown error")); - } else { - pending.resolve(raw.result); - } - return; - } - - // Server-initiated request - if (isJsonRpcRequest(raw)) { - const request = raw; - const handler = this.requestHandlers.get(request.method); - try { - const result = handler ? await handler(request.params) : {}; - this.writeJsonRpcResponse({ id: request.id, result }); - } catch (error) { - this.writeJsonRpcResponse({ - id: request.id, - error: { message: error instanceof Error ? error.message : String(error) }, - }); - } - return; - } - } - - if (isJsonRpcNotification(raw)) { - this.notificationHandler?.(raw.method, raw.params); - } - } -} - function toAgentUsage(tokenUsage: unknown): AgentUsage | undefined { const usage = toObjectRecord(tokenUsage); if (!usage) return undefined; diff --git a/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts b/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts new file mode 100644 index 000000000..ae9012cc5 --- /dev/null +++ b/packages/server/src/server/agent/providers/codex/app-server-transport.test.ts @@ -0,0 +1,52 @@ +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import { EventEmitter } from "node:events"; +import { PassThrough } from "node:stream"; +import { describe, expect, test } from "vitest"; + +import { createTestLogger } from "../../../../test-utils/test-logger.js"; +import { CodexAppServerClient } from "./app-server-transport.js"; + +function createChildProcessStub(): ChildProcessWithoutNullStreams { + const child = new EventEmitter() as ChildProcessWithoutNullStreams; + child.stdin = new PassThrough() as ChildProcessWithoutNullStreams["stdin"]; + child.stdout = new PassThrough() as ChildProcessWithoutNullStreams["stdout"]; + child.stderr = new PassThrough() as ChildProcessWithoutNullStreams["stderr"]; + child.exitCode = null; + child.signalCode = null; + child.kill = (() => true) as ChildProcessWithoutNullStreams["kill"]; + return child; +} + +describe("Codex app-server transport", () => { + test("ignores non-JSON stdout lines without dropping pending requests", async () => { + const child = createChildProcessStub(); + const client = new CodexAppServerClient(child, createTestLogger()); + + const request = client.request("model/list", {}); + child.stdout.write("Codex ha iniciado en modo localizado\n"); + child.stdout.write('{"id":1,"result":{"data":[]}}\n'); + + await expect(request).resolves.toEqual({ data: [] }); + child.stdout.end(); + child.stderr.end(); + child.stdin.end(); + }); + + test("answers server-initiated requests through registered handlers", async () => { + const child = createChildProcessStub(); + const client = new CodexAppServerClient(child, createTestLogger()); + client.setRequestHandler("tool/requestUserInput", async (params) => ({ echoed: params })); + + const response = new Promise((resolve) => { + child.stdin.once("data", (chunk) => resolve(chunk.toString())); + }); + child.stdout.write( + '{"id":7,"method":"tool/requestUserInput","params":{"prompt":"Continue?"}}\n', + ); + + await expect(response).resolves.toBe('{"id":7,"result":{"echoed":{"prompt":"Continue?"}}}\n'); + child.stdout.end(); + child.stderr.end(); + child.stdin.end(); + }); +}); diff --git a/packages/server/src/server/agent/providers/codex/app-server-transport.ts b/packages/server/src/server/agent/providers/codex/app-server-transport.ts new file mode 100644 index 000000000..27ead31bd --- /dev/null +++ b/packages/server/src/server/agent/providers/codex/app-server-transport.ts @@ -0,0 +1,230 @@ +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import readline from "node:readline"; +import type { Logger } from "pino"; + +import { terminateWithTreeKill } from "../../../../utils/tree-kill.js"; + +const DEFAULT_TIMEOUT_MS = 14 * 24 * 60 * 60 * 1000; +const APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS = 2_000; +const APP_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS = 1_000; +const STDERR_BUFFER_LIMIT = 8192; + +interface JsonRpcRequest { + id: number; + method: string; + params?: unknown; +} + +interface JsonRpcResponse { + id: number; + result?: unknown; + error?: { message?: string }; +} + +interface JsonRpcNotification { + method: string; + params?: unknown; +} + +interface PendingRequest { + resolve: (value: unknown) => void; + reject: (error: Error) => void; + timer: NodeJS.Timeout; +} + +type RequestHandler = (params: unknown) => unknown; +type NotificationHandler = (method: string, params: unknown) => void; + +function isRecord(value: unknown): value is Record { + return value != null && typeof value === "object" && !Array.isArray(value); +} + +function isJsonRpcResponse(msg: unknown): msg is JsonRpcResponse { + if (!isRecord(msg)) return false; + return typeof msg.id === "number"; +} + +function isJsonRpcRequest(msg: unknown): msg is JsonRpcRequest { + if (!isRecord(msg)) return false; + return typeof msg.id === "number" && typeof msg.method === "string"; +} + +function isJsonRpcNotification(msg: unknown): msg is JsonRpcNotification { + if (!isRecord(msg)) return false; + return typeof msg.method === "string" && msg.id === undefined; +} + +export class CodexAppServerClient { + private readonly rl: readline.Interface; + private readonly pending = new Map(); + private readonly requestHandlers = new Map(); + private notificationHandler: NotificationHandler | null = null; + private nextId = 1; + private disposed = false; + private stderrBuffer = ""; + + constructor( + private readonly child: ChildProcessWithoutNullStreams, + private readonly logger: Logger, + ) { + this.rl = readline.createInterface({ input: child.stdout }); + this.rl.on("line", (line) => { + void this.handleLine(line).catch((error) => { + this.logger.warn({ error, line }, "Failed to handle Codex app-server stdout line"); + }); + }); + + child.stderr.on("data", (chunk) => { + this.stderrBuffer += chunk.toString(); + if (this.stderrBuffer.length > STDERR_BUFFER_LIMIT) { + this.stderrBuffer = this.stderrBuffer.slice(-STDERR_BUFFER_LIMIT); + } + }); + + child.on("error", (err) => { + this.logger.error({ err }, "Codex app-server child process error"); + for (const pending of this.pending.values()) { + clearTimeout(pending.timer); + pending.reject(err); + } + this.pending.clear(); + this.disposed = true; + }); + + child.on("exit", (code, signal) => { + const message = + code === 0 && !signal + ? "Codex app-server exited" + : `Codex app-server exited with code ${code ?? "null"} and signal ${signal ?? "null"}`; + const error = new Error(`${message}\n${this.stderrBuffer}`.trim()); + for (const pending of this.pending.values()) { + clearTimeout(pending.timer); + pending.reject(error); + } + this.pending.clear(); + this.disposed = true; + }); + } + + setNotificationHandler(handler: NotificationHandler): void { + this.notificationHandler = handler; + } + + setRequestHandler(method: string, handler: RequestHandler): void { + this.requestHandlers.set(method, handler); + } + + request(method: string, params?: unknown, timeoutMs = DEFAULT_TIMEOUT_MS): Promise { + if (this.disposed) { + return Promise.reject(new Error("Codex app-server client is closed")); + } + const id = this.nextId++; + const payload: JsonRpcRequest = { id, method, params }; + const serialized = JSON.stringify(payload); + this.child.stdin.write(`${serialized}\n`); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.pending.delete(id); + reject(new Error(`Codex app-server request timed out for ${method}`)); + }, timeoutMs); + this.pending.set(id, { resolve, reject, timer }); + }); + } + + notify(method: string, params?: unknown): void { + if (this.disposed) { + return; + } + const payload: JsonRpcNotification = { method, params }; + this.child.stdin.write(`${JSON.stringify(payload)}\n`); + } + + async dispose(): Promise { + if (this.disposed) return; + this.disposed = true; + this.rl.close(); + try { + this.child.stdin.end(); + } catch { + // ignore + } + const result = await terminateWithTreeKill(this.child, { + gracefulTimeoutMs: APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS, + forceTimeoutMs: APP_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS, + onForceSignal: () => { + this.logger.warn( + { timeoutMs: APP_SERVER_GRACEFUL_SHUTDOWN_TIMEOUT_MS }, + "Codex app-server did not exit after SIGTERM; sending SIGKILL", + ); + }, + }); + if (result === "kill-timeout") { + this.logger.warn( + { timeoutMs: APP_SERVER_FORCE_SHUTDOWN_TIMEOUT_MS }, + "Codex app-server did not report exit after SIGKILL", + ); + } + } + + private writeJsonRpcResponse(response: JsonRpcResponse): void { + if (this.disposed || this.child.stdin.destroyed || !this.child.stdin.writable) { + return; + } + try { + this.child.stdin.write(`${JSON.stringify(response)}\n`); + } catch (error) { + this.logger.debug({ error }, "Failed to write Codex app-server JSON-RPC response"); + } + } + + private async handleLine(line: string): Promise { + if (!line.trim()) return; + let raw: unknown; + try { + raw = JSON.parse(line); + } catch (error) { + this.logger.warn({ error, line }, "Ignoring non-JSON Codex app-server stdout line"); + return; + } + + if (!isRecord(raw)) { + this.logger.warn({ line }, "Parsed JSON is not an object"); + return; + } + + if (isJsonRpcResponse(raw)) { + const id = raw.id; + if (raw.result !== undefined || raw.error) { + const pending = this.pending.get(id); + if (!pending) return; + clearTimeout(pending.timer); + this.pending.delete(id); + if (raw.error) { + pending.reject(new Error(raw.error.message ?? "Unknown error")); + } else { + pending.resolve(raw.result); + } + return; + } + + if (isJsonRpcRequest(raw)) { + const request = raw; + const handler = this.requestHandlers.get(request.method); + try { + const result = handler ? await handler(request.params) : {}; + this.writeJsonRpcResponse({ id: request.id, result }); + } catch (error) { + this.writeJsonRpcResponse({ + id: request.id, + error: { message: error instanceof Error ? error.message : String(error) }, + }); + } + return; + } + } + + if (isJsonRpcNotification(raw)) { + this.notificationHandler?.(raw.method, raw.params); + } + } +}