From 10fdc4330d0f2e535c70c22b4b2b4bdcde64f68d Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sun, 10 May 2026 13:03:02 +0700 Subject: [PATCH] refactor(server): extract codex app-server test peer --- .../providers/codex-app-server-agent.test.ts | 234 ++++-------------- .../codex/app-server-transport.test.ts | 41 +-- .../codex/test-utils/test-app-server-peer.ts | 183 ++++++++++++++ 3 files changed, 246 insertions(+), 212 deletions(-) create mode 100644 packages/server/src/server/agent/providers/codex/test-utils/test-app-server-peer.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 3688410a9a..3e3c31b3de 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 @@ -1,11 +1,9 @@ import { describe, expect, test, vi } from "vitest"; import type { ChildProcessWithoutNullStreams } from "node:child_process"; -import { EventEmitter } from "node:events"; import { existsSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; import { mkdtemp } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; -import { PassThrough } from "node:stream"; import type { AgentLaunchContext, @@ -18,6 +16,10 @@ import { CodexAppServerAgentClient, codexAppServerTurnInputFromPrompt, } from "./codex-app-server-agent.js"; +import { + createCodexAppServerChildProcessStub, + TestCodexAppServerPeer, +} from "./codex/test-utils/test-app-server-peer.js"; import { createTestLogger } from "../../../test-utils/test-logger.js"; import { asInternals as castInternals, createStub } from "../../test-utils/class-mocks.js"; @@ -98,120 +100,32 @@ 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 waitForNextPermission( + session: AgentSession, + events: AgentStreamEvent[], +): Promise> { + const existing = events.find( + (event): event is Extract => + event.type === "permission_requested", + ); + if (existing) { + return Promise.resolve(existing); + } -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); + 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; } - } - - 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))); - } - } + clearTimeout(timeout); + unsubscribe(); + resolve(event); + }); }); - - 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", () => { @@ -277,21 +191,15 @@ describe("Codex app-server provider", () => { test("disposes an unresponsive app-server child with SIGKILL", async () => { vi.useFakeTimers(); - 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(() => true) as ChildProcessWithoutNullStreams["kill"]; + const child = createCodexAppServerChildProcessStub({ exitOnKill: false }); const client = new __codexAppServerInternals.CodexAppServerClient(child, createTestLogger()); try { const disposePromise = client.dispose(); - expect(child.kill).toHaveBeenCalledWith("SIGTERM"); + expect(child.killSignals).toEqual(["SIGTERM"]); await vi.advanceTimersByTimeAsync(2_000); - expect(child.kill).toHaveBeenCalledWith("SIGKILL"); + expect(child.killSignals).toEqual(["SIGTERM", "SIGKILL"]); await vi.advanceTimersByTimeAsync(1_000); await expect(disposePromise).resolves.toBeUndefined(); @@ -301,8 +209,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 peer = new TestCodexAppServerPeer({ initialize: () => ({}), "collaborationMode/list": () => ({ data: [] }), "skills/list": () => ({ data: [] }), @@ -311,7 +218,7 @@ describe("Codex app-server provider", () => { createConfig({ cwd: "/workspace/project" }), null, createTestLogger(), - async () => child, + async () => peer.child, ); const events: AgentStreamEvent[] = []; session.subscribe((event) => events.push(event)); @@ -319,45 +226,19 @@ describe("Codex app-server provider", () => { await session.connect(); peer.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, events); - 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`, + peer.writeRequest( + "item/commandExecution/requestApproval", + { + itemId: "exec-approval-1", + threadId: "thread-1", + turnId: "turn-1", + command: "git restore README.md", + cwd: "/workspace/project", + reason: "requires escalated permissions", + }, + 41, ); const permissionEvent = await permissionRequested; @@ -381,15 +262,7 @@ 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({ + await expect(peer.waitForResponse(41, { decision: "accept" })).resolves.toMatchObject({ id: 41, result: { decision: "accept" }, }); @@ -407,8 +280,12 @@ describe("Codex app-server provider", () => { path.join(repoSkillDir, "SKILL.md"), "---\nname: shipper\ndescription: Ship changes carefully.\n---\n", ); + const resolvedRepoRoots: string[] = []; const workspaceGitService = { - resolveRepoRoot: vi.fn().mockResolvedValue(path.join(tempDir, "repo")), + resolveRepoRoot: async (pathToResolve: string) => { + resolvedRepoRoots.push(pathToResolve); + return path.join(tempDir, "repo"); + }, }; try { @@ -419,7 +296,7 @@ describe("Codex app-server provider", () => { description: "Ship changes carefully.", argumentHint: "", }); - expect(workspaceGitService.resolveRepoRoot).toHaveBeenCalledWith(cwd); + expect(resolvedRepoRoots).toEqual([cwd]); } finally { rmSync(tempDir, { recursive: true, force: true }); } @@ -1913,14 +1790,7 @@ describe("Codex persisted sessions", () => { castInternals<{ spawnAppServer: () => Promise }>( provider, ).spawnAppServer = async () => { - const child = new EventEmitter() as ChildProcessWithoutNullStreams; - child.exitCode = 0; - child.signalCode = null; - child.stdin = new PassThrough(); - child.stdout = new PassThrough(); - child.stderr = new PassThrough(); - child.kill = vi.fn(() => true) as ChildProcessWithoutNullStreams["kill"]; - return child; + return createCodexAppServerChildProcessStub(); }; const descriptors = await provider.listPersistedAgents({ cwd: "/workspace/project-a" }); 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 2d72266bb8..741f12edfc 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,35 +1,20 @@ -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; -} +import { TestCodexAppServerPeer } from "./test-utils/test-app-server-peer.js"; 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 peer = new TestCodexAppServerPeer(); + const client = new CodexAppServerClient(peer.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'); + peer.writeNonJsonStdout("Codex ha iniciado en modo localizado"); + peer.writeResponse(1, { data: [] }); await expect(request).resolves.toEqual({ data: [] }); - child.stdout.end(); - child.stderr.end(); - child.stdin.end(); + peer.close(); }); test.each([ @@ -38,23 +23,19 @@ describe("Codex app-server transport", () => { "item/tool/requestUserInput", "tool/requestUserInput", ])("answers server-initiated %s requests through registered handlers", async (method) => { - const child = createChildProcessStub(); - const client = new CodexAppServerClient(child, createTestLogger()); + const peer = new TestCodexAppServerPeer(); + const client = new CodexAppServerClient(peer.child, createTestLogger()); const handlerCalls: unknown[] = []; client.setRequestHandler(method, async (params) => { handlerCalls.push(params); return { ok: true }; }); - const response = new Promise((resolve) => { - child.stdin.once("data", (chunk) => resolve(chunk.toString())); - }); - child.stdout.write(`${JSON.stringify({ jsonrpc: "2.0", id: 7, method, params: {} })}\n`); + const response = peer.nextPaseoOutputLine(); + peer.writeRequest(method); await expect(response).resolves.toBe('{"id":7,"result":{"ok":true}}\n'); expect(handlerCalls).toEqual([{}]); - child.stdout.end(); - child.stderr.end(); - child.stdin.end(); + peer.close(); }); }); diff --git a/packages/server/src/server/agent/providers/codex/test-utils/test-app-server-peer.ts b/packages/server/src/server/agent/providers/codex/test-utils/test-app-server-peer.ts new file mode 100644 index 0000000000..1d09947030 --- /dev/null +++ b/packages/server/src/server/agent/providers/codex/test-utils/test-app-server-peer.ts @@ -0,0 +1,183 @@ +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import { EventEmitter } from "node:events"; +import { PassThrough } from "node:stream"; +import { isDeepStrictEqual } from "node:util"; + +interface JsonRecord { + [key: string]: unknown; +} + +type RequestHandler = (params: unknown) => unknown; +type StubbedCodexAppServerChildProcess = ChildProcessWithoutNullStreams & { + stdin: PassThrough; + stdout: PassThrough; + stderr: PassThrough; + killSignals: Array; +}; + +function isJsonRecord(value: unknown): value is JsonRecord { + return value != null && typeof value === "object" && !Array.isArray(value); +} + +function parseJsonLine(line: string): JsonRecord | null { + const parsed: unknown = JSON.parse(line); + return isJsonRecord(parsed) ? parsed : null; +} + +export function createCodexAppServerChildProcessStub( + options: { exitOnKill?: boolean } = {}, +): StubbedCodexAppServerChildProcess { + const child = new EventEmitter() as StubbedCodexAppServerChildProcess; + child.stdin = new PassThrough(); + child.stdout = new PassThrough(); + child.stderr = new PassThrough(); + child.killSignals = []; + Object.defineProperty(child, "exitCode", { value: null, configurable: true }); + Object.defineProperty(child, "signalCode", { value: null, configurable: true }); + child.kill = ((signal?: NodeJS.Signals | number) => { + child.killSignals.push(signal ?? null); + if (options.exitOnKill !== false) { + queueMicrotask(() => child.emit("exit", null, signal ?? null)); + } + return true; + }) as ChildProcessWithoutNullStreams["kill"]; + return child; +} + +export class TestCodexAppServerPeer { + readonly child: StubbedCodexAppServerChildProcess; + private readonly messages: JsonRecord[] = []; + private readonly errors: Error[] = []; + private readonly waiters = new Set<{ + predicate: (message: JsonRecord) => boolean; + resolve: (message: JsonRecord) => void; + }>(); + private buffer = ""; + + constructor(handlers: Record = {}) { + this.child = createCodexAppServerChildProcessStub(); + this.child.stdin.on("data", (chunk) => { + this.acceptPaseoOutput(chunk.toString(), handlers); + }); + } + + writeNonJsonStdout(line: string): void { + this.child.stdout.write(`${line}\n`); + } + + writeResponse(id: number, result: unknown): void { + this.child.stdout.write(`${JSON.stringify({ id, result })}\n`); + } + + writeRequest(method: string, params: unknown = {}, id = 7): void { + this.child.stdout.write(`${JSON.stringify({ jsonrpc: "2.0", id, method, params })}\n`); + } + + nextPaseoOutputLine(): Promise { + return new Promise((resolve) => { + this.child.stdin.once("data", (chunk) => resolve(chunk.toString())); + }); + } + + async waitForResponse(id: number, result: unknown): Promise { + return this.waitForMessage( + (message) => + message.id === id && !("method" in message) && isDeepStrictEqual(message.result, result), + `response ${id}`, + ); + } + + assertNoErrors(): void { + if (this.errors.length > 0) { + throw this.errors[0]; + } + } + + close(): void { + this.child.stdout.end(); + this.child.stderr.end(); + this.child.stdin.end(); + } + + private acceptPaseoOutput(chunk: string, handlers: Record): void { + this.buffer += chunk; + for (;;) { + const newlineIndex = this.buffer.indexOf("\n"); + if (newlineIndex === -1) { + break; + } + const line = this.buffer.slice(0, newlineIndex).trim(); + this.buffer = this.buffer.slice(newlineIndex + 1); + if (!line) { + continue; + } + try { + const message = parseJsonLine(line); + if (message) { + this.processPaseoMessage(message, handlers); + } + } catch (error) { + this.errors.push(error instanceof Error ? error : new Error(String(error))); + } + } + } + + private processPaseoMessage(message: JsonRecord, handlers: Record): void { + this.messages.push(message); + for (const waiter of Array.from(this.waiters)) { + if (waiter.predicate(message)) { + this.waiters.delete(waiter); + waiter.resolve(message); + } + } + + if (typeof message.id !== "number" || typeof message.method !== "string") { + return; + } + + const handler = handlers[message.method]; + if (!handler) { + this.errors.push(new Error(`Unexpected Codex app-server request: ${message.method}`)); + return; + } + + Promise.resolve(handler(message.params)) + .then((result) => { + this.writeResponse(message.id as number, result); + return undefined; + }) + .catch((error) => { + this.child.stdout.write( + `${JSON.stringify({ + id: message.id, + error: { message: error instanceof Error ? error.message : String(error) }, + })}\n`, + ); + return undefined; + }); + } + + private waitForMessage( + predicate: (message: JsonRecord) => boolean, + label: string, + ): Promise { + const existing = this.messages.find(predicate); + if (existing) { + return Promise.resolve(existing); + } + return new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + this.waiters.delete(waiter); + reject(new Error(`Timed out waiting for ${label}`)); + }, 1000); + const waiter = { + predicate, + resolve: (message: JsonRecord) => { + clearTimeout(timeout); + resolve(message); + }, + }; + this.waiters.add(waiter); + }); + } +}