fix(runtime): recover Windows lock creation and exact fixture cleanup

docs/development/testing-and-quality.md

@@ -426,6 +426,17 @@ Effect 服务先排空已接纳的 handler，再退出。启用编译缓存时
 继续普通清理。一秒额度不涵盖定位文件锁、存储等待或操作系统调度；故障注入不能
 代替原生平台资格验证，V8 CPU 样本也不测量内核 IO。
 
+Windows mutation-lock acquisition retries `EPERM` from exclusive creation only
+after observing contention, within the existing lock timeout. This covers a
+[pending-delete pathname](https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-deletefilew#remarks).
+Initial permission denial, errors after creation and other platforms retain
+their failure behavior; persistent denial preserves its original error.
+Injected filesystem failures complement actual Windows execution, not replace it.
+
+Windows 文件锁只在已观察到争用后，对独占创建返回的 `EPERM` 使用原锁等待额度重试。
+初次权限拒绝、创建后的错误和其它平台仍直接失败；持续拒绝保留原错误。
+故障注入验证重试边界，原生 Windows 运行仍须单独验收。
+
 Smoke cleanup owns every valid runtime locator in its newly created private
 temporary namespace, including runtimes loaded by temporary base checkouts.
 It uses the existing authenticated shutdown protocol, then observes actual

loopx/control_plane/effect_runtime_io.ts

@@ -321,9 +321,12 @@ export async function acquireFileMutationLock(
   const token = randomUUID();
   const deadline = Date.now() + timeoutMs;
   let retriedAfterReclaim = false;
+  let contended = false;
   while (true) {
+    let opened = false;
     try {
       const handle = await open(lockPath, "wx", 0o600);
+      opened = true;
       let identity: CreatedFileIdentity | null = null;
       try {
         try {
@@ -364,17 +367,26 @@ export async function acquireFileMutationLock(
       }
       return { targetPath, lockPath, token };
     } catch (error) {
-      if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error;
-      const reclaimed = await reclaimStaleMutationLock(lockPath);
-      // Removing a dead owner is progress, not waiting for a live owner. Allow
-      // one immediate acquisition even for a zero-wait caller. Bound the retry
-      // so repeated replacement cannot extend that caller's lock budget.
-      if (reclaimed && !retriedAfterReclaim) {
-        retriedAfterReclaim = true;
-        continue;
-      }
-      if (Date.now() >= deadline) {
-        throw new EffectRuntimeLockTimeoutError();
+      const code = (error as NodeJS.ErrnoException).code;
+      // Windows may deny exclusive creation while a contended lock is pending
+      // deletion. Retry only the unopened pathname within the original budget;
+      // initial permission errors and failures after creation remain fatal.
+      if (!opened && contended && process.platform === "win32" && code === "EPERM") {
+        if (Date.now() >= deadline) throw error;
+      } else {
+        if (code !== "EEXIST") throw error;
+        if (!opened) contended = true;
+        const reclaimed = await reclaimStaleMutationLock(lockPath);
+        // Removing a dead owner is progress, not waiting for a live owner. Allow
+        // one immediate acquisition even for a zero-wait caller. Bound the retry
+        // so repeated replacement cannot extend that caller's lock budget.
+        if (reclaimed && !retriedAfterReclaim) {
+          retriedAfterReclaim = true;
+          continue;
+        }
+        if (Date.now() >= deadline) {
+          throw new EffectRuntimeLockTimeoutError();
+        }
       }
       await new Promise((resolve) =>
         setTimeout(

tests/control_plane/test_effect_runtime_integration.py

@@ -637,10 +637,12 @@ def test_identical_release_does_not_reuse_a_retired_lazy_loader(
         effect_runtime.restart_effect_runtime()
 
 
+@pytest.mark.parametrize("cleanup_delay", [0, 0.6], ids=["serving", "idle-retired"])
 def test_runtime_decode_change_rotates_identity_and_starts_replacement(
     tmp_path: Path,
     monkeypatch,
     request,
+    cleanup_delay: float,
 ) -> None:
     staged_root, captures = observe_copied_runtime(tmp_path, monkeypatch, request)
     runtime_dir = tmp_path / "runtime"
@@ -661,9 +663,20 @@ def test_runtime_decode_change_rotates_identity_and_starts_replacement(
     assert original_info != replacement_info
     assert replacement_info.exists()
 
+    replacement_locator = effect_runtime._read_info(
+        replacement_info, fingerprint=replacement_fingerprint,
+    )
+    assert replacement_locator is not None
+    runtime_count = len(captures)
+    time.sleep(cleanup_delay)
     shutdown_started = time.monotonic()
-    effect_runtime.effect_runtime_result("runtime.shutdown", {}, retry_safe=False)
+    # Cleanup owns this exact process. Rediscovery can race its 250 ms idle
+    # retirement or start a third runtime merely to shut that new one down.
+    effect_runtime._stop_effect_runtime(
+        replacement_info, replacement_locator, timeout=2, wait_for_process_exit=True,
+    )
     shutdown_acknowledged = time.monotonic()
+    assert len(captures) == runtime_count, "cleanup must not start another runtime"
     deadline = time.monotonic() + 2
     while (
         original_info.exists() or replacement_info.exists()

tests/control_plane_ts/effect_runtime_io.test.ts

@@ -20,6 +20,77 @@ async function workspace(t: TestContext): Promise<string> {
   return root;
 }
 
+for (const scenario of [
+  {name: "Windows replacement open recovers", platform: "win32", contended: true, persistent: false, writeFault: false, timeout: 500, code: "EPERM", expected: null},
+  {name: "Windows initial permission denial fails", platform: "win32", contended: false, persistent: true, writeFault: false, timeout: 500, code: "EPERM", expected: "EPERM"},
+  {name: "Windows persistent replacement denial keeps its error", platform: "win32", contended: true, persistent: true, writeFault: false, timeout: 250, code: "EPERM", expected: "EPERM"},
+  {name: "non-Windows replacement denial fails", platform: "linux", contended: true, persistent: false, writeFault: false, timeout: 500, code: "EPERM", expected: "EPERM"},
+  {name: "unrelated Windows open failure fails", platform: "win32", contended: true, persistent: false, writeFault: false, timeout: 500, code: "EIO", expected: "EIO"},
+  {name: "Windows failure after creation is not retried", platform: "win32", contended: true, persistent: false, writeFault: true, timeout: 500, code: "EPERM", expected: "EPERM"},
+  {name: "zero-wait live contention does not gain a retry", platform: "win32", contended: true, persistent: false, writeFault: false, timeout: 0, code: "EPERM", expected: "mutation_lock_timeout"},
+]) {
+  test(scenario.name, async t => {
+    const root = await workspace(t), target = join(root, "state");
+    // Only the selected open/write failure and platform predicate are injected.
+    // All lock bytes, handles, identities, live-owner checks and release use
+    // the production filesystem owner in an isolated child.
+    const script = `
+      import assert from "node:assert/strict";
+      import * as fs from "node:fs/promises";
+      import {mock} from "node:test";
+      const [module, target, scenario] = JSON.parse(process.argv[1]);
+      const path = target + ".ts-effect.lock";
+      let attempts = 0;
+      if (scenario.contended) {
+        await fs.writeFile(path, JSON.stringify({pid: process.pid, token: "live-holder"}));
+      }
+      const {default: ignored, ...namedExports} = fs;
+      mock.module("node:fs/promises", {namedExports: {...namedExports,
+        open: async (file, flags, ...args) => {
+          if (file !== path || flags !== "wx") return fs.open(file, flags, ...args);
+          attempts++;
+          if (scenario.contended && attempts === 1) return fs.open(file, flags, ...args);
+          if (scenario.contended && attempts === 2) await fs.rm(path);
+          const fault = Object.assign(new Error("controlled filesystem failure"),
+            {code: scenario.code, syscall: scenario.writeFault ? "write" : "open"});
+          if (!scenario.writeFault && (scenario.persistent || attempts === 2)) throw fault;
+          const handle = await fs.open(file, flags, ...args);
+          if (scenario.writeFault) handle.writeFile = async () => { throw fault; };
+          return handle;
+        },
+      }});
+      const io = await import(module);
+      Object.defineProperty(process, "platform", {value: scenario.platform});
+      let code = null;
+      const started = Date.now(); // Match the acquisition deadline's clock.
+      try {
+        const lock = await io.acquireFileMutationLock(target, process.pid, scenario.timeout);
+        assert.equal((await io.mutationLockOwner(target)).token, lock.token);
+        assert.equal(await io.releaseFileMutationLock(target, lock.token), true);
+        await assert.rejects(fs.stat(path), {code: "ENOENT"});
+      } catch (error) { code = error.code ?? String(error); }
+      if (scenario.writeFault) await assert.rejects(fs.stat(path), {code: "ENOENT"});
+      console.log(JSON.stringify({code, attempts, elapsed: Date.now() - started}));
+    `;
+    const result = spawnSync(process.execPath, [
+      "--no-warnings", "--experimental-test-module-mocks", "--experimental-strip-types",
+      "--input-type=module", "-e", script,
+      JSON.stringify([new URL("../../loopx/control_plane/effect_runtime_io.ts", import.meta.url).href,
+        target, scenario]),
+    ], {encoding: "utf8", timeout: 10_000});
+    assert.equal(result.status, 0, result.stderr);
+    const observed = JSON.parse(result.stdout.trim());
+    assert.equal(observed.code, scenario.expected);
+    if (scenario.expected === null) assert.equal(observed.attempts, 3);
+    else if (!scenario.contended || scenario.timeout === 0) assert.equal(observed.attempts, 1);
+    else if (scenario.persistent) {
+      assert.ok(observed.attempts > 2);
+      assert.ok(observed.elapsed >= scenario.timeout);
+      assert.ok(observed.elapsed < 2_000, "permission denial must keep a bounded wait");
+    } else assert.equal(observed.attempts, 2);
+  });
+}
+
 test("JSONL append starts a new row after an unterminated tail", async t => {
   const root = await workspace(t);
   const path = join(root, "index.jsonl");