Merge branch 'main' into refactor-test-iter12

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 {
+function waitForNextPermission(
-  const child = new EventEmitter() as ChildProcessWithoutNullStreams;
+  session: AgentSession,
-  child.stdin = new PassThrough() as ChildProcessWithoutNullStreams["stdin"];
+  events: AgentStreamEvent[],
-  child.stdout = new PassThrough() as ChildProcessWithoutNullStreams["stdout"];
+): Promise<Extract<AgentStreamEvent, { type: "permission_requested" }>> {
-  child.stderr = new PassThrough() as ChildProcessWithoutNullStreams["stderr"];
+  const existing = events.find(
-  child.exitCode = null;
+    (event): event is Extract<AgentStreamEvent, { type: "permission_requested" }> =>
-  child.signalCode = null;
+      event.type === "permission_requested",
-  child.kill = vi.fn((signal?: NodeJS.Signals | number) => {
+  );
-    queueMicrotask(() => child.emit("exit", null, signal ?? null));
+  if (existing) {
-    return true;
+    return Promise.resolve(existing);
-  }) as ChildProcessWithoutNullStreams["kill"];
+  }
-  return child;
-}
 
-function createScriptedCodexPeer(
+  return new Promise((resolve, reject) => {
-  child: ChildProcessWithoutNullStreams,
+    const timeout = setTimeout(() => {
-  handlers: Record<string, (params: unknown) => unknown>,
+      unsubscribe();
-) {
+      reject(new Error("Timed out waiting for permission_requested"));
-  const messages: Record<string, unknown>[] = [];
+    }, 1000);
-  const errors: Error[] = [];
+    const unsubscribe = session.subscribe((event) => {
-  const waiters = new Set<{
+      if (event.type !== "permission_requested") {
-    predicate: (message: Record<string, unknown>) => boolean;
+        return;
-    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);
       }
-    }
+      clearTimeout(timeout);
-
+      unsubscribe();
-    if (typeof message.id !== "number" || typeof message.method !== "string") {
+      resolve(event);
-      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", () => {
@@ -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;
+    const child = createCodexAppServerChildProcessStub({ exitOnKill: false });
-    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 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 = new TestCodexAppServerPeer({
-    const peer = createScriptedCodexPeer(child, {
       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<
+    const permissionRequested = waitForNextPermission(session, events);
-      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);
-      });
-    });
 
-    child.stdout.write(
+    peer.writeRequest(
-      `${JSON.stringify({
+      "item/commandExecution/requestApproval",
-        jsonrpc: "2.0",
+      {
-        id: 41,
+        itemId: "exec-approval-1",
-        method: "item/commandExecution/requestApproval",
+        threadId: "thread-1",
-        params: {
+        turnId: "turn-1",
-          itemId: "exec-approval-1",
+        command: "git restore README.md",
-          threadId: "thread-1",
+        cwd: "/workspace/project",
-          turnId: "turn-1",
+        reason: "requires escalated permissions",
-          command: "git restore README.md",
+      },
-          cwd: "/workspace/project",
+      41,
-          reason: "requires escalated permissions",
-        },
-      })}\n`,
     );
 
     const permissionEvent = await permissionRequested;
@@ -381,15 +262,7 @@ describe("Codex app-server provider", () => {
 
     await session.respondToPermission(permissionEvent.request.id, { behavior: "allow" });
 
-    await expect(
+    await expect(peer.waitForResponse(41, { decision: "accept" })).resolves.toMatchObject({
-      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" },
     });
@@ -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<ChildProcessWithoutNullStreams> }>(
       provider,
     ).spawnAppServer = async () => {
-      const child = new EventEmitter() as ChildProcessWithoutNullStreams;
+      return createCodexAppServerChildProcessStub();
-      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;
     };
 
     const descriptors = await provider.listPersistedAgents({ cwd: "/workspace/project-a" });

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";
-
+import { TestCodexAppServerPeer } from "./test-utils/test-app-server-peer.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 peer = new TestCodexAppServerPeer();
-    const client = new CodexAppServerClient(child, createTestLogger());
+    const client = new CodexAppServerClient(peer.child, createTestLogger());
 
     const request = client.request("model/list", {});
-    child.stdout.write("Codex ha iniciado en modo localizado\n");
+    peer.writeNonJsonStdout("Codex ha iniciado en modo localizado");
-    child.stdout.write('{"id":1,"result":{"data":[]}}\n');
+    peer.writeResponse(1, { data: [] });
 
     await expect(request).resolves.toEqual({ data: [] });
-    child.stdout.end();
+    peer.close();
-    child.stderr.end();
-    child.stdin.end();
   });
 
   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 peer = new TestCodexAppServerPeer();
-    const client = new CodexAppServerClient(child, createTestLogger());
+    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<string>((resolve) => {
+    const response = peer.nextPaseoOutputLine();
-      child.stdin.once("data", (chunk) => resolve(chunk.toString()));
+    peer.writeRequest(method);
-    });
-    child.stdout.write(`${JSON.stringify({ jsonrpc: "2.0", id: 7, method, params: {} })}\n`);
 
     await expect(response).resolves.toBe('{"id":7,"result":{"ok":true}}\n');
     expect(handlerCalls).toEqual([{}]);
-    child.stdout.end();
+    peer.close();
-    child.stderr.end();
-    child.stdin.end();
   });
 });

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<NodeJS.Signals | number | null>;
+};
+
+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<string, RequestHandler> = {}) {
+    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<string> {
+    return new Promise((resolve) => {
+      this.child.stdin.once("data", (chunk) => resolve(chunk.toString()));
+    });
+  }
+
+  async waitForResponse(id: number, result: unknown): Promise<JsonRecord> {
+    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<string, RequestHandler>): 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<string, RequestHandler>): 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<JsonRecord> {
+    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);
+    });
+  }
+}