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..35fc17710c 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,25 @@ 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):", + "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))', + "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)", @@ -1330,7 +1347,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 370c2f4e86..2bbee5f418 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 @@ -81,18 +82,32 @@ 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() - 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._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] = {} 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( @@ -106,12 +121,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 +142,37 @@ 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() + 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. + 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 +184,186 @@ 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 + 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 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 _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. + 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] + 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 + from apm_cli.utils.content_hash import compute_file_hash + + snapshot = self._snapshot(data) + if snapshot == self._legacy_snapshot: + return + 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 + destination = data["destinations"][key] + path = Path(destination) + 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 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"): @@ -265,6 +447,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) @@ -307,16 +490,151 @@ 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: - return - committed = type(self)(self.root, self.scope, lockfile) + # 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, 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() + 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 @@ -342,7 +660,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,7 +668,12 @@ 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, } + 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", @@ -358,6 +681,49 @@ 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_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.""" @@ -368,7 +734,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/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/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..15fbaee9b1 100644 --- a/tests/integration/test_architecture_materialization_receipts.py +++ b/tests/integration/test_architecture_materialization_receipts.py @@ -23,6 +23,26 @@ 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)", + "legacy_changed = False", + ), + ( + "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:", @@ -115,9 +135,74 @@ 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", + '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_generated_lifecycle_state_machine.py b/tests/integration/test_generated_lifecycle_state_machine.py index c83ef8e86e..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/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/integration/test_receipt_version_compatibility.py b/tests/integration/test_receipt_version_compatibility.py new file mode 100644 index 0000000000..89f045a931 --- /dev/null +++ b/tests/integration/test_receipt_version_compatibility.py @@ -0,0 +1,415 @@ +"""Run the source CLI in a disposable home; never touch live APM state.""" + +from __future__ import annotations + +import hashlib +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("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, legacy_paths: 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" + ) + 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) + + 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) / ( + (".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() + 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()) + 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 + + +@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, legacy_paths: 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,)) + 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")) + 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) / ( + (".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()} + 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) + + +@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, + member_added: 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") + 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" + 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): + 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) + 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 + + +@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"] + ) + + +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/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..ff4853ee8c 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] @@ -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") == "" @@ -140,7 +149,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..cdb1cbea40 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 @@ -124,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 @@ -253,6 +255,320 @@ 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_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_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") + 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) + 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): + 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") + 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: @@ -517,6 +833,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"]) @@ -657,3 +974,341 @@ 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"] == { + 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 + 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 + + +@pytest.mark.parametrize("newer_v2", [False, True]) +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) + 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") + newer_hash = compute_file_hash(path) + session = _session(root, lock) + session.record({path: newer_hash}, [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 + + +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) + + +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) + + +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