"""Local apply adapter (contracts/apply-adapter.md §local, frozen v1 seam). Places a Solution bundle's ``payload/`` into the operator's home per ``recipe.yaml``, runs post-install checks, and records InstallState. Idempotent (re-apply of an already-applied changeset is a no-op) and reversible (rollback restores backups / removes newly-placed files) per constitution VI (canary-first, reversible deployment). Bundle layout (task T021): /solution.json # optional here; solution_id/version/content_hash if present /payload/... # files referenced by recipe targets /recipe.yaml # {targets: [{src, dest}], checks: [{cmd, expect_exit}], entrypoint} """ from __future__ import annotations import hashlib import json import os import shutil import subprocess from pathlib import Path import yaml from .base import ApplyResult, Changeset, ChangesetItem, InstallState _CHECK_KIND = "check" _PLACEMENT_KIND = "placement" class LocalApplyAdapter: """`ApplyAdapter` implementation for the "local" adapter (v1 default).""" name = "local" def __init__(self) -> None: # Set by plan(); apply()/verify() reuse it since the protocol's apply() # signature carries only the changeset, not the state. self._state: InstallState | None = None # -- ApplyAdapter protocol ------------------------------------------------- def plan(self, bundle: Path, state: InstallState) -> Changeset: self._state = state bundle = Path(bundle) recipe = self._load_recipe(bundle) solution = self._load_solution(bundle) home = Path.home().resolve() items = self._build_placement_items(bundle, recipe, home) items += self._build_check_items(recipe) listing_id = solution.get("solution_id") or bundle.name version = solution.get("version") or solution.get("solution_version") or "0" content_hash = solution.get("content_hash") changeset_hash = self._hash_items(items, content_hash) return Changeset( listing_id=listing_id, version=str(version), items=items, changeset_hash=changeset_hash, ) def apply(self, changeset: Changeset) -> ApplyResult: state = self._state or InstallState() existing = state.get(changeset.listing_id) if ( existing is not None and existing.get("changeset_hash") == changeset.changeset_hash and existing.get("status") == "applied" ): return ApplyResult(ok=True, applied=[], errors=[]) placed: list[ChangesetItem] = [] errors: list[str] = [] for item in changeset.items: if item.kind != _PLACEMENT_KIND: continue try: self._place(item) placed.append(item) except OSError as exc: errors.append(f"place {item.src} -> {item.dest}: {exc}") break if not errors: errors.extend(self._run_checks(changeset.items)) if errors: self.rollback(Changeset(changeset.listing_id, changeset.version, placed, changeset.changeset_hash)) state.record_install( changeset.listing_id, changeset.version, "", changeset.changeset_hash, "failed" ) return ApplyResult(ok=False, applied=[], errors=errors) state.record_install( changeset.listing_id, changeset.version, "", changeset.changeset_hash, "applied" ) return ApplyResult(ok=True, applied=placed, errors=[]) def verify(self, bundle: Path, state: InstallState) -> bool: bundle = Path(bundle) recipe = self._load_recipe(bundle) home = Path.home().resolve() items = self._build_placement_items(bundle, recipe, home) if not all(Path(item.dest).exists() for item in items): return False return not self._run_checks(items + self._build_check_items(recipe)) def rollback(self, changeset: Changeset) -> None: for item in changeset.items: if item.kind != _PLACEMENT_KIND: continue dest = Path(item.dest) if item.backup: backup = Path(item.backup) if backup.exists(): if dest.exists(): self._remove(dest) shutil.move(str(backup), str(dest)) item.backup = None elif dest.exists(): self._remove(dest) # -- helpers ---------------------------------------------------------------- @staticmethod def _load_recipe(bundle: Path) -> dict: recipe_path = bundle / "recipe.yaml" with recipe_path.open(encoding="utf-8") as f: return yaml.safe_load(f) or {} @staticmethod def _load_solution(bundle: Path) -> dict: solution_path = bundle / "solution.json" if not solution_path.exists(): return {} with solution_path.open(encoding="utf-8") as f: return json.load(f) @staticmethod def _resolve_dest(dest_raw: str, home: Path) -> Path: dest = Path(os.path.expanduser(dest_raw)) if not dest.is_absolute(): dest = home / dest dest = Path(os.path.normpath(str(dest))) # Resolve symlinked/bind-mounted ancestors so the containment check # compares real paths: on m2o desktops $HOME (/home/developer) is a # mount of /agent_home/home/developer, and Path.home().resolve() # already followed it — an unresolved dest would false-positive here. dest = dest.resolve() if dest != home and home not in dest.parents: raise ValueError( f"recipe target dest {dest_raw!r} resolves to {dest}, which is outside " f"the user home {home} — refusing for safety (contracts/apply-adapter.md)" ) return dest def _build_placement_items(self, bundle: Path, recipe: dict, home: Path) -> list[ChangesetItem]: items = [] for target in recipe.get("targets", []): src = (bundle / "payload" / target["src"]).resolve() dest = self._resolve_dest(target["dest"], home) items.append(ChangesetItem(kind=_PLACEMENT_KIND, src=str(src), dest=str(dest))) return items @staticmethod def _build_check_items(recipe: dict) -> list[ChangesetItem]: items = [] for check in recipe.get("checks", []): items.append( ChangesetItem( kind=_CHECK_KIND, src=check["cmd"], dest=str(check.get("expect_exit", 0)), ) ) return items @staticmethod def _hash_items(items: list[ChangesetItem], content_hash: str | None) -> str: digest = hashlib.sha256() pairs = sorted((item.src, item.dest) for item in items) for src, dest in pairs: digest.update(src.encode("utf-8")) digest.update(b"\x00") digest.update(dest.encode("utf-8")) digest.update(b"\n") if content_hash: digest.update(content_hash.encode("utf-8")) return f"sha256:{digest.hexdigest()}" @staticmethod def _remove(path: Path) -> None: if path.is_dir() and not path.is_symlink(): shutil.rmtree(path) else: path.unlink() @classmethod def _place(cls, item: ChangesetItem) -> None: src = Path(item.src) dest = Path(item.dest) dest.parent.mkdir(parents=True, exist_ok=True) if dest.exists() or dest.is_symlink(): backup = f"{dest}.m2bak-{_timestamp()}" shutil.move(str(dest), backup) item.backup = backup if src.is_dir(): shutil.copytree(src, dest) else: shutil.copy2(src, dest) @staticmethod def _run_checks(items: list[ChangesetItem]) -> list[str]: errors = [] for item in items: if item.kind != _CHECK_KIND: continue expect_exit = int(item.dest) result = subprocess.run( item.src, shell=True, capture_output=True, text=True, check=False ) if result.returncode != expect_exit: errors.append( f"check failed: {item.src!r} exit={result.returncode} " f"expected={expect_exit} stderr={result.stderr.strip()}" ) return errors def _timestamp() -> str: from datetime import datetime, timezone return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")