fix(process): retain provider output until streams close (#928)

src/javascript/recovery/WakaruCommand.test.ts

@@ -1,5 +1,8 @@
+import { EventEmitter } from "node:events";
+import { PassThrough } from "node:stream";
 import { expect, it } from "vitest";
-import { runWakaruCommand } from "./WakaruCommand.js";
+import { runWakaruCommand, type WakaruLauncher } from "./WakaruCommand.js";
+import { RECOVERY_LIMITS } from "./WakaruRelease.js";
 
 it("rejects an exhausted shared execution budget before launching the next command", async () => {
   await expect(
@@ -36,3 +39,113 @@ it("preserves caller cancellation before acquiring an expired execution budget",
     }),
   ).rejects.toMatchObject({ _tag: "AnalysisCancelledError" });
 });
+
+const commandFixture = (
+  emit: (producer: ReturnType<typeof createProducer>) => void,
+) => {
+  const producer = createProducer();
+  const launcher: WakaruLauncher = async (options) => {
+    setImmediate(() => emit(producer));
+    return {
+      process: producer,
+      ownership: {
+        runId: options.runId,
+        leaderPid: 12345,
+        processGroupId: 12345,
+        expectedParentPid: process.pid,
+      },
+      cleanup: async () => ({ cleaned: true, signaled: false }),
+    };
+  };
+  return {
+    producer,
+    context: {
+      command: "/selected/wakaru",
+      limiter: "/usr/bin/prlimit",
+      arguments: ["--version"],
+      cwd: "/tmp/unused",
+      environment: {},
+      deadline: Date.now() + 5_000,
+      launcher,
+    },
+  };
+};
+
+const createProducer = () =>
+  Object.assign(new EventEmitter(), {
+    stdout: new PassThrough(),
+    stderr: new PassThrough(),
+    exitCode: null,
+    signalCode: null,
+    kill: () => false,
+  });
+
+it("retains stdout and stderr delivered after launcher exit", async () => {
+  const { context } = commandFixture((producer) => {
+    producer.stdout.write("first");
+    producer.emit("exit", 0, null);
+    setImmediate(() => {
+      producer.stdout.end("last");
+      producer.stderr.end("diagnostic");
+      producer.emit("close", 0, null);
+    });
+  });
+  await expect(runWakaruCommand(context)).resolves.toMatchObject({
+    exit_code: 0,
+    signal: null,
+    stdout: "firstlast",
+    stderr: "diagnostic",
+  });
+});
+
+it("checks the diagnostic byte limit against output delivered after exit", async () => {
+  const { context } = commandFixture((producer) => {
+    producer.emit("exit", 0, null);
+    setImmediate(() => {
+      producer.stderr.end(Buffer.alloc(RECOVERY_LIMITS.reportBytes + 1, 97));
+      producer.stdout.end();
+      producer.emit("close", 0, null);
+    });
+  });
+  await expect(runWakaruCommand(context)).rejects.toMatchObject({
+    _tag: "AnalysisOutputError",
+    operation: "recover_javascript_sources",
+  });
+});
+
+it("enforces the shared deadline while exited output streams remain open", async () => {
+  const { context, producer } = commandFixture((process) => {
+    process.emit("exit", 0, null);
+  });
+  try {
+    await expect(
+      runWakaruCommand({ ...context, deadline: Date.now() + 50 }),
+    ).rejects.toMatchObject({
+      _tag: "AnalysisTimeoutError",
+      operation: "recover_javascript_sources",
+    });
+    expect(producer.stderr.listenerCount("data")).toBe(0);
+  } finally {
+    producer.stdout.destroy();
+    producer.stderr.destroy();
+  }
+});
+
+it("honors cancellation while waiting for output after exit", async () => {
+  const controller = new AbortController();
+  const { context, producer } = commandFixture((process) => {
+    process.emit("exit", 0, null);
+    setImmediate(() => controller.abort());
+  });
+  try {
+    await expect(
+      runWakaruCommand({ ...context, signal: controller.signal }),
+    ).rejects.toMatchObject({
+      _tag: "AnalysisCancelledError",
+    });
+    expect(producer.stderr.listenerCount("data")).toBe(0);
+  } finally {
+    producer.stdout.destroy();
+    producer.stderr.destroy();
+  }
+});

src/javascript/recovery/WakaruCommand.ts

@@ -132,7 +132,7 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => {
   );
   let failure: unknown;
   try {
-    while (!(await supervisor.waitForExit(100))) {
+    while (!(await supervisor.waitForOutputClose(100))) {
       checkRecoveryDeadline(context.deadline, context.signal);
       if (oversized)
         throw new AnalysisOutputError(