Fix action hang when descendants keep stdio open

src/runCodexExec.ts

@@ -2,6 +2,7 @@ import { spawn } from "child_process";
 import { chmod, mkdtemp, readFile, rm, writeFile } from "fs/promises";
 import path from "path";
 import os from "os";
+import type { Readable } from "stream";
 import { setOutput } from "@actions/core";
 import { checkOutput } from "./checkOutput";
 import { captureLinuxRunnerCredentials } from "./linuxCredentials";
@@ -312,14 +313,28 @@ export async function runCodexExec({
     await new Promise((resolve, reject) => {
       const child = spawn(program, command, {
         env,
-        stdio: ["pipe", "inherit", "inherit"],
+        stdio: ["pipe", "pipe", "pipe"],
       });
+      child.stdout.pipe(process.stdout, { end: false });
+      child.stderr.pipe(process.stderr, { end: false });
       child.stdin.write(input);
       child.stdin.end();
 
-      child.on("error", reject);
+      const closeOutputStreams = () => {
+        child.stdout.unpipe(process.stdout);
+        child.stderr.unpipe(process.stderr);
+        child.stdout.destroy();
+        child.stderr.destroy();
+      };
 
-      child.on("close", async (code) => {
+      child.once("error", (err) => {
+        closeOutputStreams();
+        reject(err);
+      });
+
+      child.once("exit", async (code) => {
+        await drainOutputStreams([child.stdout, child.stderr]);
+        closeOutputStreams();
         if (code !== 0) {
           reject(new Error(`${program} exited with code ${code}`));
           return;
@@ -338,6 +353,56 @@ export async function runCodexExec({
   }
 }
 
+const OUTPUT_DRAIN_QUIET_MS = 25;
+const OUTPUT_DRAIN_TIMEOUT_MS = 1_000;
+
+/**
+ * Lets libuv deliver output that was already buffered when the direct child exited. Descendants
+ * may still own the write ends, so EOF cannot be the completion condition. The absolute bound
+ * keeps a descendant that continuously writes (or a blocked runner log destination) from hanging
+ * the action forever.
+ */
+function drainOutputStreams(streams: ReadonlyArray<Readable>): Promise<void> {
+  return new Promise((resolve) => {
+    let quietHandle: NodeJS.Timeout;
+
+    const onData = () => {
+      scheduleQuietCheck();
+    };
+    for (const stream of streams) {
+      stream.on("data", onData);
+    }
+
+    const finish = () => {
+      clearTimeout(quietHandle);
+      clearTimeout(timeoutHandle);
+      for (const stream of streams) {
+        stream.off("data", onData);
+      }
+      resolve();
+    };
+
+    const scheduleQuietCheck = () => {
+      clearTimeout(quietHandle);
+      quietHandle = setTimeout(() => {
+        const streamsAreDrained = streams.every(
+          (stream) =>
+            stream.destroyed ||
+            (stream.readableLength === 0 && stream.readableFlowing !== false)
+        );
+        if (streamsAreDrained) {
+          finish();
+          return;
+        }
+        scheduleQuietCheck();
+      }, OUTPUT_DRAIN_QUIET_MS);
+    };
+
+    const timeoutHandle = setTimeout(finish, OUTPUT_DRAIN_TIMEOUT_MS);
+    scheduleQuietCheck();
+  });
+}
+
 async function finalizeExecution(
   outputFile: OutputFile,
   runAsUser: string | null

test/runCodexExec.test.mjs

@@ -25,21 +25,38 @@ function runCodexExecWithFakeCodex({
   permissionProfile = "",
   extraArgs = "",
   safetyStrategy = "unsafe",
+  holdStdioOpen = false,
+  writeLargeFinalOutput = false,
 } = {}) {
   const tempDir = mkdtempSync(path.join(tmpdir(), "codex-action-permissions-"));
   const capturePath = path.join(tempDir, "args.json");
   const outputPath = path.join(tempDir, "output.txt");
   const fakeCodexPath = path.join(tempDir, "codex.mjs");
   writeFileSync(
     fakeCodexPath,
-    `import { writeFileSync } from "node:fs";
+    `import { spawn } from "node:child_process";
+import { writeFileSync } from "node:fs";
 const args = process.argv.slice(2);
 writeFileSync(process.env.CODEX_CAPTURE_ARGS, JSON.stringify(args));
 const outputIndex = args.indexOf("--output-last-message");
 if (outputIndex < 0 || outputIndex + 1 >= args.length) {
   throw new Error("missing --output-last-message");
 }
 writeFileSync(args[outputIndex + 1], "fake final message\\n");
+if (process.env.CODEX_WRITE_LARGE_FINAL_OUTPUT === "1") {
+  process.stdout.write("stdout-start:" + "o".repeat(1024 * 1024) + ":stdout-end\\n");
+  process.stderr.write("stderr-start:" + "e".repeat(1024 * 1024) + ":stderr-end\\n");
+}
+if (process.env.CODEX_HOLD_STDIO_OPEN === "1") {
+  console.log("fake codex stdout");
+  console.error("fake codex stderr");
+  const descendant = spawn(
+    process.execPath,
+    ["-e", "setTimeout(() => {}, 5000)"],
+    { stdio: "inherit" }
+  );
+  descendant.unref();
+}
 `,
     "utf8"
   );
@@ -95,9 +112,17 @@ writeFileSync(args[outputIndex + 1], "fake final message\\n");
       encoding: "utf8",
       env: {
         ...process.env,
+        // Keep the regression focused on Codex stdout/stderr forwarding. If
+        // inherited, @actions/core writes final-message to this file instead
+        // of the captured stdout used by these assertions.
+        GITHUB_OUTPUT: undefined,
         PATH: `${tempDir}${path.delimiter}${process.env.PATH ?? ""}`,
         CODEX_CAPTURE_ARGS: capturePath,
+        CODEX_HOLD_STDIO_OPEN: holdStdioOpen ? "1" : "0",
+        CODEX_WRITE_LARGE_FINAL_OUTPUT: writeLargeFinalOutput ? "1" : "0",
       },
+      timeout: holdStdioOpen ? 2_000 : undefined,
+      maxBuffer: 10 * 1024 * 1024,
     }
   );
 
@@ -111,6 +136,34 @@ writeFileSync(args[outputIndex + 1], "fake final message\\n");
   return { result, capturedArgs };
 }
 
+test("does not wait for descendants holding stdio open", () => {
+  const { result } = runCodexExecWithFakeCodex({ holdStdioOpen: true });
+
+  assert.equal(result.error, undefined);
+  assert.equal(result.status, 0, result.error?.message ?? result.stderr);
+  assert.match(result.stdout, /fake codex stdout/);
+  assert.match(result.stderr, /fake codex stderr/);
+  assert.match(result.stdout, /fake final message/);
+});
+
+test("drains large final output without waiting for descendant stdio", () => {
+  const { result } = runCodexExecWithFakeCodex({
+    holdStdioOpen: true,
+    writeLargeFinalOutput: true,
+  });
+
+  assert.equal(result.error, undefined);
+  assert.equal(result.status, 0, result.error?.message ?? result.stderr);
+  assert.match(
+    result.stdout,
+    new RegExp(`stdout-start:${"o".repeat(1024 * 1024)}:stdout-end`)
+  );
+  assert.match(
+    result.stderr,
+    new RegExp(`stderr-start:${"e".repeat(1024 * 1024)}:stderr-end`)
+  );
+});
+
 test("preserves workspace-write as the default legacy sandbox", () => {
   const { result, capturedArgs } = runCodexExecWithFakeCodex();
 

test/runCodexExecStreams.test.mjs

@@ -0,0 +1,120 @@
+import assert from "node:assert/strict";
+import { EventEmitter } from "node:events";
+import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { PassThrough, Writable } from "node:stream";
+import { test } from "node:test";
+import { fileURLToPath } from "node:url";
+import { createRequire } from "node:module";
+import vm from "node:vm";
+import { buildSync } from "esbuild";
+
+const require = createRequire(import.meta.url);
+
+function loadRunCodexExec(spawn, stdout, stderr) {
+  const { outputFiles } = buildSync({
+    entryPoints: [
+      fileURLToPath(new URL("../src/runCodexExec.ts", import.meta.url)),
+    ],
+    bundle: true,
+    format: "cjs",
+    platform: "node",
+    write: false,
+    external: ["./checkOutput", "@actions/core"],
+  });
+  const module = { exports: {} };
+  vm.runInNewContext(outputFiles[0].text, {
+    module,
+    exports: module.exports,
+    require(name) {
+      if (name === "child_process" || name === "node:child_process") {
+        return { spawn };
+      }
+      if (name === "./checkOutput") {
+        return { checkOutput: async () => "" };
+      }
+      if (name === "@actions/core") {
+        return { setOutput() {} };
+      }
+      return require(name);
+    },
+    process: {
+      platform: "darwin",
+      env: {},
+      stdout,
+      stderr,
+    },
+    console: { log() {} },
+    setImmediate,
+    clearImmediate,
+    setTimeout,
+    clearTimeout,
+  });
+  return module.exports.runCodexExec;
+}
+
+function captureOutput() {
+  const chunks = [];
+  return {
+    chunks,
+    stream: new Writable({
+      write(chunk, _encoding, callback) {
+        chunks.push(Buffer.from(chunk));
+        callback();
+      },
+    }),
+  };
+}
+
+test("drains buffered stdout and stderr after the direct child exits", async () => {
+  const tempDir = mkdtempSync(path.join(tmpdir(), "codex-action-streams-"));
+  const outputPath = path.join(tempDir, "output.txt");
+  writeFileSync(outputPath, "fake final message\n", "utf8");
+
+  const stdout = captureOutput();
+  const stderr = captureOutput();
+  const child = new EventEmitter();
+  child.stdin = new PassThrough();
+  child.stdout = new PassThrough();
+  child.stderr = new PassThrough();
+
+  const stdoutPayload = `stdout-start:${"o".repeat(1024 * 1024)}:stdout-end`;
+  const stderrPayload = `stderr-start:${"e".repeat(1024 * 1024)}:stderr-end`;
+  const runCodexExec = loadRunCodexExec(
+    () => {
+      setImmediate(() => {
+        child.emit("exit", 0);
+        setImmediate(() => {
+          child.stdout.write(stdoutPayload);
+          child.stderr.write(stderrPayload);
+        });
+      });
+      return child;
+    },
+    stdout.stream,
+    stderr.stream
+  );
+
+  try {
+    await runCodexExec({
+      prompt: { type: "inline", content: "test" },
+      codexHome: null,
+      cd: tempDir,
+      extraArgs: [],
+      explicitOutputFile: outputPath,
+      outputSchema: null,
+      model: null,
+      effort: null,
+      safetyStrategy: "unsafe",
+      codexUser: null,
+      sandbox: null,
+      permissionProfile: null,
+    });
+
+    assert.equal(Buffer.concat(stdout.chunks).toString(), stdoutPayload);
+    assert.equal(Buffer.concat(stderr.chunks).toString(), stderrPayload);
+  } finally {
+    rmSync(tempDir, { recursive: true, force: true });
+  }
+});