fix(checkpoint): classify occupied destinations

loopx/control_plane/runtime/checkpoint_artifacts.ts

@@ -134,8 +134,25 @@ function unchanged(a: unknown, b: unknown, message = "checkpoint source moved; r
 }
 async function absentDestination(path: string): Promise<void> {
   await physical(dirname(path));
-  try {await lstat(path); throw new Error("checkpoint destination already exists");}
-  catch (error) {if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;}
+  try {await lstat(path);}
+  catch (error) {
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") return;
+    throw error;
+  }
+  throw new EffectRuntimeConflictError("checkpoint destination already exists");
+}
+async function publishCheckpointDirectory(staging: string, destination: string): Promise<void> {
+  try {await publishNewDirectory(staging, destination);}
+  catch (error) {
+    const code = (error as NodeJS.ErrnoException).code;
+    const permissionFailure = code === "EACCES" || code === "EPERM";
+    // A directory rename can report a permission error for an occupied target.
+    // Confirm the entry exists before changing the meaning of that error.
+    const occupied = code === "EEXIST" || code === "ENOTEMPTY" ||
+      (permissionFailure && await lstat(destination).then(() => true, () => false));
+    if (occupied) throw new EffectRuntimeConflictError("checkpoint destination already exists");
+    throw error;
+  }
 }
 async function sqliteFile(path: string): Promise<boolean> {
   const handle = await open(path, "r");
@@ -427,7 +444,7 @@ export async function checkpointArtifacts(value: unknown): Promise<JsonObject> {
       await copyComponent(root, join(staging, "unit"), false);
       await verify(join(staging, "unit"), request.expected_sha256);
       withinDeadline();
-      await publishNewDirectory(join(staging, "unit"), destination);
+      await publishCheckpointDirectory(join(staging, "unit"), destination);
       return {...verification, materialized: true, activation_performed: false};
     }
     const sources = components(request.components);
@@ -462,7 +479,7 @@ export async function checkpointArtifacts(value: unknown): Promise<JsonObject> {
     await verify(staging, manifest.sha256);
     unchanged(before, await witness(sources));
     withinDeadline();
-    await publishNewDirectory(staging, destination);
+    await publishCheckpointDirectory(staging, destination);
     return {...await verify(destination, manifest.sha256), source_stability_verified: true};
   } finally {await rm(staging, {recursive: true, force: true});}
 }

tests/control_plane/test_checkpoint_artifact_transport.py

@@ -9,6 +9,7 @@
 from loopx.control_plane.effect_runtime import (
     MAX_REQUEST_BYTES,
     MAX_RESPONSE_BYTES,
+    EffectRuntimeConflict,
     EffectRuntimeRejected,
     EffectRuntimeRemoteError,
     effect_runtime_result,
@@ -68,6 +69,12 @@ def test_large_member_transport_with_explicit_rpc_budget_preserves_inventory(tmp
     assert sorted(p.name for p in (destination / "components/research").iterdir()) == [
         row["path"] for row in expected[1:]]
     member = destination / "components/research" / expected[-1]["path"]
+    with pytest.raises(EffectRuntimeConflict, match="^checkpoint destination already exists$") as rejected:
+        effect_runtime_result("runtime.checkpoint_artifacts", {
+            "action": "materialize", "root": str(unit), "expected_sha256": captured["sha256"],
+            "destination": str(destination),
+        }, timeout=LARGE_ARTIFACT_RPC_SECONDS, retry_safe=False, large_local_snapshot=True)
+    assert rejected.value.diagnostic_code == "state_conflict"
     assert member.read_bytes() == content
     member.unlink()
     with pytest.raises(EffectRuntimeRemoteError):

tests/control_plane_ts/checkpoint_artifacts.test.ts

@@ -7,7 +7,7 @@ import {chmod, mkdir, mkdtemp, readFile, readdir, realpath, rm, stat, symlink, w
 import {tmpdir} from "node:os";
 import {join} from "node:path";
 import {checkpointArtifacts} from "../../loopx/control_plane/runtime/checkpoint_artifacts.ts";
-import {EffectRuntimeRequestError} from "../../loopx/control_plane/effect_runtime_errors.ts";
+import {EffectRuntimeConflictError, EffectRuntimeRequestError} from "../../loopx/control_plane/effect_runtime_errors.ts";
 import {sqliteAuthorityRuntime} from "../../loopx/control_plane/coordination/sqlite_runtime.ts";
 
 async function fixture(t: test.TestContext) {
@@ -40,7 +40,9 @@ test("capture, verify and inert materialization preserve executable code and emp
   assert.deepEqual(await readdir(join(root, "restored/components/code/empty")), []);
   assert.deepEqual(await readFile(join(root, "restored/checkpoint.json")), await readFile(join(root, "captured/checkpoint.json")));
   await assert.rejects(checkpointArtifacts({action: "materialize", root: request.destination,
-    expected_sha256: captured.sha256, destination: join(root, "restored")}), /already exists/);
+    expected_sha256: captured.sha256, destination: join(root, "restored")}), {
+    name: "EffectRuntimeConflictError", code: "state_conflict", message: "checkpoint destination already exists",
+  });
 });
 
 test("a read-only participant's copied mode survives the durability flush and materialization", async t => {
@@ -179,7 +181,9 @@ test("concurrent capture reserves one destination and cannot replace an existing
   const {root, request} = await fixture(t);
   const results = await Promise.allSettled([checkpointArtifacts(request), checkpointArtifacts(request)]);
   assert.equal(results.filter(r => r.status === "fulfilled").length, 1);
-  assert.equal(results.filter(r => r.status === "rejected").length, 1);
+  const rejected = results.find(result => result.status === "rejected");
+  assert.ok(rejected?.reason instanceof EffectRuntimeConflictError, rejected?.reason);
+  assert.equal(rejected.reason.code, "state_conflict");
   assert.deepEqual((await readdir(root)).sort(), ["captured", "code", "conversation"]);
   const bytes = await readFile(join(root, "captured/checkpoint.json"));
   await assert.rejects(checkpointArtifacts(request), /already exists/);