jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(server): extract Codex app-server transport (#866)

Browse Source
This commit is contained in:
Mohamed Boudra
2026-05-10 12:45:13 +08:00
committed by GitHub
parent 15e15b525f
commit 4165dbe8e9
3 changed files with 283 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/server/agent/providers/codex-app-server-agent.ts
+1 -212
View File
@@ -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<string, unknown> {
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<string, unknown> | undefined {
return isRecord(value) ? value : undefined;
}
interface PendingRequest {
resolve: (value: unknown) => void;
reject: (error: Error) => void;
timer: ReturnType<typeof setTimeout>;
}
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<number, PendingRequest>();
private readonly requestHandlers = new Map<string, RequestHandler>();
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<unknown> {
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<void> {
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<void> {
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;
packages/server/src/server/agent/providers/codex/app-server-transport.test.ts
+52
View File
@@ -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<string>((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();
});
});
packages/server/src/server/agent/providers/codex/app-server-transport.ts
+230
View File
@@ -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<string, unknown> {
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<number, PendingRequest>();
private readonly requestHandlers = new Map<string, RequestHandler>();
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<unknown> {
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<void> {
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<void> {
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);
}
}
}