fix(coordination): bind shadow entries and the primary guard to the lineage root digest

loopx/control_plane/coordination/local_authority_shadow_adapter.py

@@ -560,6 +560,10 @@ def _receipt(transaction: dict[str, Any]) -> dict[str, Any] | None:
     def _partition_history(
         self, transactions: list[dict[str, Any]]
     ) -> dict[int, dict[str, Any]]:
+        # Receipts carry the lineage binding's root digest, which the owner
+        # derives from the resolved path; comparing against this process's
+        # lexical spelling would reject every receipt behind a symlinked root.
+        source_root_digest = self._binding()["source_root_digest"]
         history: dict[int, dict[str, Any]] = {}
         for transaction in transactions:
             receipt = self._receipt(transaction)
@@ -570,8 +574,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") != source_root_digest
                 or receipt.get("entry_id") != transaction.get("operation_id")
             ):
                 raise outbox.OutboxError(

loopx/control_plane/coordination/local_authority_shadow_outbox.py

@@ -629,11 +629,13 @@ def next_seq(
 
 
 def runtime_root_digest(runtime_root: Path) -> str:
-    """Digest of the absolute, dot-normalized root; must match the TypeScript writer.
+    """Digest of the absolute, dot-normalized root as it was spelled.
 
-    Symlinks are deliberately not resolved: both runtimes normalize the string
-    they were given, so a root passed through the effect runtime hashes the
-    same on either side.
+    Diagnostic only. Outbox entries and drain receipts carry the lineage
+    binding's ``source_root_digest`` instead: the TypeScript owner writes that
+    digest from the resolved path (`shadow_management.ts`), so a root reached
+    through a symlink, such as the macOS temp directory, hashes differently
+    from this lexical spelling (#4892).
     """
 
     return text_digest(os.path.abspath(str(runtime_root)))
@@ -795,6 +797,7 @@ def __init__(
         self._entry_id: str | None = None
         self._event_id: str | None = None
         self._lineage_id: str | None = None
+        self._source_root_digest: str | None = None
         self.outcome = CaptureOutcome(partition=TODO_PARTITION if enabled else None)
 
     @classmethod
@@ -870,6 +873,10 @@ def prepare(self, new_text: str, *, event_id: str | None = None) -> None:
             )
             return
         self._lineage_id = str(binding_view["binding"]["capture_lineage_id"])
+        # Entries belong to the lineage: bind them to the digest the owner wrote
+        # into the binding, not to this process's spelling of the root.
+        source_root_digest = str(binding_view["binding"]["source_root_digest"])
+        self._source_root_digest = source_root_digest
         if event_id is not None:
             self.outcome.skipped_reason = "event_log_writer_not_bound"
             return
@@ -892,7 +899,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 +919,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,10 @@ 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));
+    // The entry belongs to the lineage: carry the binding's digest, which the
+    // owner derived from the resolved root, so a symlinked spelling of the same
+    // root still drains (#4892).
+    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,12 +58,35 @@ def _text(value: object) -> bool:
     return isinstance(value, str) and bool(value) and value.strip() == value
 
 
-def _binding(value: object, root_digest: str) -> bool:
+def shadow_source_root_digests(runtime_root: Path) -> frozenset[str]:
+    """The root digests the TypeScript owner accepts: the lexical and the canonical spelling.
+
+    `shadow_management.ts` writes `source_root_digest` from the real path
+    (`realpathSync`) and reads either the lexical or the real spelling
+    (`sourceRootDigests`). A root reached through a symlink, such as the macOS
+    temp directory, hashes differently per spelling, so this guard must accept the
+    same set or it rejects a journal the owner has just written (#4892). The
+    canonical spelling falls back to the lexical one when the root cannot be
+    resolved, as the TypeScript side does.
+    """
+
+    lexical = os.path.abspath(str(runtime_root))
+    try:
+        canonical = os.path.realpath(lexical, strict=True)
+    except OSError:
+        canonical = lexical
+    return frozenset(
+        "sha256:" + hashlib.sha256(spelling.encode("utf-8")).hexdigest()
+        for spelling in (lexical, canonical)
+    )
+
+
+def _binding(value: object, root_digests: frozenset[str]) -> bool:
     return (
         isinstance(value, dict) and set(value) == _BINDING_KEYS
         and all(_text(item) for item in value.values())
         and value["capture_profile"] == SHADOW_CAPTURE_PROFILE
-        and value["source_root_digest"] == root_digest
+        and value["source_root_digest"] in root_digests
         and re.fullmatch(r"file:[0-9a-f]{32}", value["store_identity"]) is not None
         and re.fullmatch(r"file:[1-9][0-9]*:[0-9a-f]{24}", value["bootstrap_provider_revision"]) is not None
     )
@@ -79,13 +102,13 @@ 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 = shadow_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")
         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 state["source_root_digest"] not in root_digests):
             raise ValueError("journal scope differs")
         status = state["status"]
         if status not in {"bootstrapping", "active", "rolling_back", "inactive"}:
@@ -112,7 +135,7 @@ def read_shadow_management_state(runtime_root: Path, goal_id: str) -> dict[str,
             raise ValueError("journal result is invalid")
         if not terminal and state["result"] is not None:
             raise ValueError("journal result is premature")
-        if state["binding"] is not None and not _binding(state["binding"], root_digest):
+        if state["binding"] is not None and not _binding(state["binding"], root_digests):
             raise ValueError("journal binding is invalid")
         if status == "active" and state["binding"] is None:
             raise ValueError("active journal has no binding")

tests/control_plane/test_local_authority_shadow_outbox.py

@@ -488,3 +488,31 @@ def test_reclaim_validates_the_complete_batch_before_any_unlink(tmp_path: Path)
         outbox.reclaim_verified_files(proof)
     assert raised.value.reason_code == "outbox_file_changed"
     assert first.exists() and second.exists()
+
+
+def test_capture_and_drain_bind_entries_to_the_lineage_root_digest_through_a_symlink(tmp_path: Path) -> None:
+    """The owner writes the binding digest from the resolved root; entries and receipts must carry it.
+
+    Before #4892 the capture hashed its own lexical spelling of the root and the
+    drain compared receipts against that spelling, so a root reached through a
+    symlink (the macOS temp directory) never drained.
+    """
+
+    registry, state, runtime_root = _fixture(tmp_path)
+    link = tmp_path / "runtime-link"
+    link.symlink_to(runtime_root, target_is_directory=True)
+    binding = require_shadow_primary_write_allowed(link, GOAL_ID)
+    assert binding is not None
+    assert binding["source_root_digest"] != outbox.runtime_root_digest(link)
+
+    _record_change(registry, state, link, "Recorded through a symlinked runtime root.")
+    [entry] = outbox.list_entries(_todo_dir(link))
+    assert entry.prepared["source_root_digest"] == binding["source_root_digest"]
+    assert entry.prepared["capture_lineage_id"] == binding["capture_lineage_id"]
+
+    result = _drain(registry, link)
+    assert result.delivered == 1, result
+    assert result.pending_after == 0, result
+    # The same outbox drained again through the real path replays, never re-delivers.
+    again = _drain(registry, runtime_root)
+    assert again.delivered == 0 and again.pending_after == 0, again

tests/control_plane/test_shadow_management.py

@@ -2,12 +2,17 @@
 
 import json
 import hashlib
+import os
 from pathlib import Path
 import subprocess
 
 import pytest
 
+from loopx.control_plane.coordination.coordination_state_contract_generated import (
+    SHADOW_MANAGEMENT_STATE_SCHEMA,
+)
 from loopx.control_plane.coordination.shadow_management import (
+    SHADOW_CAPTURE_PROFILE,
     ShadowManagementError,
     read_shadow_management_state,
     require_shadow_primary_write_allowed,
@@ -116,3 +121,97 @@ def test_bound_source_path_never_accepts_or_repairs_a_damaged_manifest(tmp_path:
     with pytest.raises(ShadowManagementError, match="shadow_management_manifest_invalid"):
         management.read_shadow_bootstrap_source_path(w.runtime, w.goal, binding)
     assert {str(path.relative_to(w.runtime)): path.read_bytes() for path in w.runtime.rglob("*") if path.is_file()} == before
+
+
+def _root_digest(path: Path, *, canonical: bool) -> str:
+    spelling = os.path.realpath(path) if canonical else os.path.abspath(path)
+    return "sha256:" + hashlib.sha256(spelling.encode("utf-8")).hexdigest()
+
+
+def _active_journal(root_digest: str) -> dict:
+    """An active journal in the shape the TypeScript owner writes, bound to one root digest."""
+
+    return {
+        "schema_version": SHADOW_MANAGEMENT_STATE_SCHEMA,
+        "goal_id": "goal-a",
+        "source_root_digest": root_digest,
+        "status": "active",
+        "binding": {
+            "capture_profile": SHADOW_CAPTURE_PROFILE,
+            "capture_lineage_id": "lineage-a",
+            "source_root_digest": root_digest,
+            "store_identity": "file:" + "a" * 32,
+            "bootstrap_operation_id": "bootstrap:guard",
+            "bootstrap_provider_revision": "file:1:" + "b" * 24,
+        },
+        "operation": {
+            "kind": "bootstrap",
+            "operation_id": "bootstrap:guard",
+            "request_digest": "sha256:" + "c" * 64,
+            "manifest_digest": "sha256:" + "d" * 64,
+            "phase": "complete",
+        },
+        "previous_operation_id": None,
+        "result": {},
+    }
+
+
+def _symlinked_root(tmp_path: Path) -> Path:
+    real = tmp_path / "real-root"
+    real.mkdir()
+    link = tmp_path / "link-root"
+    link.symlink_to(real, target_is_directory=True)
+    assert os.path.realpath(link) != os.path.abspath(link)
+    return link
+
+
+def test_guard_accepts_both_root_spellings_the_typescript_owner_accepts(tmp_path: Path) -> None:
+    """A root reached through a symlink (macOS `$TMPDIR`, `/tmp`) hashes differently per spelling.
+
+    `shadow_management.ts` writes the canonical (realpath) digest and reads either
+    spelling; the Python guard must accept the same set or it rejects a journal the
+    owner just wrote. An unrelated root must still be rejected.
+    """
+
+    link = _symlinked_root(tmp_path)
+    path = shadow_management_state_path(link, "goal-a")
+    path.parent.mkdir(parents=True)
+    for canonical in (True, False):
+        digest = _root_digest(link, canonical=canonical)
+        path.write_text(json.dumps(_active_journal(digest)))
+        binding = require_shadow_primary_write_allowed(link, "goal-a")
+        assert binding is not None and binding["source_root_digest"] == digest
+    # A journal bound to the canonical spelling is the same root when read through
+    # the real path, exactly as the TypeScript reader treats it.
+    path.write_text(json.dumps(_active_journal(_root_digest(link, canonical=True))))
+    assert require_shadow_primary_write_allowed(Path(os.path.realpath(link)), "goal-a") is not None
+
+    foreign = _root_digest(tmp_path / "elsewhere", canonical=False)
+    path.write_text(json.dumps(_active_journal(foreign)))
+    with pytest.raises(ShadowManagementError) as failure:
+        require_shadow_primary_write_allowed(link, "goal-a")
+    assert failure.value.code == "shadow_management_state_invalid"
+
+
+def test_python_reads_the_typescript_bootstrap_through_a_symlinked_root(tmp_path: Path) -> None:
+    link = _symlinked_root(tmp_path)
+    script = """
+import {bootstrapManagedShadow} from './loopx/control_plane/coordination/shadow_management.ts';
+const root = process.argv[1];
+const request = {runtime_root:root,goal_id:'goal-a',operation_id:'bootstrap:symlink',source_version:'v1',source_snapshot:{},projection:{goal_id:'goal-a',todos:[],leases:[]}};
+const result = await bootstrapManagedShadow(request,{withPrimaryLocks:async fn=>await fn(),verifySourceSnapshot:async()=>{}});
+process.stdout.write(JSON.stringify(result));
+"""
+    result = subprocess.run(
+        ["node", "--no-warnings", "--experimental-strip-types", "--input-type=module", "-e", script, str(link)],
+        check=True, capture_output=True, text=True,
+    )
+    applied = json.loads(result.stdout)
+    assert applied["status"] == "applied"
+    state = read_shadow_management_state(link, "goal-a")
+    assert state is not None and state["status"] == "active"
+    # The owner stored the canonical spelling; the guard reads it through the symlink.
+    assert state["source_root_digest"] == _root_digest(link, canonical=True)
+    binding = require_shadow_primary_write_allowed(link, "goal-a")
+    assert binding is not None
+    assert binding["capture_lineage_id"] == applied["capture_lineage_id"]

tests/control_plane/test_shared_goal_authority_e2e.py

@@ -250,6 +250,21 @@ def test_stage_2a_row_reports_specific_unverified_reasons_for_each_missing_input
     assert ladder.collect_bindings(inputs)["nokv_client_config_sha256"] is not None
 
 
+@pytest.mark.stage2c_e2e
+def test_stage_2c2_row_passes_when_the_ladder_root_is_reached_through_a_symlink(tmp_path: Path) -> None:
+    """macOS's default temp directory is a symlink; the shadow rows must not care (#4892)."""
+
+    real = tmp_path / "real-root"
+    real.mkdir()
+    link = tmp_path / "link-root"
+    link.symlink_to(real, target_is_directory=True)
+    row = ladder.row_by_id("s2c2.outbox_prepared_then_committed_entries")
+    result = ladder.run_row(row, root=link, environ=os.environ)
+    if result.status == "unverified":
+        pytest.skip(f"unverified: {result.reason_code}")
+    assert result.status == "pass", (result.reason_code, result.evidence)
+
+
 def test_nokv_sdk_pin_and_fence_checks_agree_across_helper_ladder_and_probe() -> None:
     from loopx.control_plane.coordination import nokv_jsonl_helper as helper
 

tests/control_plane_ts/local_authority_shadow_outbox.test.ts

@@ -1,7 +1,7 @@
 import assert from "node:assert/strict";
 import { execFile } from "node:child_process";
-import { readFile, writeFile, rename, unlink } from "node:fs/promises";
-import { join } from "node:path";
+import { readFile, writeFile, rename, symlink, unlink } from "node:fs/promises";
+import { dirname, join, resolve } from "node:path";
 import test from "node:test";
 import { promisify } from "node:util";
 import type { JsonObject } from "../../loopx/control_plane/effect_program.ts";
@@ -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,29 @@ 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("lease capture binds its entry to the lineage digest when the runtime root is a symlink", async (t) => {
+  const f = await fixture(t);
+  // A second spelling of the same root: the owner's binding digest comes from
+  // the resolved path, the lexical spelling of the link hashes differently (#4892).
+  const link = join(dirname(f.root), `${f.root.split("/").at(-1)}-link`);
+  await symlink(f.root, link, "dir");
+  t.after(() => unlink(link));
+  const binding = await requireShadowCaptureBinding(link, "goal-a");
+  assert.notEqual(binding.source_root_digest, sha(resolve(link)));
+  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: link, goal_id: "goal-a",
+    lease_directory: join(link, "goals", "goal-a", "task-leases"), write_class: "task_lease_acquire",
+    operation_id: "op-symlink", previous_lease: null, planned_lease: planned, active_todo_ids: null });
+  assert.equal(capture.failure, null);
+  const prepared = JSON.parse(await readFile(
+    join(link, "authority-shadow", "outbox", "goal-a", "leases",
+      `0000000001-${capture.entry_id}.prepared.json`), "utf8"));
+  assert.equal(prepared.source_root_digest, binding.source_root_digest);
+  assert.equal(prepared.entry_id, outboxEntryIdentity("goal-a", "leases", 1, String(prepared.source.bytes_digest),
+    String(prepared.capture_lineage_id), binding.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");