fix(coordination): preserve shadow lineage across root aliases

loopx/control_plane/coordination/local_authority_shadow_adapter.py

@@ -489,6 +489,7 @@ def __init__(
         self._lock_timeout = lock_timeout_seconds
         self._directory = outbox.partition_directory(runtime_root, goal_id, partition)
         self._lineage: str | None = capture_lineage_id
+        self._source_root_digest: str | None = None
         self.last_delivered_digest: str | None = None
 
     def _lock(self) -> Any:
@@ -512,6 +513,7 @@ def _binding(self) -> dict[str, Any]:
                 "stale_generation", "drain belongs to an earlier lineage"
             )
         self._lineage = lineage
+        self._source_root_digest = str(binding["source_root_digest"])
         return binding
 
     def _proof(self) -> tuple[dict[str, Any], list[dict[str, Any]]]:
@@ -570,8 +572,7 @@ def _partition_history(
                 type(seq) is not int
                 or seq != len(history) + 1
                 or receipt.get("capture_lineage_id") != self._lineage
-                or receipt.get("source_root_digest")
-                != outbox.runtime_root_digest(self._runtime_root)
+                or receipt.get("source_root_digest") != self._source_root_digest
                 or receipt.get("entry_id") != transaction.get("operation_id")
             ):
                 raise outbox.OutboxError(

loopx/control_plane/coordination/local_authority_shadow_outbox.py

@@ -869,7 +869,9 @@ def prepare(self, new_text: str, *, event_id: str | None = None) -> None:
                 binding_view.get("reason_code") or "bootstrap_required"
             )
             return
-        self._lineage_id = str(binding_view["binding"]["capture_lineage_id"])
+        binding = binding_view["binding"]
+        self._lineage_id = str(binding["capture_lineage_id"])
+        source_root_digest = str(binding["source_root_digest"])
         if event_id is not None:
             self.outcome.skipped_reason = "event_log_writer_not_bound"
             return
@@ -892,7 +894,7 @@ def prepare(self, new_text: str, *, event_id: str | None = None) -> None:
                 seq=seq,
                 source_ref=source_ref,
                 capture_lineage_id=self._lineage_id,
-                source_root_digest=runtime_root_digest(self._runtime_root),
+                source_root_digest=source_root_digest,
             )
             record = _entry_record(
                 goal_id=self._goal_id,
@@ -912,7 +914,7 @@ def prepare(self, new_text: str, *, event_id: str | None = None) -> None:
                     "lease": None,
                     "event_id": event_id,
                 },
-                source_root_digest=runtime_root_digest(self._runtime_root),
+                source_root_digest=source_root_digest,
                 capture_lineage_id=self._lineage_id,
                 projection=projection,
                 digest=digest,

loopx/control_plane/coordination/local_authority_shadow_outbox.ts

@@ -1,6 +1,6 @@
 import { createHash } from "node:crypto";
 import { readdir, readFile } from "node:fs/promises";
-import { join, resolve } from "node:path";
+import { join } from "node:path";
 
 import type { JsonObject } from "../effect_program.ts";
 import { durableWriteJson } from "../effect_runtime_io.ts";
@@ -295,7 +295,7 @@ export async function beginLeaseOutboxEntry(
     }));
     const bytesDigest = leaseRecordDigest(input.planned_lease);
     const seq = await nextSeq(directory, input.runtime_root, input.goal_id, binding.capture_lineage_id);
-    const sourceRootDigest = sha256Digest(resolve(input.runtime_root));
+    const sourceRootDigest = binding.source_root_digest;
     const entryId = outboxEntryIdentity(input.goal_id, LEASE_PARTITION, seq, bytesDigest,
       binding.capture_lineage_id, sourceRootDigest);
     const entry: JsonObject = {

loopx/control_plane/coordination/shadow_management.py

@@ -58,6 +58,17 @@ def _text(value: object) -> bool:
     return isinstance(value, str) and bool(value) and value.strip() == value
 
 
+def _source_root_digests(runtime_root: Path) -> set[str]:
+    roots = {
+        os.path.abspath(str(runtime_root)),
+        os.path.realpath(str(runtime_root)),
+    }
+    return {
+        "sha256:" + hashlib.sha256(root.encode()).hexdigest()
+        for root in roots
+    }
+
+
 def _binding(value: object, root_digest: str) -> bool:
     return (
         isinstance(value, dict) and set(value) == _BINDING_KEYS
@@ -79,13 +90,14 @@ def read_shadow_management_state(runtime_root: Path, goal_id: str) -> dict[str,
         raise ShadowManagementError("shadow_management_state_invalid") from exc
     except OSError as exc:
         raise ShadowManagementError("shadow_management_state_unavailable") from exc
-    root_digest = "sha256:" + hashlib.sha256(os.path.abspath(str(runtime_root)).encode()).hexdigest()
+    root_digests = _source_root_digests(runtime_root)
     try:
         state = json.loads(raw)
         if not isinstance(state, dict) or set(state) != _STATE_KEYS:
             raise ValueError("journal fields differ")
+        root_digest = state["source_root_digest"]
         if (state["schema_version"] != SHADOW_MANAGEMENT_STATE_SCHEMA
-                or state["goal_id"] != goal_id or state["source_root_digest"] != root_digest):
+                or state["goal_id"] != goal_id or root_digest not in root_digests):
             raise ValueError("journal scope differs")
         status = state["status"]
         if status not in {"bootstrapping", "active", "rolling_back", "inactive"}:

tests/control_plane/test_local_authority_shadow_outbox.py

@@ -226,6 +226,32 @@ def test_capture_records_prepared_then_committed_and_skips_prose_only_writes(tmp
     assert [entry.seq for entry in outbox.list_entries(_todo_dir(runtime_root))] == [1, 2]
 
 
+def test_capture_and_drain_use_the_active_binding_digest_through_a_root_alias(
+    tmp_path: Path,
+) -> None:
+    target = tmp_path / "target"
+    target.mkdir()
+    alias = tmp_path / "alias"
+    try:
+        alias.symlink_to(target, target_is_directory=True)
+    except OSError as exc:
+        pytest.skip(f"directory symlinks unavailable: {exc}")
+    registry, state, runtime_root = _fixture(alias)
+
+    capture = _record_change(
+        registry, state, runtime_root, "Capture through the runtime alias."
+    )
+    binding = require_shadow_primary_write_allowed(runtime_root, GOAL_ID)
+    assert binding is not None
+    [entry] = outbox.list_entries(_todo_dir(runtime_root))
+    assert entry.prepared["source_root_digest"] == binding["source_root_digest"]
+
+    drained = _drain(registry, runtime_root)
+    assert drained.outcome == "drained"
+    assert drained.delivered == 1
+    assert capture.outcome.entry_id == drained.entries[0]["entry_id"]
+
+
 def test_disabled_capture_creates_nothing(tmp_path: Path) -> None:
     registry, state, runtime_root = _fixture(tmp_path, bootstrap=False)
     capture = _capture(registry, state, runtime_root, original_text="", enabled=False)

tests/control_plane/test_shadow_management.py

@@ -36,8 +36,16 @@ def test_corrupt_management_holds_before_any_primary_write(tmp_path: Path, raw:
     assert path.read_bytes() == before
 
 
-def test_python_reads_typescript_binding_and_rejects_cross_root_replay(tmp_path: Path) -> None:
+def test_python_reads_typescript_binding_across_root_alias_and_rejects_cross_root_replay(
+    tmp_path: Path,
+) -> None:
     root = tmp_path / "runtime"
+    root.mkdir()
+    alias = tmp_path / "runtime-alias"
+    try:
+        alias.symlink_to(root, target_is_directory=True)
+    except OSError as exc:
+        pytest.skip(f"directory symlinks unavailable: {exc}")
     script = """
 import {bootstrapManagedShadow} from './loopx/control_plane/coordination/shadow_management.ts';
 const root = process.argv[1];
@@ -46,18 +54,29 @@ def test_python_reads_typescript_binding_and_rejects_cross_root_replay(tmp_path:
 process.stdout.write(JSON.stringify(result));
 """
     result = subprocess.run(
-        ["node", "--no-warnings", "--experimental-strip-types", "--input-type=module", "-e", script, str(root)],
+        ["node", "--no-warnings", "--experimental-strip-types", "--input-type=module", "-e", script, str(alias)],
         check=True, capture_output=True, text=True,
     )
     applied = json.loads(result.stdout)
     assert applied["status"] == "applied"
-    binding = require_shadow_primary_write_allowed(root, "goal-a")
+    binding = require_shadow_primary_write_allowed(alias, "goal-a")
     assert binding is not None
     assert binding["capture_lineage_id"] == applied["capture_lineage_id"]
+    assert require_shadow_primary_write_allowed(root, "goal-a") == binding
+    state_path = shadow_management_state_path(root, "goal-a")
+    state = json.loads(state_path.read_text())
+    lexical_digest = "sha256:" + hashlib.sha256(str(alias).encode()).hexdigest()
+    assert lexical_digest != binding["source_root_digest"]
+    state["source_root_digest"] = lexical_digest
+    state_path.write_text(json.dumps(state))
+    with pytest.raises(ShadowManagementError, match="shadow_management_state_invalid"):
+        require_shadow_primary_write_allowed(alias, "goal-a")
+    state["source_root_digest"] = binding["source_root_digest"]
+    state_path.write_text(json.dumps(state))
     other = tmp_path / "other-root"
     destination = shadow_management_state_path(other, "goal-a")
     destination.parent.mkdir(parents=True)
-    destination.write_bytes(shadow_management_state_path(root, "goal-a").read_bytes())
+    destination.write_bytes(state_path.read_bytes())
     with pytest.raises(ShadowManagementError, match="shadow_management_state_invalid"):
         require_shadow_primary_write_allowed(other, "goal-a")
 

tests/control_plane_ts/local_authority_shadow_outbox.test.ts

@@ -1,6 +1,6 @@
 import assert from "node:assert/strict";
 import { execFile } from "node:child_process";
-import { readFile, writeFile, rename, unlink } from "node:fs/promises";
+import { readFile, writeFile, rename, symlink, unlink } from "node:fs/promises";
 import { join } from "node:path";
 import test from "node:test";
 import { promisify } from "node:util";
@@ -11,6 +11,7 @@ import {
   readLocalAuthorityShadow,
 } from "../../loopx/control_plane/coordination/local_authority_shadow.ts";
 import { outboxEntryIdentity, beginLeaseOutboxEntry } from "../../loopx/control_plane/coordination/local_authority_shadow_outbox.ts";
+import { requireShadowCaptureBinding } from "../../loopx/control_plane/coordination/shadow_management.ts";
 import * as schemas from "../../loopx/control_plane/coordination/coordination_state_contract.generated.ts";
 import { fixture, pendingEntry, settleFiles, todo, sha } from "./shadow_file_fixture.ts";
 
@@ -146,6 +147,22 @@ test("a lease writer with a missing cursor obtains its next sequence from proved
   await assert.rejects(readFile(join(directory, "drain-cursor.json")), { code: "ENOENT" });
 });
 
+test("a lease writer uses the active binding digest through a runtime-root alias", async (t) => {
+  const f = await fixture(t);
+  const alias = `${f.root}-alias`;
+  t.after(() => unlink(alias));
+  await symlink(f.root, alias, process.platform === "win32" ? "junction" : "dir");
+  const planned = { schema_version: "task_lease_v0", goal_id: "goal-a", todo_id: "todo_one",
+    owner: "agent-a", version: 1, lease_epoch: 1, status: "active", updated_at: "2026-09-06T00:00:00Z" };
+  const capture = await beginLeaseOutboxEntry({ runtime_root: alias, goal_id: "goal-a",
+    lease_directory: join(alias, "goals", "goal-a", "task-leases"), write_class: "task_lease_acquire",
+    operation_id: null, previous_lease: null, planned_lease: planned, active_todo_ids: null });
+  assert.equal(capture.failure, null);
+  const prepared = JSON.parse(await readFile(join(alias, "authority-shadow", "outbox", "goal-a", "leases",
+    `0000000001-${capture.entry_id}.prepared.json`), "utf8"));
+  assert.equal(prepared.source_root_digest, (await requireShadowCaptureBinding(alias, "goal-a")).source_root_digest);
+});
+
 test("lease capture omits a lease whose Todo left the current graph", async (t) => {
   const f = await fixture(t);
   const leaseDirectory = join(f.root, "goals", "goal-a", "task-leases");