diff --git a/cli/src/m2_market/apply/local.py b/cli/src/m2_market/apply/local.py
new file mode 100644
index 0000000..6bebd11
--- /dev/null
+++ b/cli/src/m2_market/apply/local.py
@@ -0,0 +1,240 @@
+"""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):
+    <bundle>/solution.json   # optional here; solution_id/version/content_hash if present
+    <bundle>/payload/...     # files referenced by recipe targets
+    <bundle>/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)))
+
+        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")