fix(evaluation): join owned command cleanup before removing files (#5613)

Use shared owned command supervision for all evaluation helpers, join streamed output, and retain/retry incomplete cleanup before deleting run files.

Fixes #5094.

scripts/lib/agent-evaluation-commands.mjs

@@ -0,0 +1,193 @@
+import { randomUUID } from "node:crypto";
+import { createInterface } from "node:readline";
+import { PassThrough, Writable } from "node:stream";
+
+import {
+  OwnedCommandFailure,
+  runOwnedCommand,
+} from "../../dist/process/OwnedCommand.js";
+
+/** Keep command ownership through failure and join cleanup before removing a run. */
+export function createAgentEvaluationCommands({
+  codex,
+  evaluationRoot,
+  evaluationEnvironment,
+  optionalModel,
+  timeoutMs,
+  repositoryRoot,
+}) {
+  const pendingCleanup = new Map();
+  const diagnosticBytes = 32_768;
+
+  async function execute(
+    command,
+    arguments_,
+    cwd,
+    env,
+    limits,
+    timeoutMessage,
+  ) {
+    try {
+      return await runOwnedCommand(
+        {
+          command,
+          arguments: arguments_,
+          // CLI shims may execute through a shebang interpreter. Ownership
+          // still uses the launch generation and token, not an argv prefix.
+          expectedCommand: null,
+          cwd,
+          hostEnvironment: env,
+          runId: randomUUID(),
+          sweepTokenOwnedProcesses: true,
+        },
+        { diagnosticBytes, acceptNonZeroExit: true, ...limits },
+      );
+    } catch (cause) {
+      if (cause instanceof OwnedCommandFailure) {
+        if (cause.cleanupOwner !== undefined)
+          pendingCleanup.set(cause.cleanupOwner, cause);
+        if (cause.reason === "timeout")
+          throw new Error(timeoutMessage, { cause });
+      }
+      throw cause;
+    }
+  }
+
+  return {
+    async runCodex(prompt, approvedTools) {
+      const arguments_ = [
+        "--no-daemon",
+        "exec",
+        "--ephemeral",
+        "--skip-git-repo-check",
+        "--sandbox",
+        "read-only",
+        "--json",
+        "--color",
+        "never",
+        "-C",
+        evaluationRoot,
+        "-c",
+        'approval_policy="never"',
+        ...approvedTools.flatMap((name) => [
+          "-c",
+          `mcp_servers.rea.tools.${name}.approval_mode="approve"`,
+        ]),
+        ...(optionalModel === undefined ? [] : ["--model", optionalModel]),
+        prompt,
+      ];
+      const events = [];
+      const nonJson = [];
+      let nonJsonBytes = 0;
+      const stdout = new PassThrough();
+      const lines = createInterface({ input: stdout });
+      const linesClosed = new Promise((resolve) =>
+        lines.once("close", resolve),
+      );
+      let lineError;
+      lines.on("error", (cause) => {
+        // The owned pipeline reports stream failure or cancellation. Join this
+        // consumer too, without turning its forwarded error into a process crash.
+        lineError = cause;
+        lines.close();
+      });
+      lines.on("line", (line) => {
+        try {
+          events.push(JSON.parse(line));
+        } catch {
+          if (nonJsonBytes >= diagnosticBytes) return;
+          const bytes = Buffer.from(
+            `non-json stdout: ${line.slice(0, diagnosticBytes - nonJsonBytes)}`,
+          );
+          const retained = bytes.subarray(0, diagnosticBytes - nonJsonBytes);
+          if (retained.length > 0) nonJson.push(retained);
+          nonJsonBytes += retained.length;
+        }
+      });
+      try {
+        const result = await execute(
+          codex,
+          arguments_,
+          evaluationRoot,
+          evaluationEnvironment,
+          { timeoutMs, stdout },
+          `Codex evaluation timed out after ${String(timeoutMs)}ms`,
+        );
+        await linesClosed;
+        if (lineError !== undefined) throw lineError;
+        return {
+          events,
+          exitCode: result.exitCode ?? 1,
+          stderr: Buffer.concat([...nonJson, Buffer.from(result.stderr.text)])
+            .subarray(0, diagnosticBytes)
+            .toString("utf8"),
+        };
+      } finally {
+        stdout.destroy();
+        lines.close();
+        await linesClosed;
+      }
+    },
+
+    async runProcess(command, arguments_, cwd, timeout, env = process.env) {
+      const result = await execute(
+        command,
+        arguments_,
+        cwd,
+        env,
+        {
+          timeoutMs: timeout,
+          stdout: new Writable({ write: (_chunk, _encoding, done) => done() }),
+        },
+        `${command} timed out`,
+      );
+      if (result.exitCode !== 0)
+        throw new Error(
+          `${command} failed with ${String(result.exitCode)}: ${result.stderr.text}`,
+        );
+    },
+
+    async readCommandVersion(command) {
+      const result = await execute(
+        command,
+        ["--version"],
+        repositoryRoot,
+        process.env,
+        { timeoutMs: 10_000, diagnosticBytes: 8192 },
+        `${command} --version timed out`,
+      );
+      if (result.exitCode !== 0)
+        throw new Error(
+          `${command} --version failed with ${String(result.exitCode)}: ${result.stderr.text}`,
+        );
+      const version = result.stdout.text.trim();
+      if (version.length === 0)
+        throw new Error(`${command} --version returned empty output`);
+      return version;
+    },
+
+    async close() {
+      const failures = [];
+      for (const [owner, failure] of pendingCleanup) {
+        try {
+          const result = await owner.stop();
+          if (result.status === "incomplete")
+            failures.push(new Error(result.reason, { cause: failure }));
+          else pendingCleanup.delete(owner);
+        } catch (cause) {
+          failures.push(
+            new AggregateError(
+              [failure, cause],
+              "Evaluation process cleanup failed",
+            ),
+          );
+        }
+      }
+      if (failures.length > 0)
+        throw new AggregateError(
+          failures,
+          "Evaluation processes remain owned; preserve the run directory and retry cleanup through the retained owners",
+        );
+    },
+  };
+}

scripts/verify-agent-experience.mjs

@@ -1,4 +1,3 @@
-import { spawn } from "node:child_process";
 import { createHash } from "node:crypto";
 import {
   chmod,
@@ -11,7 +10,6 @@ import {
 } from "node:fs/promises";
 import { homedir } from "node:os";
 import { delimiter, dirname, join, resolve } from "node:path";
-import { createInterface } from "node:readline";
 import { fileURLToPath } from "node:url";
 import { z } from "zod";
 
@@ -26,6 +24,7 @@ import {
   summarizeFactualCorrectness,
 } from "../dist/evaluation/AgentEvaluationReport.js";
 import { PRODUCT_IDENTITY } from "../dist/identity.js";
+import { createAgentEvaluationCommands } from "./lib/agent-evaluation-commands.mjs";
 import { closeVerifierResources } from "./lib/verifier-cleanup.mjs";
 import { completeVerifierRun, createVerifierRun } from "./lib/verifier-run.mjs";
 
@@ -69,6 +68,16 @@ const evaluationEnvironment = {
   PATH: `${dirname(process.execPath)}${delimiter}${process.env.PATH ?? ""}`,
 };
 
+const commands = createAgentEvaluationCommands({
+  codex,
+  evaluationRoot,
+  evaluationEnvironment,
+  optionalModel,
+  timeoutMs,
+  repositoryRoot,
+});
+const { runCodex, runProcess, readCommandVersion } = commands;
+
 let primaryError;
 try {
   const codexVersion = await readCommandVersion(codex);
@@ -400,10 +409,16 @@ try {
   // Even a deliberately retained fixture directory must never retain auth.
   await closeVerifierResources(
     [
-      () => rm(authPath, { force: true }),
-      ...(process.env.REA_AGENT_EVAL_KEEP_FIXTURES === "true"
-        ? []
-        : [() => rm(evaluationRoot, { recursive: true, force: true })]),
+      async () => {
+        // A failed process cleanup retains both its exact owner and its files.
+        await commands.close();
+        await closeVerifierResources([
+          () => rm(authPath, { force: true }),
+          ...(process.env.REA_AGENT_EVAL_KEEP_FIXTURES === "true"
+            ? []
+            : [() => rm(evaluationRoot, { recursive: true, force: true })]),
+        ]);
+      },
     ],
     primaryError,
   );
@@ -487,137 +502,3 @@ function validTargetClarification(message) {
     throw cause;
   }
 }
-
-async function runCodex(prompt, approvedTools) {
-  const arguments_ = [
-    "--no-daemon",
-    "exec",
-    "--ephemeral",
-    "--skip-git-repo-check",
-    "--sandbox",
-    "read-only",
-    "--json",
-    "--color",
-    "never",
-    "-C",
-    evaluationRoot,
-    "-c",
-    'approval_policy="never"',
-    ...approvedTools.flatMap((name) => [
-      "-c",
-      `mcp_servers.rea.tools.${name}.approval_mode="approve"`,
-    ]),
-    ...(optionalModel === undefined ? [] : ["--model", optionalModel]),
-    prompt,
-  ];
-  const child = spawn(codex, arguments_, {
-    cwd: evaluationRoot,
-    env: evaluationEnvironment,
-    stdio: ["ignore", "pipe", "pipe"],
-  });
-  const events = [];
-  const stderr = [];
-  const lines = createInterface({ input: child.stdout });
-  const linesClosed = new Promise((resolveClose) =>
-    lines.once("close", resolveClose),
-  );
-  lines.on("line", (line) => {
-    try {
-      events.push(JSON.parse(line));
-    } catch {
-      stderr.push(`non-json stdout: ${line}`);
-    }
-  });
-  child.stderr.on("data", (chunk) => {
-    if (stderr.join("").length < 32_768) stderr.push(chunk.toString("utf8"));
-  });
-  const exitCode = await new Promise((resolveExit, reject) => {
-    const timeout = setTimeout(() => {
-      child.kill("SIGTERM");
-      reject(
-        new Error(`Codex evaluation timed out after ${String(timeoutMs)}ms`),
-      );
-    }, timeoutMs);
-    child.once("error", (cause) => {
-      clearTimeout(timeout);
-      reject(cause);
-    });
-    child.once("exit", (code) => {
-      clearTimeout(timeout);
-      resolveExit(code ?? 1);
-    });
-  });
-  await linesClosed;
-  return { events, exitCode, stderr: stderr.join("").slice(0, 32_768) };
-}
-
-async function runProcess(
-  command,
-  arguments_,
-  cwd,
-  timeout,
-  env = process.env,
-) {
-  const child = spawn(command, arguments_, {
-    cwd,
-    env,
-    stdio: ["ignore", "ignore", "pipe"],
-  });
-  let stderr = "";
-  child.stderr.on("data", (chunk) => {
-    if (stderr.length < 32_768) stderr += chunk.toString("utf8");
-  });
-  const code = await new Promise((resolveExit, reject) => {
-    const timer = setTimeout(() => {
-      child.kill("SIGTERM");
-      reject(new Error(`${command} timed out`));
-    }, timeout);
-    child.once("error", (cause) => {
-      clearTimeout(timer);
-      reject(cause);
-    });
-    child.once("exit", (value) => {
-      clearTimeout(timer);
-      resolveExit(value ?? 1);
-    });
-  });
-  if (code !== 0)
-    throw new Error(`${command} failed with ${String(code)}: ${stderr}`);
-}
-
-async function readCommandVersion(command) {
-  const child = spawn(command, ["--version"], {
-    cwd: repositoryRoot,
-    stdio: ["ignore", "pipe", "pipe"],
-  });
-  let stdout = "";
-  let stderr = "";
-  child.stdout.on("data", (chunk) => {
-    if (stdout.length < 4096) stdout += chunk.toString("utf8");
-  });
-  child.stderr.on("data", (chunk) => {
-    if (stderr.length < 4096) stderr += chunk.toString("utf8");
-  });
-  const code = await new Promise((resolveExit, reject) => {
-    const timer = setTimeout(() => {
-      child.kill("SIGTERM");
-      reject(new Error(`${command} --version timed out`));
-    }, 10_000);
-    child.once("error", (cause) => {
-      clearTimeout(timer);
-      reject(cause);
-    });
-    child.once("exit", (value) => {
-      clearTimeout(timer);
-      resolveExit(value ?? 1);
-    });
-  });
-  if (code !== 0)
-    throw new Error(
-      `${command} --version failed with ${String(code)}: ${stderr}`,
-    );
-  const version = stdout.trim();
-  if (version.length === 0)
-    throw new Error(`${command} --version returned empty output`);
-  return version;
-}