From ae379c539c29aec128644caf0b1a01472770bae3 Mon Sep 17 00:00:00 2001 From: Tyce Herrman Date: Sun, 1 Nov 2026 21:40:14 -0500 Subject: [PATCH 1/7] fix(receipts): preserve macOS identity across remounts and APM downgrades --- .../owners/install-deployment.json | 1 + ...install_base_integrator_and_contraction.py | 11 +- src/apm_cli/core/deployment_ledger.py | 209 ++++++++++++++---- src/apm_cli/utils/filesystem_identity.py | 47 ++++ ...t_architecture_materialization_receipts.py | 42 +++- .../test_generated_lifecycle_state_machine.py | 2 +- .../test_receipt_version_compatibility.py | 81 +++++++ .../test_required_lifecycle_state_machine.py | 18 +- ...skill_materialization_receipt_lifecycle.py | 4 +- .../test_pipeline_ownership_snapshot.py | 2 +- ...est_deployment_materialization_receipts.py | 181 +++++++++++++++ 11 files changed, 543 insertions(+), 55 deletions(-) create mode 100644 src/apm_cli/utils/filesystem_identity.py create mode 100644 tests/integration/test_receipt_version_compatibility.py diff --git a/.apm/architecture/owners/install-deployment.json b/.apm/architecture/owners/install-deployment.json index 42e01df344..21ca025c18 100644 --- a/.apm/architecture/owners/install-deployment.json +++ b/.apm/architecture/owners/install-deployment.json @@ -46,6 +46,7 @@ "owner": "deployment_ledger.py (DeploymentLedgerCodec and installation-bound DeploymentMaterializationSession); receipt claim keys route through install/deployed_paths.py (deployed_path_entry) and retain the admitted key through postcommit binding; authenticated same-writer target handoffs require a guarded actual copy, never adopted-receipt rebinding; live install and uninstall survivor sessions persist actual final-rendered skill writes, with final-lock-success binding; cleanup retires confirmed outcomes once per phase before survivor copies; read-only index: skill_support.build_skill_ownership_maps, exposed by SkillIntegrator._build_ownership_maps with integrate-phase snapshot lifetime", "selectors": [ "src/apm_cli/core/deployment_ledger.py", + "src/apm_cli/utils/filesystem_identity.py", "src/apm_cli/integration/skill_support.py", "src/apm_cli/install/pipeline.py" ], diff --git a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py index daf37dc3c5..84872d17f6 100644 --- a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py +++ b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py @@ -1277,8 +1277,15 @@ def _materialization_evidence_findings(provider: FactsProvider, rule_id: str) -> requirements = { _DEPLOYMENT_LEDGER: ( "class DeploymentMaterializationSession:", - 'data["installation"] != installation', - 'data["root_identity"] != self._root_identity()', + 'data["installation"] != self._installation', + "not self._matches_root_identity(data)", + 'return data["root_identity"] == [metadata.st_dev, metadata.st_ino]', + 'return data["root_identity"] == self._root_identity()', + '"deployment-receipts-v2"', + '"volume": darwin_volume_uuid(self.root)', + "self.baseline(path, [target]) != record.content_hash", + "compute_file_hash(path) != record.content_hash", + "if snapshot[key] == self._legacy_snapshot.get(key):", "record.active_owner not in self._valid_owners", "self._bindings.get(key) != [record.active_owner, self._claim_signature(value)]", "value = deployed_path_entry(path, self.root, targets)", diff --git a/src/apm_cli/core/deployment_ledger.py b/src/apm_cli/core/deployment_ledger.py index 370c2f4e86..29d541a70a 100644 --- a/src/apm_cli/core/deployment_ledger.py +++ b/src/apm_cli/core/deployment_ledger.py @@ -4,6 +4,7 @@ import hashlib import json +import sys import uuid from collections.abc import Collection, Mapping, Sequence from dataclasses import dataclass @@ -85,13 +86,16 @@ def __init__(self, project_root: Path, scope: InstallScope, lockfile: LockFile | self.root = project_root.resolve() self.scope = scope key = hashlib.sha256(f"{self.root}|{scope.value}".encode()).hexdigest() - self.state_path = Path.home() / ".apm" / "deployment-receipts" / f"{key}.json" + self.state_path = Path.home() / ".apm" / "deployment-receipts-v2" / f"{key}.json" + self.legacy_state_path = Path.home() / ".apm" / "deployment-receipts" / f"{key}.json" self.identity_path = self.root / ".apm-local" / "installation" self._records: dict[str, DeploymentRecord] = {} self._destinations: dict[str, str] = {} self._bindings: dict[str, list[str]] = {} self._written: dict[str, str] = {} self._dirty = False + self._migration_pending = False + self._legacy_snapshot: dict[str, str] = {} self._claims: dict[str, list[DeploymentRecord]] = {} if lockfile is not None: for record in DeploymentLedgerCodec.from_lockfile(lockfile).records.values(): @@ -106,12 +110,15 @@ def __init__(self, project_root: Path, scope: InstallScope, lockfile: LockFile | self._state_present = False self._load() - def _validate_paths(self) -> None: - for base, path in ( + def _validate_paths(self, *, legacy: bool = False) -> None: + paths = [ (Path.home(), self.state_path), (self.root, self.identity_path), (self.root, self.identity_path.parent / ".gitignore"), - ): + ] + if legacy: + paths.append((Path.home(), self.legacy_state_path)) + for base, path in paths: ensure_path_within(path, base) if has_symlink_component(base, path, raise_on_error=True): raise PathTraversalError( @@ -124,9 +131,32 @@ def _validate_paths(self) -> None: "Preserve this path and move it aside before retrying." ) - def _root_identity(self) -> list[int]: + def _root_identity(self) -> dict[str, Any]: metadata = self.root.stat() - return [metadata.st_dev, metadata.st_ino] + if sys.platform == "darwin": + from apm_cli.utils.filesystem_identity import darwin_volume_uuid + + # Device numbers identify a mount, not a persistent volume. Keep + # directory identity as well: a copied installation UUID alone is + # never authority to reuse a different directory's receipt. + return { + "kind": "darwin-volume", + "volume": darwin_volume_uuid(self.root), + "inode": metadata.st_ino, + "created": metadata.st_birthtime, + } + # Retain the strict existing rule on platforms without a supported + # persistent-volume implementation; never silently weaken it. + return {"kind": "stat", "device": metadata.st_dev, "inode": metadata.st_ino} + + def _matches_root_identity(self, data: dict[str, Any]) -> bool: + if data["version"] == 1: + metadata = self.root.stat() + # A legacy receipt cannot prove which volume a changed device + # number referred to. Migrate only an exact legacy binding, and + # publish the new format at a live install's commit boundary. + return data["root_identity"] == [metadata.st_dev, metadata.st_ino] + return data["root_identity"] == self._root_identity() def _load(self) -> None: self._validate_paths() @@ -138,50 +168,136 @@ def _load(self) -> None: if uuid.UUID(hex=installation).hex != installation: raise ValueError("invalid installation identity") self._installation = installation - if not self.state_path.exists(): - return - data = json.loads(self.state_path.read_text(encoding="utf-8")) - if ( - data["version"] != 1 - or data["root"] != self.root.as_posix() - or data["scope"] != self.scope.value - or data["installation"] != installation - or data["root_identity"] != self._root_identity() - ): - raise ValueError("installation binding does not match") - ledger = DeploymentLedgerCodec.from_rows(data["records"]) - if len(ledger.records) != len(data["records"]): - raise ValueError("duplicate rendered-write records") - destinations = data["destinations"] - bindings = data["bindings"] - if not isinstance(destinations, dict) or set(destinations) != set(ledger.records): - raise ValueError("destination bindings do not match") - if not isinstance(bindings, dict) or set(bindings) != set(ledger.records): - raise ValueError("ownership bindings do not match") - for record in ledger.records.values(): - if ( - record.locator.kind == LocatorKind.URI - or record.locator.scope != self.scope.value - or not _is_canonical_bundle_hash(record.content_hash) - or not isinstance(destinations[record.locator.key], str) - or not isinstance(bindings[record.locator.key], list) - or len(bindings[record.locator.key]) != 2 - or not all(isinstance(value, str) for value in bindings[record.locator.key]) + if self._state_present: + data = self._read_state(self.state_path, version=2) + self._set_state(data) + snapshot = data.get("legacy_snapshot", {}) + if not isinstance(snapshot, dict) or not all( + isinstance(key, str) and isinstance(value, str) + for key, value in snapshot.items() ): - raise ValueError("invalid rendered-write record") - self._records = dict(ledger.records) - self._destinations = destinations - self._bindings = bindings - self._physical_keys = {path: key for key, path in destinations.items()} - if len(self._physical_keys) != len(destinations): - raise ValueError("ambiguous physical write receipts") + raise ValueError("invalid legacy receipt checkpoint") + self._legacy_snapshot = snapshot + if self.legacy_state_path.exists() or self.legacy_state_path.is_symlink(): + self._validate_paths(legacy=True) + try: + legacy = self._read_state(self.legacy_state_path, version=1) + except (ValueError, KeyError, TypeError, UnicodeError): + if not self._state_present: + raise + # A valid stable receipt remains usable after remount, + # even when the older device-based receipt has expired. + return + if not self._state_present: + self._set_state(legacy) + self._legacy_snapshot = self._snapshot(legacy) + self._migration_pending = True + else: + self._merge_legacy_writes(legacy) except (ValueError, KeyError, TypeError, UnicodeError) as exc: + path = self.state_path if self._state_present else self.legacy_state_path raise ValueError( - f"Invalid local materialization evidence at {self.state_path}: {exc}. " + f"Invalid local materialization evidence at {path}: {exc}. " "Preserve local skill edits, then remove the invalid receipt file and retry. " "Existing files without trustworthy evidence will not be overwritten." ) from exc + def _read_state(self, path: Path, *, version: int) -> dict[str, Any]: + data = json.loads(path.read_text(encoding="utf-8")) + if ( + data["version"] != version + or data["root"] != self.root.as_posix() + or data["scope"] != self.scope.value + or data["installation"] != self._installation + or not self._matches_root_identity(data) + ): + raise ValueError("installation binding does not match") + ledger = DeploymentLedgerCodec.from_rows(data["records"]) + if len(ledger.records) != len(data["records"]): + raise ValueError("duplicate rendered-write records") + destinations = data["destinations"] + bindings = data["bindings"] + if not isinstance(destinations, dict) or set(destinations) != set(ledger.records): + raise ValueError("destination bindings do not match") + if not isinstance(bindings, dict) or set(bindings) != set(ledger.records): + raise ValueError("ownership bindings do not match") + for record in ledger.records.values(): + if ( + record.locator.kind == LocatorKind.URI + or record.locator.scope != self.scope.value + or not _is_canonical_bundle_hash(record.content_hash) + or not isinstance(destinations[record.locator.key], str) + or not isinstance(bindings[record.locator.key], list) + or len(bindings[record.locator.key]) != 2 + or not all(isinstance(value, str) for value in bindings[record.locator.key]) + ): + raise ValueError("invalid rendered-write record") + if len(set(destinations.values())) != len(destinations): + raise ValueError("ambiguous physical write receipts") + return data + + def _set_state(self, data: dict[str, Any]) -> None: + self._records = dict(DeploymentLedgerCodec.from_rows(data["records"]).records) + self._destinations = dict(data["destinations"]) + self._bindings = dict(data["bindings"]) + self._physical_keys = {path: key for key, path in self._destinations.items()} + + @staticmethod + def _snapshot(data: dict[str, Any]) -> dict[str, str]: + records = DeploymentLedgerCodec.from_rows(data["records"]).records + rows = DeploymentLedgerCodec.rows(DeploymentLedger(records)) + return { + key: hashlib.sha256( + json.dumps( + [row, data["destinations"][key], data["bindings"][key]], sort_keys=True + ).encode() + ).hexdigest() + for key, row in zip(sorted(records), rows, strict=True) + } + + def _merge_legacy_writes(self, data: dict[str, Any]) -> None: + """Admit changed older-writer evidence, never an unchanged stale baseline.""" + from apm_cli.integration.targets import KNOWN_TARGETS + from apm_cli.utils.content_hash import compute_file_hash + + snapshot = self._snapshot(data) + if snapshot == self._legacy_snapshot: + return + for key, record in DeploymentLedgerCodec.from_rows(data["records"]).records.items(): + if snapshot[key] == self._legacy_snapshot.get(key): + continue + destination = data["destinations"][key] + path = Path(destination) + profile = KNOWN_TARGETS.get(record.locator.target) + target = profile.for_scope(user_scope=self.scope.value == "user") if profile else None + if target is None or not path.is_file() or path.is_symlink(): + continue + previous = ( + self._records.copy(), + self._destinations.copy(), + self._bindings.copy(), + self._physical_keys.copy(), + ) + # Validate ownership and physical target through the same authority + # check used by the writer, then require actual rendered bytes. + physical = path.resolve().as_posix() + prior_key = self._physical_keys.get(physical) + if prior_key is not None and prior_key != key: + self._records.pop(prior_key, None) + self._destinations.pop(prior_key, None) + self._bindings.pop(prior_key, None) + self._records[key] = record + self._destinations[key] = destination + self._bindings[key] = data["bindings"][key] + self._physical_keys[physical] = key + if ( + self.baseline(path, [target]) != record.content_hash + or compute_file_hash(path) != record.content_hash + ): + self._records, self._destinations, self._bindings, self._physical_keys = previous + self._legacy_snapshot = snapshot + self._migration_pending = True + def _locator(self, path: Path, targets: Sequence[TargetProfile]) -> DeploymentLocator | None: for target in targets: if not target.supports("skills"): @@ -307,7 +423,7 @@ def bind_committed_claims(self, lockfile: LockFile | None) -> None: Keep that projection separate from the actual writer; never infer a writer or a baseline from the committed lock. """ - if not self._written: + if not self._written and not self._migration_pending: return committed = type(self)(self.root, self.scope, lockfile) for key, value in self._written.items(): @@ -316,6 +432,8 @@ def bind_committed_claims(self, lockfile: LockFile | None) -> None: continue committed._bindings[key] = [record.active_owner, committed._claim_signature(value)] committed._dirty = True + if committed._migration_pending: + committed._dirty = True committed.flush() def _persist(self, records: dict[str, DeploymentRecord], destinations: dict[str, str]) -> None: @@ -342,7 +460,7 @@ def _persist(self, records: dict[str, DeploymentRecord], destinations: dict[str, self.state_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) self._validate_paths() data = { - "version": 1, + "version": 2, "root": self.root.as_posix(), "root_identity": self._root_identity(), "scope": self.scope.value, @@ -350,6 +468,7 @@ def _persist(self, records: dict[str, DeploymentRecord], destinations: dict[str, "records": DeploymentLedgerCodec.rows(DeploymentLedger(records)), "destinations": destinations, "bindings": {key: self._bindings[key] for key in records}, + "legacy_snapshot": self._legacy_snapshot, } atomic_write_text( self.state_path, diff --git a/src/apm_cli/utils/filesystem_identity.py b/src/apm_cli/utils/filesystem_identity.py new file mode 100644 index 0000000000..7301f54881 --- /dev/null +++ b/src/apm_cli/utils/filesystem_identity.py @@ -0,0 +1,47 @@ +"""Native persistent volume identity, without invoking an external command.""" + +from __future__ import annotations + +import ctypes +import os +import uuid +from pathlib import Path + + +class _AttrList(ctypes.Structure): + _fields_ = [ + ("bitmapcount", ctypes.c_uint16), + ("reserved", ctypes.c_uint16), + ("commonattr", ctypes.c_uint32), + ("volattr", ctypes.c_uint32), + ("dirattr", ctypes.c_uint32), + ("fileattr", ctypes.c_uint32), + ("forkattr", ctypes.c_uint32), + ] + + +def darwin_volume_uuid(path: Path) -> str: + """Read ATTR_VOL_UUID; fail closed if a stable identity is unavailable.""" + libc = ctypes.CDLL("/usr/lib/libSystem.B.dylib", use_errno=True) + getattrlist = libc.getattrlist + getattrlist.argtypes = [ + ctypes.c_char_p, + ctypes.POINTER(_AttrList), + ctypes.c_void_p, + ctypes.c_size_t, + ctypes.c_ulong, + ] + getattrlist.restype = ctypes.c_int + # ATTR_VOL_INFO | ATTR_VOL_UUID; the packed result is a 32-bit length + # followed by the 16-byte UUID. getattrlist accepts a path within a volume. + attrs = _AttrList(5, 0, 0, 0x80040000, 0, 0, 0) + result = ctypes.create_string_buffer(20) + if getattrlist(os.fsencode(path), ctypes.byref(attrs), result, len(result), 0): + error = ctypes.get_errno() + raise OSError(error, os.strerror(error), str(path)) + if int.from_bytes(result.raw[:4], "little") != 20: + raise ValueError("Invalid persistent volume identity response") + volume = uuid.UUID(bytes=result.raw[4:20]) + if volume.int == 0: + raise ValueError("Persistent volume identity unavailable") + return str(volume) diff --git a/tests/integration/test_architecture_materialization_receipts.py b/tests/integration/test_architecture_materialization_receipts.py index 725a168e1e..cb2d044e18 100644 --- a/tests/integration/test_architecture_materialization_receipts.py +++ b/tests/integration/test_architecture_materialization_receipts.py @@ -115,9 +115,49 @@ def test_receipt_guard_accepts_current_owner_routing() -> None: ), ( "core/deployment_ledger.py", - 'data["root_identity"] != self._root_identity()', + "not self._matches_root_identity(data)", "False", ), + ( + "core/deployment_ledger.py", + 'data["installation"] != self._installation', + "False", + ), + ( + "core/deployment_ledger.py", + 'return data["root_identity"] == [metadata.st_dev, metadata.st_ino]', + "return True", + ), + ( + "core/deployment_ledger.py", + 'return data["root_identity"] == self._root_identity()', + "return True", + ), + ( + "core/deployment_ledger.py", + '"deployment-receipts-v2"', + '"deployment-receipts"', + ), + ( + "core/deployment_ledger.py", + '"volume": darwin_volume_uuid(self.root)', + '"volume": str(metadata.st_dev)', + ), + ( + "core/deployment_ledger.py", + "self.baseline(path, [target]) != record.content_hash", + "False", + ), + ( + "core/deployment_ledger.py", + "compute_file_hash(path) != record.content_hash", + "False", + ), + ( + "core/deployment_ledger.py", + "if snapshot[key] == self._legacy_snapshot.get(key):", + "if False:", + ), ( "core/deployment_ledger.py", "record.active_owner not in self._valid_owners", diff --git a/tests/integration/test_generated_lifecycle_state_machine.py b/tests/integration/test_generated_lifecycle_state_machine.py index c83ef8e86e..7ff1ff32ef 100644 --- a/tests/integration/test_generated_lifecycle_state_machine.py +++ b/tests/integration/test_generated_lifecycle_state_machine.py @@ -946,7 +946,7 @@ def test_generated_identity_write_scope_is_exact( if extra_write == "identity-sibling": unexpected = identity.parent / "unrelated.txt" elif extra_write == "receipt-sibling": - unexpected = home / ".apm/deployment-receipts/unrelated.json" + unexpected = home / ".apm/deployment-receipts-v2/unrelated.json" else: wrong_root = project if model.fixture.global_scope else home unexpected = wrong_root / ".apm-local/installation" diff --git a/tests/integration/test_receipt_version_compatibility.py b/tests/integration/test_receipt_version_compatibility.py new file mode 100644 index 0000000000..9e0da4ffcb --- /dev/null +++ b/tests/integration/test_receipt_version_compatibility.py @@ -0,0 +1,81 @@ +"""Run the source CLI in a disposable home; never touch live APM state.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from tests.utils.apm_lifecycle_runner import ApmLifecycleRunner +from tests.utils.isolated_apm_environment import IsolatedApmEnvironment + +pytestmark = [pytest.mark.integration] + + +@pytest.mark.parametrize("user_scope", [False, True]) +@pytest.mark.parametrize("edited", [False, True]) +def test_real_legacy_cli_upgrade_downgrade_upgrade( + tmp_path: Path, apm_engine_command, user_scope: bool, edited: bool +): + legacy_binary = os.environ.get("APM_LEGACY_BINARY") + if not legacy_binary: + pytest.skip("Set APM_LEGACY_BINARY to the unpatched APM 0.35.0 executable") + isolated = IsolatedApmEnvironment.create(tmp_path / "environment", base_env=dict(os.environ)) + env = isolated.subprocess_env() + project = isolated.work_root / "consumer" + project.mkdir(parents=True) + root = isolated.config_root if user_scope else project + root.mkdir(parents=True, exist_ok=True) + package = isolated.work_root / "package" + source = package / ".apm/skills/demo" + source.mkdir(parents=True) + skill = source / "SKILL.md" + skill.write_text("---\nname: demo\ndescription: Compatibility fixture\n---\nv1\n") + (package / "apm.yml").write_text("name: demo-pkg\nversion: 1.0.0\n") + (root / "apm.yml").write_text( + f"name: consumer\nversion: 1.0.0\ndependencies:\n apm:\n - path: {package}\ntargets:\n- agent-skills\n" + ) + args = ("install", *(("-g",) if user_scope else ()), "--target", "agent-skills", "--no-policy") + legacy = ApmLifecycleRunner((legacy_binary,)) + fixed_binary = os.environ.get("APM_FIXED_BINARY") + fixed = ApmLifecycleRunner((fixed_binary,) if fixed_binary else apm_engine_command) + + def install(runner, *extra): + result = runner.run((*args, *extra), cwd=project, env=env) + assert result.returncode == 0, result.stdout + result.stderr + return result + + install(legacy) + old_receipt = next((isolated.home / ".apm/deployment-receipts").glob("*.json")) + old_bytes = old_receipt.read_bytes() + new_receipt = isolated.home / ".apm/deployment-receipts-v2" / old_receipt.name + deployed = (isolated.home if user_scope else project) / ".agents/skills/demo" + before = {p.name: (p.read_bytes(), p.stat().st_mtime_ns) for p in deployed.iterdir()} + install(fixed, "--dry-run") + assert not new_receipt.exists() + assert old_receipt.read_bytes() == old_bytes + install(fixed) + assert new_receipt.exists(), "An unchanged live install must migrate the valid old receipt" + assert old_receipt.read_bytes() == old_bytes + assert {p.name: (p.read_bytes(), p.stat().st_mtime_ns) for p in deployed.iterdir()} == before + if edited: + (deployed / "SKILL.md").write_text("local edits\n") + # Old APM must still read its own receipt and make real, witnessed writes. + skill.write_text(skill.read_text().replace("v1", "v2")) + (source / "asset.md").write_text("v2\n") + install(legacy) + legacy_after_downgrade = old_receipt.read_bytes() + assert json.loads(legacy_after_downgrade)["version"] == 1 + assert (deployed / "asset.md").read_text() == "v2\n" + assert (deployed / "SKILL.md").read_text() == ("local edits\n" if edited else skill.read_text()) + # Returning to fixed APM must recognize changed older-writer evidence, + # while an unchanged older baseline must not authorize a user's edits. + skill.write_text(skill.read_text().replace("v2", "v3")) + (source / "asset.md").write_text("v3\n") + install(fixed) + assert (deployed / "asset.md").read_text() == "v3\n" + assert (deployed / "SKILL.md").read_text() == ("local edits\n" if edited else skill.read_text()) + assert old_receipt.read_bytes() == legacy_after_downgrade + assert json.loads(new_receipt.read_text())["version"] == 2 diff --git a/tests/integration/test_required_lifecycle_state_machine.py b/tests/integration/test_required_lifecycle_state_machine.py index 17308eda5c..ebc6b4339f 100644 --- a/tests/integration/test_required_lifecycle_state_machine.py +++ b/tests/integration/test_required_lifecycle_state_machine.py @@ -5,6 +5,7 @@ import json import os import shutil +import sys import zipfile from collections.abc import Callable, Mapping from dataclasses import dataclass @@ -131,7 +132,7 @@ def _assert_skill_receipt( bound: bool, ) -> Path: """Check successful-write evidence and its expected committed-claim binding.""" - receipts = list((scenario.isolated.config_root / "deployment-receipts").glob("*.json")) + receipts = list((scenario.isolated.config_root / "deployment-receipts-v2").glob("*.json")) assert len(receipts) == 1 receipt = receipts[0] assert receipt.is_file() and not receipt.is_symlink() @@ -141,9 +142,20 @@ def _assert_skill_receipt( assert UUID(hex=data["installation"]).hex == data["installation"] assert identity.read_bytes() == (data["installation"] + "\n").encode("ascii") assert (identity.parent / ".gitignore").read_bytes() == b"*\n" - assert data["version"] == 1 + assert data["version"] == 2 assert data["root"] == root.resolve().as_posix() - assert data["root_identity"] == [root.stat().st_dev, root.stat().st_ino] + metadata = root.stat() + if sys.platform == "darwin": + assert data["root_identity"]["kind"] == "darwin-volume" + assert UUID(data["root_identity"]["volume"]).int != 0 + assert data["root_identity"]["inode"] == metadata.st_ino + assert data["root_identity"]["created"] == metadata.st_birthtime + else: + assert data["root_identity"] == { + "kind": "stat", + "device": metadata.st_dev, + "inode": metadata.st_ino, + } assert data["scope"] == scope ledger = DeploymentLedgerCodec.from_rows(data["records"]) assert len(ledger.records) == len(expected) diff --git a/tests/integration/test_skill_materialization_receipt_lifecycle.py b/tests/integration/test_skill_materialization_receipt_lifecycle.py index c99c29e56d..ad28b8aafc 100644 --- a/tests/integration/test_skill_materialization_receipt_lifecycle.py +++ b/tests/integration/test_skill_materialization_receipt_lifecycle.py @@ -18,7 +18,7 @@ def _receipt(scenario) -> Path: receipts = list( - (Path(scenario.environment["HOME"]) / ".apm/deployment-receipts").glob("*.json") + (Path(scenario.environment["HOME"]) / ".apm/deployment-receipts-v2").glob("*.json") ) assert len(receipts) == 1 return receipts[0] @@ -140,7 +140,7 @@ def test_receipts_capture_rendered_links_and_preserve_repeat_bytes( scenario.install("a") lock = scenario.lock_root / "apm.lock.yaml" before = lock.read_bytes() - receipts = list((scenario.environment.home / ".apm/deployment-receipts").glob("*.json")) + receipts = list((scenario.environment.home / ".apm/deployment-receipts-v2").glob("*.json")) assert len(receipts) == 1 data = json.loads(receipts[0].read_text(encoding="utf-8")) rows = [row for row in data["records"] if row["value"].endswith("SKILL.md")] diff --git a/tests/unit/install/test_pipeline_ownership_snapshot.py b/tests/unit/install/test_pipeline_ownership_snapshot.py index cb9d0cb0e0..7d40f31bfa 100644 --- a/tests/unit/install/test_pipeline_ownership_snapshot.py +++ b/tests/unit/install/test_pipeline_ownership_snapshot.py @@ -321,7 +321,7 @@ def test_receipt_persistence_failure_is_a_pipeline_failure( def fail_receipt(path: Path, text: str, **kwargs: Any) -> None: nonlocal writes, failed_path, prior_receipt - if path.parent.name == "deployment-receipts": + if path.parent.name == "deployment-receipts-v2": writes += 1 if writes == failed_write: failed_path = path diff --git a/tests/unit/test_deployment_materialization_receipts.py b/tests/unit/test_deployment_materialization_receipts.py index fcc008ec0b..6d3c1bd32f 100644 --- a/tests/unit/test_deployment_materialization_receipts.py +++ b/tests/unit/test_deployment_materialization_receipts.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import sys from pathlib import Path import pytest @@ -253,6 +254,186 @@ def test_recreated_checkout_cannot_reuse_old_receipt( assert _session(root, lock).baseline(path, [TARGET]) is None +@pytest.mark.skipif(sys.platform != "darwin", reason="Darwin volume identity") +@pytest.mark.parametrize("symlinked", [False, True]) +def test_darwin_receipt_survives_changed_device_number(tmp_path, monkeypatch, symlinked): + from types import SimpleNamespace + + if symlinked: + real = tmp_path / "real" + real.mkdir() + alias = tmp_path / "alias" + alias.symlink_to(real, target_is_directory=True) + tmp_path = alias + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + content_hash = compute_file_hash(path) + session.record({path: content_hash}, [TARGET], "pkg") + session.flush() + original_stat = Path.stat + resolved_root = root.resolve() + changed = [] + + def changed_device(self, *args, **kwargs): + metadata = original_stat(self, *args, **kwargs) + if self == resolved_root: + changed.append(self) + return SimpleNamespace( + st_dev=metadata.st_dev + 100, + st_ino=metadata.st_ino, + st_birthtime=metadata.st_birthtime, + st_mode=metadata.st_mode, + ) + return metadata + + monkeypatch.setattr(Path, "stat", changed_device) + assert _session(root, lock).baseline(path, [TARGET]) == content_hash + assert changed, "The session must observe the simulated changed device" + + +@pytest.mark.parametrize("field", ["volume", "inode", "created"]) +def test_persistent_receipt_rejects_different_directory_identity(tmp_path, monkeypatch, field): + root, path, lock = _fixture(tmp_path, monkeypatch) + identity = {"kind": "darwin-volume", "volume": "volume-a", "inode": 42, "created": 123.5} + monkeypatch.setattr( + DeploymentMaterializationSession, "_root_identity", lambda self: identity.copy() + ) + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + identity[field] = "different" + with pytest.raises(ValueError, match="installation binding does not match"): + _session(root, lock) + + +@pytest.mark.parametrize("changed_device", [False, True]) +def test_legacy_receipt_migration_requires_exact_old_identity( + tmp_path, monkeypatch, changed_device +): + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + content_hash = compute_file_hash(path) + session.record({path: content_hash}, [TARGET], "pkg") + session.flush() + data = json.loads(session.state_path.read_text()) + metadata = root.stat() + data.update(version=1, root_identity=[metadata.st_dev + int(changed_device), metadata.st_ino]) + session.legacy_state_path.parent.mkdir(parents=True, exist_ok=True) + session.legacy_state_path.write_text(json.dumps(data)) + session.state_path.unlink() + before = session.legacy_state_path.read_bytes() + if changed_device: + with pytest.raises(ValueError, match="installation binding does not match"): + _session(root, lock) + else: + migrated = _session(root, lock) + assert migrated.baseline(path, [TARGET]) == content_hash + assert not migrated.state_path.exists() + migrated.flush() + assert not migrated.state_path.exists() + migrated.bind_committed_claims(lock) + assert json.loads(migrated.state_path.read_text())["version"] == 2 + assert migrated.legacy_state_path.read_bytes() == before + + +@pytest.mark.skipif(sys.platform != "darwin", reason="Darwin volume identity") +def test_volume_identity_failure_does_not_fall_back_to_device_number(tmp_path, monkeypatch): + from apm_cli.utils import filesystem_identity + + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + + def unavailable(path): + raise OSError("volume identity unavailable") + + monkeypatch.setattr(filesystem_identity, "darwin_volume_uuid", unavailable) + with pytest.raises(OSError, match="volume identity unavailable"): + _session(root, lock) + + +def _move_to_legacy(session): + data = json.loads(session.state_path.read_text()) + metadata = session.root.stat() + data.update(version=1, root_identity=[metadata.st_dev, metadata.st_ino]) + data.pop("legacy_snapshot", None) + session.legacy_state_path.parent.mkdir(parents=True, exist_ok=True) + session.legacy_state_path.write_text(json.dumps(data)) + session.state_path.unlink() + return data + + +def test_legacy_changes_cannot_adopt_an_edit_equal_to_an_older_version(tmp_path, monkeypatch): + root, path, lock = _fixture(tmp_path, monkeypatch) + asset = path.with_name("asset.md") + asset.write_text("old asset\n") + locator = DeploymentLedgerCodec.locator_for_path( + asset, project_root=root, target=TARGET, scope=InstallScope.PROJECT + ) + ledger = DeploymentLedgerCodec.from_lockfile(lock) + ledger.records[locator.key] = DeploymentRecord( + locator, ("pkg",), "pkg", compute_file_hash(asset) + ) + DeploymentLedgerCodec.apply_to_lockfile(ledger, lock, write_legacy_views=False) + session = _session(root, lock) + old_bytes = path.read_bytes() + session.record( + {path: compute_file_hash(path), asset: compute_file_hash(asset)}, [TARGET], "pkg" + ) + session.flush() + legacy = _move_to_legacy(session) + _session(root, lock).bind_committed_claims(lock) + path.write_text("new rendered\n") + new_hash = compute_file_hash(path) + session = _session(root, lock) + session.record({path: new_hash}, [TARGET], "pkg") + session.flush() + # A real older-writer change to another member must not revive this + # unchanged older baseline when the user restores the old skill bytes. + path.write_bytes(old_bytes) + asset.write_text("new asset\n") + for row in legacy["records"]: + if row["value"].endswith("asset.md"): + row["content_hash"] = compute_file_hash(asset) + session.legacy_state_path.write_text(json.dumps(legacy)) + reloaded = _session(root, lock) + assert reloaded.baseline(path, [TARGET]) == new_hash + assert reloaded.baseline(asset, [TARGET]) == compute_file_hash(asset) + reloaded.bind_committed_claims(lock) + assert _session(root, lock).baseline(path, [TARGET]) == new_hash + + +@pytest.mark.parametrize("damage", ["owner", "destination", "content", "device"]) +def test_returning_from_legacy_does_not_admit_unverified_changes(tmp_path, monkeypatch, damage): + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + content_hash = compute_file_hash(path) + session.record({path: content_hash}, [TARGET], "pkg") + session.flush() + legacy = _move_to_legacy(session) + _session(root, lock).bind_committed_claims(lock) + if damage == "owner": + legacy["records"][0]["active_owner"] = "foreign" + legacy["records"][0]["owners"] = ["foreign"] + elif damage == "destination": + foreign = root / "foreign.md" + foreign.write_bytes(path.read_bytes()) + key = next(iter(legacy["destinations"])) + legacy["destinations"][key] = str(foreign) + elif damage == "device": + legacy["root_identity"][0] += 1 + else: + legacy["records"][0]["content_hash"] = "sha256:" + "e" * 64 + session.legacy_state_path.write_text(json.dumps(legacy)) + proof = session.legacy_state_path.read_bytes() + reloaded = _session(root, lock) + assert reloaded.baseline(path, [TARGET]) == content_hash + reloaded.bind_committed_claims(lock) + assert _session(root, lock).baseline(path, [TARGET]) == content_hash + assert session.legacy_state_path.read_bytes() == proof + + def test_atomic_failure_cannot_publish_new_receipt( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: From 5c19275be430825eaa7534868b4e90994c802e22 Mon Sep 17 00:00:00 2001 From: Tyce Herrman Date: Mon, 2 Nov 2026 06:32:24 -0500 Subject: [PATCH 2/7] fix(receipts): publish safe legacy evidence for fresh-install downgrades --- ...install_base_integrator_and_contraction.py | 5 + src/apm_cli/core/deployment_ledger.py | 84 ++++++++++++ ...t_architecture_materialization_receipts.py | 25 ++++ .../test_receipt_version_compatibility.py | 61 +++++++++ ...skill_materialization_receipt_lifecycle.py | 15 +- ...est_deployment_materialization_receipts.py | 129 ++++++++++++++++++ 6 files changed, 316 insertions(+), 3 deletions(-) diff --git a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py index 84872d17f6..31d9bb10f6 100644 --- a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py +++ b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py @@ -1286,6 +1286,11 @@ def _materialization_evidence_findings(provider: FactsProvider, rule_id: str) -> "self.baseline(path, [target]) != record.content_hash", "compute_file_hash(path) != record.content_hash", "if snapshot[key] == self._legacy_snapshot.get(key):", + 'data["stable_root_identity"] != self._root_identity()', + "committed._publish_legacy_compatibility()", + 'data["legacy_snapshot"] = self._snapshot(self._legacy_projection(data))', + "self.baseline(Path(self._destinations[key]), [target]) != record.content_hash", + "new_file_mode=0o600,\n create_only=True,", "record.active_owner not in self._valid_owners", "self._bindings.get(key) != [record.active_owner, self._claim_signature(value)]", "value = deployed_path_entry(path, self.root, targets)", diff --git a/src/apm_cli/core/deployment_ledger.py b/src/apm_cli/core/deployment_ledger.py index 29d541a70a..fd2e0246c2 100644 --- a/src/apm_cli/core/deployment_ledger.py +++ b/src/apm_cli/core/deployment_ledger.py @@ -93,6 +93,7 @@ def __init__(self, project_root: Path, scope: InstallScope, lockfile: LockFile | self._destinations: dict[str, str] = {} self._bindings: dict[str, list[str]] = {} self._written: dict[str, str] = {} + self._recorded_targets: dict[str, TargetProfile] = {} self._dirty = False self._migration_pending = False self._legacy_snapshot: dict[str, str] = {} @@ -152,6 +153,11 @@ def _root_identity(self) -> dict[str, Any]: def _matches_root_identity(self, data: dict[str, Any]) -> bool: if data["version"] == 1: metadata = self.root.stat() + if ( + "stable_root_identity" in data + and data["stable_root_identity"] != self._root_identity() + ): + return False # A legacy receipt cannot prove which volume a changed device # number referred to. Migrate only an exact legacy binding, and # publish the new format at a live install's commit boundary. @@ -381,6 +387,7 @@ def record( if not written: return + self._recorded_targets.update((target.name, target) for target in targets) prepared: list[tuple[DeploymentRecord, str, str]] = [] for path, content_hash in written.items(): locator = self._locator(path, targets) @@ -424,6 +431,7 @@ def bind_committed_claims(self, lockfile: LockFile | None) -> None: writer or a baseline from the committed lock. """ if not self._written and not self._migration_pending: + self._publish_legacy_compatibility() return committed = type(self)(self.root, self.scope, lockfile) for key, value in self._written.items(): @@ -435,6 +443,77 @@ def bind_committed_claims(self, lockfile: LockFile | None) -> None: if committed._migration_pending: committed._dirty = True committed.flush() + committed._recorded_targets.update(self._recorded_targets) + committed._publish_legacy_compatibility() + + def _legacy_projection(self, data: dict[str, Any]) -> dict[str, Any]: + """Project recorded rendered-write evidence, never hashes adopted from disk.""" + metadata = self.root.stat() + return { + **{ + key: data[key] + for key in ("root", "scope", "installation", "records", "destinations", "bindings") + }, + "version": 1, + "root_identity": [metadata.st_dev, metadata.st_ino], + "stable_root_identity": data["root_identity"], + } + + def _publish_legacy_compatibility(self) -> None: + """Create missing old-reader evidence only at the committed-claims boundary.""" + from apm_cli.integration.targets import KNOWN_TARGETS + from apm_cli.utils.atomic_io import atomic_write_text + + if ( + not self._state_present + or not self._records + or self.legacy_state_path.exists() + or self.legacy_state_path.is_symlink() + ): + return + # A no-write install cannot use changed claims to promote old evidence. + # Preserve recorded hashes even for edited files: the old writer must + # compare those prior hashes, not adopt the user's current bytes. + for key, record in self._records.items(): + profile = KNOWN_TARGETS.get(record.locator.target) + target = self._recorded_targets.get(record.locator.target) + if target is None: + target = ( + profile.for_scope(user_scope=self.scope.value == "user") if profile else None + ) + if ( + target is None + or self.baseline(Path(self._destinations[key]), [target]) != record.content_hash + ): + return + data = self._read_state(self.state_path, version=2) + if ( + DeploymentLedgerCodec.from_rows(data["records"]).records != self._records + or data["destinations"] != self._destinations + or data["bindings"] != self._bindings + ): + raise ValueError("Materialization evidence changed before compatibility publication") + projection = self._legacy_projection(data) + snapshot = self._snapshot(projection) + if snapshot != self._legacy_snapshot: + # Repair pre-projection v2 receipts once; newly written receipts + # already contain this checkpoint, so repeat installs do not churn. + self._legacy_snapshot = snapshot + self._persist(self._records, self._destinations) + self._validate_paths(legacy=True) + self.legacy_state_path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + self._validate_paths(legacy=True) + try: + atomic_write_text( + self.legacy_state_path, + json.dumps(projection, sort_keys=True, indent=2) + "\n", + new_file_mode=0o600, + create_only=True, + ) + except FileExistsError: + # Another writer won the create race. Its evidence is not ours to + # replace; the normal loader validates it on the next install. + return def _persist(self, records: dict[str, DeploymentRecord], destinations: dict[str, str]) -> None: from apm_cli.utils.atomic_io import atomic_write_text @@ -470,6 +549,10 @@ def _persist(self, records: dict[str, DeploymentRecord], destinations: dict[str, "bindings": {key: self._bindings[key] for key in records}, "legacy_snapshot": self._legacy_snapshot, } + if not self.legacy_state_path.exists() and not self.legacy_state_path.is_symlink(): + # Checkpoint the expected compatibility projection before it can + # appear, so its first observation is not mistaken for older writes. + data["legacy_snapshot"] = self._snapshot(self._legacy_projection(data)) atomic_write_text( self.state_path, json.dumps(data, sort_keys=True, indent=2) + "\n", @@ -477,6 +560,7 @@ def _persist(self, records: dict[str, DeploymentRecord], destinations: dict[str, create_only=not self._state_present, ) self._state_present = True + self._legacy_snapshot = data["legacy_snapshot"] def retire(self, deleted: Sequence[Path]) -> None: """Forget successfully removed physical outputs without granting cleanup authority.""" diff --git a/tests/integration/test_architecture_materialization_receipts.py b/tests/integration/test_architecture_materialization_receipts.py index cb2d044e18..d91671d121 100644 --- a/tests/integration/test_architecture_materialization_receipts.py +++ b/tests/integration/test_architecture_materialization_receipts.py @@ -158,6 +158,31 @@ def test_receipt_guard_accepts_current_owner_routing() -> None: "if snapshot[key] == self._legacy_snapshot.get(key):", "if False:", ), + ( + "core/deployment_ledger.py", + 'data["stable_root_identity"] != self._root_identity()', + "False", + ), + ( + "core/deployment_ledger.py", + "committed._publish_legacy_compatibility()", + "pass", + ), + ( + "core/deployment_ledger.py", + 'data["legacy_snapshot"] = self._snapshot(self._legacy_projection(data))', + 'data["legacy_snapshot"] = {}', + ), + ( + "core/deployment_ledger.py", + "self.baseline(Path(self._destinations[key]), [target]) != record.content_hash", + "False", + ), + ( + "core/deployment_ledger.py", + "new_file_mode=0o600,\n create_only=True,", + "new_file_mode=0o600,\n create_only=False,", + ), ( "core/deployment_ledger.py", "record.active_owner not in self._valid_owners", diff --git a/tests/integration/test_receipt_version_compatibility.py b/tests/integration/test_receipt_version_compatibility.py index 9e0da4ffcb..a4a51da184 100644 --- a/tests/integration/test_receipt_version_compatibility.py +++ b/tests/integration/test_receipt_version_compatibility.py @@ -79,3 +79,64 @@ def install(runner, *extra): assert (deployed / "SKILL.md").read_text() == ("local edits\n" if edited else skill.read_text()) assert old_receipt.read_bytes() == legacy_after_downgrade assert json.loads(new_receipt.read_text())["version"] == 2 + + +@pytest.mark.parametrize("user_scope", [False, True]) +@pytest.mark.parametrize("edited", [False, True]) +def test_fresh_fixed_install_can_downgrade_without_losing_skills( + tmp_path: Path, apm_engine_command, user_scope: bool, edited: bool +): + legacy_binary = os.environ.get("APM_LEGACY_BINARY") + if not legacy_binary: + pytest.skip("Set APM_LEGACY_BINARY to the unpatched APM 0.35.0 executable") + isolated = IsolatedApmEnvironment.create(tmp_path / "environment", base_env=dict(os.environ)) + env = isolated.subprocess_env() + project = isolated.work_root / "consumer" + project.mkdir(parents=True) + root = isolated.config_root if user_scope else project + root.mkdir(parents=True, exist_ok=True) + package = isolated.work_root / "package" + source = package / ".apm/skills/demo" + source.mkdir(parents=True) + (package / "apm.yml").write_text("name: demo-pkg\nversion: 1.0.0\n") + skill_text = "---\nname: demo\ndescription: Fresh downgrade fixture\n---\nv1\n" + (source / "SKILL.md").write_text(skill_text) + (source / "asset.md").write_text("v1\n") + (root / "apm.yml").write_text( + f"name: consumer\nversion: 1.0.0\ndependencies:\n apm:\n - path: {package}\ntargets:\n- agent-skills\n" + ) + fixed_binary = os.environ.get("APM_FIXED_BINARY") + fixed = ApmLifecycleRunner((fixed_binary,) if fixed_binary else apm_engine_command) + legacy = ApmLifecycleRunner((legacy_binary,)) + args = ("install", *(("-g",) if user_scope else ()), "--target", "agent-skills", "--no-policy") + result = fixed.run((*args, "--dry-run"), cwd=project, env=env) + assert result.returncode == 0, result.stdout + result.stderr + assert not list((isolated.home / ".apm").glob("deployment-receipts*/*.json")) + result = fixed.run(args, cwd=project, env=env) + assert result.returncode == 0, result.stdout + result.stderr + receipts = list((isolated.home / ".apm").glob("deployment-receipts*/*.json")) + proof = {p: (p.read_bytes(), p.stat().st_mtime_ns) for p in receipts} + result = fixed.run(args, cwd=project, env=env) + assert result.returncode == 0, result.stdout + result.stderr + assert {p: (p.read_bytes(), p.stat().st_mtime_ns) for p in receipts} == proof + deployed = (isolated.home if user_scope else project) / ".agents/skills/demo" + if edited: + (deployed / "SKILL.md").write_text("local edits\n") + before = {p.name: p.read_bytes() for p in deployed.iterdir()} + lock_before = (root / "apm.lock.yaml").read_bytes() + (package / "apm.yml").write_text("name: demo-pkg\nversion: 1.1.0\n") + (source / "SKILL.md").write_text(skill_text.replace("v1", "v2")) + (source / "asset.md").write_text("v2\n") + result = legacy.run(args, cwd=project, env=env) + assert deployed.is_dir(), result.stdout + result.stderr + assert all((deployed / name).is_file() for name in before), result.stdout + result.stderr + if result.returncode: + assert {p.name: p.read_bytes() for p in deployed.iterdir()} == before + assert (root / "apm.lock.yaml").read_bytes() == lock_before + else: + assert (deployed / "SKILL.md").read_text() == ( + "local edits\n" if edited else (source / "SKILL.md").read_text() + ) + assert (deployed / "asset.md").read_bytes() == (source / "asset.md").read_bytes() + assert result.returncode == 0, result.stdout + result.stderr + assert any(p.parent.name == "deployment-receipts" for p in receipts) diff --git a/tests/integration/test_skill_materialization_receipt_lifecycle.py b/tests/integration/test_skill_materialization_receipt_lifecycle.py index ad28b8aafc..ff4853ee8c 100644 --- a/tests/integration/test_skill_materialization_receipt_lifecycle.py +++ b/tests/integration/test_skill_materialization_receipt_lifecycle.py @@ -53,6 +53,7 @@ def test_untrusted_receipt_does_not_replace_legacy_deployment( receipt = _receipt(scenario) if damage == "missing": receipt.unlink() + (receipt.parent.parent / "deployment-receipts" / receipt.name).unlink(missing_ok=True) elif damage == "corrupt": receipt.write_text("{", encoding="utf-8") else: @@ -88,8 +89,13 @@ def test_read_only_commands_never_promote_live_receipts( tmp_path: Path, apm_binary_path: Path, command: tuple[str, ...] ) -> None: scenario = _branch_switch(tmp_path, apm_binary_path, "local") - state = _receipt(scenario).parent - before = {path.name: path.read_bytes() for path in state.iterdir()} + state = _receipt(scenario).parent.parent + for legacy in (state / "deployment-receipts").glob("*.json"): + legacy.unlink() + before = { + path.relative_to(state): path.read_bytes() + for path in state.glob("deployment-receipts*/*.json") + } old = scenario.deployed_skill.read_bytes() result = scenario.runner.run( command, @@ -102,7 +108,10 @@ def test_read_only_commands_never_promote_live_receipts( assert isinstance(json.loads(result.stdout), dict) else: assert "dry-run" in result.stdout.lower() - assert {path.name: path.read_bytes() for path in state.iterdir()} == before + assert { + path.relative_to(state): path.read_bytes() + for path in state.glob("deployment-receipts*/*.json") + } == before assert scenario.deployed_skill.read_bytes() == old assert _git(scenario.root, scenario.environment, "ls-files", ".apm-local") == "" diff --git a/tests/unit/test_deployment_materialization_receipts.py b/tests/unit/test_deployment_materialization_receipts.py index 6d3c1bd32f..a2863f1c18 100644 --- a/tests/unit/test_deployment_materialization_receipts.py +++ b/tests/unit/test_deployment_materialization_receipts.py @@ -125,6 +125,7 @@ def test_external_user_receipt_uses_the_deployed_compatibility_claim( == content_hash ) session.bind_committed_claims(lock) + assert session.legacy_state_path.exists() assert ( DeploymentMaterializationSession(home, InstallScope.USER, lock).baseline(path, [target]) == content_hash @@ -364,6 +365,133 @@ def _move_to_legacy(session): return data +def test_compatibility_receipt_is_committed_once_and_keeps_recorded_hashes(tmp_path, monkeypatch): + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + original_hash = compute_file_hash(path) + session.record({path: original_hash}, [TARGET], "pkg") + session.flush() + assert not session.legacy_state_path.exists() + path.write_text("local edits\n") + session.bind_committed_claims(lock) + legacy = json.loads(session.legacy_state_path.read_text()) + assert legacy["version"] == 1 + assert legacy["root_identity"] == [root.stat().st_dev, root.stat().st_ino] + assert legacy["stable_root_identity"] == session._root_identity() + assert legacy["records"][0]["content_hash"] == original_hash + assert compute_file_hash(path) != original_hash + proof = { + p: (p.read_bytes(), p.stat().st_mtime_ns) + for p in (session.state_path, session.legacy_state_path) + } + _session(root, lock).bind_committed_claims(lock) + assert {p: (p.read_bytes(), p.stat().st_mtime_ns) for p in proof} == proof + + +@pytest.mark.parametrize("stale", [False, True]) +def test_compatibility_publication_never_replaces_existing_receipt(tmp_path, monkeypatch, stale): + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + session.bind_committed_claims(lock) + if stale: + legacy = json.loads(session.legacy_state_path.read_text()) + legacy["root_identity"][0] += 1 + session.legacy_state_path.write_text(json.dumps(legacy)) + proof = session.legacy_state_path.read_bytes() + path.write_text("new actual rendered write\n") + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + session.bind_committed_claims(lock) + assert session.legacy_state_path.read_bytes() == proof + + +def test_compatibility_receipt_exclusive_create_preserves_a_racing_writer(tmp_path, monkeypatch): + from apm_cli.utils import atomic_io + + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + proof = session.state_path.read_bytes() + write = atomic_io.atomic_write_text + + def race(destination, text, **kwargs): + if destination == session.legacy_state_path: + assert kwargs["create_only"] is True + destination.write_bytes(b"another writer's evidence\n") + raise FileExistsError(str(destination)) + return write(destination, text, **kwargs) + + monkeypatch.setattr(atomic_io, "atomic_write_text", race) + session.bind_committed_claims(lock) + assert session.legacy_state_path.read_bytes() == b"another writer's evidence\n" + assert session.state_path.read_bytes() == proof + + +def test_failed_checkpoint_repair_cannot_publish_compatibility_receipt(tmp_path, monkeypatch): + from apm_cli.utils import atomic_io + + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + old = json.loads(session.state_path.read_text()) + old.pop("legacy_snapshot") + session.state_path.write_text(json.dumps(old)) + proof = session.state_path.read_bytes() + write = atomic_io.atomic_write_text + + def refuse(destination, text, **kwargs): + if destination == session.state_path: + raise OSError("checkpoint failure") + return write(destination, text, **kwargs) + + monkeypatch.setattr(atomic_io, "atomic_write_text", refuse) + with pytest.raises(OSError, match="checkpoint failure"): + _session(root, lock).bind_committed_claims(lock) + assert not session.legacy_state_path.exists() + assert session.state_path.read_bytes() == proof + + +def test_pre_projection_v2_receipt_is_repaired_once_without_skill_writes(tmp_path, monkeypatch): + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + data = json.loads(session.state_path.read_text()) + data.pop("legacy_snapshot") + session.state_path.write_text(json.dumps(data)) + skill_proof = (path.read_bytes(), path.stat().st_mtime_ns) + _session(root, lock).bind_committed_claims(lock) + assert session.legacy_state_path.exists() + proof = { + p: (p.read_bytes(), p.stat().st_mtime_ns) + for p in (session.state_path, session.legacy_state_path) + } + _session(root, lock).bind_committed_claims(lock) + assert {p: (p.read_bytes(), p.stat().st_mtime_ns) for p in proof} == proof + assert (path.read_bytes(), path.stat().st_mtime_ns) == skill_proof + + +def test_compatibility_fallback_checks_stable_identity_without_v2(tmp_path, monkeypatch): + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + content_hash = compute_file_hash(path) + session.record({path: content_hash}, [TARGET], "pkg") + session.flush() + session.bind_committed_claims(lock) + session.state_path.unlink() + assert _session(root, lock).baseline(path, [TARGET]) == content_hash + legacy = json.loads(session.legacy_state_path.read_text()) + legacy["stable_root_identity"] = {"kind": "foreign-root"} + session.legacy_state_path.write_text(json.dumps(legacy)) + with pytest.raises(ValueError, match="installation binding does not match"): + _session(root, lock) + + def test_legacy_changes_cannot_adopt_an_edit_equal_to_an_older_version(tmp_path, monkeypatch): root, path, lock = _fixture(tmp_path, monkeypatch) asset = path.with_name("asset.md") @@ -698,6 +826,7 @@ def test_changed_canonical_claims_cannot_rebind_without_actual_writes( new_session.bind_committed_claims(lock) assert session.state_path.read_bytes() == before assert _session(root, lock).baseline(path, [TARGET]) is None + assert not session.legacy_state_path.exists() @pytest.mark.parametrize("binding", ["identity", "state", "ignore"]) From 97d973ab5a7edb7c4324b0526928e93519d7ab6a Mon Sep 17 00:00:00 2001 From: Tyce Herrman Date: Mon, 2 Nov 2026 07:05:12 -0500 Subject: [PATCH 3/7] fix: reconcile receipt retirement and committed compatibility state --- src/apm_cli/core/deployment_ledger.py | 11 ++- .../test_generated_lifecycle_state_machine.py | 18 +++-- ...est_deployment_materialization_receipts.py | 75 +++++++++++++++++++ 3 files changed, 94 insertions(+), 10 deletions(-) diff --git a/src/apm_cli/core/deployment_ledger.py b/src/apm_cli/core/deployment_ledger.py index fd2e0246c2..0e0372402e 100644 --- a/src/apm_cli/core/deployment_ledger.py +++ b/src/apm_cli/core/deployment_ledger.py @@ -98,6 +98,7 @@ def __init__(self, project_root: Path, scope: InstallScope, lockfile: LockFile | self._migration_pending = False self._legacy_snapshot: dict[str, str] = {} self._claims: dict[str, list[DeploymentRecord]] = {} + self._has_ownership_context = lockfile is not None if lockfile is not None: for record in DeploymentLedgerCodec.from_lockfile(lockfile).records.values(): self._claims.setdefault( @@ -198,7 +199,10 @@ def _load(self) -> None: self._set_state(legacy) self._legacy_snapshot = self._snapshot(legacy) self._migration_pending = True - else: + elif self._has_ownership_context: + # Cleanup-only retirement cannot validate older-writer + # imports. Leave their checkpoint for an ownership-aware + # session instead of consuming evidence it cannot admit. self._merge_legacy_writes(legacy) except (ValueError, KeyError, TypeError, UnicodeError) as exc: path = self.state_path if self._state_present else self.legacy_state_path @@ -430,9 +434,8 @@ def bind_committed_claims(self, lockfile: LockFile | None) -> None: Keep that projection separate from the actual writer; never infer a writer or a baseline from the committed lock. """ - if not self._written and not self._migration_pending: - self._publish_legacy_compatibility() - return + # Cleanup may have retired evidence since integration captured this + # session, even when integration performed no new skill writes. committed = type(self)(self.root, self.scope, lockfile) for key, value in self._written.items(): record = committed._records.get(key) diff --git a/tests/integration/test_generated_lifecycle_state_machine.py b/tests/integration/test_generated_lifecycle_state_machine.py index 7ff1ff32ef..dbeacb4d99 100644 --- a/tests/integration/test_generated_lifecycle_state_machine.py +++ b/tests/integration/test_generated_lifecycle_state_machine.py @@ -252,15 +252,17 @@ def exact_paths(self) -> dict[str, frozenset[str]]: if not self.global_scope: home = self.scenario.isolated.home with patch("apm_cli.core.deployment_ledger.Path.home", return_value=home): - receipt = DeploymentMaterializationSession( + session = DeploymentMaterializationSession( self.project.root, InstallScope.PROJECT, None - ).state_path + ) + receipts = (session.state_path, session.legacy_state_path) return { **_INSTALL_EXACT_PATHS, "user": _USER_WRITE_PATHS | { - receipt.parent.relative_to(home).as_posix(), - receipt.relative_to(home).as_posix(), + path.relative_to(home).as_posix() + for receipt in receipts + for path in (receipt.parent, receipt) }, } return { @@ -919,6 +921,7 @@ def test_generated_lifecycle_mandatory_replay( [ *itertools.product(("identity-sibling", "wrong-root"), _VARIANTS), ("receipt-sibling", "project"), + ("legacy-receipt-sibling", "project"), ], ) def test_generated_identity_write_scope_is_exact( @@ -945,8 +948,11 @@ def test_generated_identity_write_scope_is_exact( before = model._capture() if extra_write == "identity-sibling": unexpected = identity.parent / "unrelated.txt" - elif extra_write == "receipt-sibling": - unexpected = home / ".apm/deployment-receipts-v2/unrelated.json" + elif extra_write in ("receipt-sibling", "legacy-receipt-sibling"): + namespace = ( + "deployment-receipts-v2" if extra_write == "receipt-sibling" else "deployment-receipts" + ) + unexpected = home / ".apm" / namespace / "unrelated.json" else: wrong_root = project if model.fixture.global_scope else home unexpected = wrong_root / ".apm-local/installation" diff --git a/tests/unit/test_deployment_materialization_receipts.py b/tests/unit/test_deployment_materialization_receipts.py index a2863f1c18..32b2cf6325 100644 --- a/tests/unit/test_deployment_materialization_receipts.py +++ b/tests/unit/test_deployment_materialization_receipts.py @@ -967,3 +967,78 @@ def test_identical_unowned_members_do_not_create_local_authority( ) assert result.deployed_files assert not session.state_path.exists() + + +def _two_member_receipt(tmp_path, monkeypatch): + root, path, lock = _fixture(tmp_path, monkeypatch) + asset = path.with_name("asset.md") + asset.write_text("retired asset\n") + locator = DeploymentLedgerCodec.locator_for_path( + asset, project_root=root, target=TARGET, scope=InstallScope.PROJECT + ) + ledger = DeploymentLedgerCodec.from_lockfile(lock) + ledger.records[locator.key] = DeploymentRecord( + locator, ("pkg",), "pkg", compute_file_hash(asset) + ) + DeploymentLedgerCodec.apply_to_lockfile(ledger, lock, write_legacy_views=False) + session = _session(root, lock) + session.record({p: compute_file_hash(p) for p in (path, asset)}, [TARGET], "pkg") + session.flush() + session.bind_committed_claims(lock) + return root, path, asset, lock, session + + +def test_claim_free_retirement_preserves_unconsumed_legacy_update(tmp_path, monkeypatch): + root, path, asset, lock, session = _two_member_receipt(tmp_path, monkeypatch) + checkpoint = json.loads(session.state_path.read_text())["legacy_snapshot"] + path.write_text("legitimate older-writer update\n") + updated_hash = compute_file_hash(path) + legacy = json.loads(session.legacy_state_path.read_text()) + for row in legacy["records"]: + if row["value"].endswith("SKILL.md"): + row["content_hash"] = updated_hash + session.legacy_state_path.write_text(json.dumps(legacy)) + legacy_proof = session.legacy_state_path.read_bytes() + asset.unlink() + DeploymentMaterializationSession.retire_removed(root, [asset], user_scope=False) + assert json.loads(session.state_path.read_text())["legacy_snapshot"] == checkpoint + for _ in range(2): + reloaded = _session(root, lock) + assert reloaded.baseline(path, [TARGET]) == updated_hash + reloaded.bind_committed_claims(lock) + assert session.legacy_state_path.read_bytes() == legacy_proof + + +@pytest.mark.parametrize("retire_all", [False, True]) +def test_no_write_compatibility_publication_reloads_retired_state( + tmp_path, monkeypatch, retire_all +): + root, path, asset, lock, session = _two_member_receipt(tmp_path, monkeypatch) + session.legacy_state_path.unlink() + captured = _session(root, lock) + removed = [path, asset] if retire_all else [asset] + for member in removed: + member.unlink() + DeploymentMaterializationSession.retire_removed(root, removed, user_scope=False) + ledger = DeploymentLedgerCodec.from_lockfile(lock) + ledger = DeploymentLedger( + { + key: row + for key, row in ledger.records.items() + if not any(row.locator.value.endswith(member.name) for member in removed) + } + ) + DeploymentLedgerCodec.apply_to_lockfile(ledger, lock, write_legacy_views=False) + captured.bind_committed_claims(lock) + if retire_all: + assert not session.legacy_state_path.exists() + else: + legacy = json.loads(session.legacy_state_path.read_text()) + assert len(legacy["records"]) == 1 + assert legacy["records"][0]["content_hash"] == compute_file_hash(path) + proof = session.state_path.read_bytes() + legacy_proof = session.legacy_state_path.read_bytes() if not retire_all else None + _session(root, lock).bind_committed_claims(lock) + assert session.state_path.read_bytes() == proof + if legacy_proof is not None: + assert session.legacy_state_path.read_bytes() == legacy_proof From 6d64a549026cb9cc0447c23beec3e86d976875b4 Mon Sep 17 00:00:00 2001 From: Tyce Herrman Date: Mon, 2 Nov 2026 07:40:25 -0500 Subject: [PATCH 4/7] fix: reconcile legacy removals and resolved skill paths --- ...install_base_integrator_and_contraction.py | 2 +- src/apm_cli/commands/uninstall/engine.py | 2 +- src/apm_cli/core/deployment_ledger.py | 39 +++++++++++++++++-- src/apm_cli/install/pipeline.py | 4 +- .../test_receipt_version_compatibility.py | 38 +++++++++++++++--- ...est_deployment_materialization_receipts.py | 29 ++++++++++++++ 6 files changed, 101 insertions(+), 13 deletions(-) diff --git a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py index 31d9bb10f6..c53048de27 100644 --- a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py +++ b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py @@ -1342,7 +1342,7 @@ def _materialization_evidence_findings(provider: FactsProvider, rule_id: str) -> ), "src/apm_cli/integration/skill_support.py": ("removed_paths.append(skill_subdir)",), "src/apm_cli/commands/uninstall/engine.py": ( - "DeploymentMaterializationSession(project_root, _rebuild_scope, lockfile)", + "DeploymentMaterializationSession(project_root, _rebuild_scope, lockfile, targets=_targets)", 'with _integrators["skills"].ownership_snapshot(materialization_session=materialization_session):', "materialization_session=materialization_session,", "removed_paths=removed_skill_paths", diff --git a/src/apm_cli/commands/uninstall/engine.py b/src/apm_cli/commands/uninstall/engine.py index 0176bfae31..2a83df9bb1 100644 --- a/src/apm_cli/commands/uninstall/engine.py +++ b/src/apm_cli/commands/uninstall/engine.py @@ -1388,7 +1388,7 @@ def _sync_integrations_after_uninstall( _allow_executables = getattr(apm_package, "allow_executables", None) reintegration_diagnostics = DiagnosticCollector() materialization_session = ( - DeploymentMaterializationSession(project_root, _rebuild_scope, lockfile) + DeploymentMaterializationSession(project_root, _rebuild_scope, lockfile, targets=_targets) if target_survivor_plan else None ) diff --git a/src/apm_cli/core/deployment_ledger.py b/src/apm_cli/core/deployment_ledger.py index 0e0372402e..a0abb6fd46 100644 --- a/src/apm_cli/core/deployment_ledger.py +++ b/src/apm_cli/core/deployment_ledger.py @@ -82,7 +82,14 @@ class DeploymentMaterializationSession: create a prior-local baseline. """ - def __init__(self, project_root: Path, scope: InstallScope, lockfile: LockFile | None) -> None: + def __init__( + self, + project_root: Path, + scope: InstallScope, + lockfile: LockFile | None, + *, + targets: Sequence[TargetProfile] = (), + ) -> None: self.root = project_root.resolve() self.scope = scope key = hashlib.sha256(f"{self.root}|{scope.value}".encode()).hexdigest() @@ -93,7 +100,9 @@ def __init__(self, project_root: Path, scope: InstallScope, lockfile: LockFile | self._destinations: dict[str, str] = {} self._bindings: dict[str, list[str]] = {} self._written: dict[str, str] = {} - self._recorded_targets: dict[str, TargetProfile] = {} + self._recorded_targets: dict[str, TargetProfile] = { + target.name: target for target in targets + } self._dirty = False self._migration_pending = False self._legacy_snapshot: dict[str, str] = {} @@ -273,13 +282,33 @@ def _merge_legacy_writes(self, data: dict[str, Any]) -> None: snapshot = self._snapshot(data) if snapshot == self._legacy_snapshot: return + current = self._snapshot( + { + "records": DeploymentLedgerCodec.rows(DeploymentLedger(self._records)), + "destinations": self._destinations, + "bindings": self._bindings, + } + ) + # An older writer can retire only the evidence we last observed from + # it. Independently changed v2 writes must survive a legacy deletion. + for key in self._legacy_snapshot.keys() - snapshot.keys(): + if current.get(key) == self._legacy_snapshot[key]: + self._records.pop(key, None) + destination = self._destinations.pop(key, None) + self._bindings.pop(key, None) + if destination is not None: + self._physical_keys.pop(destination, None) for key, record in DeploymentLedgerCodec.from_rows(data["records"]).records.items(): if snapshot[key] == self._legacy_snapshot.get(key): continue destination = data["destinations"][key] path = Path(destination) profile = KNOWN_TARGETS.get(record.locator.target) - target = profile.for_scope(user_scope=self.scope.value == "user") if profile else None + target = self._recorded_targets.get(record.locator.target) + if target is None: + target = ( + profile.for_scope(user_scope=self.scope.value == "user") if profile else None + ) if target is None or not path.is_file() or path.is_symlink(): continue previous = ( @@ -436,7 +465,9 @@ def bind_committed_claims(self, lockfile: LockFile | None) -> None: """ # Cleanup may have retired evidence since integration captured this # session, even when integration performed no new skill writes. - committed = type(self)(self.root, self.scope, lockfile) + committed = type(self)( + self.root, self.scope, lockfile, targets=tuple(self._recorded_targets.values()) + ) for key, value in self._written.items(): record = committed._records.get(key) if record is None or record != self._records.get(key): diff --git a/src/apm_cli/install/pipeline.py b/src/apm_cli/install/pipeline.py index 1674628d41..d18b2279e0 100644 --- a/src/apm_cli/install/pipeline.py +++ b/src/apm_cli/install/pipeline.py @@ -112,7 +112,9 @@ def _run_integration_phase(ctx: InstallContext) -> DeploymentMaterializationSess session = ( None if ctx.lockfile_only or ctx.dry_run - else DeploymentMaterializationSession(ctx.project_root, ctx.scope, ctx.existing_lockfile) + else DeploymentMaterializationSession( + ctx.project_root, ctx.scope, ctx.existing_lockfile, targets=ctx.targets + ) ) with ( contextlib.nullcontext() diff --git a/tests/integration/test_receipt_version_compatibility.py b/tests/integration/test_receipt_version_compatibility.py index a4a51da184..7bb90457d4 100644 --- a/tests/integration/test_receipt_version_compatibility.py +++ b/tests/integration/test_receipt_version_compatibility.py @@ -14,10 +14,11 @@ pytestmark = [pytest.mark.integration] +@pytest.mark.parametrize("legacy_paths", [False, True]) @pytest.mark.parametrize("user_scope", [False, True]) @pytest.mark.parametrize("edited", [False, True]) def test_real_legacy_cli_upgrade_downgrade_upgrade( - tmp_path: Path, apm_engine_command, user_scope: bool, edited: bool + tmp_path: Path, apm_engine_command, user_scope: bool, edited: bool, legacy_paths: bool ): legacy_binary = os.environ.get("APM_LEGACY_BINARY") if not legacy_binary: @@ -37,7 +38,15 @@ def test_real_legacy_cli_upgrade_downgrade_upgrade( (root / "apm.yml").write_text( f"name: consumer\nversion: 1.0.0\ndependencies:\n apm:\n - path: {package}\ntargets:\n- agent-skills\n" ) - args = ("install", *(("-g",) if user_scope else ()), "--target", "agent-skills", "--no-policy") + target = "copilot" if legacy_paths else "agent-skills" + args = ( + "install", + *(("-g",) if user_scope else ()), + "--target", + target, + "--no-policy", + *(("--legacy-skill-paths",) if legacy_paths else ()), + ) legacy = ApmLifecycleRunner((legacy_binary,)) fixed_binary = os.environ.get("APM_FIXED_BINARY") fixed = ApmLifecycleRunner((fixed_binary,) if fixed_binary else apm_engine_command) @@ -51,7 +60,11 @@ def install(runner, *extra): old_receipt = next((isolated.home / ".apm/deployment-receipts").glob("*.json")) old_bytes = old_receipt.read_bytes() new_receipt = isolated.home / ".apm/deployment-receipts-v2" / old_receipt.name - deployed = (isolated.home if user_scope else project) / ".agents/skills/demo" + deployed = (isolated.home if user_scope else project) / ( + (".copilot/skills/demo" if user_scope else ".github/skills/demo") + if legacy_paths + else ".agents/skills/demo" + ) before = {p.name: (p.read_bytes(), p.stat().st_mtime_ns) for p in deployed.iterdir()} install(fixed, "--dry-run") assert not new_receipt.exists() @@ -81,10 +94,11 @@ def install(runner, *extra): assert json.loads(new_receipt.read_text())["version"] == 2 +@pytest.mark.parametrize("legacy_paths", [False, True]) @pytest.mark.parametrize("user_scope", [False, True]) @pytest.mark.parametrize("edited", [False, True]) def test_fresh_fixed_install_can_downgrade_without_losing_skills( - tmp_path: Path, apm_engine_command, user_scope: bool, edited: bool + tmp_path: Path, apm_engine_command, user_scope: bool, edited: bool, legacy_paths: bool ): legacy_binary = os.environ.get("APM_LEGACY_BINARY") if not legacy_binary: @@ -108,7 +122,15 @@ def test_fresh_fixed_install_can_downgrade_without_losing_skills( fixed_binary = os.environ.get("APM_FIXED_BINARY") fixed = ApmLifecycleRunner((fixed_binary,) if fixed_binary else apm_engine_command) legacy = ApmLifecycleRunner((legacy_binary,)) - args = ("install", *(("-g",) if user_scope else ()), "--target", "agent-skills", "--no-policy") + target = "copilot" if legacy_paths else "agent-skills" + args = ( + "install", + *(("-g",) if user_scope else ()), + "--target", + target, + "--no-policy", + *(("--legacy-skill-paths",) if legacy_paths else ()), + ) result = fixed.run((*args, "--dry-run"), cwd=project, env=env) assert result.returncode == 0, result.stdout + result.stderr assert not list((isolated.home / ".apm").glob("deployment-receipts*/*.json")) @@ -119,7 +141,11 @@ def test_fresh_fixed_install_can_downgrade_without_losing_skills( result = fixed.run(args, cwd=project, env=env) assert result.returncode == 0, result.stdout + result.stderr assert {p: (p.read_bytes(), p.stat().st_mtime_ns) for p in receipts} == proof - deployed = (isolated.home if user_scope else project) / ".agents/skills/demo" + deployed = (isolated.home if user_scope else project) / ( + (".copilot/skills/demo" if user_scope else ".github/skills/demo") + if legacy_paths + else ".agents/skills/demo" + ) if edited: (deployed / "SKILL.md").write_text("local edits\n") before = {p.name: p.read_bytes() for p in deployed.iterdir()} diff --git a/tests/unit/test_deployment_materialization_receipts.py b/tests/unit/test_deployment_materialization_receipts.py index 32b2cf6325..6f23ed374f 100644 --- a/tests/unit/test_deployment_materialization_receipts.py +++ b/tests/unit/test_deployment_materialization_receipts.py @@ -1042,3 +1042,32 @@ def test_no_write_compatibility_publication_reloads_retired_state( assert session.state_path.read_bytes() == proof if legacy_proof is not None: assert session.legacy_state_path.read_bytes() == legacy_proof + + +@pytest.mark.parametrize("newer_v2", [False, True]) +def test_legacy_deletion_retires_only_unchanged_v2_baseline(tmp_path, monkeypatch, newer_v2): + root, path, lock = _fixture(tmp_path, monkeypatch) + old_bytes = path.read_bytes() + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + session.bind_committed_claims(lock) + expected = None + if newer_v2: + path.write_text("independently newer v2 write\n") + expected = compute_file_hash(path) + session = _session(root, lock) + session.record({path: expected}, [TARGET], "pkg") + session.flush() + # An older executable deletes the member and retires its legacy row. + path.unlink() + legacy = json.loads(session.legacy_state_path.read_text()) + legacy.update(records=[], destinations={}, bindings={}) + session.legacy_state_path.write_text(json.dumps(legacy)) + # A branch restores the same claims and the user recreates the former bytes. + path.write_bytes(old_bytes) + for _ in range(2): + reloaded = _session(root, lock) + assert reloaded.baseline(path, [TARGET]) == expected + reloaded.bind_committed_claims(lock) + assert path.read_bytes() == old_bytes From 7432e29622da4c2ed53f0121030f16bd01090ca2 Mon Sep 17 00:00:00 2001 From: Tyce Herrman Date: Mon, 2 Nov 2026 10:12:31 -0500 Subject: [PATCH 5/7] fix: revoke legacy-deleted overwrite authority across target handoffs --- ...install_base_integrator_and_contraction.py | 2 + src/apm_cli/core/deployment_ledger.py | 48 ++++++++---- ...t_architecture_materialization_receipts.py | 5 ++ .../test_receipt_version_compatibility.py | 77 +++++++++++++++++++ ...est_deployment_materialization_receipts.py | 69 ++++++++++++++++- 5 files changed, 182 insertions(+), 19 deletions(-) diff --git a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py index c53048de27..07056c98c0 100644 --- a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py +++ b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py @@ -1286,6 +1286,8 @@ def _materialization_evidence_findings(provider: FactsProvider, rule_id: str) -> "self.baseline(path, [target]) != record.content_hash", "compute_file_hash(path) != record.content_hash", "if snapshot[key] == self._legacy_snapshot.get(key):", + "for key in self._legacy_snapshot.keys() - snapshot.keys():", + "self._retire_legacy_deletions(snapshot)", 'data["stable_root_identity"] != self._root_identity()', "committed._publish_legacy_compatibility()", 'data["legacy_snapshot"] = self._snapshot(self._legacy_projection(data))', diff --git a/src/apm_cli/core/deployment_ledger.py b/src/apm_cli/core/deployment_ledger.py index a0abb6fd46..5be1b6b15c 100644 --- a/src/apm_cli/core/deployment_ledger.py +++ b/src/apm_cli/core/deployment_ledger.py @@ -274,6 +274,37 @@ def _snapshot(data: dict[str, Any]) -> dict[str, str]: for key, row in zip(sorted(records), rows, strict=True) } + def _retire_legacy_deletions(self, snapshot: dict[str, str]) -> None: + """Revoke a deleted member even if its writer or rendered hash changed.""" + from apm_cli.integration.targets import KNOWN_TARGETS + + # A different v2 hash cannot prove that its write happened after this + # deletion: an older uninstall may have deleted those newer bytes using + # the current lock. Revoke prior authority regardless of hash equality. + # Only a witnessed write after reconciliation can establish it again. + for key in self._legacy_snapshot.keys() - snapshot.keys(): + retired = {key} + target_name = key.split("|", 1)[0] + target = self._recorded_targets.get(target_name) + if target is None: + profile = KNOWN_TARGETS.get(target_name) + target = ( + profile.for_scope(user_scope=self.scope.value == "user") if profile else None + ) + if target is not None: + # Canonical target aliases can share one physical directory. + # Changing the target name must not evade physical retirement. + for current_key, destination in self._destinations.items(): + locator = self._locator(Path(destination), [target]) + if locator is not None and locator.key == key: + retired.add(current_key) + for retired_key in retired: + self._records.pop(retired_key, None) + destination = self._destinations.pop(retired_key, None) + self._bindings.pop(retired_key, None) + if destination is not None: + self._physical_keys.pop(destination, None) + def _merge_legacy_writes(self, data: dict[str, Any]) -> None: """Admit changed older-writer evidence, never an unchanged stale baseline.""" from apm_cli.integration.targets import KNOWN_TARGETS @@ -282,22 +313,7 @@ def _merge_legacy_writes(self, data: dict[str, Any]) -> None: snapshot = self._snapshot(data) if snapshot == self._legacy_snapshot: return - current = self._snapshot( - { - "records": DeploymentLedgerCodec.rows(DeploymentLedger(self._records)), - "destinations": self._destinations, - "bindings": self._bindings, - } - ) - # An older writer can retire only the evidence we last observed from - # it. Independently changed v2 writes must survive a legacy deletion. - for key in self._legacy_snapshot.keys() - snapshot.keys(): - if current.get(key) == self._legacy_snapshot[key]: - self._records.pop(key, None) - destination = self._destinations.pop(key, None) - self._bindings.pop(key, None) - if destination is not None: - self._physical_keys.pop(destination, None) + self._retire_legacy_deletions(snapshot) for key, record in DeploymentLedgerCodec.from_rows(data["records"]).records.items(): if snapshot[key] == self._legacy_snapshot.get(key): continue diff --git a/tests/integration/test_architecture_materialization_receipts.py b/tests/integration/test_architecture_materialization_receipts.py index d91671d121..3f161a458e 100644 --- a/tests/integration/test_architecture_materialization_receipts.py +++ b/tests/integration/test_architecture_materialization_receipts.py @@ -23,6 +23,11 @@ def test_receipt_guard_accepts_current_owner_routing() -> None: @pytest.mark.parametrize( ("path", "before", "after"), [ + ( + "core/deployment_ledger.py", + "for key in self._legacy_snapshot.keys() - snapshot.keys():", + "for key in ():", + ), ( "commands/uninstall/cli.py", "else:\n if integration_cleanup.materialization_session is not None:", diff --git a/tests/integration/test_receipt_version_compatibility.py b/tests/integration/test_receipt_version_compatibility.py index 7bb90457d4..8a5fa9ce22 100644 --- a/tests/integration/test_receipt_version_compatibility.py +++ b/tests/integration/test_receipt_version_compatibility.py @@ -166,3 +166,80 @@ def test_fresh_fixed_install_can_downgrade_without_losing_skills( assert (deployed / "asset.md").read_bytes() == (source / "asset.md").read_bytes() assert result.returncode == 0, result.stdout + result.stderr assert any(p.parent.name == "deployment-receipts" for p in receipts) + + +@pytest.mark.parametrize("user_scope", [False, True]) +@pytest.mark.parametrize("legacy_paths", [False, True]) +def test_legacy_uninstall_revokes_newer_v2_authority_after_claim_restore( + tmp_path: Path, apm_engine_command, monkeypatch, user_scope: bool, legacy_paths: bool +): + """Actual old uninstall must not leave authority for a recreated newer file.""" + from apm_cli.core.deployment_ledger import DeploymentMaterializationSession + from apm_cli.core.scope import InstallScope + from apm_cli.deps.lockfile import LockFile + from apm_cli.integration.targets import KNOWN_TARGETS, apply_legacy_skill_paths + + legacy_binary = os.environ.get("APM_LEGACY_BINARY") + if not legacy_binary: + pytest.skip("Set APM_LEGACY_BINARY to the unpatched APM 0.35.0 executable") + isolated = IsolatedApmEnvironment.create(tmp_path / "environment", base_env=dict(os.environ)) + env = isolated.subprocess_env() + if legacy_paths: + env["APM_LEGACY_SKILL_PATHS"] = "1" + project = isolated.work_root / "consumer" + project.mkdir() + root = isolated.config_root if user_scope else project + package = isolated.work_root / "package" + source = package / ".apm/skills/demo" + source.mkdir(parents=True) + (source / "SKILL.md").write_text("---\nname: demo\ndescription: Uninstall fixture\n---\nv1\n") + (source / "asset.md").write_text("v1\n") + (package / "apm.yml").write_text("name: demo-pkg\nversion: 1.0.0\n") + target_name = "copilot" if legacy_paths else "agent-skills" + manifest = root / "apm.yml" + manifest.write_text( + f"name: consumer\nversion: 1.0.0\ndependencies:\n apm:\n - path: {package}\ntargets:\n- {target_name}\n" + ) + scope_args = ("-g",) if user_scope else () + fixed = ApmLifecycleRunner(apm_engine_command) + legacy = ApmLifecycleRunner((legacy_binary,)) + args = ("install", *scope_args, "--target", target_name, "--no-policy") + for version in (1, 2): + (source / "asset.md").write_text(f"v{version}\n") + (package / "apm.yml").write_text(f"name: demo-pkg\nversion: 1.{version}.0\n") + result = fixed.run(args, cwd=project, env=env) + assert result.returncode == 0, result.stdout + result.stderr + lock_path = root / "apm.lock.yaml" + lock_bytes, manifest_bytes = lock_path.read_bytes(), manifest.read_bytes() + target = KNOWN_TARGETS[target_name].for_scope(user_scope=user_scope) + assert target is not None + if legacy_paths: + target = apply_legacy_skill_paths([target])[0] + deploy_root = isolated.home if user_scope else project + asset = target.skills_deploy_path(deploy_root) / "demo/asset.md" + assert asset.read_text() == "v2\n" + result = legacy.run(("uninstall", *scope_args, str(package)), cwd=project, env=env) + assert result.returncode == 0, result.stdout + result.stderr + assert not asset.exists(), "The old executable must actually delete the newer bytes" + # Restore historical branch claims and recreate the former newer bytes. + manifest.write_bytes(manifest_bytes) + lock_path.write_bytes(lock_bytes) + asset.parent.mkdir(parents=True, exist_ok=True) + asset.write_text("v2\n") + (source / "asset.md").write_text("v3\n") + legacy_receipt = next((isolated.config_root / "deployment-receipts").glob("*.json")) + legacy_proof = legacy_receipt.read_bytes() + assert json.loads(legacy_proof)["records"] == [] + scope = InstallScope.USER if user_scope else InstallScope.PROJECT + # Check the authority boundary directly: #3439 remains independent of the + # refusal/cleanup error gate supplied separately by #3440. + with monkeypatch.context() as patch: + patch.setattr(Path, "home", classmethod(lambda cls: isolated.home)) + for _ in range(2): + lock = LockFile.read(lock_path) + session = DeploymentMaterializationSession(deploy_root, scope, lock, targets=[target]) + assert session.baseline(asset, [target]) is None + session.bind_committed_claims(lock) + assert asset.read_text() == "v2\n" + assert lock_path.read_bytes() == lock_bytes + assert legacy_receipt.read_bytes() == legacy_proof diff --git a/tests/unit/test_deployment_materialization_receipts.py b/tests/unit/test_deployment_materialization_receipts.py index 6f23ed374f..fedf53029b 100644 --- a/tests/unit/test_deployment_materialization_receipts.py +++ b/tests/unit/test_deployment_materialization_receipts.py @@ -1045,7 +1045,7 @@ def test_no_write_compatibility_publication_reloads_retired_state( @pytest.mark.parametrize("newer_v2", [False, True]) -def test_legacy_deletion_retires_only_unchanged_v2_baseline(tmp_path, monkeypatch, newer_v2): +def test_legacy_deletion_revokes_even_differing_v2_baseline(tmp_path, monkeypatch, newer_v2): root, path, lock = _fixture(tmp_path, monkeypatch) old_bytes = path.read_bytes() session = _session(root, lock) @@ -1055,9 +1055,9 @@ def test_legacy_deletion_retires_only_unchanged_v2_baseline(tmp_path, monkeypatc expected = None if newer_v2: path.write_text("independently newer v2 write\n") - expected = compute_file_hash(path) + newer_hash = compute_file_hash(path) session = _session(root, lock) - session.record({path: expected}, [TARGET], "pkg") + session.record({path: newer_hash}, [TARGET], "pkg") session.flush() # An older executable deletes the member and retires its legacy row. path.unlink() @@ -1071,3 +1071,66 @@ def test_legacy_deletion_retires_only_unchanged_v2_baseline(tmp_path, monkeypatc assert reloaded.baseline(path, [TARGET]) == expected reloaded.bind_committed_claims(lock) assert path.read_bytes() == old_bytes + + +def test_witnessed_write_after_legacy_retirement_can_establish_new_authority(tmp_path, monkeypatch): + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + session.bind_committed_claims(lock) + legacy = json.loads(session.legacy_state_path.read_text()) + legacy.update(records=[], destinations={}, bindings={}) + session.legacy_state_path.write_text(json.dumps(legacy)) + path.unlink() + retired = _session(root, lock) + assert retired.baseline(path, [TARGET]) is None + retired.bind_committed_claims(lock) + # A real write after observing retirement has unambiguous ordering. It + # establishes new evidence without reviving any pre-retirement baseline. + path.write_text("fresh witnessed write\n") + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + session.bind_committed_claims(lock) + assert _session(root, lock).baseline(path, [TARGET]) == compute_file_hash(path) + + +@pytest.mark.parametrize("first,second", [("codex", "agent-skills"), ("agent-skills", "codex")]) +def test_legacy_deletion_revokes_same_physical_member_after_target_handoff( + tmp_path, monkeypatch, first, second +): + old_target, new_target = KNOWN_TARGETS[first], KNOWN_TARGETS[second] + root, path, lock = _fixture(tmp_path, monkeypatch, target=old_target) + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [old_target], "pkg") + session.flush() + session.bind_committed_claims(lock) + locator = DeploymentLedgerCodec.locator_for_path( + path, project_root=root, target=new_target, scope=InstallScope.PROJECT + ) + path.write_text("new target's witnessed write\n") + content_hash = compute_file_hash(path) + DeploymentLedgerCodec.apply_to_lockfile( + DeploymentLedger({locator.key: DeploymentRecord(locator, ("pkg",), "pkg", content_hash)}), + lock, + write_legacy_views=False, + ) + session = DeploymentMaterializationSession( + root, InstallScope.PROJECT, lock, targets=[new_target] + ) + session.record({path: content_hash}, [new_target], "pkg") + session.flush() + session.bind_committed_claims(lock) + assert _session(root, lock).baseline(path, [new_target]) == content_hash + legacy = json.loads(session.legacy_state_path.read_text()) + legacy.update(records=[], destinations={}, bindings={}) + session.legacy_state_path.write_text(json.dumps(legacy)) + path.unlink() + path.write_text("new target's witnessed write\n") + for _ in range(2): + reloaded = DeploymentMaterializationSession( + root, InstallScope.PROJECT, lock, targets=[new_target] + ) + assert reloaded.baseline(path, [new_target]) is None + reloaded.bind_committed_claims(lock) From b297c15a9f8f0f4bcdaa631a8ac4a32bacc4def4 Mon Sep 17 00:00:00 2001 From: Tyce Herrman Date: Mon, 2 Nov 2026 13:15:07 -0500 Subject: [PATCH 6/7] fix: propagate confirmed retirements to validated v1 receipts --- ...install_base_integrator_and_contraction.py | 1 + src/apm_cli/core/deployment_ledger.py | 45 +++++++- ...t_architecture_materialization_receipts.py | 5 + .../test_receipt_version_compatibility.py | 75 +++++++++++++ ...est_deployment_materialization_receipts.py | 101 +++++++++++++++++- 5 files changed, 225 insertions(+), 2 deletions(-) diff --git a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py index 07056c98c0..b598cda9fa 100644 --- a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py +++ b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py @@ -1288,6 +1288,7 @@ def _materialization_evidence_findings(provider: FactsProvider, rule_id: str) -> "if snapshot[key] == self._legacy_snapshot.get(key):", "for key in self._legacy_snapshot.keys() - snapshot.keys():", "self._retire_legacy_deletions(snapshot)", + "legacy_changed = self._retire_legacy_members(removed)", 'data["stable_root_identity"] != self._root_identity()', "committed._publish_legacy_compatibility()", 'data["legacy_snapshot"] = self._snapshot(self._legacy_projection(data))', diff --git a/src/apm_cli/core/deployment_ledger.py b/src/apm_cli/core/deployment_ledger.py index 5be1b6b15c..48adc8a6ca 100644 --- a/src/apm_cli/core/deployment_ledger.py +++ b/src/apm_cli/core/deployment_ledger.py @@ -612,6 +612,48 @@ def _persist(self, records: dict[str, DeploymentRecord], destinations: dict[str, self._state_present = True self._legacy_snapshot = data["legacy_snapshot"] + def _retire_legacy_members(self, removed: set[str]) -> bool: + """Subtract confirmed deletions from valid old-reader evidence only.""" + from apm_cli.utils.atomic_io import atomic_write_text + + self._validate_paths(legacy=True) + if not self.legacy_state_path.exists(): + return False + proof = self.legacy_state_path.read_bytes() + try: + data = self._read_state(self.legacy_state_path, version=1) + except (ValueError, KeyError, TypeError, UnicodeError): + # Leave invalid evidence for explicit recovery instead of silently + # repairing or rebinding it. Stable evidence may still be retired. + return False + retired = { + key + for key, value in data["destinations"].items() + if value in removed + or any(parent.as_posix() in removed for parent in Path(value).parents) + } + if not retired: + return False + records = DeploymentLedgerCodec.from_rows(data["records"]).records + data["records"] = DeploymentLedgerCodec.rows( + DeploymentLedger({key: row for key, row in records.items() if key not in retired}) + ) + for field in ("destinations", "bindings"): + data[field] = {key: value for key, value in data[field].items() if key not in retired} + self._validate_paths(legacy=True) + if self.legacy_state_path.read_bytes() != proof: + raise ValueError("Legacy materialization evidence changed before retirement") + atomic_write_text( + self.legacy_state_path, + json.dumps(data, sort_keys=True, indent=2) + "\n", + new_file_mode=0o600, + ) + # Do not consume unrelated pending older-writer updates while recording + # our own removals. They still need ownership-aware reconciliation. + for key in retired: + self._legacy_snapshot.pop(key, None) + return True + def retire(self, deleted: Sequence[Path]) -> None: """Forget successfully removed physical outputs without granting cleanup authority.""" removed = {path.resolve().as_posix() for path in deleted} @@ -621,7 +663,8 @@ def retire(self, deleted: Sequence[Path]) -> None: if value in removed or any(parent.as_posix() in removed for parent in Path(value).parents) } - if keys: + legacy_changed = self._retire_legacy_members(removed) + if keys or legacy_changed: records = {key: record for key, record in self._records.items() if key not in keys} destinations = { key: value for key, value in self._destinations.items() if key not in keys diff --git a/tests/integration/test_architecture_materialization_receipts.py b/tests/integration/test_architecture_materialization_receipts.py index 3f161a458e..c37c1286c9 100644 --- a/tests/integration/test_architecture_materialization_receipts.py +++ b/tests/integration/test_architecture_materialization_receipts.py @@ -23,6 +23,11 @@ def test_receipt_guard_accepts_current_owner_routing() -> None: @pytest.mark.parametrize( ("path", "before", "after"), [ + ( + "core/deployment_ledger.py", + "legacy_changed = self._retire_legacy_members(removed)", + "legacy_changed = False", + ), ( "core/deployment_ledger.py", "for key in self._legacy_snapshot.keys() - snapshot.keys():", diff --git a/tests/integration/test_receipt_version_compatibility.py b/tests/integration/test_receipt_version_compatibility.py index 8a5fa9ce22..4422bc2d84 100644 --- a/tests/integration/test_receipt_version_compatibility.py +++ b/tests/integration/test_receipt_version_compatibility.py @@ -243,3 +243,78 @@ def test_legacy_uninstall_revokes_newer_v2_authority_after_claim_restore( assert asset.read_text() == "v2\n" assert lock_path.read_bytes() == lock_bytes assert legacy_receipt.read_bytes() == legacy_proof + + +@pytest.mark.parametrize("user_scope", [False, True]) +@pytest.mark.parametrize("legacy_paths", [False, True]) +def test_new_uninstall_retires_v1_before_downgrade_recreation( + tmp_path: Path, apm_engine_command, user_scope: bool, legacy_paths: bool +): + """An actual downgraded reader cannot reuse a retired compatibility baseline.""" + from apm_cli.integration.targets import KNOWN_TARGETS, apply_legacy_skill_paths + + legacy_binary = os.environ.get("APM_LEGACY_BINARY") + if not legacy_binary: + pytest.skip("Set APM_LEGACY_BINARY to the unpatched APM 0.35.0 executable") + isolated = IsolatedApmEnvironment.create(tmp_path / "environment", base_env=dict(os.environ)) + env = isolated.subprocess_env() + if legacy_paths: + env["APM_LEGACY_SKILL_PATHS"] = "1" + project = isolated.work_root / "consumer" + project.mkdir() + root = isolated.config_root if user_scope else project + package = isolated.work_root / "package" + source = package / ".apm/skills/demo" + source.mkdir(parents=True) + (source / "SKILL.md").write_text("---\nname: demo\ndescription: Uninstall fixture\n---\nv1\n") + (source / "asset.md").write_text("v1\n") + (package / "apm.yml").write_text("name: demo-pkg\nversion: 1.0.0\n") + target_name = "copilot" if legacy_paths else "agent-skills" + manifest = root / "apm.yml" + manifest.write_text( + f"name: consumer\nversion: 1.0.0\ndependencies:\n apm:\n - path: {package}\ntargets:\n- {target_name}\n" + ) + scope_args = ("-g",) if user_scope else () + fixed = ApmLifecycleRunner(apm_engine_command) + legacy = ApmLifecycleRunner((legacy_binary,)) + args = ("install", *scope_args, "--target", target_name, "--no-policy") + for version in (1, 2): + (source / "asset.md").write_text(f"v{version}\n") + (package / "apm.yml").write_text(f"name: demo-pkg\nversion: 1.{version}.0\n") + result = fixed.run(args, cwd=project, env=env) + assert result.returncode == 0, result.stdout + result.stderr + lock_path = root / "apm.lock.yaml" + lock_bytes, manifest_bytes = lock_path.read_bytes(), manifest.read_bytes() + target = KNOWN_TARGETS[target_name].for_scope(user_scope=user_scope) + assert target is not None + if legacy_paths: + target = apply_legacy_skill_paths([target])[0] + deploy_root = isolated.home if user_scope else project + asset = target.skills_deploy_path(deploy_root) / "demo/asset.md" + assert asset.read_text() == "v2\n" + result = fixed.run(("uninstall", *scope_args, str(package)), cwd=project, env=env) + assert result.returncode == 0, result.stdout + result.stderr + assert not asset.exists() + receipts = list(isolated.config_root.glob("deployment-receipts*/*.json")) + assert len(receipts) == 2 + assert all(json.loads(receipt.read_text())["records"] == [] for receipt in receipts) + manifest.write_bytes(manifest_bytes) + lock_path.write_bytes(lock_bytes) + asset.parent.mkdir(parents=True, exist_ok=True) + # Recreate the bytes formerly authorized by v1, distinct from the later + # lock's v2 hash. This isolates stale overwrite authority from the old + # executable's separate refusal/cleanup bug repaired by #3440. + asset.write_text("v1\n") + (source / "asset.md").write_text("v3\n") + (package / "apm.yml").write_text("name: demo-pkg\nversion: 1.3.0\n") + result = legacy.run(args, cwd=project, env=env) + assert asset.is_file(), result.stdout + result.stderr + assert asset.read_text() == "v1\n", result.stdout + result.stderr + assert "no trustworthy local deployment baseline" in " ".join( + (result.stdout + result.stderr).split() + ) + legacy_receipt = next((isolated.config_root / "deployment-receipts").glob("*.json")) + assert all( + not row["value"].endswith("asset.md") + for row in json.loads(legacy_receipt.read_text())["records"] + ) diff --git a/tests/unit/test_deployment_materialization_receipts.py b/tests/unit/test_deployment_materialization_receipts.py index fedf53029b..6cd35f2eb9 100644 --- a/tests/unit/test_deployment_materialization_receipts.py +++ b/tests/unit/test_deployment_materialization_receipts.py @@ -1001,7 +1001,14 @@ def test_claim_free_retirement_preserves_unconsumed_legacy_update(tmp_path, monk legacy_proof = session.legacy_state_path.read_bytes() asset.unlink() DeploymentMaterializationSession.retire_removed(root, [asset], user_scope=False) - assert json.loads(session.state_path.read_text())["legacy_snapshot"] == checkpoint + assert json.loads(session.state_path.read_text())["legacy_snapshot"] == { + key: value for key, value in checkpoint.items() if not key.endswith("asset.md") + } + retired_legacy = json.loads(session.legacy_state_path.read_text()) + assert retired_legacy["records"] == [ + row for row in json.loads(legacy_proof)["records"] if row["value"].endswith("SKILL.md") + ] + legacy_proof = session.legacy_state_path.read_bytes() for _ in range(2): reloaded = _session(root, lock) assert reloaded.baseline(path, [TARGET]) == updated_hash @@ -1134,3 +1141,95 @@ def test_legacy_deletion_revokes_same_physical_member_after_target_handoff( ) assert reloaded.baseline(path, [new_target]) is None reloaded.bind_committed_claims(lock) + + +def test_confirmed_retirement_removes_only_matching_legacy_evidence(tmp_path, monkeypatch): + root, path, asset, _lock, session = _two_member_receipt(tmp_path, monkeypatch) + before = json.loads(session.legacy_state_path.read_text()) + survivor = next(row for row in before["records"] if row["value"].endswith("SKILL.md")) + asset.unlink() + DeploymentMaterializationSession.retire_removed(root, [asset], user_scope=False) + after = json.loads(session.legacy_state_path.read_text()) + assert after["records"] == [survivor] + assert len(after["destinations"]) == len(after["bindings"]) == 1 + for field in ("root", "root_identity", "stable_root_identity", "scope", "installation"): + assert after[field] == before[field] + proof = session.legacy_state_path.read_bytes(), session.legacy_state_path.stat().st_mtime_ns + DeploymentMaterializationSession.retire_removed(root, [asset], user_scope=False) + assert ( + session.legacy_state_path.read_bytes(), + session.legacy_state_path.stat().st_mtime_ns, + ) == proof + assert path.is_file() + + +@pytest.mark.parametrize("damage", ["corrupt", "identity", "device"]) +def test_retirement_preserves_invalid_legacy_receipt(tmp_path, monkeypatch, damage): + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + session.bind_committed_claims(lock) + data = json.loads(session.legacy_state_path.read_text()) + if damage == "identity": + data["installation"] = "foreign" + elif damage == "device": + data["root_identity"][0] += 1 + session.legacy_state_path.write_text("{" if damage == "corrupt" else json.dumps(data)) + proof = session.legacy_state_path.read_bytes() + path.unlink() + DeploymentMaterializationSession.retire_removed(root, [path], user_scope=False) + assert session.legacy_state_path.read_bytes() == proof + assert json.loads(session.state_path.read_text())["records"] == [] + + +@pytest.mark.parametrize("failed_version", [1, 2]) +def test_retirement_write_failure_keeps_old_reader_safe_or_reports_failure( + tmp_path, monkeypatch, failed_version +): + from apm_cli.utils import atomic_io + + root, path, lock = _fixture(tmp_path, monkeypatch) + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + session.bind_committed_claims(lock) + old_v1, old_v2 = session.legacy_state_path.read_bytes(), session.state_path.read_bytes() + original = atomic_io.atomic_write_text + failed_path = session.legacy_state_path if failed_version == 1 else session.state_path + + def fail(destination, text, **kwargs): + if destination == failed_path: + raise OSError("retirement persistence failed") + return original(destination, text, **kwargs) + + monkeypatch.setattr(atomic_io, "atomic_write_text", fail) + path.unlink() + with pytest.raises(OSError, match="retirement persistence failed"): + DeploymentMaterializationSession.retire_removed(root, [path], user_scope=False) + assert session.state_path.read_bytes() == old_v2 + if failed_version == 1: + assert session.legacy_state_path.read_bytes() == old_v1 + else: + assert json.loads(session.legacy_state_path.read_text())["records"] == [] + # A failed v2 flush cannot restore authority already removed from v1. + path.write_text("rendered old\n") + assert _session(root, lock).baseline(path, [TARGET]) is None + + +def test_retirement_removes_legacy_only_members_without_creating_claims(tmp_path, monkeypatch): + root, path, asset, lock, session = _two_member_receipt(tmp_path, monkeypatch) + stable = json.loads(session.state_path.read_text()) + for field in ("destinations", "bindings", "legacy_snapshot"): + stable[field] = { + key: value for key, value in stable[field].items() if not key.endswith("asset.md") + } + stable["records"] = [row for row in stable["records"] if not row["value"].endswith("asset.md")] + session.state_path.write_text(json.dumps(stable)) + asset.unlink() + DeploymentMaterializationSession.retire_removed(root, [asset], user_scope=False) + legacy = json.loads(session.legacy_state_path.read_text()) + assert len(legacy["records"]) == 1 + assert legacy["records"][0]["value"].endswith("SKILL.md") + assert json.loads(session.state_path.read_text())["records"] == stable["records"] + assert _session(root, lock).baseline(path, [TARGET]) == compute_file_hash(path) From 4a560dff1df1fabdc9bae3bc8918796db2372d2e Mon Sep 17 00:00:00 2001 From: Tyce Herrman Date: Mon, 2 Nov 2026 14:15:17 -0500 Subject: [PATCH 7/7] fix: keep legacy coverage aligned with witnessed member writes --- ...install_base_integrator_and_contraction.py | 2 + src/apm_cli/core/deployment_ledger.py | 71 ++++++++++++ ...t_architecture_materialization_receipts.py | 10 ++ .../test_receipt_version_compatibility.py | 103 +++++++++++++++++- ...est_deployment_materialization_receipts.py | 83 +++++++++++++- 5 files changed, 263 insertions(+), 6 deletions(-) diff --git a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py index b598cda9fa..35fc17710c 100644 --- a/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py +++ b/scripts/architecture_linter/checks/install_base_integrator_and_contraction.py @@ -1289,6 +1289,8 @@ def _materialization_evidence_findings(provider: FactsProvider, rule_id: str) -> "for key in self._legacy_snapshot.keys() - snapshot.keys():", "self._retire_legacy_deletions(snapshot)", "legacy_changed = self._retire_legacy_members(removed)", + "committed._refresh_legacy_writes(witnessed)", + "for current_key in self._records.keys() - snapshot.keys():", 'data["stable_root_identity"] != self._root_identity()', "committed._publish_legacy_compatibility()", 'data["legacy_snapshot"] = self._snapshot(self._legacy_projection(data))', diff --git a/src/apm_cli/core/deployment_ledger.py b/src/apm_cli/core/deployment_ledger.py index 48adc8a6ca..2bbee5f418 100644 --- a/src/apm_cli/core/deployment_ledger.py +++ b/src/apm_cli/core/deployment_ledger.py @@ -282,6 +282,17 @@ def _retire_legacy_deletions(self, snapshot: dict[str, str]) -> None: # deletion: an older uninstall may have deleted those newer bytes using # the current lock. Revoke prior authority regardless of hash equality. # Only a witnessed write after reconciliation can establish it again. + deleted_keys = self._legacy_snapshot.keys() - snapshot.keys() + if deleted_keys: + # Earlier v2 writers could add members absent from v1. A legacy + # uninstall can delete those through the lock without naming them + # in its receipt. Missing legacy coverage cannot preserve authority. + for current_key in self._records.keys() - snapshot.keys(): + self._records.pop(current_key, None) + destination = self._destinations.pop(current_key, None) + self._bindings.pop(current_key, None) + if destination is not None: + self._physical_keys.pop(destination, None) for key in self._legacy_snapshot.keys() - snapshot.keys(): retired = {key} target_name = key.split("|", 1)[0] @@ -484,18 +495,78 @@ def bind_committed_claims(self, lockfile: LockFile | None) -> None: committed = type(self)( self.root, self.scope, lockfile, targets=tuple(self._recorded_targets.values()) ) + witnessed = set() for key, value in self._written.items(): record = committed._records.get(key) if record is None or record != self._records.get(key): continue committed._bindings[key] = [record.active_owner, committed._claim_signature(value)] committed._dirty = True + witnessed.add(key) if committed._migration_pending: committed._dirty = True committed.flush() committed._recorded_targets.update(self._recorded_targets) + committed._refresh_legacy_writes(witnessed) committed._publish_legacy_compatibility() + def _refresh_legacy_writes(self, witnessed: set[str]) -> None: + """Keep old-reader coverage current using committed actual writes only.""" + from apm_cli.integration.targets import KNOWN_TARGETS + from apm_cli.utils.atomic_io import atomic_write_text + + if not witnessed or not self.legacy_state_path.exists(): + return + self._validate_paths(legacy=True) + proof = self.legacy_state_path.read_bytes() + try: + data = self._read_state(self.legacy_state_path, version=1) + except (ValueError, KeyError, TypeError, UnicodeError): + return + records = dict(DeploymentLedgerCodec.from_rows(data["records"]).records) + updated, retired = set(), set() + for key in witnessed: + record = self._records.get(key) + if record is None: + continue + profile = KNOWN_TARGETS.get(record.locator.target) + target = self._recorded_targets.get(record.locator.target) + if target is None: + target = ( + profile.for_scope(user_scope=self.scope.value == "user") if profile else None + ) + destination = self._destinations[key] + if target is None or self.baseline(Path(destination), [target]) != record.content_hash: + continue + for old_key, old_destination in list(data["destinations"].items()): + if old_destination == destination and old_key != key: + records.pop(old_key, None) + data["destinations"].pop(old_key) + data["bindings"].pop(old_key) + retired.add(old_key) + records[key] = record + data["destinations"][key] = destination + data["bindings"][key] = self._bindings[key] + updated.add(key) + if not updated: + return + data["records"] = DeploymentLedgerCodec.rows(DeploymentLedger(records)) + self._validate_paths(legacy=True) + if self.legacy_state_path.read_bytes() != proof: + raise ValueError("Legacy materialization evidence changed before write publication") + text = json.dumps(data, sort_keys=True, indent=2) + "\n" + if json.loads(proof) != data: + atomic_write_text(self.legacy_state_path, text, new_file_mode=0o600) + # Publish the receipt before advancing its checkpoint. A failed write + # must not make unchanged old evidence look like a new older-writer edit. + snapshot = self._snapshot(data) + checkpoint = self._legacy_snapshot.copy() + for key in retired: + self._legacy_snapshot.pop(key, None) + self._legacy_snapshot.update({key: snapshot[key] for key in updated}) + if checkpoint != self._legacy_snapshot: + self._persist(self._records, self._destinations) + def _legacy_projection(self, data: dict[str, Any]) -> dict[str, Any]: """Project recorded rendered-write evidence, never hashes adopted from disk.""" metadata = self.root.stat() diff --git a/tests/integration/test_architecture_materialization_receipts.py b/tests/integration/test_architecture_materialization_receipts.py index c37c1286c9..15fbaee9b1 100644 --- a/tests/integration/test_architecture_materialization_receipts.py +++ b/tests/integration/test_architecture_materialization_receipts.py @@ -23,6 +23,16 @@ def test_receipt_guard_accepts_current_owner_routing() -> None: @pytest.mark.parametrize( ("path", "before", "after"), [ + ( + "core/deployment_ledger.py", + "committed._refresh_legacy_writes(witnessed)", + "pass", + ), + ( + "core/deployment_ledger.py", + "for current_key in self._records.keys() - snapshot.keys():", + "for current_key in ():", + ), ( "core/deployment_ledger.py", "legacy_changed = self._retire_legacy_members(removed)", diff --git a/tests/integration/test_receipt_version_compatibility.py b/tests/integration/test_receipt_version_compatibility.py index 4422bc2d84..89f045a931 100644 --- a/tests/integration/test_receipt_version_compatibility.py +++ b/tests/integration/test_receipt_version_compatibility.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import json import os from pathlib import Path @@ -90,7 +91,13 @@ def install(runner, *extra): install(fixed) assert (deployed / "asset.md").read_text() == "v3\n" assert (deployed / "SKILL.md").read_text() == ("local edits\n" if edited else skill.read_text()) - assert old_receipt.read_bytes() == legacy_after_downgrade + refreshed = json.loads(old_receipt.read_text()) + assert any( + row["value"].endswith("asset.md") + and row["content_hash"] == "sha256:" + hashlib.sha256(b"v3\n").hexdigest() + for row in refreshed["records"] + ) + assert old_receipt.read_bytes() != legacy_after_downgrade assert json.loads(new_receipt.read_text())["version"] == 2 @@ -170,8 +177,14 @@ def test_fresh_fixed_install_can_downgrade_without_losing_skills( @pytest.mark.parametrize("user_scope", [False, True]) @pytest.mark.parametrize("legacy_paths", [False, True]) +@pytest.mark.parametrize("member_added", [False, True]) def test_legacy_uninstall_revokes_newer_v2_authority_after_claim_restore( - tmp_path: Path, apm_engine_command, monkeypatch, user_scope: bool, legacy_paths: bool + tmp_path: Path, + apm_engine_command, + monkeypatch, + user_scope: bool, + legacy_paths: bool, + member_added: bool, ): """Actual old uninstall must not leave authority for a recreated newer file.""" from apm_cli.core.deployment_ledger import DeploymentMaterializationSession @@ -193,7 +206,8 @@ def test_legacy_uninstall_revokes_newer_v2_authority_after_claim_restore( source = package / ".apm/skills/demo" source.mkdir(parents=True) (source / "SKILL.md").write_text("---\nname: demo\ndescription: Uninstall fixture\n---\nv1\n") - (source / "asset.md").write_text("v1\n") + if not member_added: + (source / "asset.md").write_text("v1\n") (package / "apm.yml").write_text("name: demo-pkg\nversion: 1.0.0\n") target_name = "copilot" if legacy_paths else "agent-skills" manifest = root / "apm.yml" @@ -205,10 +219,18 @@ def test_legacy_uninstall_revokes_newer_v2_authority_after_claim_restore( legacy = ApmLifecycleRunner((legacy_binary,)) args = ("install", *scope_args, "--target", target_name, "--no-policy") for version in (1, 2): - (source / "asset.md").write_text(f"v{version}\n") + if not member_added or version == 2: + (source / "asset.md").write_text(f"v{version}\n") (package / "apm.yml").write_text(f"name: demo-pkg\nversion: 1.{version}.0\n") result = fixed.run(args, cwd=project, env=env) assert result.returncode == 0, result.stdout + result.stderr + receipt = next((isolated.config_root / "deployment-receipts").glob("*.json")) + asset_rows = [ + row + for row in json.loads(receipt.read_text())["records"] + if row["value"].endswith("asset.md") + ] + assert bool(asset_rows) == (not member_added or version == 2) lock_path = root / "apm.lock.yaml" lock_bytes, manifest_bytes = lock_path.read_bytes(), manifest.read_bytes() target = KNOWN_TARGETS[target_name].for_scope(user_scope=user_scope) @@ -318,3 +340,76 @@ def test_new_uninstall_retires_v1_before_downgrade_recreation( not row["value"].endswith("asset.md") for row in json.loads(legacy_receipt.read_text())["records"] ) + + +def test_legacy_uninstall_revokes_relocated_user_target( + tmp_path: Path, apm_engine_command, monkeypatch +): + from apm_cli.core.deployment_ledger import DeploymentMaterializationSession + from apm_cli.core.scope import InstallScope + from apm_cli.deps.lockfile import LockFile + from apm_cli.integration.targets import KNOWN_TARGETS + + legacy_binary = os.environ.get("APM_LEGACY_BINARY") + if not legacy_binary: + pytest.skip("Set APM_LEGACY_BINARY to the unpatched APM 0.35.0 executable") + isolated = IsolatedApmEnvironment.create(tmp_path / "environment", base_env=dict(os.environ)) + env = isolated.subprocess_env() + project = isolated.work_root / "consumer" + project.mkdir() + package = isolated.work_root / "package" + source = package / ".apm/skills/demo" + source.mkdir(parents=True) + (source / "SKILL.md").write_text("---\nname: demo\ndescription: Relocation fixture\n---\nv1\n") + (source / "asset.md").write_text("v2\n") + (package / "apm.yml").write_text("name: pkg\nversion: 1.0.0\n") + manifest = isolated.config_root / "apm.yml" + manifest.write_text( + f"name: consumer\nversion: 1.0.0\ndependencies:\n apm:\n - path: {package}\ntargets:\n- claude\n" + ) + fixed, old = ApmLifecycleRunner(apm_engine_command), ApmLifecycleRunner((legacy_binary,)) + args = ("install", "-g", "--target", "claude", "--no-policy") + for name in ("original-target", "moved-target"): + env["CLAUDE_CONFIG_DIR"] = str(isolated.home / name) + result = fixed.run(args, cwd=project, env=env) + assert result.returncode == 0, result.stdout + result.stderr + lock_path = isolated.config_root / "apm.lock.yaml" + lock_bytes, manifest_bytes = lock_path.read_bytes(), manifest.read_bytes() + asset = isolated.home / "moved-target/skills/demo/asset.md" + assert asset.read_text() == "v2\n" + receipt = next((isolated.config_root / "deployment-receipts").glob("*.json")) + assert str(asset) in json.loads(receipt.read_text())["destinations"].values() + before_outputs = { + str(path): path.read_bytes() + for name in ("original-target", "moved-target") + for path in (isolated.home / name / "skills/demo").glob("*.md") + } + result = old.run(("uninstall", "-g", str(package)), cwd=project, env=env) + if result.returncode: + # Some legacy configurations reject these relocated target paths + # before deletion. That is a safe stop, not retirement evidence. + assert result.returncode == 1 + assert "could not remove tracked target files" in result.stdout + assert lock_path.read_bytes() == lock_bytes + assert manifest.read_bytes() == manifest_bytes + assert {path: Path(path).read_bytes() for path in before_outputs} == before_outputs + return + assert not asset.exists() + manifest.write_bytes(manifest_bytes) + lock_path.write_bytes(lock_bytes) + asset.parent.mkdir(parents=True, exist_ok=True) + asset.write_text("v2\n") + (source / "asset.md").write_text("v3\n") + with monkeypatch.context() as patch: + patch.setattr(Path, "home", classmethod(lambda cls: isolated.home)) + patch.setenv("CLAUDE_CONFIG_DIR", env["CLAUDE_CONFIG_DIR"]) + target = KNOWN_TARGETS["claude"].for_scope(user_scope=True) + assert target is not None + for _ in range(2): + lock = LockFile.read(lock_path) + session = DeploymentMaterializationSession( + isolated.home, InstallScope.USER, lock, targets=[target] + ) + assert session.baseline(asset, [target]) is None + session.bind_committed_claims(lock) + assert asset.read_text() == "v2\n" diff --git a/tests/unit/test_deployment_materialization_receipts.py b/tests/unit/test_deployment_materialization_receipts.py index 6cd35f2eb9..cdb1cbea40 100644 --- a/tests/unit/test_deployment_materialization_receipts.py +++ b/tests/unit/test_deployment_materialization_receipts.py @@ -389,7 +389,9 @@ def test_compatibility_receipt_is_committed_once_and_keeps_recorded_hashes(tmp_p @pytest.mark.parametrize("stale", [False, True]) -def test_compatibility_publication_never_replaces_existing_receipt(tmp_path, monkeypatch, stale): +def test_compatibility_refresh_requires_witnessed_writes_and_valid_evidence( + tmp_path, monkeypatch, stale +): root, path, lock = _fixture(tmp_path, monkeypatch) session = _session(root, lock) session.record({path: compute_file_hash(path)}, [TARGET], "pkg") @@ -405,7 +407,12 @@ def test_compatibility_publication_never_replaces_existing_receipt(tmp_path, mon session.record({path: compute_file_hash(path)}, [TARGET], "pkg") session.flush() session.bind_committed_claims(lock) - assert session.legacy_state_path.read_bytes() == proof + if stale: + assert session.legacy_state_path.read_bytes() == proof + else: + refreshed = json.loads(session.legacy_state_path.read_text()) + assert refreshed["records"][0]["content_hash"] == compute_file_hash(path) + assert _session(root, lock).baseline(path, [TARGET]) == compute_file_hash(path) def test_compatibility_receipt_exclusive_create_preserves_a_racing_writer(tmp_path, monkeypatch): @@ -1233,3 +1240,75 @@ def test_retirement_removes_legacy_only_members_without_creating_claims(tmp_path assert legacy["records"][0]["value"].endswith("SKILL.md") assert json.loads(session.state_path.read_text())["records"] == stable["records"] assert _session(root, lock).baseline(path, [TARGET]) == compute_file_hash(path) + + +def test_legacy_deletion_revokes_uncovered_members_but_keeps_covered_survivors( + tmp_path, monkeypatch +): + root, path, asset, lock, session = _two_member_receipt(tmp_path, monkeypatch) + added = path.with_name("added.md") + added.write_text("v2-only prior write\n") + locator = DeploymentLedgerCodec.locator_for_path( + added, project_root=root, target=TARGET, scope=InstallScope.PROJECT + ) + ledger = DeploymentLedgerCodec.from_lockfile(lock) + ledger.records[locator.key] = DeploymentRecord( + locator, ("pkg",), "pkg", compute_file_hash(added) + ) + DeploymentLedgerCodec.apply_to_lockfile(ledger, lock, write_legacy_views=False) + session = _session(root, lock) + session.record({added: compute_file_hash(added)}, [TARGET], "pkg") + session.flush() # Model older v2 evidence predating compatibility refresh. + legacy = json.loads(session.legacy_state_path.read_text()) + retired = {key for key in legacy["destinations"] if key.endswith("asset.md")} + legacy["records"] = [row for row in legacy["records"] if not row["value"].endswith("asset.md")] + for field in ("destinations", "bindings"): + legacy[field] = {key: value for key, value in legacy[field].items() if key not in retired} + session.legacy_state_path.write_text(json.dumps(legacy)) + asset.unlink() + for _ in range(2): + reloaded = _session(root, lock) + assert reloaded.baseline(added, [TARGET]) is None + assert reloaded.baseline(path, [TARGET]) == compute_file_hash(path) + reloaded.bind_committed_claims(lock) + + +@pytest.mark.parametrize("failed_version", [1, 2]) +def test_failed_legacy_refresh_cannot_revive_unchanged_older_evidence( + tmp_path, monkeypatch, failed_version +): + from apm_cli.utils import atomic_io + + root, path, lock = _fixture(tmp_path, monkeypatch) + old_bytes = path.read_bytes() + session = _session(root, lock) + session.record({path: compute_file_hash(path)}, [TARGET], "pkg") + session.flush() + session.bind_committed_claims(lock) + proof = session.legacy_state_path.read_bytes() + old_checkpoint = json.loads(session.state_path.read_text())["legacy_snapshot"] + path.write_text("new witnessed bytes\n") + expected = compute_file_hash(path) + session = _session(root, lock) + session.record({path: expected}, [TARGET], "pkg") + session.flush() + write = atomic_io.atomic_write_text + + def fail(destination, text, **kwargs): + if destination == session.legacy_state_path and failed_version == 1: + raise OSError("refresh failed") + if ( + destination == session.state_path + and failed_version == 2 + and json.loads(text)["legacy_snapshot"] != old_checkpoint + ): + raise OSError("refresh failed") + return write(destination, text, **kwargs) + + monkeypatch.setattr(atomic_io, "atomic_write_text", fail) + with pytest.raises(OSError, match="refresh failed"): + session.bind_committed_claims(lock) + if failed_version == 1: + assert session.legacy_state_path.read_bytes() == proof + path.write_bytes(old_bytes) + assert _session(root, lock).baseline(path, [TARGET]) == expected