fix(worker): reap deterministic station descendants on timeout and exit (#10, #17)

docs/harness-containment.md

@@ -53,10 +53,17 @@ hold it together:
   backgrounds a grandchild, and must show that the grandchild's pid is gone and its
   sentinel file stops changing after the timeout.
   `src/worker/harness-containment-registry.test.ts` fails if an adapter ships without a
-  conformance call. The deterministic station runner is registered against the same suite
-  but does not pass it yet ([#10](https://github.com/theaiteam-dev/conduit/issues/10),
-  [#17](https://github.com/theaiteam-dev/conduit/issues/17)); that is a separate path from
-  harness stations, and its test is marked as a known failure until those are fixed.
+  conformance call. The deterministic station runner and the harness runner, two separate
+  spawn paths, both pass the same suite, and both also run its exit scenarios: when the
+  command exits 0 or nonzero before its timeout, the runner kills the command's process
+  group before it returns, so a grandchild does not outlive a station or a harness
+  invocation that finished on its own
+  ([#10](https://github.com/theaiteam-dev/conduit/issues/10),
+  [#17](https://github.com/theaiteam-dev/conduit/issues/17)). For the harness runner this
+  also keeps a grandchild that inherited the stdout/stderr pipes from stalling the output
+  drains past its actual finish. Both runners also kill every live station process group
+  when the kernel receives SIGINT, SIGTERM or SIGHUP, or exits; a SIGKILLed kernel cannot do
+  this, which is why the container boundary below still matters.
   Deployment guidance additionally recommends running the container as a dedicated non-root
   user for agentic/harness flows.
 - **The ADR-0003 container wall.** [ADR-0003](../adr/0003-packaging-and-distribution.md)'s

src/controller/sync-deterministic-failure.test.ts

@@ -23,6 +23,10 @@
  * build TERMINATE (the no-progress watchdog halts it, leaving the card
  * un-scrapped) instead of hanging — so this test fails fast (red) pre-fix rather
  * than wedging the suite. The fixed build scraps well before the watchdog.
+ *
+ * The second describe covers containment on the same path (#17): a station
+ * whose command backgrounds a grandchild and exits 0 must not leave that
+ * grandchild running once the card has advanced.
  */
 
 import { describe, it, expect, beforeEach, afterEach } from 'bun:test';
@@ -36,6 +40,12 @@ import type { FlowConfig } from '../types/kernel';
 import { runExecutor } from './executor';
 import type { RunEngineArgs } from '../cli/main';
 import type { ModelAdapter } from '../worker/adapter';
+import {
+  CONTAINMENT_FIXTURE,
+  containmentFixtureExitArgs,
+  expectGrandchildReaped,
+  killRecordedGrandchild,
+} from '../worker/harness-containment.conformance';
 
 const io = { out: (_l: string) => {}, err: (_l: string) => {} };
 
@@ -120,6 +130,7 @@ afterEach(() => {
     db.close();
     db = null;
   }
+  killRecordedGrandchild(projectDir);
   process.chdir(originalCwd);
   rmSync(projectDir, { recursive: true, force: true });
 });
@@ -170,3 +181,51 @@ describe('WI-686 — synchronous-path deterministic failure routes to scrap', ()
     expect(runs).toBeLessThanOrEqual(MAX_ATTEMPTS);
   });
 });
+
+describe('#17: synchronous-path deterministic station reaps its descendants on exit 0', () => {
+  it('advances the card to done and leaves no backgrounded grandchild running', async () => {
+    db = openDb();
+    // The containment fixture backgrounds a grandchild that touches a sentinel,
+    // waits for its first touch, then exits 0. It writes into its cwd, which
+    // the executor sets to the project root.
+    const args = [CONTAINMENT_FIXTURE, ...containmentFixtureExitArgs(0)].map((a) => JSON.stringify(a));
+    writeFileSync(
+      join(projectDir, 'flow.yaml'),
+      `
+flow: sync-deterministic-reap
+project_root: .
+flow_version: 1
+terminal_lanes: [done, scrap, hold]
+defaults:
+  cap_policy: scrap
+budgets:
+  per_card: { max_execution_attempts: 1 }
+stations:
+  - id: boom
+    next: done
+    worker:
+      kind: deterministic
+      role: spawner
+      command: sh
+      args: [${args.join(', ')}]
+    wip: 1
+    inputs: []
+    outputs: []
+`,
+    );
+    const loaded = loadFlow(join(projectDir, 'flow.yaml'));
+    if (!loaded.ok) throw new Error(`fixture flow invalid: ${JSON.stringify(loaded.errors)}`);
+    seedBoomCard(db);
+
+    await runExecutor({
+      db,
+      flow: loaded.flow,
+      now: () => Math.floor(Date.now() / 1000),
+      adapter: throwingModel,
+      io,
+    } as RunEngineArgs);
+
+    expect(db.getCard(DEFAULT_RUN_ID, 'entry')?.lane).toBe('done');
+    await expectGrandchildReaped(projectDir);
+  }, 20_000);
+});

src/worker/containment-fixture.sh

@@ -10,6 +10,12 @@
 # until killed. It prints nothing, so the invocation can only end at its
 # timeout.
 #
+# Exit mode: `containment-fixture.sh --exit <code>` waits until the grandchild
+# has touched the sentinel once, then exits with <code> instead of blocking.
+# The suite uses it for the normal-exit and nonzero-exit scenarios, where the
+# invocation ends on its own and the grandchild must still not outlive it.
+# Only spawn paths that pass the suite's `fixtureArgs` through use this mode.
+#
 # The grandchild's stdio goes to /dev/null so it does not hold the spawn
 # path's stdout/stderr pipes open. A path that fails to reap it then returns
 # at its timeout and fails the suite's assertions, instead of hanging.
@@ -25,4 +31,10 @@ export PATH
   done
 ) </dev/null >/dev/null 2>&1 &
 echo "$!" > containment.pid
+if [ "$1" = "--exit" ]; then
+  while [ ! -f containment.sentinel ]; do
+    sleep 0.05
+  done
+  exit "${2:-0}"
+fi
 wait

src/worker/containment-signal-runner.ts

@@ -0,0 +1,43 @@
+/**
+ * Stand-in kernel for the signal containment tests (./process-group.test.ts).
+ *
+ *   bun containment-signal-runner.ts <deterministic|harness> <projectRoot> [exit]
+ *
+ * Runs the containment fixture (./containment-fixture.sh) through the named
+ * runner with a timeout far longer than any test, so only the kernel's own
+ * signal or exit handling can end the station. Prints `ready` once the fixture
+ * has recorded its grandchild's pid, by which point the runner has registered
+ * the group. With `exit`, it then calls process.exit(0) mid-station.
+ *
+ * This file is not a test file. Bun only runs it when a test spawns it.
+ */
+import { existsSync } from 'node:fs';
+import { join } from 'node:path';
+import { runDeterministic } from './deterministic';
+import { runHarnessProcess } from './harness-runner';
+
+const [runner, projectRoot, mode] = process.argv.slice(2);
+if ((runner !== 'deterministic' && runner !== 'harness') || projectRoot === undefined) {
+  console.error('usage: containment-signal-runner.ts <deterministic|harness> <projectRoot> [exit]');
+  process.exit(2);
+}
+
+const fixture = join(import.meta.dir, 'containment-fixture.sh');
+const timeoutMs = 600_000;
+
+const poll = setInterval(() => {
+  if (!existsSync(join(projectRoot, 'containment.pid'))) return;
+  clearInterval(poll);
+  console.log('ready');
+  if (mode === 'exit') process.exit(0);
+}, 20);
+
+if (runner === 'deterministic') {
+  await runDeterministic({ command: fixture, args: [] }, { allowlist: [fixture], cwd: projectRoot, timeoutMs });
+} else {
+  await runHarnessProcess({ command: fixture, args: [] }, { projectRoot, timeoutMs });
+}
+// A fixture that never recorded its pid leaves the poll running, which would
+// keep this process alive after the station returned.
+clearInterval(poll);
+console.log('station returned');

src/worker/deterministic.test.ts

@@ -25,7 +25,7 @@
  *     // MUST call checkCommandAllowed first and THROW (refuse) before spawning if denied.
  */
 import { describe, it, expect, beforeEach, afterEach } from 'bun:test';
-import { existsSync, mkdtempSync, rmSync, writeFileSync, chmodSync } from 'node:fs';
+import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync, chmodSync } from 'node:fs';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import {
@@ -193,7 +193,7 @@ describe('runDeterministic — timeout enforcement (punch-list #8)', () => {
 
   // Code-review fix #1 — a SIGKILL that is NOT from our timeout (here: the child
   // kills itself well before the deadline) must NOT be mislabeled `timedOut`.
-  // The elapsed-time guard keeps the timeout diagnostic honest. The self-kill
+  // `timedOut` requires that our own timer fired, which keeps the diagnostic honest. The self-kill
   // lives in a SCRIPT FILE (not argv) so the Law-lite metacharacter scan — which
   // only inspects argv — admits it; the path is plain safe-charset.
   it('does NOT report timedOut for a SIGKILL that landed well before the deadline', async () => {
@@ -291,22 +291,143 @@ describe('runDeterministic — env injection', () => {
 });
 
 // ---------------------------------------------------------------------------
-// Containment conformance (issue #27). runDeterministic is the other spawn
-// path, and it does not reap descendants yet: Bun.spawn's native timeout kills
-// only the immediate child, so the fixture's grandchild survives (#10 for the
-// timeout, #17 for a normal exit). The reaping test is registered with
-// `test.failing` until that is fixed. It goes red once the grandchild dies,
-// and the fix should then drop `knownLeak`.
+// Containment conformance (issues #27, #10, #17). runDeterministic spawns the
+// command as its own process group and SIGKILLs the group on timeout and after
+// a normal or nonzero exit, so the fixture's grandchild dies on every path.
+// `reapsOnExit` adds the exit-0 and nonzero-exit scenarios to the timeout one.
 // ---------------------------------------------------------------------------
 
+/**
+ * Test-local label for `result.timedOut === true`. DeterministicResult carries
+ * a boolean, not a classification code, so the suite compares against this.
+ */
+const DETERMINISTIC_TIMEOUT_LABEL = 'timedOut';
+
 describeContainmentConformance(
   'deterministic',
-  async ({ projectRoot, fixture, timeoutMs }) => {
+  async ({ projectRoot, fixture, timeoutMs, fixtureArgs }) => {
     const result = await runDeterministic(
-      { command: fixture, args: [] },
+      { command: fixture, args: fixtureArgs },
       { allowlist: [fixture], cwd: projectRoot, timeoutMs },
     );
-    return result.timedOut === true ? 'timedOut' : undefined;
+    return result.timedOut === true ? DETERMINISTIC_TIMEOUT_LABEL : undefined;
   },
-  { timeoutClass: 'timedOut', knownLeak: '#10 and #17' },
+  { timeoutClass: DETERMINISTIC_TIMEOUT_LABEL, reapsOnExit: true },
 );
+
+// ---------------------------------------------------------------------------
+// #10 / #17: a descendant that keeps the inherited stdout/stderr pipes open.
+// The conformance fixture sends its grandchild's stdio to /dev/null; these
+// scripts do not, so a surviving grandchild would keep the drains pending.
+// The group kill must close the pipes, and output written before the exit or
+// the timeout must still be captured.
+// ---------------------------------------------------------------------------
+
+describe('runDeterministic: descendants holding the output pipes (#10, #17)', () => {
+  /** Grandchild pids a test recorded; afterEach SIGKILLs any that survived. */
+  const recorded: number[] = [];
+
+  afterEach(() => {
+    // Single pids only, never a group: a leaked grandchild of an undetached
+    // spawn shares the test runner's process group.
+    for (const pid of recorded.splice(0)) {
+      try {
+        process.kill(pid, 'SIGKILL');
+      } catch {
+        /* already gone: the expected case */
+      }
+    }
+  });
+
+  /**
+   * A script that writes to stdout and stderr, backgrounds a `sleep 30` that
+   * inherits both pipes, records its pid, then runs `tail`.
+   */
+  function pipeHoldingScript(tail: string): string {
+    const script = join(dir, 'hold.sh');
+    writeFileSync(
+      script,
+      ['echo before-out', 'echo before-err >&2', 'sleep 30 &', 'echo "$!" > grandchild.pid', tail, ''].join('\n'),
+    );
+    return script;
+  }
+
+  function grandchildPid(): number {
+    const pid = Number(readFileSync(join(dir, 'grandchild.pid'), 'utf-8').trim());
+    expect(Number.isInteger(pid) && pid > 1).toBe(true);
+    recorded.push(pid);
+    return pid;
+  }
+
+  async function waitForPidGone(pid: number): Promise<boolean> {
+    const deadline = Date.now() + 3_000;
+    while (Date.now() < deadline) {
+      try {
+        process.kill(pid, 0);
+      } catch {
+        return true;
+      }
+      await new Promise((r) => setTimeout(r, 25));
+    }
+    return false;
+  }
+
+  for (const [label, code] of [
+    ['exits 0', 0],
+    ['exits nonzero', 4],
+  ] as const) {
+    it(`returns promptly with the output written before it ${label}, and the grandchild is gone`, async () => {
+      const script = pipeHoldingScript(`exit ${code}`);
+      const start = Date.now();
+      const result = await runDeterministic({ command: 'sh', args: [script] }, { allowlist: ['sh'], cwd: dir });
+      const elapsed = Date.now() - start;
+
+      // Without the group kill the drains wait out the 30s sleep.
+      expect(elapsed).toBeLessThan(10_000);
+      expect(result.exitCode).toBe(code);
+      expect(result.ok).toBe(code === 0);
+      expect(result.timedOut).toBeFalsy();
+      expect(result.stdout).toBe('before-out\n');
+      expect(result.stderr).toBe('before-err\n');
+      expect(await waitForPidGone(grandchildPid())).toBe(true);
+    }, 40_000);
+  }
+
+  it('does not report a timeout when the group is killed after a normal exit', async () => {
+    const script = pipeHoldingScript('exit 0');
+    const start = Date.now();
+    const result = await runDeterministic(
+      { command: 'sh', args: [script] },
+      { allowlist: ['sh'], cwd: dir, timeoutMs: 20_000 },
+    );
+
+    // Without the post-exit group kill the drains wait for the 20s timer,
+    // which reaps the grandchild and reports no timeout, so only the elapsed
+    // time tells the two apart.
+    expect(Date.now() - start).toBeLessThan(10_000);
+    expect(result.ok).toBe(true);
+    expect(result.exitCode).toBe(0);
+    expect(result.timedOut).toBeFalsy();
+    expect(result.stderr).not.toMatch(/exceeded its timeout/i);
+    expect(await waitForPidGone(grandchildPid())).toBe(true);
+  }, 40_000);
+
+  it('returns promptly after the deadline and keeps the output written before the timeout', async () => {
+    const script = pipeHoldingScript('wait');
+    const start = Date.now();
+    const result = await runDeterministic(
+      { command: 'sh', args: [script] },
+      { allowlist: ['sh'], cwd: dir, timeoutMs: 300 },
+    );
+    const elapsed = Date.now() - start;
+
+    // #10's reproduction returned only when the descendant exited.
+    expect(elapsed).toBeLessThan(10_000);
+    expect(result.ok).toBe(false);
+    expect(result.timedOut).toBe(true);
+    expect(result.stdout).toBe('before-out\n');
+    expect(result.stderr).toStartWith('before-err\n');
+    expect(result.stderr).toMatch(/exceeded its timeout of 300ms/);
+    expect(await waitForPidGone(grandchildPid())).toBe(true);
+  }, 40_000);
+});