fix(test): preserve lane output after timeouts and stabilize the Cursor stream-health watchdog

docs-site/src/content/docs/contributing.md

@@ -42,6 +42,12 @@ bun run prepare:package           # refresh package launchers/assets
 `origin/dev`, then local `dev`. It reports that ref and the exact `git merge-base HEAD <ref>`
 commit, then passes the merge-base SHA to Bun.
 
+If a test lane times out, the runner prints the stdout and stderr it has already
+captured and exits with code 124. After a process exits, captured pipes have a
+one-second drain limit so a descendant holding a pipe open cannot stall the runner.
+Incomplete capture is reported explicitly and does not count as a successful run,
+even if the direct child exited with code 0.
+
 Tests are Bun tests in domain directories that mirror `src/`: `tests/server/`, `tests/providers/`,
 `tests/adapters/openai/`, `tests/cli/` and so on. `scripts/test-layout/layout.json` is the map
 and `tests/test-layout.test.ts` enforces it, so a new test goes into its domain directory and gets

scripts/test.ts

@@ -394,11 +394,78 @@ function waitWithTimeout<T>(promise: Promise<T>, timeoutMs: number): Promise<T |
   });
 }
 
-async function runTestLane(
+/** Read continuously so a timeout can still report output received before EOF. */
+export function captureTestOutput(
+  stdout: ReadableStream<Uint8Array>,
+  stderr: ReadableStream<Uint8Array>,
+) {
+  const collect = (stream: ReadableStream<Uint8Array>) => {
+    const reader = stream.getReader();
+    const decoder = new TextDecoder();
+    let text = "";
+    let reading = true;
+    let complete = false;
+    const done = (async () => {
+      try {
+        while (reading) {
+          const chunk = await reader.read();
+          if (!reading) break;
+          if (chunk.done) {
+            complete = true;
+            break;
+          }
+          text += decoder.decode(chunk.value, { stream: true });
+        }
+      } catch {
+        // Retain the prefix without turning a pipe error into an unhandled rejection.
+      } finally {
+        if (reading) text += decoder.decode();
+        reading = false;
+        reader.releaseLock();
+      }
+    })();
+    return {
+      done,
+      snapshot: () => ({ text, complete }),
+      cancel() {
+        if (!reading) return;
+        reading = false;
+        text += decoder.decode();
+        // A descendant may own a pipe, or a stream's cancellation may never settle.
+        // Cancellation is best effort; neither it nor EOF may extend the drain bound.
+        void reader.cancel().catch(() => {});
+      },
+    };
+  };
+  const out = collect(stdout);
+  const err = collect(stderr);
+  return {
+    async finish(timeoutMs: number) {
+      const drained = await waitWithTimeout(Promise.all([out.done, err.done]), timeoutMs);
+      if (drained === null) {
+        out.cancel();
+        err.cancel();
+      }
+      const stdout = out.snapshot();
+      const stderr = err.snapshot();
+      return {
+        stdout: stdout.text,
+        stderr: stderr.text,
+        complete: drained !== null && stdout.complete && stderr.complete,
+      };
+    },
+  };
+}
+
+export async function runTestLane(
   lane: BunTestLane,
   runId: string,
   inheritedLock: { lockPath: string; ownerToken: string } | undefined,
   capture = false,
+  writers = {
+    stdout: (value: string) => { process.stdout.write(value); },
+    stderr: (value: string) => { process.stderr.write(value); },
+  },
 ): Promise<{ exitCode: number; output: string }> {
   const isolated = createIsolatedTestEnvironment({
     ...process.env,
@@ -418,8 +485,7 @@ async function runTestLane(
     stdout: capture ? "pipe" : "inherit",
     stderr: capture ? "pipe" : "inherit",
   });
-  const stdoutP = capture ? new Response(child.stdout).text() : Promise.resolve("");
-  const stderrP = capture ? new Response(child.stderr).text() : Promise.resolve("");
+  const captured = capture ? captureTestOutput(child.stdout!, child.stderr!) : undefined;
   const forward = (signal: NodeJS.Signals) => {
     interrupted = signal;
     try { child.kill(signal); } catch { /* child already exited */ }
@@ -431,7 +497,7 @@ async function runTestLane(
 
   const exited = child.exited;
   try {
-    const exitCode = await waitWithTimeout(exited, lane.timeoutMs);
+    let exitCode = await waitWithTimeout(exited, lane.timeoutMs);
     if (exitCode === null) {
       console.error(`[test] ${lane.label} exceeded ${Math.round(lane.timeoutMs / 1000)}s; terminating pid ${child.pid}.`);
       try { child.kill("SIGTERM"); } catch { /* child already exited */ }
@@ -440,12 +506,19 @@ async function runTestLane(
         try { child.kill("SIGKILL"); } catch { /* child already exited */ }
         await waitWithTimeout(exited, 2_000);
       }
-      return { exitCode: 124, output: "" };
     }
-    const [stdout, stderr] = await Promise.all([stdoutP, stderrP]);
-    if (stdout) process.stdout.write(stdout);
-    if (stderr) process.stderr.write(stderr);
+    // Process exit does not guarantee EOF when a descendant inherited the pipe.
+    const result = await captured?.finish(1_000);
+    const stdout = result?.stdout ?? "";
+    const stderr = result?.stderr ?? "";
+    if (stdout) writers.stdout(stdout);
+    if (stderr) writers.stderr(stderr);
     const output = stdout + "\n" + stderr;
+    if (result && !result.complete) {
+      console.error("[test] captured output is incomplete; collected output is shown above.");
+      if (exitCode === 0) exitCode = 1;
+    }
+    if (exitCode === null) return { exitCode: 124, output };
     if (interrupted === "SIGINT") return { exitCode: 130, output };
     if (interrupted === "SIGTERM") return { exitCode: 143, output };
     const seconds = ((Date.now() - startedAt) / 1000).toFixed(1);

tests/ci-workflows/test-runner.test.ts

@@ -1,15 +1,17 @@
-import { describe, expect, test } from "bun:test";
+import { describe, expect, spyOn, test } from "bun:test";
 import { spawnSync } from "node:child_process";
 import { existsSync, mkdtempSync, readFileSync, statSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { basename, dirname, isAbsolute, join, posix, win32 } from "node:path";
 import {
   changedSelectionFailure,
+  captureTestOutput,
   createIsolatedTestEnvironment,
   ensureGuiDependencies,
   inspectChangedRun,
   resolveBunTestArgs,
   resolveBunTestPlan,
+  runTestLane,
   selectChangedComparisonRef,
   SERIAL_FULL_SUITE_FILES,
 } from "../../scripts/test";
@@ -98,6 +100,150 @@ function initChangedRunFixture(): { cwd: string; base: string } {
   return { cwd, base };
 }
 
+describe("test runner captured output", () => {
+  test("preserves both streams and UTF-8 characters split across chunks", async () => {
+    const bytes = new TextEncoder().encode("before 한글 after\n");
+    const stdout = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(bytes.slice(0, 8));
+        controller.enqueue(bytes.slice(8));
+        controller.close();
+      },
+    });
+    const stderr = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("diagnostic\n"));
+        controller.close();
+      },
+    });
+    expect(await captureTestOutput(stdout, stderr).finish(1_000)).toEqual({
+      stdout: "before 한글 after\n", stderr: "diagnostic\n", complete: true,
+    });
+  });
+
+  test.each(["pending", "rejected"] as const)(
+    "bounds an open pipe even when cancellation is %s",
+    async cancellation => {
+      let controller!: ReadableStreamDefaultController<Uint8Array>;
+      let cancelled = false;
+      const stdout = new ReadableStream<Uint8Array>({
+        start(value) {
+          controller = value;
+          value.enqueue(new TextEncoder().encode("retained prefix\n"));
+        },
+        cancel() {
+          cancelled = true;
+          return cancellation === "pending"
+            ? new Promise<void>(() => {})
+            : Promise.reject(new Error("fixture cancellation failure"));
+        },
+      });
+      const stderr = new ReadableStream<Uint8Array>({ start(value) { value.close(); } });
+      let timer: ReturnType<typeof setTimeout> | undefined;
+      try {
+        const result = await Promise.race([
+          captureTestOutput(stdout, stderr).finish(20),
+          new Promise<null>(resolve => { timer = setTimeout(() => resolve(null), 2_000); }),
+        ]);
+        expect(result).toEqual({ stdout: "retained prefix\n", stderr: "", complete: false });
+        expect(cancelled).toBe(true);
+      } finally {
+        clearTimeout(timer);
+        try { controller.close(); } catch { /* cancellation already closed it */ }
+      }
+    },
+  );
+
+  test("retains a prefix when reading the pipe fails", async () => {
+    let reads = 0;
+    const stdout = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        if (reads++ === 0) controller.enqueue(new TextEncoder().encode("before error\n"));
+        else controller.error(new Error("fixture read failure"));
+      },
+    });
+    const stderr = new ReadableStream<Uint8Array>({ start(controller) { controller.close(); } });
+    expect(await captureTestOutput(stdout, stderr).finish(1_000)).toEqual({
+      stdout: "before error\n", stderr: "", complete: false,
+    });
+  });
+
+  test("an exited child with an open pipe reports incomplete capture instead of success", async () => {
+    let cancelled = false;
+    const stdout = new ReadableStream<Uint8Array>({
+      start(controller) { controller.enqueue(new TextEncoder().encode("partial output\n")); },
+      cancel() { cancelled = true; },
+    });
+    const stderr = new ReadableStream<Uint8Array>({ start(controller) { controller.close(); } });
+    const spawn = spyOn(Bun, "spawn").mockReturnValue({
+      pid: 0,
+      stdout,
+      stderr,
+      exited: Promise.resolve(0),
+      kill() { throw new Error("the fixture child already exited"); },
+    } as unknown as ReturnType<typeof Bun.spawn>);
+    const emitted: string[] = [];
+    try {
+      const pending = runTestLane(
+        { label: "open pipe fixture", args: [], timeoutMs: 2_000 },
+        "capture-fixture",
+        undefined,
+        true,
+        { stdout: value => { emitted.push(value); }, stderr: value => { emitted.push(value); } },
+      );
+      // Only the synchronous spawn is mocked; no other test or later subprocess uses it.
+      spawn.mockRestore();
+      expect(await pending).toEqual({ exitCode: 1, output: "partial output\n\n" });
+      expect(emitted).toEqual(["partial output\n"]);
+      expect(cancelled).toBe(true);
+    } finally {
+      spawn.mockRestore();
+    }
+  });
+
+  test.each(["pass", "fail", "timeout"] as const)(
+    "returns and prints a %s lane's output exactly once",
+    async outcome => {
+      const root = mkdtempSync(join(tmpdir(), "opencodex-capture-lane-"));
+      const fixture = join(root, "capture.test.ts");
+      const stdout: string[] = [];
+      const stderr: string[] = [];
+      writeFileSync(fixture, `
+        import { test } from "bun:test";
+        test("capture fixture", async () => {
+          process.stdout.write("OCX_CAPTURE_STDOUT_MARKER\\n");
+          process.stderr.write("OCX_CAPTURE_STDERR_MARKER\\n");
+          ${outcome === "timeout" ? "await new Promise(() => {});" : ""}
+          ${outcome === "fail" ? 'throw new Error("fixture assertion failure");' : ""}
+        }, 60_000);
+      `);
+      try {
+        const runId = process.env[TEST_RUN_ID_ENV]!;
+        const result = await runTestLane(
+          { label: "capture fixture", args: [fixture], timeoutMs: INTERNAL_DEADLINE_MS },
+          runId,
+          resolveInheritedTestRunLock({ wrappedRunId: runId, env: process.env }),
+          true,
+          { stdout: value => { stdout.push(value); }, stderr: value => { stderr.push(value); } },
+        );
+        expect(result.exitCode).toBe(outcome === "timeout" ? 124 : outcome === "fail" ? 1 : 0);
+        expect(result.output).toContain("OCX_CAPTURE_STDOUT_MARKER\n");
+        expect(result.output).toContain("OCX_CAPTURE_STDERR_MARKER\n");
+        // A failed Bun assertion may quote the fixture source containing the marker.
+        // Count emitted marker lines, not mentions inside the error's code frame.
+        expect(stdout.join("").split(/\r?\n/).filter(line => line === "OCX_CAPTURE_STDOUT_MARKER"))
+          .toHaveLength(1);
+        expect(stderr.join("").split(/\r?\n/).filter(line => line === "OCX_CAPTURE_STDERR_MARKER"))
+          .toHaveLength(1);
+        expect(result.output).toBe(stdout.join("") + "\n" + stderr.join(""));
+      } finally {
+        removeTreeWithRetry(root);
+      }
+    },
+    { timeout: SPAWN_BUDGET_MS },
+  );
+});
+
 describe("test runner isolation", () => {
   test("redirects user homes to a disposable root", () => {
     const isolated = createIsolatedTestEnvironment({ PATH: "/test/bin", HOME: "/real/home" });