fix(native): drain bounded runtime requests to EOF with the shared exit bound

src/backends/native-project-process.ts

@@ -12,6 +12,7 @@ import {
 import type { NativeAuthoredStorageTool } from "./native-authored-storage-tool.ts";
 import {
   invokeNativeRuntime,
+  NATIVE_EXIT_DRAIN_BOUND_MS,
   type NativeRuntimeSelection,
   readNativeFailureCodes,
   rethrowNativeInputFailure,
@@ -22,20 +23,6 @@ const RUN = /^[a-f0-9]{32}$/;
 const MAX_OUTPUT = 16 * 1024 * 1024;
 const MAX_INPUT = 256 * 1024;
 
-/**
- * Exit-relative bound on draining the owned child's stdout and stderr.
- *
- * The native runtime writes its closed error code (and any readiness line)
- * before it exits, so those bytes are already in the pipes when the exit is
- * observed; with no other writer, EOF follows immediately and the drain ends
- * there. This bound only ends a drain that EOF cannot end: a descendant that
- * inherited a pipe and keeps it open. It is measured from the observed exit,
- * not from spawn, and is deliberately generous so that a heavily
- * oversubscribed host still schedules the pipe readers before it expires.
- * A 250 ms bound lost the closed code under load (HACK-1350).
- */
-export const NATIVE_EXIT_DRAIN_BOUND_MS = 5000;
-
 export type NativeExitDiagnostic = {
   readonly exitCode: number;
   readonly nativeCode?: string;
@@ -61,6 +48,12 @@ type ProcessOptions = {
   readonly onStopFailure?: () => void;
   /** Durable capture in the existing frontend owner, before private input delivery. */
   readonly onGroup?: (group: number) => Promise<void>;
+  /**
+   * Test seam: how long after the owned child's observed exit a pipe a
+   * descendant still holds keeps being read. Production callers leave it at
+   * {@link NATIVE_EXIT_DRAIN_BOUND_MS}; EOF ends the drain earlier.
+   */
+  readonly exitDrainBoundMs?: number;
 };
 
 /**
@@ -275,9 +268,11 @@ function nativeSnapshotReady(
 }
 
 function assertProcessSelection(opts: ProcessOptions): void {
+  const exitDrainBoundMs = opts.exitDrainBoundMs ?? NATIVE_EXIT_DRAIN_BOUND_MS;
   if (
     !(RUN.test(opts.run) && Number.isSafeInteger(opts.startupTimeoutMs)) ||
     opts.startupTimeoutMs < 1 ||
+    !(Number.isSafeInteger(exitDrainBoundMs) && exitDrainBoundMs >= 1) ||
     opts.startupTimeoutMs > 600_000 ||
     (opts.privateInput?.byteLength ?? 0) > MAX_INPUT ||
     canceledAtEntry(opts)
@@ -339,7 +334,10 @@ async function serveGraphProcess(
   let drainTimer: ReturnType<typeof setTimeout> | undefined;
   void child.exited.then(() => {
     owner.abort();
-    drainTimer = setTimeout(() => drain.abort(), NATIVE_EXIT_DRAIN_BOUND_MS);
+    drainTimer = setTimeout(
+      () => drain.abort(),
+      opts.exitDrainBoundMs ?? NATIVE_EXIT_DRAIN_BOUND_MS
+    );
   });
   const failureCodes = readNativeFailureCodes(child.stderr, drain.signal);
   const failureCode = failureCodes.then((failure) => failure?.code);

src/backends/native-runtime-client.ts

@@ -90,6 +90,22 @@ export interface NativeRuntimeSelection {
 
 const MAX_OUTPUT_BYTES = 16 * 1024 * 1024;
 const MAX_PRIVATE_INPUT_BYTES = 256 * 1024;
+
+/**
+ * Exit-relative bound on draining an owned native child's stdout and stderr.
+ *
+ * The native runtime writes its closed error code (and any response or
+ * readiness line) before it exits, so those bytes are already in the pipes
+ * when the exit is observed; with no other writer, EOF follows immediately
+ * and the drain ends there. This bound only ends a drain that EOF cannot end:
+ * a descendant that inherited a pipe and keeps it open. It is measured from
+ * the observed exit, not from spawn, and is deliberately generous so that a
+ * heavily oversubscribed host still schedules the pipe readers before it
+ * expires. A 250 ms bound lost the closed code under load for the foreground
+ * graph owner (HACK-1350) and for bounded one-shot requests (HACK-1387).
+ * Shared by {@link invokeNativeRuntime} and the foreground graph owner.
+ */
+export const NATIVE_EXIT_DRAIN_BOUND_MS = 5000;
 const REQUEST_BUDGET_REFUSAL =
   "Native runtime request exceeds its input or time budget.";
 const SYSTEM_ENVIRONMENT = ["PATH", "HOME", "TMPDIR", "USER", "LOGNAME"];
@@ -137,6 +153,12 @@ export async function invokeNativeRuntime(opts: {
   readonly boundNativeAuthoredReadDrain?: boolean;
   /** Explicit recovery requests have their own closed argv domain and never accept private input. */
   readonly boundNativeAuthoredRecoveryDrain?: "selection" | "cleanup";
+  /**
+   * Test seam for a bound drain: how long after the child's observed exit a
+   * pipe a descendant still holds keeps being read. Production callers leave
+   * it at {@link NATIVE_EXIT_DRAIN_BOUND_MS}; EOF ends the drain earlier.
+   */
+  readonly exitDrainBoundMs?: number;
 }): Promise<unknown> {
   if (opts.signal?.aborted) {
     throw new NativeRuntimeRequestError({
@@ -146,14 +168,17 @@ export async function invokeNativeRuntime(opts: {
   }
   const timeoutMs = opts.timeoutMs ?? 180_000;
   const bound = boundDrainSelection(opts);
+  const exitDrainBoundMs = exitDrainBound(opts.exitDrainBoundMs);
   if (
     !(
       validExecResponseSelection(opts.args, opts.serviceExecResponse) &&
       validBoundDrainSelection(
         opts.args,
         bound,
         opts.privateInput !== undefined
-      )
+      ) &&
+      Number.isSafeInteger(exitDrainBoundMs) &&
+      exitDrainBoundMs >= 1
     )
   ) {
     // Internal invariant: callers build these argv shapes; no user input selects them.
@@ -188,11 +213,15 @@ export async function invokeNativeRuntime(opts: {
       stderr: "pipe",
     }
   );
+  // A descendant may inherit a pipe after the owned child exits. The child's
+  // exit, not that writer's EOF, bounds draining (see
+  // NATIVE_EXIT_DRAIN_BOUND_MS); its own closed code and response are already
+  // in the pipes at exit, so EOF normally ends the drain long before the bound.
   const drain = bound === undefined ? undefined : new AbortController();
   let drainTimer: ReturnType<typeof setTimeout> | undefined;
   if (drain) {
     void child.exited.then(() => {
-      drainTimer = setTimeout(() => drain.abort(), 250);
+      drainTimer = setTimeout(() => drain.abort(), exitDrainBoundMs);
     });
   }
   const failureDetails = readNativeFailure(child.stderr, drain?.signal);
@@ -259,6 +288,11 @@ export async function invokeNativeRuntime(opts: {
   }
 }
 
+/** The exit-relative drain bound in effect: the test seam's value, else the shared constant. */
+function exitDrainBound(selected: number | undefined): number {
+  return selected ?? NATIVE_EXIT_DRAIN_BOUND_MS;
+}
+
 /** One bound output drain per request; the public flags remain the callers' contract. */
 type BoundDrain =
   | "status"

tests/native-authored-project-exec.test.ts

@@ -21,7 +21,10 @@ import {
   saveNativeAuthoredProjectRun,
   withNativeAuthoredProjectStatus,
 } from "../src/backends/native-authored-project-run.ts";
-import { invokeNativeRuntime } from "../src/backends/native-runtime-client.ts";
+import {
+  invokeNativeRuntime,
+  NATIVE_EXIT_DRAIN_BOUND_MS,
+} from "../src/backends/native-runtime-client.ts";
 import { tryNativeAuthoredCommand } from "../src/lib/native-authored-command.ts";
 import type { NativeComposeCommandOptions } from "../src/lib/native-compose-command.ts";
 import { captureCompletedJobFixtureCommand } from "./e2e/scenarios/native-compose-adoption-job-worktrees.ts";
@@ -34,8 +37,8 @@ import {
 const DEFAULT_BUDGET_MS = 4000;
 /** The cancelled request child sleeps this long after writing its identity. */
 const REQUEST_SLEEP_MS = 3000;
-/** The inherited-stream keeper child lives this long after its parent exits. */
-const KEEPER_SLEEP_MS = 800;
+/** An inherited-stream keeper its case never releases still retires after this cap. */
+const KEEPER_HOLD_CAP_MS = 60_000;
 /** Explicit keeper join allowance on top of its own lifetime. */
 const KEEPER_JOIN_MS = 3000;
 
@@ -596,10 +599,15 @@ test(
       "exited exec request joins finite keeper",
       async () => {
         const f = await fixture(),
-          marker = join(f.runtime.home, "keeper.pid");
+          marker = join(f.runtime.home, "keeper.pid"),
+          release = join(f.runtime.home, "keeper.release");
+        // The keeper holds the inherited streams until the test releases it
+        // after the refusal, so the output cannot settle to EOF and the
+        // exit-relative drain bound (NATIVE_EXIT_DRAIN_BOUND_MS) ends the drain.
+        const keeper = `import {existsSync} from 'node:fs';const end=Date.now()+${KEEPER_HOLD_CAP_MS};while(!existsSync(${JSON.stringify(release)})&&Date.now()<end)await Bun.sleep(20);`;
         await Bun.write(
           f.runtime.binary,
-          `#!${process.execPath}\nconst child=Bun.spawn([process.execPath,'--eval','await Bun.sleep(${KEEPER_SLEEP_MS})'],{stdin:'ignore',stdout:'inherit',stderr:'inherit'});await Bun.write(${JSON.stringify(marker)},String(child.pid));console.log('{}');process.exit(0);\n`
+          `#!${process.execPath}\nconst child=Bun.spawn([process.execPath,'--eval',${JSON.stringify(keeper)}],{stdin:'ignore',stdout:'inherit',stderr:'inherit'});await Bun.write(${JSON.stringify(marker)},String(child.pid));console.log('{}');process.exit(0);\n`
         );
         caseAdmission();
         try {
@@ -612,13 +620,14 @@ test(
             })
           ).rejects.toThrow("did not settle");
         } finally {
+          await Bun.write(release, "release");
           // The keeper identity was registered by the fixture; join it explicitly.
           await settlement.settle({ budgetMs: KEEPER_JOIN_MS });
         }
       },
-      KEEPER_SLEEP_MS + KEEPER_JOIN_MS + DEFAULT_BUDGET_MS
+      NATIVE_EXIT_DRAIN_BOUND_MS + KEEPER_JOIN_MS + DEFAULT_BUDGET_MS
     ),
-  10_000
+  NATIVE_EXIT_DRAIN_BOUND_MS + KEEPER_JOIN_MS + DEFAULT_BUDGET_MS + 5000
 );
 macTest.each([
   { extra: [], code: 17 },

tests/native-authored-project-logs.test.ts

@@ -23,6 +23,7 @@ import {
 } from "../src/backends/native-authored-project-run.ts";
 import {
   invokeNativeRuntime,
+  NATIVE_EXIT_DRAIN_BOUND_MS,
   NativeRuntimeRequestError,
 } from "../src/backends/native-runtime-client.ts";
 import { tryNativeAuthoredCommand } from "../src/lib/native-authored-command.ts";
@@ -453,28 +454,40 @@ async function joinKeeper(pidPath: string) {
     await Bun.sleep(10);
   }
 }
-test("leader exit with an inherited pipe refuses output; the exact finite keeper is joined", async () => {
-  return owned(async () => {
-    const f = await fixture(),
-      pidPath = join(f.runtime.home, "keeper.pid");
-    await Bun.write(
-      f.runtime.binary,
-      `#!${process.execPath}\nconst c=Bun.spawn([process.execPath,'--eval','await Bun.sleep(800)'],{stdin:'ignore',stdout:'inherit',stderr:'inherit'});await Bun.write(${JSON.stringify(pidPath)},String(c.pid));console.log('{}');process.exit(0);\n`
-    );
-    try {
-      await expect(
-        invokeNativeRuntime({
-          ...f,
-          cwd: f.scope.projectRoot,
-          args: f.args,
-          boundNativeAuthoredLogsDrain: true,
-        })
-      ).rejects.toThrow("did not settle");
-    } finally {
-      await joinKeeper(pidPath);
-    }
-  });
-}, 10_000);
+/** An inherited-pipe keeper its case never releases still retires after this cap. */
+const KEEPER_HOLD_CAP_MS = 60_000;
+test(
+  "leader exit with an inherited pipe refuses output; the exact finite keeper is joined",
+  async () => {
+    return owned(async () => {
+      const f = await fixture(),
+        pidPath = join(f.runtime.home, "keeper.pid"),
+        release = join(f.runtime.home, "keeper.release");
+      // The keeper holds the inherited stdout and stderr until the test releases
+      // it after the refusal, so the leader's output cannot settle to EOF and the
+      // exit-relative drain bound (NATIVE_EXIT_DRAIN_BOUND_MS) ends the drain.
+      const keeper = `import {existsSync} from 'node:fs';const end=Date.now()+${KEEPER_HOLD_CAP_MS};while(!existsSync(${JSON.stringify(release)})&&Date.now()<end)await Bun.sleep(20);`;
+      await Bun.write(
+        f.runtime.binary,
+        `#!${process.execPath}\nconst c=Bun.spawn([process.execPath,'--eval',${JSON.stringify(keeper)}],{stdin:'ignore',stdout:'inherit',stderr:'inherit'});await Bun.write(${JSON.stringify(pidPath)},String(c.pid));console.log('{}');process.exit(0);\n`
+      );
+      try {
+        await expect(
+          invokeNativeRuntime({
+            ...f,
+            cwd: f.scope.projectRoot,
+            args: f.args,
+            boundNativeAuthoredLogsDrain: true,
+          })
+        ).rejects.toThrow("did not settle");
+      } finally {
+        await Bun.write(release, "release");
+        await joinKeeper(pidPath);
+      }
+    });
+  },
+  NATIVE_EXIT_DRAIN_BOUND_MS + 15_000
+);
 
 test("cancellation after the real held-read owner closes withholds the result", async () => {
   return owned(async () => {