test(windows): close the four residual shard-2 Windows failures

src/storage/cleanup.ts

@@ -2578,6 +2578,11 @@ export interface RestoreTestHooks {
    * reconcile, so cleanup can race an in-flight restore.
    */
   holdAfterFileMovesMs?: number;
+  /**
+   * Test-only: publish a ready file after rollout moves, then wait until the
+   * release file exists. This makes cross-thread race tests phase-driven.
+   */
+  pauseAfterFileMoves?: { readyPath: string; releasePath: string };
 }
 
 /**
@@ -2917,6 +2922,11 @@ export function restoreTrashEntry(
     return { ok: false, trashDir: id, ...partialCounts, error };
   };
 
+  if (hooks?.pauseAfterFileMoves) {
+    writeFileSync(hooks.pauseAfterFileMoves.readyPath, "ready\n");
+    while (!existsSync(hooks.pauseAfterFileMoves.releasePath)) Bun.sleepSync(10);
+  }
+
   if (hooks?.holdAfterFileMovesMs !== undefined) {
     const holdMs = Math.max(0, Math.floor(hooks.holdAfterFileMovesMs));
     if (holdMs > 0) {

src/storage/storage-mutation-coordinator.ts

@@ -16,6 +16,12 @@ export type StorageMutationBusyError = "storage_mutation_busy";
 export interface StorageMutationCoordinatorTestHooks {
   /** Block after acquiring the slot, before mutation work (race tests). */
   blockMs?: number;
+  /** Test-only cross-thread handshake after one mutation kind acquires its slot. */
+  pauseAfterAcquire?: {
+    kind: StorageMutationKind;
+    readyPath: string;
+    releasePath: string;
+  };
 }
 
 interface ActiveSlot {
@@ -93,7 +99,12 @@ export function endStorageMutation(codexHome?: string): void {
   slot.lease.release();
 }
 
-async function applyCoordinatorBlock(): Promise<void> {
+async function applyCoordinatorBlock(kind: StorageMutationKind): Promise<void> {
+  const pause = testHooks?.pauseAfterAcquire;
+  if (pause?.kind === kind) {
+    await Bun.write(pause.readyPath, "ready\n");
+    while (!Bun.file(pause.releasePath).size) await Bun.sleep(10);
+  }
   const blockMs = testHooks?.blockMs;
   if (typeof blockMs === "number" && Number.isFinite(blockMs) && blockMs > 0) {
     await Bun.sleep(Math.floor(blockMs));
@@ -114,7 +125,7 @@ export async function runPolicyStorageMutation<T>(
     return { ok: false, error: "storage_mutation_busy" };
   }
   try {
-    await applyCoordinatorBlock();
+    await applyCoordinatorBlock("policy");
     return await work();
   } finally {
     gate.lease.release();
@@ -131,7 +142,7 @@ export async function withStorageMutationSlot<T>(
     return { ok: false, error: "storage_mutation_busy" };
   }
   try {
-    await applyCoordinatorBlock();
+    await applyCoordinatorBlock(kind);
     return await work();
   } finally {
     gate.lease.release();

tests/init-eof.test.ts

@@ -3,6 +3,24 @@ import { existsSync, mkdtempSync, rmSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 
+async function waitForOutput(
+  stream: ReadableStream<Uint8Array>,
+  expected: string,
+): Promise<void> {
+  const reader = stream.getReader();
+  const decoder = new TextDecoder();
+  let output = "";
+  try {
+    while (!output.includes(expected)) {
+      const { value, done } = await reader.read();
+      if (done) throw new Error(`init exited before writing ${JSON.stringify(expected)}`);
+      output += decoder.decode(value, { stream: true });
+    }
+  } finally {
+    reader.releaseLock();
+  }
+}
+
 describe("ocx init piped stdin (#754)", () => {
   const dirs: string[] = [];
   afterEach(() => {
@@ -20,14 +38,20 @@ describe("ocx init piped stdin (#754)", () => {
       stdout: "pipe",
       stderr: "pipe",
     });
-    proc.stdin.end();
-    const exit = await Promise.race([
-      proc.exited,
-      new Promise<number>((_, reject) => setTimeout(() => reject(new Error("init did not exit after stdin EOF")), 8_000)),
-    ]);
-    expect(exit).toBe(1);
-    const stderr = await new Response(proc.stderr).text();
-    expect(stderr.toLowerCase()).toMatch(/stdin (closed|reached eof)/);
-    expect(existsSync(join(home, "config.json"))).toBe(false);
-  });
+    const stderrPromise = new Response(proc.stderr).text();
+    try {
+      // Synchronize on the behavior under test, not Windows process startup/import time.
+      // EOF now arrives while readline is waiting for the first answer.
+      await waitForOutput(proc.stdout, "Select default provider (number):");
+      proc.stdin.end();
+
+      expect(await proc.exited).toBe(1);
+      const stderr = await stderrPromise;
+      expect(stderr.toLowerCase()).toMatch(/stdin (closed|reached eof)/);
+      expect(existsSync(join(home, "config.json"))).toBe(false);
+    } finally {
+      if (proc.exitCode === null) proc.kill();
+      await proc.exited.catch(() => {});
+    }
+  }, 30_000);
 });

tests/service.test.ts

@@ -3,6 +3,7 @@ import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, wri
 import { execFileSync } from "node:child_process";
 import { tmpdir } from "node:os";
 import { delimiter, isAbsolute, join, posix, win32 } from "node:path";
+import { pathToFileURL } from "node:url";
 import * as serviceModule from "../src/service";
 import { saveConfig } from "../src/config";
 import { windowsEnvIndirectBatchValue } from "../src/lib/win-paths";
@@ -956,15 +957,24 @@ describe("launchd service plist", () => {
     const shimDir = join(root, "shims");
     const v1 = join(root, "installs", "2.35.0 package's");
     const v2 = join(root, "installs", "2.36.0 package's");
-    const quoteForSh = (value: string): string => `'${value.replaceAll("'", "'\"'\"'")}'`;
     mkdirSync(shimDir, { recursive: true });
     mkdirSync(v1, { recursive: true });
     mkdirSync(v2, { recursive: true });
-    writeFileSync(join(v1, "ocx"), "#!/bin/sh\necho V1 \"$@\"\n", { mode: 0o755 });
-    writeFileSync(join(v2, "ocx"), "#!/bin/sh\necho V2 \"$@\"\n", { mode: 0o755 });
+    const v1Entry = join(v1, "ocx");
+    const v2Entry = join(v2, "ocx");
+    writeFileSync(v1Entry, 'console.log("V1", Bun.argv.slice(2).join(" "));\n');
+    writeFileSync(v2Entry, 'console.log("V2", Bun.argv.slice(2).join(" "));\n');
 
     const shim = join(shimDir, "ocx");
-    writeFileSync(shim, `#!/bin/sh\nexec ${quoteForSh(join(v1, "ocx"))} "\$@"\n`, { mode: 0o755 });
+    const retargetShim = (target: string): void => {
+      writeFileSync(shim, `await import(${JSON.stringify(pathToFileURL(target).href)});\n`);
+    };
+    const runShim = (): string => execFileSync(
+      process.execPath,
+      [shim, "start", "--port", "1"],
+      { encoding: "utf8" },
+    );
+    retargetShim(v1Entry);
 
     // stableLauncherEntry finds the shim lexically from PATH — not its versioned target.
     const found = buildUnit(resolvedProxyEnv({}), { launcher: shim });
@@ -977,13 +987,15 @@ describe("launchd service plist", () => {
     const windowsUnit = buildUnit(resolvedProxyEnv({}), { launcher: windowsShim });
     expectTextToContainPath(windowsUnit, windowsShim);
 
-    expect(execFileSync(shim, ["start", "--port", "1"], { encoding: "utf8" })).toContain("V1");
+    // Exercise the retarget through Bun on every host. Directly executing the old
+    // extensionless #!/bin/sh fixture was itself a POSIX-only assumption.
+    expect(runShim()).toContain("V1");
 
     // The upgrade: shim retargeted, old version removed.
-    writeFileSync(shim, `#!/bin/sh\nexec ${quoteForSh(join(v2, "ocx"))} "\$@"\n`, { mode: 0o755 });
+    retargetShim(v2Entry);
     rmSync(v1, { recursive: true, force: true });
-    expect(existsSync(join(v1, "ocx"))).toBe(false);
-    expect(execFileSync(shim, ["start", "--port", "1"], { encoding: "utf8" })).toContain("V2");
+    expect(existsSync(v1Entry)).toBe(false);
+    expect(runShim()).toContain("V2");
 
     rmSync(root, { recursive: true, force: true });
   });

tests/storage-mutation-race.test.ts

@@ -144,6 +144,18 @@ async function waitForPolicyJob(
   throw new Error("policy job did not finish");
 }
 
+async function waitForCondition(
+  description: string,
+  condition: () => boolean,
+  timeoutMs = 8_000,
+): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (!condition()) {
+    if (Date.now() >= deadline) throw new Error(`timed out waiting for ${description}`);
+    await Bun.sleep(20);
+  }
+}
+
 /** Windows can keep SQLite/job handles briefly after stop; retry only transient cleanup codes. */
 function removeTree(path: string): void {
   let lastError: unknown;
@@ -313,13 +325,20 @@ describe("storage mutation coordinator", () => {
 
   test("cleanup quarantine and permanent are rejected while restore holds slot after file moves", async () => {
     const home = isolatedCodexHome!.path;
-    const holdMs = 2500;
+    const movedReadyPath = join(testDir, "restore-files-moved.ready");
+    const releaseRestorePath = join(testDir, "release-restore");
     setRestoreTrashJobTestHooks({
-      restoreTest: { holdAfterFileMovesMs: holdMs },
+      restoreTest: {
+        pauseAfterFileMoves: {
+          readyPath: movedReadyPath,
+          releasePath: releaseRestorePath,
+        },
+      },
     });
     seedArchivedPair(home);
 
     const server = startServer(0);
+    let restorePromise: Promise<Response> | null = null;
     try {
       const preview = await previewDigest(server.url, 50);
       const cleanupRes = await fetch(new URL("/api/storage/cleanup", server.url), {
@@ -336,17 +355,17 @@ describe("storage mutation coordinator", () => {
       const remainingPreview = await previewDigest(server.url, 50);
       expect(remainingPreview.count).toBe(1);
 
-      const restorePromise = fetch(new URL("/api/storage/trash/restore", server.url), {
+      restorePromise = fetch(new URL("/api/storage/trash/restore", server.url), {
         method: "POST",
         headers: { "content-type": "application/json" },
         body: JSON.stringify({ id: trashId }),
       });
 
       const restoredPath = join(home, "archived_sessions", "rollout-old.jsonl");
-      const movedDeadline = Date.now() + 8000;
-      while (!existsSync(restoredPath) && Date.now() < movedDeadline) {
-        await Bun.sleep(20);
-      }
+      await waitForCondition(
+        "restore worker to finish file moves",
+        () => existsSync(movedReadyPath),
+      );
       expect(existsSync(restoredPath)).toBe(true);
       expect(existsSync(join(trashStage, "rollout-old.jsonl"))).toBe(false);
       expect(existsSync(join(trashStage, "restore-pending.json"))).toBe(true);
@@ -378,6 +397,7 @@ describe("storage mutation coordinator", () => {
       expect(permanentDuring.status).toBe(409);
       expect((await permanentDuring.json()).error).toBe("storage_mutation_busy");
 
+      writeFileSync(releaseRestorePath, "release\n");
       const restoreRes = await restorePromise;
       expect(restoreRes.status).toBe(200);
       const restored = await restoreRes.json();
@@ -386,26 +406,41 @@ describe("storage mutation coordinator", () => {
       expect(threadCount(home)).toBe(2);
       expect(readFileSync(restoredPath, "utf8")).toBe("o".repeat(100));
     } finally {
+      // Never strand the Worker if an assertion above fails; also consume the
+      // request so its rejection cannot leak into the next isolated test.
+      writeFileSync(releaseRestorePath, "release\n");
+      if (restorePromise) await restorePromise.catch(() => undefined);
       await stopRaceServer(server);
     }
   }, { timeout: 45_000 });
 
   test("restore is rejected while cleanup holds the shared mutation slot", async () => {
     const home = isolatedCodexHome!.path;
-    const blockMs = 1200;
-    setArchivedCleanupJobTestHooks({ blockMs });
+    const cleanupReadyPath = join(testDir, "cleanup-slot-acquired.ready");
+    const releaseCleanupPath = join(testDir, "release-cleanup");
+    setArchivedCleanupJobTestHooks({
+      pauseAfterAcquire: {
+        kind: "cleanup",
+        readyPath: cleanupReadyPath,
+        releasePath: releaseCleanupPath,
+      },
+    });
     seedArchivedPair(home);
 
     const server = startServer(0);
+    let cleanupPromise: Promise<Response> | null = null;
     try {
       const preview = await previewDigest(server.url, 50);
-      const cleanupPromise = fetch(new URL("/api/storage/cleanup", server.url), {
+      cleanupPromise = fetch(new URL("/api/storage/cleanup", server.url), {
         method: "POST",
         headers: { "content-type": "application/json" },
         body: JSON.stringify({ percent: 50, mode: "quarantine", digest: preview.digest }),
       });
 
-      await Bun.sleep(80);
+      await waitForCondition(
+        "manual cleanup to acquire the storage mutation slot",
+        () => existsSync(cleanupReadyPath),
+      );
 
       const restoreAttempt = await fetch(new URL("/api/storage/trash/restore", server.url), {
         method: "POST",
@@ -418,6 +453,7 @@ describe("storage mutation coordinator", () => {
       expect(existsSync(join(home, "archived_sessions", "rollout-new.jsonl"))).toBe(true);
       expect(trashStageCount(home)).toBe(0);
 
+      writeFileSync(releaseCleanupPath, "release\n");
       const cleanupRes = await cleanupPromise;
       expect(cleanupRes.status).toBe(200);
       const cleanup = await cleanupRes.json();
@@ -427,6 +463,8 @@ describe("storage mutation coordinator", () => {
       expect(existsSync(join(home, "archived_sessions", "rollout-new.jsonl"))).toBe(true);
       expect(threadCount(home)).toBe(1);
     } finally {
+      writeFileSync(releaseCleanupPath, "release\n");
+      if (cleanupPromise) await cleanupPromise.catch(() => undefined);
       await stopRaceServer(server);
     }
   }, { timeout: 30_000 });