fix(storage): wait for storage workers to exit before the test file boundary

Windows CI died with `panic(...): Internal assertion failure` and no `(fail)`
line — four runs across `dev`, #771, #780 and #781. The crash header names the
shape: `workers_spawned(9) workers_terminated(8)`, one worker never reclaimed,
and it lands exactly at the `api-storage-policy` -> `api-storage` file
boundary. That suite is the only one that spawns policy workers.

`Worker.terminate()` returns void and does not wait for the thread to go away.
Under `bun test --isolate` the harness reclaims a file's realm at the boundary,
so a worker still exiting at that instant trips a Bun-internal assertion and
takes the whole run down. macOS and Linux tolerate the same race, which is why
it reads as a Windows-only flake.

`worker-lifecycle.ts` tracks live workers and terminates them through the
`close` event, so callers can await the thread actually being gone rather than
a timer they hope is long enough; a 5s cap keeps a wedged worker from hanging
teardown. Both storage jobs settle their run promise after that await, and the
policy suite's `afterEach` now drains.

devlog `_plan/260730_remote_issue_merge_round/150` reproduced this panic twice
against an unchanged tree, confirmed the same tree passed in PR context, and
left this defence as the follow-up if it recurred. It recurred.

Evidence: 6190 pass / 0 fail across 445 files, typecheck clean, privacy scan
green. The regression test asserts a worker was really spawned before checking
it is gone, so it cannot pass vacuously.

Author: lidge-jun

src/storage/policy-job.ts

@@ -18,6 +18,11 @@ import {
   type PolicyRunResult,
   type PolicySkipReason,
 } from "./policy";
+import {
+  drainStorageWorkers,
+  registerStorageWorker,
+  terminateStorageWorker,
+} from "./worker-lifecycle";
 
 export type PolicyJobStatus = "idle" | "running";
 
@@ -127,7 +132,7 @@ function disownActiveRun(): void {
 export function resetStorageCleanupPolicyJobForTests(): void {
   disownActiveRun();
   if (activeWorker) {
-    try { activeWorker.terminate(); } catch { /* */ }
+    void terminateStorageWorker(activeWorker);
     activeWorker = null;
   }
   inflight = null;
@@ -136,11 +141,24 @@ export function resetStorageCleanupPolicyJobForTests(): void {
   state = { status: "idle" };
 }
 
+/**
+ * Await-able sibling of the reset above, for test teardown.
+ *
+ * `bun test --isolate` reclaims a file's realm at the file boundary. A storage
+ * worker still exiting at that moment trips a Bun-internal assertion on Windows
+ * and takes the whole run down, so a suite that spawns workers must be able to
+ * wait for them rather than fire-and-forget.
+ */
+export async function resetStorageCleanupPolicyJobForTestsAsync(): Promise<void> {
+  resetStorageCleanupPolicyJobForTests();
+  await drainStorageWorkers();
+}
+
 /** Terminate an in-flight worker during process shutdown. */
 export function abortStorageCleanupPolicyJob(): void {
   disownActiveRun();
   if (activeWorker) {
-    try { activeWorker.terminate(); } catch { /* */ }
+    void terminateStorageWorker(activeWorker);
     activeWorker = null;
   }
   releaseHeldMutationSlot();
@@ -218,13 +236,14 @@ function runInWorker(opts: RequestPolicyRunOptions & { blockMs?: number }): Prom
     const requestId = crypto.randomUUID();
     let settled = false;
     const worker = new Worker(new URL("./policy-worker.ts", import.meta.url).href);
+    registerStorageWorker(worker);
     activeWorker = worker;
 
     const timer = setTimeout(() => {
       if (settled) return;
       settled = true;
       cancelActiveRun = null;
-      try { worker.terminate(); } catch { /* */ }
+      void terminateStorageWorker(worker);
       if (activeWorker === worker) activeWorker = null;
       reject(new Error("storage_cleanup_worker_timeout"));
     }, WORKER_TIMEOUT_MS);
@@ -235,8 +254,10 @@ function runInWorker(opts: RequestPolicyRunOptions & { blockMs?: number }): Prom
       cancelActiveRun = null;
       clearTimeout(timer);
       if (activeWorker === worker) activeWorker = null;
-      try { worker.terminate(); } catch { /* */ }
-      fn();
+      // Settle the caller only after the thread is actually gone, so a suite
+      // that awaits its request cannot reach the next test file with a worker
+      // still exiting behind it.
+      void terminateStorageWorker(worker).then(fn, fn);
     };
 
     cancelActiveRun = () => {

src/storage/restore-job.ts

@@ -16,6 +16,11 @@ import {
   withStorageMutationSlot,
   type StorageMutationCoordinatorTestHooks,
 } from "./storage-mutation-coordinator";
+import {
+  drainStorageWorkers,
+  registerStorageWorker,
+  terminateStorageWorker,
+} from "./worker-lifecycle";
 
 export interface RestoreJobTestHooks extends StorageMutationCoordinatorTestHooks {
   /**
@@ -69,7 +74,7 @@ export function setRestoreTrashJobTestHooks(hooks: RestoreJobTestHooks | null):
 
 export function resetRestoreTrashJobForTests(): void {
   if (activeWorker) {
-    try { activeWorker.terminate(); } catch { /* */ }
+    void terminateStorageWorker(activeWorker);
     activeWorker = null;
   }
   cancelActiveRun?.();
@@ -78,10 +83,16 @@ export function resetRestoreTrashJobForTests(): void {
   resetStorageMutationCoordinatorForTests();
 }
 
+/** Await-able reset for test teardown; see policy-job's equivalent for why. */
+export async function resetRestoreTrashJobForTestsAsync(): Promise<void> {
+  resetRestoreTrashJobForTests();
+  await drainStorageWorkers();
+}
+
 /** Terminate an in-flight worker during process shutdown. */
 export function abortRestoreTrashJob(): void {
   if (activeWorker) {
-    try { activeWorker.terminate(); } catch { /* */ }
+    void terminateStorageWorker(activeWorker);
     activeWorker = null;
   }
   cancelActiveRun?.();
@@ -117,13 +128,14 @@ function runInWorker(opts: {
     const requestId = crypto.randomUUID();
     let settled = false;
     const worker = new Worker(new URL("./restore-worker.ts", import.meta.url).href);
+    registerStorageWorker(worker);
     activeWorker = worker;
 
     const timer = setTimeout(() => {
       if (settled) return;
       settled = true;
       cancelActiveRun = null;
-      try { worker.terminate(); } catch { /* */ }
+      void terminateStorageWorker(worker);
       if (activeWorker === worker) activeWorker = null;
       reject(new Error("restore_worker_timeout"));
     }, WORKER_TIMEOUT_MS);
@@ -134,8 +146,7 @@ function runInWorker(opts: {
       cancelActiveRun = null;
       clearTimeout(timer);
       if (activeWorker === worker) activeWorker = null;
-      try { worker.terminate(); } catch { /* */ }
-      fn();
+      void terminateStorageWorker(worker).then(fn, fn);
     };
 
     cancelActiveRun = () => {

src/storage/worker-lifecycle.ts

@@ -0,0 +1,81 @@
+/**
+ * Deterministic teardown for the storage Bun Workers.
+ *
+ * `Worker.terminate()` returns void and does NOT wait for the thread to be
+ * reclaimed. Every caller here used to fire it and move on, which is fine for
+ * the proxy but not for `bun test --isolate`: the harness tears a test file's
+ * realm down at the file boundary, and on Windows a worker that has not
+ * finished exiting by then trips a Bun-internal assertion and kills the whole
+ * run.
+ *
+ * The crash header names the shape exactly — `workers_spawned(9)
+ * workers_terminated(8)`, one worker still alive — and it lands right at the
+ * `api-storage-policy` → `api-storage` file boundary, the only suite that
+ * spawns policy workers. See `devlog/_plan/260730_remote_issue_merge_round/150`,
+ * which reproduced the panic twice against an unchanged tree and left this
+ * defence as the follow-up if it ever recurred. It recurred.
+ *
+ * So: keep a registry of live workers, and give shutdown/reset paths something
+ * they can actually await. `close` is Bun's post-exit event for a worker
+ * thread, so awaiting it is the real "the thread is gone" signal rather than a
+ * timer we hope is long enough. The timeout only exists so a wedged worker
+ * cannot hang a test teardown forever.
+ */
+
+const liveWorkers = new Set<Worker>();
+
+/** Track a freshly spawned worker so teardown can wait for it later. */
+export function registerStorageWorker(worker: Worker): void {
+  liveWorkers.add(worker);
+}
+
+/**
+ * Terminate a worker and resolve once its thread has actually exited.
+ *
+ * Safe to call twice: the second call finds the worker already deregistered and
+ * resolves immediately.
+ */
+export function terminateStorageWorker(worker: Worker, timeoutMs = 5_000): Promise<void> {
+  if (!liveWorkers.has(worker)) {
+    try { worker.terminate(); } catch { /* already gone */ }
+    return Promise.resolve();
+  }
+  liveWorkers.delete(worker);
+
+  return new Promise<void>(resolve => {
+    let done = false;
+    const settle = (): void => {
+      if (done) return;
+      done = true;
+      clearTimeout(timer);
+      resolve();
+    };
+    // A worker that refuses to exit must not wedge teardown; the proxy path
+    // never awaits this, and a test teardown would rather continue than hang.
+    const timer = setTimeout(settle, timeoutMs);
+    try {
+      worker.addEventListener("close", settle, { once: true });
+    } catch {
+      // No close event available — fall back to the timeout above.
+    }
+    try {
+      worker.terminate();
+    } catch {
+      settle();
+    }
+  });
+}
+
+/**
+ * Await every worker this module still tracks. Used by test resets so no
+ * storage worker outlives the file that spawned it.
+ */
+export async function drainStorageWorkers(timeoutMs = 5_000): Promise<void> {
+  const pending = [...liveWorkers];
+  await Promise.all(pending.map(worker => terminateStorageWorker(worker, timeoutMs)));
+}
+
+/** Live worker count — exported so a regression test can assert the invariant. */
+export function liveStorageWorkerCount(): number {
+  return liveWorkers.size;
+}

tests/api-storage-policy.test.ts

@@ -14,6 +14,7 @@ import {
 } from "../src/storage/cleanup-job";
 import {
   resetStorageCleanupPolicyJobForTests,
+  resetStorageCleanupPolicyJobForTestsAsync,
   setStorageCleanupPolicyJobTestHooks,
 } from "../src/storage/policy-job";
 import { stopStorageCleanupScheduler } from "../src/storage/policy-scheduler";
@@ -114,9 +115,12 @@ beforeEach(() => {
   resetArchivedCleanupJobForTests();
 });
 
-afterEach(() => {
+afterEach(async () => {
   stopStorageCleanupScheduler();
-  resetStorageCleanupPolicyJobForTests();
+  // Await the worker teardown: under `bun test --isolate` an exiting Bun Worker
+  // that outlives its test file crashes the Windows runner at the file
+  // boundary (workers_spawned > workers_terminated in the panic header).
+  await resetStorageCleanupPolicyJobForTestsAsync();
   setStorageCleanupPolicyJobTestHooks(null);
   resetArchivedCleanupJobForTests();
   setArchivedCleanupJobTestHooks(null);

tests/storage-worker-lifecycle.test.ts

@@ -0,0 +1,109 @@
+/**
+ * The storage workers must be GONE, not merely asked to stop, by the time a
+ * test file's realm is torn down.
+ *
+ * `bun test --isolate` reclaims the realm at the file boundary. A Bun Worker
+ * still exiting at that instant trips an internal assertion on Windows and
+ * kills the entire run — no `(fail)` line, just
+ * `panic(...): Internal assertion failure` with `workers_spawned(N)
+ * workers_terminated(N-1)` in the header. That crashed `dev` and three PRs at
+ * the `api-storage-policy` → `api-storage` boundary, the only place policy
+ * workers are spawned.
+ *
+ * These assertions are the invariant the fix rests on: after a run settles, and
+ * after a reset, nothing is left tracked.
+ */
+import { afterEach, beforeEach, expect, test } from "bun:test";
+import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { Database } from "bun:sqlite";
+import {
+  requestStorageCleanupPolicyRun,
+  getStorageCleanupPolicyJobState,
+  resetStorageCleanupPolicyJobForTestsAsync,
+  setStorageCleanupPolicyJobTestHooks,
+} from "../src/storage/policy-job";
+import { liveStorageWorkerCount } from "../src/storage/worker-lifecycle";
+import { installIsolatedCodexHome, type IsolatedCodexHome } from "./helpers/isolated-codex-home";
+
+let isolatedCodexHome: IsolatedCodexHome | null = null;
+let testDir = "";
+let previousHome: string | undefined;
+
+function seedArchived(codexHome: string): void {
+  mkdirSync(join(codexHome, "archived_sessions"), { recursive: true });
+  writeFileSync(join(codexHome, "archived_sessions", "rollout-old.jsonl"), "o".repeat(100));
+  const db = new Database(join(codexHome, "state_5.sqlite"));
+  db.exec(`CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT NOT NULL, archived INTEGER)`);
+  db.exec(`INSERT INTO threads VALUES ('told','archived_sessions/rollout-old.jsonl',1)`);
+  db.close();
+}
+
+beforeEach(() => {
+  previousHome = process.env.OPENCODEX_HOME;
+  isolatedCodexHome = installIsolatedCodexHome("ocx-worker-lifecycle-codex-");
+  testDir = mkdtempSync(join(tmpdir(), "ocx-worker-lifecycle-"));
+  process.env.OPENCODEX_HOME = testDir;
+});
+
+afterEach(async () => {
+  await resetStorageCleanupPolicyJobForTestsAsync();
+  setStorageCleanupPolicyJobTestHooks(null);
+  if (previousHome === undefined) delete process.env.OPENCODEX_HOME;
+  else process.env.OPENCODEX_HOME = previousHome;
+  isolatedCodexHome?.restore();
+  isolatedCodexHome = null;
+  if (testDir) rmSync(testDir, { recursive: true, force: true });
+  testDir = "";
+});
+
+async function waitForIdle(timeoutMs = 20_000): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (getStorageCleanupPolicyJobState().status === "idle") return;
+    await Bun.sleep(25);
+  }
+  throw new Error("policy job did not settle");
+}
+
+/** Guards against a vacuous pass: assert we really did spawn a worker. */
+async function waitForLiveWorker(timeoutMs = 10_000): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (liveStorageWorkerCount() > 0) return;
+    await Bun.sleep(5);
+  }
+  throw new Error("no storage worker was ever spawned; this test would prove nothing");
+}
+
+test("a settled policy worker leaves nothing alive behind it", async () => {
+  seedArchived(isolatedCodexHome!.path);
+  const started = requestStorageCleanupPolicyRun({
+    reason: "manual",
+    codexHome: isolatedCodexHome!.path,
+  });
+  expect(started.accepted).toBe(true);
+
+  await waitForLiveWorker();
+  await waitForIdle();
+  // The run reached a terminal state, so its worker thread must already be
+  // reclaimed — not merely sent a terminate() that has yet to land.
+  expect(liveStorageWorkerCount()).toBe(0);
+}, { timeout: 30_000 });
+
+test("reset drains a worker that is still blocked mid-run", async () => {
+  // A worker held inside its run is exactly the state that outlived the file
+  // boundary in CI, so tear it down while it is still busy.
+  setStorageCleanupPolicyJobTestHooks({ blockMs: 1_500 });
+  seedArchived(isolatedCodexHome!.path);
+  const started = requestStorageCleanupPolicyRun({
+    reason: "manual",
+    codexHome: isolatedCodexHome!.path,
+  });
+  expect(started.accepted).toBe(true);
+
+  await waitForLiveWorker();
+  await resetStorageCleanupPolicyJobForTestsAsync();
+  expect(liveStorageWorkerCount()).toBe(0);
+}, { timeout: 30_000 });