fix(cli): exit deliberately after bounded teardown instead of riding the 10s backstop (#2084)

Root cause: bounded teardown (endPoolBounded, #2015) RESOLVES, but lingering
sockets — embedding-provider fetch keep-alive, PgBouncer txn-mode sockets the
bound raced past — keep Bun's event loop alive, so every `gbrain query` paid
a flat 10s tax exiting via the hard-deadline force-exit banner.

Three changes, one contract:

- flushStdoutThenExit (cli-force-exit.ts): when main() resolves and the
  command is not a daemon, exit deliberately — after stdout AND stderr drain
  (writableLength===0, 'drain'-event + poll loop, 2s unref'd guard for a
  blocked pipe). Incident #1959 (force-exit truncating piped stdout) is the
  regression class; pinned by a 256KB real-pipe subprocess test.

- drainThenDisconnect (cli.ts): ONE owner-disconnect helper at all 8 sites
  (op-dispatch, CLI_ONLY fall-through, search dashboard, doctor remediation
  x3, ze-switch, dream, read-only timeout path). Drains the background-work
  registry, then disconnect (best-effort), bounded by the 10s unref'd
  hard-deadline — which is now armed around the TEARDOWN window only, not
  before the op handler (the old placement would have force-killed any op
  slower than 10s). Closes the filed TODOS P3 drain-hoist: six sites
  previously skipped the drain entirely and had no hang timer at all.

- Inner process.exit sweep: mid-handler exits in engine-owning/output-bearing
  paths (status, friction, claw-test, smoke-test, eval cross-modal /
  takes-quality replay / conversation-parser / whoknows-thin, status-thin)
  become process.exitCode + return so they flow through the drains and the
  flush-exit. Pre-engine usage/parse/refusal exits stay as-is.

BrainRegistry.disconnectAll deliberately unchanged: zero production callers
in src/, per-engine disconnects already bounded, and the kernel reclaims
sockets on exit (src/core/timeout.ts doctrine).

DAEMON_COMMANDS gains 'watch' ahead of the #2095 push transport.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/cli.ts

@@ -26,7 +26,7 @@ import type { BrainEngine } from './core/engine.ts';
 import { operations, OperationError } from './core/operations.ts';
 import type { Operation, OperationContext } from './core/operations.ts';
 import { drainAllBackgroundWorkForCliExit } from './core/background-work.ts';
-import { shouldForceExitAfterMain } from './core/cli-force-exit.ts';
+import { shouldForceExitAfterMain, flushStdoutThenExit } from './core/cli-force-exit.ts';
 import { serializeMarkdown } from './core/markdown.ts';
 import { parseGlobalFlags, setCliOptions, getCliOptions } from './core/cli-options.ts';
 import type { CliOptions } from './core/cli-options.ts';
@@ -260,7 +260,9 @@ async function main() {
         await withTimeout(runSearch(engine, subArgs), timeoutMs, label);
       }
     } finally {
-      await engine.disconnect();
+      // search stats/tune read the query_cache — a pending cache write must
+      // drain, not get killed by the deliberate exit (#2084 drain hoist).
+      await drainThenDisconnect(engine);
     }
     return;
   }
@@ -350,37 +352,6 @@ async function main() {
 
   // Local engine path (unchanged behavior for local installs).
   const engine = await connectEngine();
-  // v0.41.8.0 (#1247, #1269, #1290): the search / query / get_page
-  // op handlers fire-and-forget `bumpLastRetrievedAt` after returning
-  // results. On PGLite that IIFE keeps Bun's event loop alive past
-  // engine.disconnect(), hanging the CLI at ~95-98% CPU until SIGKILL.
-  // Drain the fire-and-forget set BEFORE disconnect; force-exit only
-  // if the drain itself times out (preserves stderr diagnostic signal
-  // AND guarantees the CLI doesn't re-hang at the disconnect layer).
-  //
-  // Defense-in-depth (adversarial-review C13): `engine.disconnect()` itself
-  // can hang on PGLite (db.close() or releaseLock racing OS-level FS state).
-  // Install an unref'd setTimeout hard-exit fallback BEFORE entering the
-  // try/catch/finally so a hung disconnect cannot defeat the force-exit
-  // contract. Daemons (`serve`) are excluded so they stay alive.
-  const DISCONNECT_HARD_DEADLINE_MS = 10_000;
-  let forceExitTimer: ReturnType<typeof setTimeout> | undefined;
-  if (shouldForceExitAfterMain()) {
-    forceExitTimer = setTimeout(() => {
-      console.warn(
-        `[cli] engine.disconnect() did not return within ${DISCONNECT_HARD_DEADLINE_MS}ms — force-exiting`,
-      );
-      // v0.42.20.0 (codex): honor an exit code an errored op already set —
-      // a bare process.exit(0) here would mask a failed op as success if the
-      // drain/disconnect then hangs.
-      process.exit(process.exitCode ?? 0);
-    }, DISCONNECT_HARD_DEADLINE_MS);
-    // unref so the timer itself doesn't keep the event loop alive — only
-    // the actual pending work (PGLite WASM handle) does. Without unref,
-    // we'd block a clean exit by 10s on every successful CLI run.
-    forceExitTimer.unref?.();
-  }
-
   try {
     const ctx = await makeContext(engine, params);
     const rawResult = await op.handler(ctx, params);
@@ -405,18 +376,64 @@ async function main() {
     }
     process.exitCode = 1;
   } finally {
-    // v0.42.20.0 — drain ALL fire-and-forget sinks (facts, last-retrieved,
-    // search-cache, eval-capture) via the background-work registry BEFORE
-    // disconnect, so a PGLite db.close() can't race in-flight work into the
-    // re-pump busy-loop (#1762). facts drains first (order 0) so its abort-path
-    // DB logIngest gets the freshest live-engine window. 1s per-sink timeout:
-    // read paths with no pending work pay the ~0ms fast path; capture/import
-    // that DO enqueue pay up to 1s (+ facts shutdown grace) while in-flight
-    // Haiku finishes. The unref'd hard-deadline timer above is the backstop if
-    // disconnect or a lingering socket keeps Bun's loop alive.
-    await drainAllBackgroundWorkForCliExit({ timeoutMs: 1000 });
-    await engine.disconnect();
-    if (forceExitTimer) clearTimeout(forceExitTimer);
+    // 1s per-sink drain timeout: read paths with no pending work pay the
+    // ~0ms fast path; capture/import that DO enqueue pay up to 1s (+ facts
+    // shutdown grace) while in-flight Haiku finishes.
+    await drainThenDisconnect(engine, { drainTimeoutMs: 1000 });
+  }
+}
+
+/**
+ * v0.43 (#2084, closes the TODOS P3 drain-hoist) — THE owner-disconnect for
+ * every CLI exit path. One helper, all sites, so a future teardown bug has
+ * one place to be wrong in.
+ *
+ *   drainThenDisconnect(engine)
+ *     ├── arm unref'd 10s hard-deadline (HUNG-teardown backstop — PGLite
+ *     │     db.close()/releaseLock can hang on OS-level FS state; the
+ *     │     entrypoint flush-exit can never fire if main() never resolves)
+ *     ├── drain background-work registry (facts → last-retrieved →
+ *     │     search-cache → eval-capture → volunteer-events), so a PGLite
+ *     │     db.close() can't race in-flight work into the re-pump
+ *     │     busy-loop (#1762)
+ *     └── engine.disconnect() (best-effort), then clear the deadline
+ *
+ * The 10s timer is armed HERE — around the teardown window only — not before
+ * the op handler (the pre-v0.43 placement would have force-killed any op
+ * slower than 10s). On the happy path the timer never fires: main() resolves
+ * and the entrypoint's flushStdoutThenExit ends the process deliberately
+ * (#2084's fix for lingering embedding/PgBouncer sockets riding the backstop).
+ */
+const DISCONNECT_HARD_DEADLINE_MS = 10_000;
+export async function drainThenDisconnect(
+  engine: BrainEngine,
+  opts?: { drainTimeoutMs?: number },
+): Promise<void> {
+  let deadlineTimer: ReturnType<typeof setTimeout> | undefined;
+  if (shouldForceExitAfterMain()) {
+    deadlineTimer = setTimeout(() => {
+      console.warn(
+        `[cli] engine.disconnect() did not return within ${DISCONNECT_HARD_DEADLINE_MS}ms — force-exiting`,
+      );
+      // Honor an exit code an errored op already set — a bare process.exit(0)
+      // here would mask a failed op as success if the drain/disconnect hangs.
+      process.exit(process.exitCode ?? 0);
+    }, DISCONNECT_HARD_DEADLINE_MS);
+    // unref so the timer itself doesn't keep the event loop alive — only
+    // the actual pending work (PGLite WASM handle) does.
+    deadlineTimer.unref?.();
+  }
+  try {
+    await drainAllBackgroundWorkForCliExit(
+      opts?.drainTimeoutMs !== undefined ? { timeoutMs: opts.drainTimeoutMs } : undefined,
+    );
+    try {
+      await engine.disconnect();
+    } catch {
+      /* best-effort — kernel reclaims sockets on exit (timeout.ts doctrine) */
+    }
+  } finally {
+    if (deadlineTimer) clearTimeout(deadlineTimer);
   }
 }
 
@@ -1123,11 +1140,16 @@ async function handleCliOnly(command: string, args: string[]) {
   }
   if (command === 'friction') {
     const { runFriction } = await import('./commands/friction.ts');
-    process.exit(runFriction(args));
+    // v0.43 (#2084 inner-exit sweep): exitCode + return instead of a
+    // mid-handler process.exit — flows through the entrypoint flush-exit
+    // so buffered stdout is never truncated.
+    process.exitCode = runFriction(args);
+    return;
   }
   if (command === 'claw-test') {
     const { runClawTest } = await import('./commands/claw-test.ts');
-    process.exit(await runClawTest(args));
+    process.exitCode = await runClawTest(args);
+    return;
   }
   if (command === 'report') {
     const { runReport } = await import('./commands/report.ts');
@@ -1173,13 +1195,13 @@ async function handleCliOnly(command: string, args: string[]) {
     if (args.includes('--remediation-plan')) {
       const { runRemediationPlan } = await import('./commands/doctor.ts');
       const eng = await connectEngine();
-      try { await runRemediationPlan(eng, args); } finally { await eng.disconnect(); }
+      try { await runRemediationPlan(eng, args); } finally { await drainThenDisconnect(eng); }
       return;
     }
     if (args.includes('--remediate')) {
       const { runRemediate } = await import('./commands/doctor.ts');
       const eng = await connectEngine();
-      try { await runRemediate(eng, args); } finally { await eng.disconnect(); }
+      try { await runRemediate(eng, args); } finally { await drainThenDisconnect(eng); }
       return;
     }
 
@@ -1195,7 +1217,7 @@ async function handleCliOnly(command: string, args: string[]) {
       try {
         const eng = await connectEngine();
         await runDoctor(eng, args);
-        await eng.disconnect();
+        await drainThenDisconnect(eng);
       } catch {
         // DB unavailable — still run filesystem checks
         await runDoctor(null, args, getDbUrlSource());
@@ -1212,7 +1234,7 @@ async function handleCliOnly(command: string, args: string[]) {
     try {
       await runZeSwitch(args, eng);
     } finally {
-      await eng.disconnect();
+      await drainThenDisconnect(eng);
     }
     return;
   }
@@ -1228,7 +1250,7 @@ async function handleCliOnly(command: string, args: string[]) {
       execSync(`bash "${scriptPath}"`, { stdio: 'inherit', env: { ...process.env } });
     } catch (e: any) {
       // Non-zero exit = some tests failed (exit code = failure count)
-      process.exit(e.status ?? 1);
+      process.exitCode = e.status ?? 1;
     }
     return;
   }
@@ -1262,7 +1284,9 @@ async function handleCliOnly(command: string, args: string[]) {
       // lifetime strictly dominating every borrower (lint/doctor probe engines
       // created mid-cycle). Do NOT disconnect `eng` before runDream returns, or
       // a borrower could outlive the owner and lose the shared singleton.
-      if (eng) await eng.disconnect();
+      // #2084 drain hoist: dream's cycle enqueues facts/last-retrieved
+      // writes — drain them against the live owner engine before teardown.
+      if (eng) await drainThenDisconnect(eng);
     }
     return;
   }
@@ -1274,7 +1298,8 @@ async function handleCliOnly(command: string, args: string[]) {
   // The handler self-configures the AI gateway from loadConfig() + process.env.
   if (command === 'eval' && args[0] === 'cross-modal') {
     const { runEvalCrossModal } = await import('./commands/eval-cross-modal.ts');
-    process.exit(await runEvalCrossModal(args.slice(1)));
+    process.exitCode = await runEvalCrossModal(args.slice(1));
+    return;
   }
 
   // v0.32 EXP-5 (codex review #10): `eval takes-quality replay <receipt>`
@@ -1285,7 +1310,8 @@ async function handleCliOnly(command: string, args: string[]) {
   // engine-required path below.
   if (command === 'eval' && args[0] === 'takes-quality' && args[1] === 'replay') {
     const { runReplayNoBrain } = await import('./commands/eval-takes-quality.ts');
-    process.exit(await runReplayNoBrain(args.slice(2)));
+    process.exitCode = await runReplayNoBrain(args.slice(2));
+    return;
   }
 
   // v0.28.8: longmemeval brings its own in-memory PGLite. Bypassing
@@ -1317,7 +1343,8 @@ async function handleCliOnly(command: string, args: string[]) {
   // gate runs on machines with no `~/.gbrain/config.json`.
   if (command === 'eval' && args[0] === 'conversation-parser') {
     const { runEvalConversationParser } = await import('./commands/eval-conversation-parser.ts');
-    process.exit(await runEvalConversationParser(args.slice(1)));
+    process.exitCode = await runEvalConversationParser(args.slice(1));
+    return;
   }
 
   // v0.41.13.0: `gbrain conversation-parser list-builtins | validate
@@ -1345,7 +1372,8 @@ async function handleCliOnly(command: string, args: string[]) {
     const cfgPre = loadConfig();
     if (isThinClient(cfgPre)) {
       const { runEvalWhoknows } = await import('./commands/eval-whoknows.ts');
-      process.exit(await runEvalWhoknows(null, args.slice(1)));
+      process.exitCode = await runEvalWhoknows(null, args.slice(1));
+      return;
     }
   }
 
@@ -1359,7 +1387,8 @@ async function handleCliOnly(command: string, args: string[]) {
     if (cfgPre && isThinClient(cfgPre)) {
       const { runStatus } = await import('./commands/status.ts');
       const result = await runStatus(null, args);
-      process.exit(result.exitCode);
+      process.exitCode = result.exitCode;
+      return;
     }
   }
 
@@ -1438,7 +1467,9 @@ async function handleCliOnly(command: string, args: string[]) {
       }
       throw e;
     } finally {
-      try { await engine.disconnect(); } catch { /* best-effort */ }
+      // #2084 drain hoist: `gbrain search` writes the query cache + bumps
+      // last_retrieved_at fire-and-forget — drain before disconnect.
+      await drainThenDisconnect(engine);
     }
     return;
   }
@@ -1672,8 +1703,10 @@ async function handleCliOnly(command: string, args: string[]) {
       case 'status': {
         const { runStatus } = await import('./commands/status.ts');
         const result = await runStatus(engine, args);
-        process.exit(result.exitCode);
-        // eslint-disable-next-line no-unreachable
+        // v0.43 (#2084 inner-exit sweep): a mid-switch process.exit skipped
+        // the finally's drain + disconnect entirely. exitCode + break flows
+        // through both, then the entrypoint flush-exit ends the process.
+        process.exitCode = result.exitCode;
         break;
       }
       // v0.38 — Capture: single human-facing entrypoint for ingestion.
@@ -1926,18 +1959,7 @@ async function handleCliOnly(command: string, args: string[]) {
     // #1471: this is also the fall-through OWNER-disconnect — the owner is torn
     // down LAST (after the drain), so module-singleton borrowers never outlive it.
     if (command !== 'serve') {
-      const forceExit = shouldForceExitAfterMain();
-      let hardExitTimer: ReturnType<typeof setTimeout> | undefined;
-      if (forceExit) {
-        hardExitTimer = setTimeout(() => {
-          console.warn('[cli] engine.disconnect() did not return within 10000ms — force-exiting');
-          process.exit(process.exitCode ?? 0);
-        }, 10_000);
-        hardExitTimer.unref?.();
-      }
-      await drainAllBackgroundWorkForCliExit();
-      await engine.disconnect();
-      if (hardExitTimer) clearTimeout(hardExitTimer);
+      await drainThenDisconnect(engine);
     }
   }
 }
@@ -2250,8 +2272,20 @@ Run gbrain <command> --help for command-specific help.
 // `bun src/cli.ts`). Guarded so tests can import cliAliases / printOpHelp
 // without triggering argv parsing + main(). v114 (#1941).
 if (import.meta.main) {
-  main().catch(e => {
-    console.error(e.message || e);
-    process.exit(1);
-  });
+  // v0.43 (#2084): exit DELIBERATELY when main() resolves. Lingering sockets
+  // (embedding fetch keep-alive, PgBouncer txn-mode sockets endPoolBounded
+  // raced past) keep Bun's event loop alive after bounded teardown resolves —
+  // pre-fix, every `gbrain query` paid a flat 10s tax riding the hard-deadline
+  // backstop. flushStdoutThenExit drains stdout/stderr first (incident #1959:
+  // a force-exit truncated piped stdout). Daemons (serve; interactive watch)
+  // are excluded by shouldForceExitAfterMain and keep the event loop.
+  main().then(
+    () => {
+      if (shouldForceExitAfterMain()) void flushStdoutThenExit(Number(process.exitCode ?? 0));
+    },
+    (e) => {
+      console.error(e instanceof Error ? (e.message || String(e)) : String(e));
+      void flushStdoutThenExit(1);
+    },
+  );
 }

src/core/cli-force-exit.ts

@@ -12,12 +12,14 @@
  * only force-exits after the drain timed out, NOT unconditionally for
  * every non-serve command.
  *
- * Daemon list is currently just `serve` (both stdio and HTTP forms use
- * the same command). If a future long-running command is added (e.g.
- * `gbrain watch` or `gbrain daemon`), add it here.
+ * Daemon list: `serve` (stdio + HTTP) and `watch` (stdin-follow push
+ * transport, v0.43 #2095 — interactive TTY sessions stay alive until
+ * Ctrl-C; its piped/EOF path exits via its own drain-then-exit lifecycle,
+ * not this gate). If a future long-running command is added (e.g.
+ * `gbrain daemon`), add it here.
  */
 
-const DAEMON_COMMANDS: ReadonlySet<string> = new Set(['serve']);
+const DAEMON_COMMANDS: ReadonlySet<string> = new Set(['serve', 'watch']);
 
 export function shouldForceExitAfterMain(
   argv: string[] = process.argv.slice(2),
@@ -26,3 +28,69 @@ export function shouldForceExitAfterMain(
   if (!command) return true;
   return !DAEMON_COMMANDS.has(command);
 }
+
+/**
+ * v0.43 (#2084) — deliberate exit after bounded teardown.
+ *
+ * Lingering sockets (embedding-provider fetch keep-alive, PgBouncer txn-mode
+ * sockets `endPoolBounded` raced past) keep Bun's event loop alive after
+ * teardown RESOLVES, so the CLI used to ride the 10s hard-deadline backstop
+ * on every `gbrain query`. The fix is to exit on purpose the moment main()
+ * resolves — but only after stdout/stderr have actually drained: incident
+ * #1959 (see src/core/db.ts) was a force-exit truncating piped stdout
+ * mid-payload.
+ *
+ * Flush contract: a stream is drained when `writableLength === 0` — bytes
+ * queued in the JS-side buffer are the only ones `process.exit()` can lose
+ * (the kernel owns anything already written to the fd). `writableNeedDrain`
+ * is NOT sufficient (it only says "below high-water mark"). We wake on
+ * 'drain' when it fires, and poll on a short tick because 'drain' is only
+ * emitted after a write() returned false — a small queued chunk can flush
+ * without ever signalling. A blocked pipe (reader stopped consuming) is
+ * bounded by `guardMs`: partial output to a stalled reader is unavoidable,
+ * hanging the process is not.
+ */
+export interface FlushableStream {
+  writableLength?: number;
+  once(event: 'drain', listener: () => void): unknown;
+  off?(event: 'drain', listener: () => void): unknown;
+  removeListener?(event: 'drain', listener: () => void): unknown;
+}
+
+export async function flushStdoutThenExit(
+  code: number,
+  deps?: {
+    streams?: FlushableStream[];
+    exit?: (code: number) => void;
+    guardMs?: number;
+  },
+): Promise<void> {
+  const streams = deps?.streams ?? [
+    process.stdout as unknown as FlushableStream,
+    process.stderr as unknown as FlushableStream,
+  ];
+  const exit = deps?.exit ?? ((c: number) => process.exit(c));
+  const guardMs = deps?.guardMs ?? 2000;
+  const deadline = Date.now() + guardMs;
+
+  for (const stream of streams) {
+    while ((stream.writableLength ?? 0) > 0 && Date.now() < deadline) {
+      await new Promise<void>((resolve) => {
+        const onDrain = () => {
+          clearTimeout(tick);
+          resolve();
+        };
+        // Poll tick: 'drain' only fires after a backpressured write, so a
+        // buffer that empties without one needs the re-check. unref'd so the
+        // wait itself never holds the loop open.
+        const tick = setTimeout(() => {
+          (stream.off ?? stream.removeListener)?.call(stream, 'drain', onDrain);
+          resolve();
+        }, 25);
+        (tick as { unref?: () => void }).unref?.();
+        stream.once('drain', onDrain);
+      });
+    }
+  }
+  exit(code);
+}

test/cli-drain-then-disconnect.test.ts

@@ -0,0 +1,99 @@
+/**
+ * v0.43 (#2084 drain hoist, closes TODOS P3) — drainThenDisconnect unit tests.
+ *
+ * One helper owns every CLI owner-disconnect: drain the background-work
+ * registry FIRST, then disconnect (best-effort), bounded by the unref'd
+ * hard-deadline. Pre-hoist, six bare sites (search dashboard, doctor
+ * remediation, ze-switch, dream, read-only timeout path) skipped the drain —
+ * an in-flight fire-and-forget write was killed at disconnect/exit.
+ */
+
+import { describe, test, expect } from 'bun:test';
+import { drainThenDisconnect } from '../src/cli.ts';
+import { __registerDrainerForTest } from '../src/core/background-work.ts';
+import type { BrainEngine } from '../src/core/engine.ts';
+
+function makeEngine(events: string[], opts?: { throwOnDisconnect?: boolean }): BrainEngine {
+  return {
+    disconnect: async () => {
+      events.push('disconnect');
+      if (opts?.throwOnDisconnect) throw new Error('disconnect blew up');
+    },
+  } as unknown as BrainEngine;
+}
+
+describe('drainThenDisconnect — drain registry, then disconnect, bounded', () => {
+  test('drains registered sinks BEFORE engine.disconnect()', async () => {
+    const events: string[] = [];
+    const unregister = __registerDrainerForTest({
+      name: 'test-sink-order',
+      order: 99,
+      drain: async () => {
+        events.push('drain');
+        return { unfinished: 0 };
+      },
+    });
+    try {
+      await drainThenDisconnect(makeEngine(events));
+    } finally {
+      unregister();
+    }
+    expect(events.indexOf('drain')).toBeGreaterThanOrEqual(0);
+    expect(events.indexOf('disconnect')).toBeGreaterThan(events.indexOf('drain'));
+  });
+
+  test('a pending fire-and-forget write survives (drained, not killed)', async () => {
+    // Simulates the search-stats path: a cache write is in flight when the
+    // command finishes. The hoisted drain must let it settle before teardown.
+    const events: string[] = [];
+    let settled = false;
+    const pending = new Promise<void>((r) =>
+      setTimeout(() => {
+        settled = true;
+        events.push('write-settled');
+        r();
+      }, 30),
+    );
+    const unregister = __registerDrainerForTest({
+      name: 'test-sink-pending-write',
+      order: 99,
+      drain: async () => {
+        await pending;
+        return { unfinished: 0 };
+      },
+    });
+    try {
+      await drainThenDisconnect(makeEngine(events));
+    } finally {
+      unregister();
+    }
+    expect(settled).toBe(true);
+    expect(events).toEqual(['write-settled', 'disconnect']);
+  });
+
+  test('disconnect failure is swallowed (best-effort; kernel reclaims on exit)', async () => {
+    const events: string[] = [];
+    await expect(
+      drainThenDisconnect(makeEngine(events, { throwOnDisconnect: true })),
+    ).resolves.toBeUndefined();
+    expect(events).toEqual(['disconnect']);
+  });
+
+  test('honors the per-site drain timeout passthrough', async () => {
+    const seen: number[] = [];
+    const unregister = __registerDrainerForTest({
+      name: 'test-sink-timeout',
+      order: 99,
+      drain: async (timeoutMs: number) => {
+        seen.push(timeoutMs);
+        return { unfinished: 0 };
+      },
+    });
+    try {
+      await drainThenDisconnect(makeEngine([]), { drainTimeoutMs: 1000 });
+    } finally {
+      unregister();
+    }
+    expect(seen).toEqual([1000]);
+  });
+});

test/cli-flush-exit.test.ts

@@ -0,0 +1,139 @@
+/**
+ * v0.43 (#2084) — flushStdoutThenExit unit tests.
+ *
+ * The deliberate-exit path must never truncate buffered output (incident
+ * #1959: a force-exit cut piped stdout mid-payload) and must never hang on
+ * a blocked pipe. Streams + exit are injected so the tests drive both
+ * properties without killing the test process.
+ */
+
+import { describe, test, expect } from 'bun:test';
+import { flushStdoutThenExit, type FlushableStream } from '../src/core/cli-force-exit.ts';
+
+/** Minimal fake of the stdout/stderr surface flushStdoutThenExit touches. */
+function makeStream(initialLength: number): FlushableStream & {
+  setLength(n: number): void;
+  emitDrain(): void;
+  drainListeners(): number;
+} {
+  let length = initialLength;
+  const listeners: Array<() => void> = [];
+  return {
+    get writableLength() {
+      return length;
+    },
+    once(_event: 'drain', listener: () => void) {
+      listeners.push(listener);
+      return this;
+    },
+    off(_event: 'drain', listener: () => void) {
+      const i = listeners.indexOf(listener);
+      if (i >= 0) listeners.splice(i, 1);
+      return this;
+    },
+    setLength(n: number) {
+      length = n;
+    },
+    emitDrain() {
+      const current = listeners.splice(0, listeners.length);
+      for (const l of current) l();
+    },
+    drainListeners() {
+      return listeners.length;
+    },
+  };
+}
+
+describe('flushStdoutThenExit — deliberate exit after output drains', () => {
+  test('exits immediately with the given code when both streams are drained', async () => {
+    const calls: number[] = [];
+    await flushStdoutThenExit(0, {
+      streams: [makeStream(0), makeStream(0)],
+      exit: (c) => calls.push(c),
+    });
+    expect(calls).toEqual([0]);
+  });
+
+  test('honors a non-zero exit code (errored op sets process.exitCode=1)', async () => {
+    const calls: number[] = [];
+    await flushStdoutThenExit(1, {
+      streams: [makeStream(0)],
+      exit: (c) => calls.push(c),
+    });
+    expect(calls).toEqual([1]);
+  });
+
+  test('waits for a pending buffer to drain before exiting', async () => {
+    const stream = makeStream(4096);
+    const calls: number[] = [];
+    const done = flushStdoutThenExit(0, {
+      streams: [stream],
+      exit: (c) => calls.push(c),
+      guardMs: 5000,
+    });
+    // Not exited while bytes are queued.
+    await new Promise((r) => setTimeout(r, 10));
+    expect(calls).toEqual([]);
+    stream.setLength(0);
+    stream.emitDrain();
+    await done;
+    expect(calls).toEqual([0]);
+  });
+
+  test('drains stdout AND stderr — stderr alone can hold the exit', async () => {
+    const stdout = makeStream(0);
+    const stderr = makeStream(2048);
+    const calls: number[] = [];
+    const done = flushStdoutThenExit(1, {
+      streams: [stdout, stderr],
+      exit: (c) => calls.push(c),
+      guardMs: 5000,
+    });
+    await new Promise((r) => setTimeout(r, 10));
+    expect(calls).toEqual([]);
+    stderr.setLength(0);
+    stderr.emitDrain();
+    await done;
+    expect(calls).toEqual([1]);
+  });
+
+  test('poll tick drains without a drain event ("drain" only fires after backpressure)', async () => {
+    const stream = makeStream(512);
+    const calls: number[] = [];
+    const done = flushStdoutThenExit(0, {
+      streams: [stream],
+      exit: (c) => calls.push(c),
+      guardMs: 5000,
+    });
+    // Buffer empties WITHOUT an emitted drain event — the 25ms poll must see it.
+    setTimeout(() => stream.setLength(0), 30);
+    await done;
+    expect(calls).toEqual([0]);
+  });
+
+  test('blocked pipe: the guard deadline exits anyway (never hangs)', async () => {
+    const stream = makeStream(65536); // reader stopped consuming; never drains
+    const calls: number[] = [];
+    const t0 = Date.now();
+    await flushStdoutThenExit(0, {
+      streams: [stream],
+      exit: (c) => calls.push(c),
+      guardMs: 120,
+    });
+    expect(calls).toEqual([0]);
+    expect(Date.now() - t0).toBeLessThan(2000);
+  });
+
+  test('removes its drain listener when the poll tick wins (no listener leak)', async () => {
+    const stream = makeStream(256);
+    const calls: number[] = [];
+    const done = flushStdoutThenExit(0, {
+      streams: [stream],
+      exit: (c) => calls.push(c),
+      guardMs: 5000,
+    });
+    setTimeout(() => stream.setLength(0), 30);
+    await done;
+    expect(stream.drainListeners()).toBe(0);
+  });
+});

test/cli-pipe-truncation.test.ts

@@ -0,0 +1,108 @@
+/**
+ * v0.43 (#2084) — CRITICAL regression: piped stdout is never truncated by the
+ * deliberate flush-exit (eng-review D3; the incident #1959 class).
+ *
+ * Two layers:
+ *  1. A subprocess that writes 256KB to a REAL pipe (4x the 64KB kernel pipe
+ *     buffer, so write() backpressures and bytes sit in the JS-side stream
+ *     buffer) and then calls the production `flushStdoutThenExit`. If the
+ *     flush gate is wrong, the tail of the payload is sheared off — this is
+ *     exactly how incident #1959 presented ("relational query came back
+ *     empty" — output truncated by a force-exit).
+ *  2. The real CLI (`bun src/cli.ts --tools-json`, engine-free) over a pipe:
+ *     parseable, byte-stable across runs, and exits deliberately (well under
+ *     the 10s backstop).
+ */
+
+import { describe, test, expect } from 'bun:test';
+import { spawnSync } from 'child_process';
+import { mkdtempSync, writeFileSync, rmSync } from 'fs';
+import { join, resolve } from 'path';
+import { tmpdir } from 'os';
+
+const REPO = resolve(import.meta.dir, '..');
+const CLI = join(REPO, 'src', 'cli.ts');
+
+const PAYLOAD_BYTES = 256 * 1024;
+
+describe('cli pipe truncation — deliberate exit flushes piped stdout (#2084)', () => {
+  test('256KB through a real pipe survives flushStdoutThenExit byte-exactly', () => {
+    const dir = mkdtempSync(join(tmpdir(), 'gbrain-flush-'));
+    const script = join(dir, 'flush-child.ts');
+    try {
+      writeFileSync(
+        script,
+        [
+          `import { flushStdoutThenExit } from ${JSON.stringify(join(REPO, 'src', 'core', 'cli-force-exit.ts'))};`,
+          `const payload = 'x'.repeat(${PAYLOAD_BYTES}) + 'END\\n';`,
+          // No await — mirrors the cli.ts entrypoint, which fires the flush
+          // exit as a dangling promise after main() resolves.
+          `process.stdout.write(payload);`,
+          `void flushStdoutThenExit(0);`,
+        ].join('\n'),
+      );
+      const res = spawnSync('bun', [script], {
+        stdio: ['ignore', 'pipe', 'pipe'],
+        encoding: 'utf-8',
+        timeout: 30_000,
+        maxBuffer: 16 * 1024 * 1024,
+      });
+      expect(res.status).toBe(0);
+      const out = res.stdout ?? '';
+      expect(Buffer.byteLength(out, 'utf-8')).toBe(PAYLOAD_BYTES + 4);
+      expect(out.endsWith('END\n')).toBe(true);
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+    }
+  }, 60_000);
+
+  test('exit code survives the flush (errored command stays non-zero)', () => {
+    const dir = mkdtempSync(join(tmpdir(), 'gbrain-flush-code-'));
+    const script = join(dir, 'flush-code-child.ts');
+    try {
+      writeFileSync(
+        script,
+        [
+          `import { flushStdoutThenExit } from ${JSON.stringify(join(REPO, 'src', 'core', 'cli-force-exit.ts'))};`,
+          `process.stdout.write('partial output before failure\\n');`,
+          `void flushStdoutThenExit(3);`,
+        ].join('\n'),
+      );
+      const res = spawnSync('bun', [script], {
+        stdio: ['ignore', 'pipe', 'pipe'],
+        encoding: 'utf-8',
+        timeout: 30_000,
+      });
+      expect(res.status).toBe(3);
+      expect(res.stdout).toBe('partial output before failure\n');
+    } finally {
+      rmSync(dir, { recursive: true, force: true });
+    }
+  }, 60_000);
+
+  test('real CLI: --tools-json over a pipe is complete, parseable, byte-stable, and prompt', () => {
+    const run = () => {
+      const t0 = Date.now();
+      const res = spawnSync('bun', [CLI, '--tools-json'], {
+        stdio: ['ignore', 'pipe', 'pipe'],
+        encoding: 'utf-8',
+        timeout: 60_000,
+        env: { ...process.env, GBRAIN_SKIP_STARTUP_HOOKS: '1' },
+        maxBuffer: 64 * 1024 * 1024,
+      });
+      return { stdout: res.stdout ?? '', status: res.status, ms: Date.now() - t0 };
+    };
+    const first = run();
+    expect(first.status).toBe(0);
+    expect(Buffer.byteLength(first.stdout, 'utf-8')).toBeGreaterThan(16 * 1024);
+    // Truncated JSON does not parse — the strongest single-run completeness check.
+    const parsed = JSON.parse(first.stdout);
+    expect(Array.isArray(parsed)).toBe(true);
+    // Deliberate exit, not the 10s hard-deadline backstop.
+    expect(first.ms).toBeLessThan(9_000);
+
+    const second = run();
+    expect(second.status).toBe(0);
+    expect(second.stdout).toBe(first.stdout);
+  }, 180_000);
+});

test/cli-should-force-exit.test.ts

@@ -70,4 +70,25 @@ describe('shouldForceExitAfterMain — daemon survival gate', () => {
     expect(shouldForceExitAfterMain(['serves'])).toBe(true);
     expect(shouldForceExitAfterMain(['serve-cluster'])).toBe(true);
   });
+
+  test('returns false for `watch` (v0.43 #2095 stdin-follow push daemon)', () => {
+    expect(shouldForceExitAfterMain(['watch'])).toBe(false);
+    expect(shouldForceExitAfterMain(['watch', '--json'])).toBe(false);
+    expect(shouldForceExitAfterMain(['--quiet', 'watch'])).toBe(false);
+  });
+
+  test('awaited long-runners exit deliberately when their handler resolves', () => {
+    // `jobs work`, `jobs watch --follow`, and `autopilot` BLOCK inside their
+    // awaited handler until done — when main() resolves for them, the work is
+    // over and the deliberate exit is correct (v0.43 #2084 contract). Only
+    // commands that RETURN from main() while the event loop carries the
+    // daemon (`serve`; interactive `watch`) belong in DAEMON_COMMANDS.
+    expect(shouldForceExitAfterMain(['jobs', 'work'])).toBe(true);
+    expect(shouldForceExitAfterMain(['jobs', 'watch', '--follow'])).toBe(true);
+    expect(shouldForceExitAfterMain(['autopilot'])).toBe(true);
+  });
+
+  test('substring match avoidance: `watcher` is NOT `watch`', () => {
+    expect(shouldForceExitAfterMain(['watcher'])).toBe(true);
+  });
 });