From 6220b47073b5185aca78311412863bc0166b2463 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 10 May 2026 17:01:50 +0800 Subject: [PATCH] refactor(server): extract codex app-server fake (#873) --- .../providers/codex-app-server-agent.test.ts | 191 ++---------------- .../codex/app-server-transport.test.ts | 19 +- .../codex/test-utils/fake-app-server.ts | 191 ++++++++++++++++++ 3 files changed, 212 insertions(+), 189 deletions(-) create mode 100644 packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 3688410a9..79fb5b9b2 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -18,6 +18,10 @@ import { CodexAppServerAgentClient, codexAppServerTurnInputFromPrompt, } from "./codex-app-server-agent.js"; +import { + createFakeCodexAppServer, + waitForNextPermission, +} from "./codex/test-utils/fake-app-server.js"; import { createTestLogger } from "../../../test-utils/test-logger.js"; import { asInternals as castInternals, createStub } from "../../test-utils/class-mocks.js"; @@ -98,122 +102,6 @@ function markdownImageSource(markdown: string): string { return match[1].replace(/\\\)/g, ")"); } -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 = vi.fn((signal?: NodeJS.Signals | number) => { - queueMicrotask(() => child.emit("exit", null, signal ?? null)); - return true; - }) as ChildProcessWithoutNullStreams["kill"]; - return child; -} - -function createScriptedCodexPeer( - child: ChildProcessWithoutNullStreams, - handlers: Record unknown>, -) { - const messages: Record[] = []; - const errors: Error[] = []; - const waiters = new Set<{ - predicate: (message: Record) => boolean; - resolve: (message: Record) => void; - }>(); - let buffer = ""; - - const processMessage = (message: Record) => { - messages.push(message); - for (const waiter of Array.from(waiters)) { - if (waiter.predicate(message)) { - waiters.delete(waiter); - waiter.resolve(message); - } - } - - if (typeof message.id !== "number" || typeof message.method !== "string") { - return; - } - - const handler = handlers[message.method]; - if (!handler) { - errors.push(new Error(`Unexpected Codex app-server request: ${message.method}`)); - return; - } - - Promise.resolve(handler(message.params)) - .then((result) => { - child.stdout.write(`${JSON.stringify({ id: message.id, result })}\n`); - return undefined; - }) - .catch((error) => { - child.stdout.write( - `${JSON.stringify({ - id: message.id, - error: { message: error instanceof Error ? error.message : String(error) }, - })}\n`, - ); - return undefined; - }); - }; - - child.stdin.on("data", (chunk) => { - buffer += chunk.toString(); - for (;;) { - const newlineIndex = buffer.indexOf("\n"); - if (newlineIndex === -1) { - break; - } - const line = buffer.slice(0, newlineIndex).trim(); - buffer = buffer.slice(newlineIndex + 1); - if (!line) { - continue; - } - try { - const parsed: unknown = JSON.parse(line); - if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) { - processMessage(parsed as Record); - } - } catch (error) { - errors.push(error instanceof Error ? error : new Error(String(error))); - } - } - }); - - return { - assertNoErrors() { - if (errors.length > 0) { - throw errors[0]; - } - }, - waitForMessage( - predicate: (message: Record) => boolean, - label: string, - ): Promise> { - const existing = messages.find(predicate); - if (existing) { - return Promise.resolve(existing); - } - return new Promise((resolve, reject) => { - const timeout = setTimeout(() => { - waiters.delete(waiter); - reject(new Error(`Timed out waiting for ${label}`)); - }, 1000); - const waiter = { - predicate, - resolve: (message: Record) => { - clearTimeout(timeout); - resolve(message); - }, - }; - waiters.add(waiter); - }); - }, - }; -} - describe("Codex app-server provider", () => { test("passes ephemeral: true to thread/start when constructed as ephemeral", async () => { const requests: Array<{ method: string; params: unknown }> = []; @@ -301,8 +189,7 @@ describe("Codex app-server provider", () => { }); test("round-trips server-initiated command approvals through the real app-server transport", async () => { - const child = createChildProcessStub(); - const peer = createScriptedCodexPeer(child, { + const appServer = createFakeCodexAppServer({ initialize: () => ({}), "collaborationMode/list": () => ({ data: [] }), "skills/list": () => ({ data: [] }), @@ -311,55 +198,22 @@ describe("Codex app-server provider", () => { createConfig({ cwd: "/workspace/project" }), null, createTestLogger(), - async () => child, + async () => appServer.child, ); - const events: AgentStreamEvent[] = []; - session.subscribe((event) => events.push(event)); await session.connect(); - peer.assertNoErrors(); + appServer.assertNoErrors(); - const permissionRequested = new Promise< - Extract - >((resolve, reject) => { - const existing = events.find( - (event): event is Extract => - event.type === "permission_requested", - ); - if (existing) { - resolve(existing); - return; - } - const timeout = setTimeout(() => { - unsubscribe(); - reject(new Error("Timed out waiting for permission_requested")); - }, 1000); - const unsubscribe = session.subscribe((event) => { - if (event.type !== "permission_requested") { - return; - } - clearTimeout(timeout); - unsubscribe(); - resolve(event); - }); + const permissionRequested = waitForNextPermission(session); + appServer.requestCommandApproval({ + itemId: "exec-approval-1", + threadId: "thread-1", + turnId: "turn-1", + command: "git restore README.md", + cwd: "/workspace/project", + reason: "requires escalated permissions", }); - child.stdout.write( - `${JSON.stringify({ - jsonrpc: "2.0", - id: 41, - method: "item/commandExecution/requestApproval", - params: { - itemId: "exec-approval-1", - threadId: "thread-1", - turnId: "turn-1", - command: "git restore README.md", - cwd: "/workspace/project", - reason: "requires escalated permissions", - }, - })}\n`, - ); - const permissionEvent = await permissionRequested; expect(permissionEvent.request).toMatchObject({ id: "permission-exec-approval-1", @@ -381,19 +235,10 @@ describe("Codex app-server provider", () => { await session.respondToPermission(permissionEvent.request.id, { behavior: "allow" }); - await expect( - peer.waitForMessage( - (message) => - message.id === 41 && - !("method" in message) && - JSON.stringify(message.result) === JSON.stringify({ decision: "accept" }), - "command approval response", - ), - ).resolves.toMatchObject({ - id: 41, - result: { decision: "accept" }, + await expect(appServer.waitForCommandApprovalDecision("exec-approval-1")).resolves.toEqual({ + decision: "accept", }); - peer.assertNoErrors(); + appServer.assertNoErrors(); await session.close(); }); 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 index 2d72266bb..d85db22a7 100644 --- 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 @@ -1,25 +1,12 @@ -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 { createCodexAppServerChildProcess } from "./test-utils/fake-app-server.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 child = createCodexAppServerChildProcess(); const client = new CodexAppServerClient(child, createTestLogger()); const request = client.request("model/list", {}); @@ -38,7 +25,7 @@ describe("Codex app-server transport", () => { "item/tool/requestUserInput", "tool/requestUserInput", ])("answers server-initiated %s requests through registered handlers", async (method) => { - const child = createChildProcessStub(); + const child = createCodexAppServerChildProcess(); const client = new CodexAppServerClient(child, createTestLogger()); const handlerCalls: unknown[] = []; client.setRequestHandler(method, async (params) => { diff --git a/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts new file mode 100644 index 000000000..471a00f7e --- /dev/null +++ b/packages/server/src/server/agent/providers/codex/test-utils/fake-app-server.ts @@ -0,0 +1,191 @@ +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import { EventEmitter } from "node:events"; +import { PassThrough } from "node:stream"; + +import type { AgentSession, AgentStreamEvent } from "../../../agent-sdk-types.js"; + +type JsonObject = Record; +type CodexAppServerChildProcess = ChildProcessWithoutNullStreams & { + stdin: PassThrough; + stdout: PassThrough; + stderr: PassThrough; +}; + +export interface FakeCodexAppServer { + readonly child: CodexAppServerChildProcess; + assertNoErrors(): void; + requestCommandApproval(params: { + itemId: string; + threadId: string; + turnId: string; + command: string; + cwd: string; + reason: string; + }): void; + waitForCommandApprovalDecision(itemId: string): Promise; +} + +export function createCodexAppServerChildProcess(): CodexAppServerChildProcess { + const child = Object.assign(new EventEmitter(), { + stdin: new PassThrough(), + stdout: new PassThrough(), + stderr: new PassThrough(), + exitCode: null, + signalCode: null, + }) as CodexAppServerChildProcess; + child.kill = ((signal?: NodeJS.Signals | number) => { + queueMicrotask(() => child.emit("exit", null, signal ?? null)); + return true; + }) as ChildProcessWithoutNullStreams["kill"]; + return child; +} + +export function createFakeCodexAppServer( + handlers: Record unknown>, +): FakeCodexAppServer { + const child = createCodexAppServerChildProcess(); + const messages: JsonObject[] = []; + const errors: Error[] = []; + const approvalRequestIds = new Map(); + const waiters = new Set<{ + predicate: (message: JsonObject) => boolean; + resolve: (message: JsonObject) => void; + }>(); + let buffer = ""; + let nextServerRequestId = 1; + + function processMessage(message: JsonObject): void { + messages.push(message); + for (const waiter of Array.from(waiters)) { + if (waiter.predicate(message)) { + waiters.delete(waiter); + waiter.resolve(message); + } + } + + if (typeof message.id !== "number" || typeof message.method !== "string") { + return; + } + + const handler = handlers[message.method]; + if (!handler) { + errors.push(new Error(`Unexpected Codex app-server request: ${message.method}`)); + return; + } + + Promise.resolve(handler(message.params)) + .then((result) => { + child.stdout.write(`${JSON.stringify({ id: message.id, result })}\n`); + return undefined; + }) + .catch((error) => { + child.stdout.write( + `${JSON.stringify({ + id: message.id, + error: { message: error instanceof Error ? error.message : String(error) }, + })}\n`, + ); + return undefined; + }); + } + + child.stdin.on("data", (chunk) => { + buffer += chunk.toString(); + for (;;) { + const newlineIndex = buffer.indexOf("\n"); + if (newlineIndex === -1) { + break; + } + const line = buffer.slice(0, newlineIndex).trim(); + buffer = buffer.slice(newlineIndex + 1); + if (!line) { + continue; + } + try { + const parsed: unknown = JSON.parse(line); + if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) { + processMessage(parsed as JsonObject); + } + } catch (error) { + errors.push(error instanceof Error ? error : new Error(String(error))); + } + } + }); + + function waitForMessage( + predicate: (message: JsonObject) => boolean, + label: string, + ): Promise { + const existing = messages.find(predicate); + if (existing) { + return Promise.resolve(existing); + } + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + waiters.delete(waiter); + reject(new Error(`Timed out waiting for ${label}`)); + }, 1000); + const waiter = { + predicate, + resolve: (message: JsonObject) => { + clearTimeout(timeout); + resolve(message); + }, + }; + waiters.add(waiter); + }); + } + + return { + child, + assertNoErrors() { + if (errors.length > 0) { + throw errors[0]; + } + }, + requestCommandApproval(params) { + const requestId = nextServerRequestId; + nextServerRequestId += 1; + approvalRequestIds.set(params.itemId, requestId); + child.stdout.write( + `${JSON.stringify({ + jsonrpc: "2.0", + id: requestId, + method: "item/commandExecution/requestApproval", + params, + })}\n`, + ); + }, + async waitForCommandApprovalDecision(itemId) { + const requestId = approvalRequestIds.get(itemId); + if (requestId === undefined) { + throw new Error(`No pending fake Codex app-server approval for ${itemId}`); + } + const message = await waitForMessage( + (candidate) => + candidate.id === requestId && !("method" in candidate) && "result" in candidate, + "command approval response", + ); + return message.result; + }, + }; +} + +export function waitForNextPermission( + session: AgentSession, +): Promise> { + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + unsubscribe(); + reject(new Error("Timed out waiting for permission_requested")); + }, 1000); + const unsubscribe = session.subscribe((event) => { + if (event.type !== "permission_requested") { + return; + } + clearTimeout(timeout); + unsubscribe(); + resolve(event); + }); + }); +}