refactor(server): extract codex app-server fake (#873)

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<string, (params: unknown) => unknown>,
-) {
-  const messages: Record<string, unknown>[] = [];
-  const errors: Error[] = [];
-  const waiters = new Set<{
-    predicate: (message: Record<string, unknown>) => boolean;
-    resolve: (message: Record<string, unknown>) => void;
-  }>();
-  let buffer = "";
-
-  const processMessage = (message: Record<string, unknown>) => {
-    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<string, unknown>);
-        }
-      } 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<string, unknown>) => boolean,
-      label: string,
-    ): Promise<Record<string, unknown>> {
-      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<string, unknown>) => {
-            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<AgentStreamEvent, { type: "permission_requested" }>
-    >((resolve, reject) => {
-      const existing = events.find(
-        (event): event is Extract<AgentStreamEvent, { type: "permission_requested" }> =>
-          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();
   });
 

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) => {

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<string, unknown>;
+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<unknown>;
+}
+
+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<string, (params: unknown) => unknown>,
+): FakeCodexAppServer {
+  const child = createCodexAppServerChildProcess();
+  const messages: JsonObject[] = [];
+  const errors: Error[] = [];
+  const approvalRequestIds = new Map<string, number>();
+  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<JsonObject> {
+    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<Extract<AgentStreamEvent, { type: "permission_requested" }>> {
+  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);
+    });
+  });
+}