From c0cb1b3878e2f30baff7665ae3eb072e83d349a0 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Tue, 22 Sep 2026 18:05:07 +0800 Subject: [PATCH 1/2] fix(coordination): preserve shadow lineage across root aliases Accept both lexical and canonical runtime-root digests when Python reads the TypeScript-owned management journal. Bind Python and TypeScript outbox entries, plus drain receipt validation, to the active lineage digest instead of recomputing a conflicting path identity. Add explicit symlink-root regression coverage so macOS temporary roots exercise the cross-runtime path. Signed-off-by: duanjialing.777 --- .../local_authority_shadow_adapter.py | 5 ++-- .../local_authority_shadow_outbox.py | 8 +++--- .../local_authority_shadow_outbox.ts | 4 +-- .../coordination/shadow_management.py | 16 ++++++++++-- .../test_local_authority_shadow_outbox.py | 26 +++++++++++++++++++ tests/control_plane/test_shadow_management.py | 15 ++++++++--- .../local_authority_shadow_outbox.test.ts | 19 +++++++++++++- 7 files changed, 80 insertions(+), 13 deletions(-) diff --git a/loopx/control_plane/coordination/local_authority_shadow_adapter.py b/loopx/control_plane/coordination/local_authority_shadow_adapter.py index 4f973a0ba8..b5a6dd0fb5 100644 --- a/loopx/control_plane/coordination/local_authority_shadow_adapter.py +++ b/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( diff --git a/loopx/control_plane/coordination/local_authority_shadow_outbox.py b/loopx/control_plane/coordination/local_authority_shadow_outbox.py index 43455c80ea..e0b4b3fee8 100644 --- a/loopx/control_plane/coordination/local_authority_shadow_outbox.py +++ b/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, diff --git a/loopx/control_plane/coordination/local_authority_shadow_outbox.ts b/loopx/control_plane/coordination/local_authority_shadow_outbox.ts index d6686152e2..66fb1b21d2 100644 --- a/loopx/control_plane/coordination/local_authority_shadow_outbox.ts +++ b/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 = { diff --git a/loopx/control_plane/coordination/shadow_management.py b/loopx/control_plane/coordination/shadow_management.py index 738907142e..445b66d6e2 100644 --- a/loopx/control_plane/coordination/shadow_management.py +++ b/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"}: diff --git a/tests/control_plane/test_local_authority_shadow_outbox.py b/tests/control_plane/test_local_authority_shadow_outbox.py index 2436e4dee9..32e0c92b9d 100644 --- a/tests/control_plane/test_local_authority_shadow_outbox.py +++ b/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) diff --git a/tests/control_plane/test_shadow_management.py b/tests/control_plane/test_shadow_management.py index cc247e05b1..0ddcb26829 100644 --- a/tests/control_plane/test_shadow_management.py +++ b/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,14 +54,15 @@ 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 other = tmp_path / "other-root" destination = shadow_management_state_path(other, "goal-a") destination.parent.mkdir(parents=True) diff --git a/tests/control_plane_ts/local_authority_shadow_outbox.test.ts b/tests/control_plane_ts/local_authority_shadow_outbox.test.ts index 495293b04d..0ed1ff74bc 100644 --- a/tests/control_plane_ts/local_authority_shadow_outbox.test.ts +++ b/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"); From fa26a5cfe75322c1786b6a4e91d4dbb54ca24762 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Tue, 22 Sep 2026 18:29:54 +0800 Subject: [PATCH 2/2] test(coordination): reject mixed shadow root digests Signed-off-by: duanjialing.777 --- tests/control_plane/test_shadow_management.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/tests/control_plane/test_shadow_management.py b/tests/control_plane/test_shadow_management.py index 0ddcb26829..458f9c578d 100644 --- a/tests/control_plane/test_shadow_management.py +++ b/tests/control_plane/test_shadow_management.py @@ -63,10 +63,20 @@ def test_python_reads_typescript_binding_across_root_alias_and_rejects_cross_roo 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")