diff --git a/backend/cortex_backend/core/paths.py b/backend/cortex_backend/core/paths.py
index 5339580..231043b 100644
--- a/backend/cortex_backend/core/paths.py
+++ b/backend/cortex_backend/core/paths.py
@@ -94,6 +94,11 @@ def execution_artifacts(self) -> Path:
         """Generated artifact root; callers must still enforce per-artifact limits."""
         return self.data_dir / "execution_artifacts"
 
+    @property
+    def recipe_bundle_store(self) -> Path:
+        """Durable signed recipe generations and their verified activation state."""
+        return self.data_dir / "recipe_bundles"
+
     @property
     def webview_profile(self) -> Path:
         """Keep native webview state isolated from every installed browser profile."""
diff --git a/backend/cortex_backend/execution/__init__.py b/backend/cortex_backend/execution/__init__.py
index ef9058a..27de659 100644
--- a/backend/cortex_backend/execution/__init__.py
+++ b/backend/cortex_backend/execution/__init__.py
@@ -1,8 +1,9 @@
 """Durable execution primitives and provider-independent safety contracts.
 
-Only the deterministic fake provider and reviewed broker contracts are exposed in
-this phase. The native adapter remains transport-only; real runtime providers stay
-absent until later ADR gates are approved.
+Only deterministic contracts, the storage-only signed bundle installer, and the
+reviewed broker transports are exposed in this phase. Native transport and bundle
+storage never load a provider; real runtime providers stay absent until later ADR
+gates are approved.
 """
 
 from .broker import (
@@ -21,6 +22,16 @@
     encode_message,
 )
 from .fake import FakeExecutionPlan, FakeExecutionProvider
+from .bundle_installer import (
+    BundleInstallError,
+    BundleRecord,
+    InstalledBundle,
+    KeyringUpdate,
+    RollbackAuthorizer,
+    SignedBundleInstaller,
+    parse_keyring_update,
+    verify_keyring_update,
+)
 from .lifecycle import ExecutionLifecycle, LifecycleSnapshot, RuntimeHealth
 from .manifest import (
     ManifestEntry,
@@ -31,6 +42,7 @@
     VerifiedRecipeManifest,
     parse_signed_manifest,
     verify_bundle_files,
+    verify_manifest_signature,
     verify_signed_manifest,
 )
 from .native_broker import (
@@ -77,6 +89,8 @@
     "BrokerPeerPolicy",
     "BrokerProtocolError",
     "BrokerSessionKeys",
+    "BundleInstallError",
+    "BundleRecord",
     "DEFAULT_NATIVE_CONNECT_TIMEOUT_MS",
     "DEFAULT_NATIVE_PIPE_BUFFER_BYTES",
     "ExecutionRepository",
@@ -87,6 +101,8 @@
     "FakeExecutionPlan",
     "FakeExecutionProvider",
     "ImageTransformPlan",
+    "InstalledBundle",
+    "KeyringUpdate",
     "LeaseConflict",
     "LifecycleSnapshot",
     "ManifestEntry",
@@ -102,6 +118,8 @@
     "PrimitiveEvaluationError",
     "RecipeValidationError",
     "RuntimeHealth",
+    "RollbackAuthorizer",
+    "SignedBundleInstaller",
     "SignedRecipeManifest",
     "TrustedRecipeKeys",
     "VerifiedRecipeManifest",
@@ -112,6 +130,9 @@
     "parse_check",
     "parse_image_transform",
     "verify_bundle_files",
+    "verify_manifest_signature",
+    "parse_keyring_update",
+    "verify_keyring_update",
     "verify_signed_manifest",
     "PeerIdentity",
     "authorize_message",
diff --git a/backend/cortex_backend/execution/bundle_installer.py b/backend/cortex_backend/execution/bundle_installer.py
new file mode 100644
index 0000000..5b5dd66
--- /dev/null
+++ b/backend/cortex_backend/execution/bundle_installer.py
@@ -0,0 +1,906 @@
+"""Fail-closed installation and recovery for verified recipe bundles.
+
+The installer is deliberately a storage boundary, not a provider.  It verifies a
+signed manifest and every declared source byte, copies only those bytes into a
+private staging directory, commits the immutable generation, and finally replaces
+one small JSON pointer atomically.  It never imports, decodes, executes, or loads a
+bundle.  Keyring updates are chained to the packaged bootstrap keyring, and a
+previous verified generation is retained for an explicit recovery decision.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+from collections.abc import Callable, Mapping
+from contextlib import contextmanager
+from dataclasses import dataclass
+from hashlib import sha256
+import json
+import os
+from pathlib import Path
+import re
+import secrets
+import shutil
+import stat
+from typing import Any
+
+from cryptography.exceptions import InvalidSignature
+
+from .manifest import (
+    MAX_MANIFEST_BYTES,
+    ManifestState,
+    ManifestVerificationError,
+    SignedRecipeManifest,
+    TrustedRecipeKeys,
+    VerifiedRecipeManifest,
+    verify_bundle_files,
+    verify_manifest_signature,
+    verify_signed_manifest,
+)
+
+
+MAX_INSTALL_STATE_BYTES = 512 * 1024
+MAX_KEYRING_HISTORY = 128
+MAX_KEYRING_KEYS = 128
+MAX_COPY_CHUNK_BYTES = 1024 * 1024
+INSTALL_STATE_SCHEMA = "recipe.install.v1"
+KEYRING_UPDATE_SCHEMA = "recipe.keyring.update.v1"
+_SAFE_KEY_ID = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")
+_SAFE_GENERATION = re.compile(r"^bundle-[0-9a-f]{64}$")
+_SHA256 = re.compile(r"^[0-9a-f]{64}$")
+_BASE64 = re.compile(r"^[A-Za-z0-9_-]+={0,2}$")
+
+
+class BundleInstallError(ValueError):
+    """Stable, non-sensitive bundle installation/recovery failure."""
+
+    def __init__(self, code: str) -> None:
+        if re.fullmatch(r"[a-z][a-z0-9_]{0,63}", code) is None:
+            raise ValueError("invalid bundle installation code")
+        self.code = code
+        super().__init__("The signed recipe bundle could not be installed safely.")
+
+
+def _canonical(value: Mapping[str, Any]) -> bytes:
+    try:
+        return json.dumps(
+            value,
+            ensure_ascii=True,
+            sort_keys=True,
+            separators=(",", ":"),
+            allow_nan=False,
+        ).encode("ascii")
+    except (TypeError, ValueError, OverflowError, UnicodeEncodeError):
+        raise BundleInstallError("bundle_install_state_invalid") from None
+
+
+def _decode_b64(value: Any, *, expected_bytes: int, code: str) -> bytes:
+    if not isinstance(value, str) or _BASE64.fullmatch(value) is None:
+        raise BundleInstallError(code)
+    padded = value + "=" * (-len(value) % 4)
+    try:
+        decoded = base64.b64decode(padded, altchars=b"-_", validate=True)
+    except (ValueError, binascii.Error):
+        raise BundleInstallError(code) from None
+    if len(decoded) != expected_bytes:
+        raise BundleInstallError(code)
+    return decoded
+
+
+def _encode_b64(value: bytes) -> str:
+    return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=")
+
+
+def _is_reparse_point(path: Path) -> bool:
+    if path.is_symlink():
+        return True
+    is_junction = getattr(path, "is_junction", None)
+    return bool(is_junction is not None and is_junction())
+
+
+def _keyring_payload(sequence: int, keyring: TrustedRecipeKeys) -> dict[str, Any]:
+    return {
+        "schema_version": "recipe.keyring.state.v1",
+        "sequence": sequence,
+        "keys": {key_id: _encode_b64(keyring.keys[key_id]) for key_id in sorted(keyring.keys)},
+        "revoked": sorted(keyring.revoked),
+    }
+
+
+def _keyring_digest(sequence: int, keyring: TrustedRecipeKeys) -> str:
+    return sha256(_canonical(_keyring_payload(sequence, keyring))).hexdigest()
+
+
+@dataclass(frozen=True, slots=True)
+class KeyringUpdate:
+    """A signed replacement keyring accepted only from the current keyring."""
+
+    sequence: int
+    signer_key_id: str
+    previous_digest: str
+    keyring: TrustedRecipeKeys
+    signature: str
+
+    def signed_payload(self) -> bytes:
+        payload = {
+            "schema_version": KEYRING_UPDATE_SCHEMA,
+            "sequence": self.sequence,
+            "signer_key_id": self.signer_key_id,
+            "previous_digest": self.previous_digest,
+            "keys": {
+                key_id: _encode_b64(self.keyring.keys[key_id])
+                for key_id in sorted(self.keyring.keys)
+            },
+            "revoked": sorted(self.keyring.revoked),
+        }
+        return _canonical(payload)
+
+    def as_payload(self) -> dict[str, Any]:
+        payload = json.loads(self.signed_payload().decode("ascii"))
+        payload["signature"] = self.signature
+        return payload
+
+
+def parse_keyring_update(payload: Mapping[str, Any]) -> KeyringUpdate:
+    """Parse a strict signed keyring update without accepting unknown fields."""
+
+    if not isinstance(payload, Mapping):
+        raise BundleInstallError("keyring_update_invalid")
+    expected_fields = {
+        "schema_version",
+        "sequence",
+        "signer_key_id",
+        "previous_digest",
+        "keys",
+        "revoked",
+        "signature",
+    }
+    if set(payload) != expected_fields:
+        raise BundleInstallError("keyring_update_invalid")
+    if payload.get("schema_version") != KEYRING_UPDATE_SCHEMA:
+        raise BundleInstallError("keyring_update_invalid")
+    sequence = payload.get("sequence")
+    signer = payload.get("signer_key_id")
+    previous_digest = payload.get("previous_digest")
+    if type(sequence) is not int or sequence < 1:
+        raise BundleInstallError("keyring_update_invalid")
+    if not isinstance(signer, str) or _SAFE_KEY_ID.fullmatch(signer) is None:
+        raise BundleInstallError("keyring_update_invalid")
+    if not isinstance(previous_digest, str) or _SHA256.fullmatch(previous_digest) is None:
+        raise BundleInstallError("keyring_update_invalid")
+    raw_keys = payload.get("keys")
+    revoked = payload.get("revoked")
+    if not isinstance(raw_keys, Mapping) or not isinstance(revoked, list):
+        raise BundleInstallError("keyring_update_invalid")
+    if not 1 <= len(raw_keys) <= MAX_KEYRING_KEYS or len(revoked) > len(raw_keys):
+        raise BundleInstallError("keyring_update_invalid")
+    keys: dict[str, bytes] = {}
+    for key_id, encoded in raw_keys.items():
+        if not isinstance(key_id, str) or _SAFE_KEY_ID.fullmatch(key_id) is None:
+            raise BundleInstallError("keyring_update_invalid")
+        if key_id in keys:
+            raise BundleInstallError("keyring_update_invalid")
+        keys[key_id] = _decode_b64(encoded, expected_bytes=32, code="keyring_update_invalid")
+    if any(
+        not isinstance(key_id, str)
+        or _SAFE_KEY_ID.fullmatch(key_id) is None
+        or key_id not in keys
+        for key_id in revoked
+    ) or len(set(revoked)) != len(revoked):
+        raise BundleInstallError("keyring_update_invalid")
+    signature = payload.get("signature")
+    if not isinstance(signature, str) or _BASE64.fullmatch(signature) is None:
+        raise BundleInstallError("keyring_update_invalid")
+    try:
+        keyring = TrustedRecipeKeys(keys, revoked=frozenset(revoked))
+    except ValueError:
+        raise BundleInstallError("keyring_update_invalid") from None
+    if all(key_id in keyring.revoked for key_id in keyring.keys):
+        raise BundleInstallError("keyring_no_active_keys")
+    return KeyringUpdate(
+        sequence=sequence,
+        signer_key_id=signer,
+        previous_digest=previous_digest,
+        keyring=keyring,
+        signature=signature,
+    )
+
+
+def verify_keyring_update(
+    payload: Mapping[str, Any],
+    trusted_keys: TrustedRecipeKeys,
+    *,
+    current_sequence: int,
+    current_digest: str,
+) -> KeyringUpdate:
+    """Verify a chained keyring update against the currently trusted keys."""
+
+    update = parse_keyring_update(payload)
+    if update.sequence <= current_sequence:
+        raise BundleInstallError("keyring_replay")
+    if update.previous_digest != current_digest:
+        raise BundleInstallError("keyring_state_mismatch")
+    if update.signer_key_id not in trusted_keys.keys or update.signer_key_id in trusted_keys.revoked:
+        raise BundleInstallError("keyring_signer_untrusted")
+    try:
+        public_key = trusted_keys.public_key(update.signer_key_id)
+        signature = _decode_b64(
+            update.signature,
+            expected_bytes=64,
+            code="keyring_signature_invalid",
+        )
+        public_key.verify(signature, update.signed_payload())
+    except BundleInstallError:
+        raise
+    except (InvalidSignature, ValueError, TypeError):
+        raise BundleInstallError("keyring_signature_invalid") from None
+    return update
+
+
+@dataclass(frozen=True, slots=True)
+class BundleRecord:
+    generation: str
+    state: ManifestState
+    key_id: str
+
+
+@dataclass(frozen=True, slots=True)
+class InstalledBundle:
+    """A verified immutable generation; no provider is loaded by this object."""
+
+    bundle_root: Path
+    state: ManifestState
+    key_id: str
+    rollback: bool
+
+
+@dataclass(frozen=True, slots=True)
+class _KeyringSnapshot:
+    sequence: int
+    digest: str
+    keyring: TrustedRecipeKeys
+    history: tuple[Mapping[str, Any], ...]
+
+
+@dataclass(frozen=True, slots=True)
+class _InstallSnapshot:
+    keyring: _KeyringSnapshot
+    current: BundleRecord | None
+    previous: BundleRecord | None
+
+
+RollbackAuthorizer = Callable[[str, str], bool]
+
+
+def _record_payload(record: BundleRecord | None) -> dict[str, Any] | None:
+    if record is None:
+        return None
+    return {
+        "generation": record.generation,
+        "sequence": record.state.sequence,
+        "bundle_version": record.state.bundle_version,
+        "digest": record.state.digest,
+        "key_id": record.key_id,
+    }
+
+
+def _parse_record(payload: Any) -> BundleRecord | None:
+    if payload is None:
+        return None
+    if not isinstance(payload, Mapping) or set(payload) != {
+        "generation",
+        "sequence",
+        "bundle_version",
+        "digest",
+        "key_id",
+    }:
+        raise BundleInstallError("bundle_install_state_invalid")
+    generation = payload.get("generation")
+    key_id = payload.get("key_id")
+    if not isinstance(generation, str) or _SAFE_GENERATION.fullmatch(generation) is None:
+        raise BundleInstallError("bundle_install_state_invalid")
+    if not isinstance(key_id, str) or _SAFE_KEY_ID.fullmatch(key_id) is None:
+        raise BundleInstallError("bundle_install_state_invalid")
+    try:
+        state = ManifestState(
+            sequence=payload["sequence"],
+            bundle_version=payload["bundle_version"],
+            digest=payload["digest"],
+        )
+    except (KeyError, TypeError, ValueError):
+        raise BundleInstallError("bundle_install_state_invalid") from None
+    if generation != f"bundle-{state.digest}":
+        raise BundleInstallError("bundle_install_state_invalid")
+    return BundleRecord(generation=generation, state=state, key_id=key_id)
+
+
+def _copy_stat_identity(path: Path) -> tuple[int, int, int, int]:
+    try:
+        info = path.lstat()
+    except OSError:
+        raise BundleInstallError("bundle_entry_unavailable") from None
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        raise BundleInstallError("bundle_path_invalid")
+    if getattr(info, "st_nlink", 1) != 1:
+        raise BundleInstallError("bundle_hardlink_rejected")
+    return (
+        int(getattr(info, "st_dev", 0)),
+        int(getattr(info, "st_ino", 0)),
+        int(info.st_size),
+        int(getattr(info, "st_mtime_ns", 0)),
+    )
+
+
+def _fsync_directory(path: Path) -> None:
+    """Best-effort directory durability; file bytes are always fsynced first."""
+
+    try:
+        descriptor = os.open(str(path), os.O_RDONLY)
+    except OSError:
+        return
+    try:
+        os.fsync(descriptor)
+    except OSError:
+        pass
+    finally:
+        os.close(descriptor)
+
+
+def _write_exclusive(path: Path, payload: bytes) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        with path.open("xb") as stream:
+            stream.write(payload)
+            stream.flush()
+            os.fsync(stream.fileno())
+    except FileExistsError:
+        raise BundleInstallError("bundle_install_conflict") from None
+    except OSError:
+        raise BundleInstallError("bundle_install_io_failed") from None
+
+
+def _atomic_write(path: Path, payload: bytes) -> None:
+    temporary = path.with_name(f".{path.name}.tmp-{secrets.token_hex(12)}")
+    try:
+        _write_exclusive(temporary, payload)
+        os.replace(temporary, path)
+        _fsync_directory(path.parent)
+    except BundleInstallError:
+        try:
+            temporary.unlink(missing_ok=True)
+        except OSError:
+            raise BundleInstallError("bundle_install_cleanup_pending") from None
+        raise
+    except OSError:
+        try:
+            temporary.unlink(missing_ok=True)
+        except OSError:
+            raise BundleInstallError("bundle_install_cleanup_pending") from None
+        raise BundleInstallError("bundle_install_io_failed") from None
+
+
+class SignedBundleInstaller:
+    """Install signed recipe bytes with an atomic pointer and explicit recovery."""
+
+    def __init__(
+        self,
+        store_root: str | os.PathLike[str],
+        bootstrap_keys: TrustedRecipeKeys,
+        *,
+        max_keyring_history: int = MAX_KEYRING_HISTORY,
+    ) -> None:
+        if not isinstance(bootstrap_keys, TrustedRecipeKeys):
+            raise TypeError("bootstrap_keys must be TrustedRecipeKeys")
+        if not 1 <= max_keyring_history <= MAX_KEYRING_HISTORY:
+            raise ValueError("max_keyring_history is invalid")
+        root = Path(store_root).expanduser()
+        if root.exists() and _is_reparse_point(root):
+            raise BundleInstallError("bundle_install_root_invalid")
+        self.store_root = root.resolve(strict=False)
+        self.bundle_root = self.store_root / "bundles"
+        self.state_path = self.store_root / "state.json"
+        self._bootstrap_keys = bootstrap_keys
+        self._bootstrap_digest = _keyring_digest(0, bootstrap_keys)
+        self._max_keyring_history = max_keyring_history
+        try:
+            self.bundle_root.mkdir(parents=True, exist_ok=True)
+        except OSError:
+            raise BundleInstallError("bundle_install_root_unavailable") from None
+
+    def _initial_keyring(self) -> _KeyringSnapshot:
+        return _KeyringSnapshot(
+            sequence=0,
+            digest=self._bootstrap_digest,
+            keyring=self._bootstrap_keys,
+            history=(),
+        )
+
+    def _load_snapshot(self, *, validate_current: bool = True) -> _InstallSnapshot:
+        if _is_reparse_point(self.state_path):
+            raise BundleInstallError("bundle_install_state_invalid")
+        if not self.state_path.exists():
+            snapshot = _InstallSnapshot(self._initial_keyring(), None, None)
+            return snapshot
+        try:
+            raw_bytes = self.state_path.read_bytes()
+        except OSError:
+            raise BundleInstallError("bundle_install_state_unavailable") from None
+        if len(raw_bytes) > MAX_INSTALL_STATE_BYTES:
+            raise BundleInstallError("bundle_install_state_invalid")
+        try:
+            raw = json.loads(raw_bytes.decode("ascii"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            raise BundleInstallError("bundle_install_state_invalid") from None
+        if not isinstance(raw, Mapping) or set(raw) != {
+            "schema_version",
+            "keyring_root_digest",
+            "keyring_history",
+            "current",
+            "previous",
+        } or raw.get("schema_version") != INSTALL_STATE_SCHEMA:
+            raise BundleInstallError("bundle_install_state_invalid")
+        if raw.get("keyring_root_digest") != self._bootstrap_digest:
+            raise BundleInstallError("keyring_root_mismatch")
+        history = raw.get("keyring_history")
+        if not isinstance(history, list) or len(history) > self._max_keyring_history:
+            raise BundleInstallError("keyring_history_invalid")
+        keyring = self._initial_keyring()
+        parsed_history: list[Mapping[str, Any]] = []
+        for item in history:
+            if not isinstance(item, Mapping):
+                raise BundleInstallError("keyring_history_invalid")
+            update = verify_keyring_update(
+                item,
+                keyring.keyring,
+                current_sequence=keyring.sequence,
+                current_digest=keyring.digest,
+            )
+            keyring = _KeyringSnapshot(
+                sequence=update.sequence,
+                digest=_keyring_digest(update.sequence, update.keyring),
+                keyring=update.keyring,
+                history=tuple(parsed_history + [dict(update.as_payload())]),
+            )
+            parsed_history.append(dict(update.as_payload()))
+        current = _parse_record(raw.get("current"))
+        previous = _parse_record(raw.get("previous"))
+        snapshot = _InstallSnapshot(keyring, current, previous)
+        if validate_current and current is not None:
+            self._validate_record(current, keyring.keyring)
+        return snapshot
+
+    def _snapshot_payload(self, snapshot: _InstallSnapshot) -> dict[str, Any]:
+        return {
+            "schema_version": INSTALL_STATE_SCHEMA,
+            "keyring_root_digest": self._bootstrap_digest,
+            "keyring_history": [dict(item) for item in snapshot.keyring.history],
+            "current": _record_payload(snapshot.current),
+            "previous": _record_payload(snapshot.previous),
+        }
+
+    def _persist_snapshot(self, snapshot: _InstallSnapshot) -> None:
+        encoded = _canonical(self._snapshot_payload(snapshot))
+        if len(encoded) > MAX_INSTALL_STATE_BYTES:
+            raise BundleInstallError("bundle_install_state_too_large")
+        _atomic_write(self.state_path, encoded)
+
+    @contextmanager
+    def _store_lock(self):
+        """Serialize state transitions across threads and installer processes."""
+
+        lock_path = self.store_root / ".install.lock"
+        try:
+            self.store_root.mkdir(parents=True, exist_ok=True)
+            descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR)
+        except OSError:
+            raise BundleInstallError("bundle_install_lock_unavailable") from None
+        try:
+            if os.fstat(descriptor).st_size == 0:
+                os.write(descriptor, b"0")
+            os.lseek(descriptor, 0, os.SEEK_SET)
+            if os.name == "nt":
+                import msvcrt
+
+                msvcrt.locking(descriptor, msvcrt.LK_LOCK, 1)
+            else:
+                import fcntl
+
+                fcntl.flock(descriptor, fcntl.LOCK_EX)
+            yield
+        except BundleInstallError:
+            raise
+        except OSError:
+            raise BundleInstallError("bundle_install_lock_unavailable") from None
+        finally:
+            try:
+                if os.name == "nt":
+                    import msvcrt
+
+                    os.lseek(descriptor, 0, os.SEEK_SET)
+                    msvcrt.locking(descriptor, msvcrt.LK_UNLCK, 1)
+                else:
+                    import fcntl
+
+                    fcntl.flock(descriptor, fcntl.LOCK_UN)
+            except OSError:
+                pass
+            os.close(descriptor)
+
+    @staticmethod
+    def _bundle_path(root: Path, record: BundleRecord) -> Path:
+        return root / record.generation
+
+    @staticmethod
+    def _reject_reserved_entries(manifest: SignedRecipeManifest) -> None:
+        if any(entry.bundle_path == "manifest.json" for entry in manifest.entries):
+            raise BundleInstallError("bundle_install_reserved_path")
+
+    def _validate_exact_tree(self, root: Path, manifest: SignedRecipeManifest) -> None:
+        expected_files = {"manifest.json", *(entry.bundle_path for entry in manifest.entries)}
+        expected_dirs = {Path(".")}
+        for relative in expected_files:
+            expected_dirs.update(Path(relative).parents)
+        try:
+            if _is_reparse_point(root) or not root.is_dir():
+                raise BundleInstallError("bundle_install_recovery_failed")
+            for path in root.rglob("*"):
+                relative = path.relative_to(root).as_posix()
+                if _is_reparse_point(path):
+                    raise BundleInstallError("bundle_install_recovery_failed")
+                if path.is_dir():
+                    if Path(relative) not in expected_dirs:
+                        raise BundleInstallError("bundle_extra_path")
+                    continue
+                if not path.is_file() or relative not in expected_files:
+                    raise BundleInstallError("bundle_extra_path")
+                if getattr(path.stat(), "st_nlink", 1) != 1:
+                    raise BundleInstallError("bundle_hardlink_rejected")
+        except BundleInstallError:
+            raise
+        except OSError:
+            raise BundleInstallError("bundle_install_recovery_failed") from None
+
+    def _validate_record_at(
+        self,
+        root: Path,
+        record: BundleRecord,
+        keyring: TrustedRecipeKeys,
+    ) -> SignedRecipeManifest:
+        manifest_path = root / "manifest.json"
+        try:
+            if _is_reparse_point(root) or not root.is_dir() or _is_reparse_point(manifest_path):
+                raise BundleInstallError("bundle_install_recovery_failed")
+            raw = manifest_path.read_bytes()
+        except BundleInstallError:
+            raise
+        except OSError:
+            raise BundleInstallError("bundle_install_recovery_failed") from None
+        if len(raw) > MAX_MANIFEST_BYTES:
+            raise BundleInstallError("bundle_install_recovery_failed")
+        try:
+            payload = json.loads(raw.decode("ascii"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            raise BundleInstallError("bundle_install_recovery_failed") from None
+        try:
+            verified = verify_manifest_signature(payload, keyring)
+            self._reject_reserved_entries(verified.manifest)
+            if (
+                verified.digest != record.state.digest
+                or verified.manifest.key_id != record.key_id
+                or verified.state != record.state
+            ):
+                raise BundleInstallError("bundle_install_recovery_failed")
+            self._validate_exact_tree(root, verified.manifest)
+            verify_bundle_files(verified.manifest, root)
+        except (ManifestVerificationError, BundleInstallError):
+            raise BundleInstallError("bundle_install_recovery_failed") from None
+        return verified.manifest
+
+    def _validate_record(self, record: BundleRecord, keyring: TrustedRecipeKeys) -> SignedRecipeManifest:
+        return self._validate_record_at(
+            self._bundle_path(self.bundle_root, record),
+            record,
+            keyring,
+        )
+
+    def status(self) -> InstalledBundle | None:
+        """Validate and return the active generation, or ``None`` before first install."""
+
+        with self._store_lock():
+            snapshot = self._load_snapshot()
+            if snapshot.current is None:
+                return None
+            return InstalledBundle(
+                bundle_root=self._bundle_path(self.bundle_root, snapshot.current),
+                state=snapshot.current.state,
+                key_id=snapshot.current.key_id,
+                rollback=False,
+            )
+
+    def _source_path(self, source_root: Path, relative: str) -> Path:
+        candidate = source_root / relative
+        try:
+            resolved = candidate.resolve(strict=True)
+            root = source_root.resolve(strict=True)
+        except (OSError, RuntimeError):
+            raise BundleInstallError("bundle_entry_unavailable") from None
+        cursor = root
+        for component in Path(relative).parts:
+            cursor = cursor / component
+            if _is_reparse_point(cursor):
+                raise BundleInstallError("bundle_path_invalid")
+        if not resolved.is_relative_to(root) or _is_reparse_point(resolved):
+            raise BundleInstallError("bundle_path_invalid")
+        return candidate
+
+    def _copy_entry(self, source: Path, destination: Path, expected_size: int, expected_digest: str) -> None:
+        before = _copy_stat_identity(source)
+        digest = sha256()
+        total = 0
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            with source.open("rb") as source_stream, destination.open("xb") as destination_stream:
+                while True:
+                    chunk = source_stream.read(MAX_COPY_CHUNK_BYTES)
+                    if not chunk:
+                        break
+                    total += len(chunk)
+                    if total > expected_size or total > 128 * 1024 * 1024:
+                        raise BundleInstallError("bundle_size_mismatch")
+                    digest.update(chunk)
+                    destination_stream.write(chunk)
+                destination_stream.flush()
+                os.fsync(destination_stream.fileno())
+        except BundleInstallError:
+            raise
+        except OSError:
+            raise BundleInstallError("bundle_entry_unavailable") from None
+        after = _copy_stat_identity(source)
+        if before != after or total != expected_size or digest.hexdigest() != expected_digest:
+            raise BundleInstallError("bundle_source_changed")
+
+    def _stage_bundle(
+        self,
+        source_root: Path,
+        verified: VerifiedRecipeManifest,
+    ) -> Path:
+        source_root = Path(source_root)
+        if _is_reparse_point(source_root):
+            raise BundleInstallError("bundle_root_unavailable")
+        try:
+            source_root = source_root.resolve(strict=True)
+        except (OSError, RuntimeError):
+            raise BundleInstallError("bundle_root_unavailable") from None
+        if not source_root.is_dir():
+            raise BundleInstallError("bundle_root_unavailable")
+        verify_bundle_files(verified.manifest, source_root)
+        self._reject_reserved_entries(verified.manifest)
+        staging = self.bundle_root / f".staging-{secrets.token_hex(16)}"
+        final = self.bundle_root / f"bundle-{verified.digest}"
+        try:
+            staging.mkdir()
+            for entry in verified.manifest.entries:
+                source = self._source_path(source_root, entry.bundle_path)
+                self._copy_entry(
+                    source,
+                    staging / entry.bundle_path,
+                    entry.size,
+                    entry.sha256,
+                )
+            _write_exclusive(
+                staging / "manifest.json",
+                verified.manifest.canonical_json().encode("ascii"),
+            )
+            candidate = BundleRecord(
+                generation=f"bundle-{verified.digest}",
+                state=verified.state,
+                key_id=verified.manifest.key_id,
+            )
+            self._validate_record_at(staging, candidate, self._load_snapshot().keyring.keyring)
+            if final.exists():
+                if _is_reparse_point(final) or not final.is_dir():
+                    raise BundleInstallError("bundle_install_conflict")
+                self._validate_record(candidate, self._load_snapshot().keyring.keyring)
+                shutil.rmtree(staging)
+                return final
+            os.replace(staging, final)
+            _fsync_directory(self.bundle_root)
+            return final
+        except BundleInstallError:
+            try:
+                if staging.exists():
+                    shutil.rmtree(staging)
+            except OSError:
+                raise BundleInstallError("bundle_install_cleanup_pending") from None
+            raise
+        except OSError:
+            try:
+                if staging.exists():
+                    shutil.rmtree(staging)
+            except OSError:
+                raise BundleInstallError("bundle_install_cleanup_pending") from None
+            raise BundleInstallError("bundle_install_io_failed") from None
+
+    @staticmethod
+    def _authorize_rollback(
+        authorizer: RollbackAuthorizer | None,
+        current_digest: str,
+        candidate_digest: str,
+    ) -> bool:
+        if authorizer is None:
+            return False
+        try:
+            return bool(authorizer(current_digest, candidate_digest))
+        except Exception:
+            raise BundleInstallError("bundle_install_rollback_authorization_failed") from None
+
+    def _install_locked(
+        self,
+        manifest_payload: Mapping[str, Any],
+        source_root: str | os.PathLike[str],
+        *,
+        rollback_authorizer: RollbackAuthorizer | None = None,
+    ) -> InstalledBundle:
+        """Verify, stage, and atomically activate one signed bundle generation."""
+
+        snapshot = self._load_snapshot()
+        current_state = snapshot.current.state if snapshot.current is not None else None
+        signature_verified = verify_manifest_signature(manifest_payload, snapshot.keyring.keyring)
+        rollback_authorized = False
+        if current_state is not None and signature_verified.manifest.rollback_of is not None:
+            rollback_authorized = self._authorize_rollback(
+                rollback_authorizer,
+                current_state.digest,
+                signature_verified.digest,
+            )
+        verified = verify_signed_manifest(
+            manifest_payload,
+            snapshot.keyring.keyring,
+            current=current_state,
+            rollback_authorized=rollback_authorized,
+        )
+        final = self._stage_bundle(Path(source_root), verified)
+        record = BundleRecord(
+            generation=f"bundle-{verified.digest}",
+            state=verified.state,
+            key_id=verified.manifest.key_id,
+        )
+        next_snapshot = _InstallSnapshot(
+            keyring=snapshot.keyring,
+            current=record,
+            previous=snapshot.current,
+        )
+        self._persist_snapshot(next_snapshot)
+        return InstalledBundle(
+            bundle_root=final,
+            state=record.state,
+            key_id=record.key_id,
+            rollback=verified.rollback,
+        )
+
+    def install(
+        self,
+        manifest_payload: Mapping[str, Any],
+        source_root: str | os.PathLike[str],
+        *,
+        rollback_authorizer: RollbackAuthorizer | None = None,
+    ) -> InstalledBundle:
+        """Verify, stage, and atomically activate one signed bundle generation."""
+
+        with self._store_lock():
+            return self._install_locked(
+                manifest_payload,
+                source_root,
+                rollback_authorizer=rollback_authorizer,
+            )
+
+    def _rotate_keyring_locked(self, payload: Mapping[str, Any]) -> None:
+        """Atomically persist a chained keyring update before new-key installs."""
+
+        snapshot = self._load_snapshot()
+        update = verify_keyring_update(
+            payload,
+            snapshot.keyring.keyring,
+            current_sequence=snapshot.keyring.sequence,
+            current_digest=snapshot.keyring.digest,
+        )
+        if snapshot.current is not None and snapshot.current.key_id in update.keyring.revoked:
+            raise BundleInstallError("keyring_active_key_revoked")
+        next_keyring = _KeyringSnapshot(
+            sequence=update.sequence,
+            digest=_keyring_digest(update.sequence, update.keyring),
+            keyring=update.keyring,
+            history=snapshot.keyring.history + (update.as_payload(),),
+        )
+        if len(next_keyring.history) > self._max_keyring_history:
+            raise BundleInstallError("keyring_history_invalid")
+        if snapshot.current is not None:
+            self._validate_record(snapshot.current, next_keyring.keyring)
+        self._persist_snapshot(
+            _InstallSnapshot(next_keyring, snapshot.current, snapshot.previous)
+        )
+
+    def rotate_keyring(self, payload: Mapping[str, Any]) -> None:
+        """Atomically persist a chained keyring update before new-key installs."""
+
+        with self._store_lock():
+            self._rotate_keyring_locked(payload)
+
+    def _recover_previous_locked(
+        self,
+        rollback_authorizer: RollbackAuthorizer | None = None,
+    ) -> InstalledBundle:
+        """Explicitly activate the retained previous generation after validation."""
+
+        snapshot = self._load_snapshot(validate_current=False)
+        if snapshot.current is None or snapshot.previous is None:
+            raise BundleInstallError("bundle_recovery_unavailable")
+        if not self._authorize_rollback(
+            rollback_authorizer,
+            snapshot.current.state.digest,
+            snapshot.previous.state.digest,
+        ):
+            raise BundleInstallError("bundle_recovery_not_authorized")
+        self._validate_record(snapshot.previous, snapshot.keyring.keyring)
+        next_snapshot = _InstallSnapshot(
+            keyring=snapshot.keyring,
+            current=snapshot.previous,
+            previous=None,
+        )
+        self._persist_snapshot(next_snapshot)
+        return InstalledBundle(
+            bundle_root=self._bundle_path(self.bundle_root, snapshot.previous),
+            state=snapshot.previous.state,
+            key_id=snapshot.previous.key_id,
+            rollback=True,
+        )
+
+    def recover_previous(self, rollback_authorizer: RollbackAuthorizer | None = None) -> InstalledBundle:
+        """Explicitly activate the retained previous generation after validation."""
+
+        with self._store_lock():
+            return self._recover_previous_locked(rollback_authorizer)
+
+    def _cleanup_locked(self) -> None:
+        """Remove only installer-owned orphan staging/generation directories."""
+
+        snapshot = self._load_snapshot(validate_current=False)
+        protected = {
+            record.generation
+            for record in (snapshot.current, snapshot.previous)
+            if record is not None
+        }
+        try:
+            for path in self.bundle_root.iterdir():
+                if path.name.startswith(".staging-") or (
+                    path.name.startswith("bundle-") and path.name not in protected
+                ):
+                    if _is_reparse_point(path):
+                        raise BundleInstallError("bundle_cleanup_unsafe")
+                    if path.is_dir():
+                        shutil.rmtree(path)
+                    else:
+                        raise BundleInstallError("bundle_cleanup_unsafe")
+        except BundleInstallError:
+            raise
+        except OSError:
+            raise BundleInstallError("bundle_cleanup_pending") from None
+
+    def cleanup(self) -> None:
+        """Remove only installer-owned orphan staging/generation directories."""
+
+        with self._store_lock():
+            self._cleanup_locked()
+
+
+__all__ = [
+    "BundleInstallError",
+    "BundleRecord",
+    "InstalledBundle",
+    "KeyringUpdate",
+    "RollbackAuthorizer",
+    "SignedBundleInstaller",
+    "parse_keyring_update",
+    "verify_keyring_update",
+]
diff --git a/backend/cortex_backend/execution/manifest.py b/backend/cortex_backend/execution/manifest.py
index 680de80..cdcb42c 100644
--- a/backend/cortex_backend/execution/manifest.py
+++ b/backend/cortex_backend/execution/manifest.py
@@ -1,8 +1,9 @@
 """Fail-closed verification for signed, pinned recipe bundles.
 
 Manifest verification authenticates a bundle description and its bytes.  It does not
-install, load, decode, execute, or publish a recipe provider.  Those actions remain
-separate sandbox and release gates.
+install, load, decode, execute, or publish a recipe provider.  Installation is kept
+in the separate ``SignedBundleInstaller`` storage gate; provider and sandbox actions
+remain release gates.
 """
 
 from __future__ import annotations
@@ -216,6 +217,31 @@ def public_key(self, key_id: str) -> Ed25519PublicKey:
         except (TypeError, ValueError, UnsupportedAlgorithm):
             raise ManifestVerificationError("manifest_key_untrusted") from None
 
+    @property
+    def keys(self) -> Mapping[str, bytes]:
+        """Return a defensive snapshot of the configured public keys."""
+
+        return dict(self._keys)
+
+    @property
+    def revoked(self) -> frozenset[str]:
+        """Return the immutable revoked-key snapshot."""
+
+        return self._revoked
+
+    def digest(self) -> str:
+        """Return a stable digest of the key identifiers and raw public keys."""
+
+        payload = {
+            "keys": {
+                key_id: base64.urlsafe_b64encode(self._keys[key_id]).decode("ascii").rstrip("=")
+                for key_id in sorted(self._keys)
+            },
+            "revoked": sorted(self._revoked),
+        }
+        encoded = json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
+        return sha256(encoded.encode("ascii")).hexdigest()
+
 
 def _parse_version(version: str) -> tuple[int, int, int]:
     match = _SEMVER.fullmatch(version)
@@ -288,6 +314,24 @@ def verify_signed_manifest(
     current: ManifestState | None = None,
     rollback_authorized: bool = False,
 ) -> VerifiedRecipeManifest:
+    verified = verify_manifest_signature(payload, trusted_keys)
+    digest = verified.digest
+    manifest = verified.manifest
+    rollback = _validate_transition(
+        manifest,
+        digest,
+        current,
+        rollback_authorized=rollback_authorized,
+    )
+    return VerifiedRecipeManifest(manifest=manifest, digest=digest, rollback=rollback)
+
+
+def verify_manifest_signature(
+    payload: Mapping[str, Any],
+    trusted_keys: TrustedRecipeKeys,
+) -> VerifiedRecipeManifest:
+    """Verify signature and digest without applying update/rollback transition rules."""
+
     manifest = parse_signed_manifest(payload)
     public_key = trusted_keys.public_key(manifest.key_id)
     signature = _decode_signature(manifest.signature)
@@ -295,14 +339,11 @@ def verify_signed_manifest(
         public_key.verify(signature, manifest.signed_payload())
     except (InvalidSignature, ValueError, TypeError):
         raise ManifestVerificationError("manifest_signature_invalid") from None
-    digest = manifest.manifest_digest()
-    rollback = _validate_transition(
-        manifest,
-        digest,
-        current,
-        rollback_authorized=rollback_authorized,
+    return VerifiedRecipeManifest(
+        manifest=manifest,
+        digest=manifest.manifest_digest(),
+        rollback=False,
     )
-    return VerifiedRecipeManifest(manifest=manifest, digest=digest, rollback=rollback)
 
 
 def verify_bundle_files(manifest: SignedRecipeManifest, bundle_root: Path) -> None:
@@ -349,5 +390,6 @@ def verify_bundle_files(manifest: SignedRecipeManifest, bundle_root: Path) -> No
     "VerifiedRecipeManifest",
     "parse_signed_manifest",
     "verify_bundle_files",
+    "verify_manifest_signature",
     "verify_signed_manifest",
 ]
diff --git a/docs/adr/0001-phase2-bundle-installation.md b/docs/adr/0001-phase2-bundle-installation.md
new file mode 100644
index 0000000..2976961
--- /dev/null
+++ b/docs/adr/0001-phase2-bundle-installation.md
@@ -0,0 +1,100 @@
+# ADR-0001 Phase 2 signed bundle installation and recovery
+
+- **Status:** Implemented and verified; provider, codec, and execution enablement remain blocked
+- **Parent:** [Phase 2 signed recipe manifest and rollback gate](0001-phase2-signed-manifest.md)
+- **Scope:** Verified bundle staging, atomic activation, durable keyring rotation, and explicit recovery
+
+## Decision
+
+`SignedBundleInstaller` is the only storage boundary allowed to turn a verified
+recipe bundle into an installed generation. It is deliberately not a provider: it
+never imports, decodes, executes, loads, or exposes a bundle to a model. The caller
+provides a source directory and a signed `recipe.manifest.v1` payload; the installer
+returns only an immutable, verified generation record.
+
+The durable layout is private to the application data directory:
+
+```text
+recipe_bundles/
+  state.json
+  .install.lock
+  bundles/
+    bundle-<manifest-sha256>/
+      manifest.json
+      <only manifest-declared files>
+```
+
+The generation name is the full manifest digest. Source files are checked by the
+existing manifest verifier and then copied into an exclusive staging directory. The
+copy rejects symlinks/junctions, non-regular files, hardlinks, path escapes, size or
+digest changes, and source identity changes during the copy. Only declared files are
+copied; `manifest.json` is reserved for the canonical signed payload. Staged files
+are flushed before the generation is renamed into the same-volume bundle directory.
+
+Activation is a two-step commit. First, the complete generation is durably written
+and renamed without replacing an existing digest directory. Second, the small
+`state.json` pointer is written to a unique temporary file, flushed, and atomically
+replaced. The pointer records the current generation, the previous verified
+generation, and the keyring update chain. State is never advanced before the
+generation exists and passes a full signature, exact-tree, size, and hash check.
+An interrupted copy or state replacement therefore leaves the old pointer active or
+leaves execution unavailable; an unverified directory is never selected.
+
+The state transition is serialized by an application-owned lock file for both
+threads and installer processes. Startup validates the keyring chain and the current
+generation before reporting it active. Orphan staging and unreferenced generations
+are removed only by the explicit installer-owned cleanup operation; cleanup failures
+become a stable pending category rather than deleting arbitrary paths.
+
+## Key rotation
+
+Key rotation uses a separate strict `recipe.keyring.update.v1` payload. It contains a
+monotonic sequence, the exact previous keyring digest, a signer key id, a bounded map
+of raw Ed25519 public keys, revoked ids, and a signature over canonical JSON without
+the signature field. The signer must be an active key in the current keyring. Every
+update is replayed and verified from the packaged bootstrap keyring on startup; a
+missing link, changed root digest, bad signature, unknown signer, sequence replay,
+or empty active key set fails closed. The active bundle's signing key cannot be
+revoked by a standalone update, preventing a rotation from bricking the installed
+generation. New-key bundles are installed only after the signed rotation has been
+atomically persisted.
+
+## Rollback and recovery
+
+Normal updates must pass the manifest sequence and non-decreasing semantic version
+rules. A downgrade requires both the manifest's exact `rollback_of` digest and a
+trusted local `RollbackAuthorizer(current_digest, candidate_digest)` decision. The
+authorization is evaluated after signature verification and before any staging.
+
+The previous generation is retained in durable state. If the current generation is
+missing, tampered, or fails startup verification, the installer reports
+`bundle_install_recovery_failed`; it does not silently fall back. An operator or
+trusted lifecycle controller must call `recover_previous` with a separate rollback
+authorization. Recovery verifies the retained generation again and atomically points
+state to it, clearing the invalid current pointer.
+
+## Failure contract
+
+Stable categories include `bundle_root_*`, `bundle_path_invalid`,
+`bundle_hardlink_rejected`, `bundle_source_changed`, `bundle_size_mismatch`,
+`bundle_install_*`, `bundle_cleanup_*`, `bundle_recovery_*`, `keyring_*`, and the
+existing manifest verification categories. No path, key material, signature, source
+payload, or OS exception detail is returned across the model/API boundary.
+
+## Explicit non-goals
+
+This ADR does not provide remote update transport, SBOM or Authenticode validation,
+image decoding, copy-in from user-owned conversation artifacts, output validation or
+publication, a sandbox/provider, process launch, broker dispatch, or lifecycle
+enablement. Those remain later gates.
+
+## Verification
+
+`tests/test_phase2_bundle_installer.py` covers restart-stable installation, staged
+copy and exact generation contents, state-commit failure preservation, explicit
+rollback and previous-generation recovery, signed chained key rotation and replay,
+active-key revocation protection, hardlink rejection, and keyring-state tampering.
+The atomic replacement model follows the Windows guidance for [ReplaceFile](https://learn.microsoft.com/en-us/windows/win32/api/winbase/nf-winbase-replacefilea)
+and same-volume [MoveFileEx](https://learn.microsoft.com/en-us/windows/win32/api/winbase/nf-winbase-movefileexa)
+operations; the provider remains disabled until its own sandbox and lifecycle gates
+pass.
diff --git a/docs/adr/0001-phase2-evidence.md b/docs/adr/0001-phase2-evidence.md
index 20982db..4d824bb 100644
--- a/docs/adr/0001-phase2-evidence.md
+++ b/docs/adr/0001-phase2-evidence.md
@@ -1,7 +1,7 @@
 # ADR-0001 Phase 2 evidence log
 
 - **Phase:** 2 — signed image recipes and calculator/check primitives
-- **Status:** Typed contract, signed-manifest verification, and native broker transport complete; installation, staging, provider, and release gates remain open
+- **Status:** Typed contract, signed-manifest verification, native broker transport, and signed bundle installation complete; artifact staging, provider, and release gates remain open
 - **Scope:** Provider-independent validation and deterministic trusted primitives only
 - **Source decision:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md)
 - **Contract ADR:** [Phase 2 typed recipe and primitive contract](0001-phase2-recipe-contract.md)
@@ -15,10 +15,10 @@
 | Calculator/check primitives | **Complete (trusted pure helpers)** | Decimal-only calculator operations and explicit comparisons are deterministic, bounded, and have no I/O or code execution surface. |
 | Canonical plan identity | **Complete** | Validated plans expose stable canonical JSON and SHA-256 digests for future idempotency/signature binding. |
 | Signed recipe manifest | **Complete (verification only)** | Ed25519 signature verification uses a pinned key-id allowlist; every declared bundle entry is path-, size-, and SHA-256-verified; monotonic updates and explicit rollback authorization are enforced. |
-| Signed bundle installation/update | **Blocked / next gate** | Verification does not install, load, atomically replace, persist state, rotate keys, or enable a provider. |
+| Signed bundle installation/update | **Complete (storage-only)** | Digest-named immutable generations, exclusive staging, atomic activation state, chained keyring rotation, explicit rollback authorization, and previous-generation recovery are covered by installer tests. No provider is loaded. |
 | Authenticated broker contract | **Complete (transport-neutral)** | Bounded versioned frames, direction-specific HMAC keys, canonical messages, peer ACL/integrity policy, and owner-scoped authorization are covered by adversarial tests. |
 | Native named-pipe adapter/DACL/peer-token binding | **Complete (transport-only)** | Protected local pipe, expected PID, OS token identity, X25519/HKDF handshake, direction keys, and close-on-error lifecycle are covered by native broker tests. |
-| Copy-in, image decoding, output validation, publication | **Blocked / next gate** | No codec or provider path has been enabled; Phase 1 artifact storage remains the only publication mechanism. |
+| User-artifact copy-in, image decoding, output validation, publication | **Blocked / next gate** | The signed bundle installer copies only its declared bundle files; user-owned artifact snapshotting, codec validation, output publication, and source non-overwrite proofs remain unimplemented. |
 | Production sandbox provider and execution route | **Blocked / next gate** | The native pipe is transport-only; no Wasmtime, AppContainer, Job Object, subprocess, lifecycle provider, or production execution route was added. |
 
 ## Security invariants
@@ -49,6 +49,12 @@
 12. Native transport uses a protected local-only DACL, rejects remote clients, requires
     expected process binding, and closes on identity or handshake failure; it never
     falls back to a default ACL, alternate transport, or provider.
+13. Bundle installation copies only verified declared bytes into an exclusive staging
+    tree, rejects reparse points/hardlinks and source mutation, and activates only a
+    complete digest-named generation.
+14. The activation pointer is atomically replaced only after the generation is
+    verified; keyring updates are signature-chained and rollback/recovery always need
+    a separate trusted local decision.
 
 ## Re-run target
 
@@ -57,6 +63,7 @@ python -m pytest tests/test_phase2_recipe_contract.py -q
 python -m pytest tests/test_phase2_manifest.py -q
 python -m pytest tests/test_phase2_broker.py -q
 python -m pytest tests/test_phase2_native_broker.py -q
+python -m pytest tests/test_phase2_bundle_installer.py -q
 python -m compileall -q backend\cortex_backend\execution tests
 python -m pytest -q
 python tools/generate_contracts.py
@@ -67,8 +74,8 @@ npm.cmd test --prefix frontend -- --run
 ```
 
 **Validation result (2026-07-21):** 16 Phase 2 contract tests, 9 signed-manifest tests,
-7 broker-contract tests, 9 native-broker tests, and the full Python suite passed (164 tests total)
-with one native-platform skip and one pre-existing
+7 broker-contract tests, 9 native-broker tests, 7 bundle-installer tests, and the full
+Python suite passed (171 tests total) with one native-platform skip and one pre-existing
 `pytest-asyncio` deprecation warning.
 Frontend lint, typecheck, production build, and all 39 frontend tests passed. Contract
 generation, compileall, and `git diff --check` passed. No production execution
diff --git a/docs/adr/0001-phase2-native-broker.md b/docs/adr/0001-phase2-native-broker.md
index bddcf48..34480d7 100644
--- a/docs/adr/0001-phase2-native-broker.md
+++ b/docs/adr/0001-phase2-native-broker.md
@@ -1,6 +1,6 @@
 # ADR-0001 Phase 2 native broker adapter
 
-- **Status:** Implemented and verified; bundle installation, staging, and provider enablement remain blocked
+- **Status:** Implemented and verified; user-artifact staging and provider enablement remain blocked
 - **Parent:** [Phase 2 authenticated broker contract](0001-phase2-broker-contract.md)
 - **Scope:** Windows named-pipe transport, protected DACL, OS peer identity, and
   authenticated session-key establishment
diff --git a/docs/adr/0001-phase2-recipe-contract.md b/docs/adr/0001-phase2-recipe-contract.md
index d328655..499e3fe 100644
--- a/docs/adr/0001-phase2-recipe-contract.md
+++ b/docs/adr/0001-phase2-recipe-contract.md
@@ -1,6 +1,6 @@
 # ADR-0001 Phase 2 typed recipe and primitive contract
 
-- **Status:** Typed contract, signed-manifest verification, and native broker transport implemented and verified; bundle installation, staging, and provider enablement remain blocked
+- **Status:** Typed contract, signed-manifest verification, native broker transport, and signed bundle installation implemented and verified; copy-in/output and provider enablement remain blocked
 - **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md)
 - **Depends on:** [Phase 1 production lifecycle gate](0001-phase1-production-lifecycle.md)
 - **Scope:** Typed fixed-function image plans, calculator/check primitives, canonical
@@ -50,10 +50,10 @@ sandbox gates have succeeded.
 
 This ADR does not authorize:
 
-- installation or loading of a signed recipe/runtime bundle; manifest signature and
-  byte verification are implemented separately in
-  [the signed-manifest ADR](0001-phase2-signed-manifest.md), while atomic install,
-  persistent state, key rotation, and recovery remain separate;
+- production loading of a signed recipe/runtime bundle; manifest signature and byte
+  verification plus storage installation are implemented separately in
+  [the signed-manifest ADR](0001-phase2-signed-manifest.md) and
+  [the bundle installation ADR](0001-phase2-bundle-installation.md);
 - image codecs, thumbnails, or decompression handling;
 - artifact copy-in, output validation, atomic publication, or source ownership binding;
 - production execution beyond the transport-only
@@ -66,11 +66,9 @@ packaged application remains on the explicitly disabled lifecycle from Phase 1.
 
 ## Required next gates
 
-1. Implement signed bundle installation/update with atomic replacement, persistent
-   verified state, key rotation, and explicit rollback recovery.
-2. Implement trusted copy-in/output validation and artifact publication tests,
+1. Implement trusted user-artifact copy-in/output validation and artifact publication tests,
    including parser fuzzing and source non-overwrite proofs.
-3. Qualify the fixed-function provider inside the OS sandbox and wire it only through
+2. Qualify the fixed-function provider inside the OS sandbox and wire it only through
    a passing lifecycle health check after external review.
 
 ## Verification
diff --git a/docs/adr/0001-phase2-signed-manifest.md b/docs/adr/0001-phase2-signed-manifest.md
index cb28f5d..428e6d0 100644
--- a/docs/adr/0001-phase2-signed-manifest.md
+++ b/docs/adr/0001-phase2-signed-manifest.md
@@ -1,6 +1,6 @@
 # ADR-0001 Phase 2 signed recipe manifest and rollback gate
 
-- **Status:** Implemented and verified; installation and provider enablement remain blocked
+- **Status:** Implemented and verified; the storage installer is complete and provider enablement remains blocked
 - **Parent:** [Phase 2 typed recipe contract](0001-phase2-recipe-contract.md)
 - **Scope:** Canonical Ed25519 manifest signatures, pinned bundle bytes, monotonic
   update policy, and explicit rollback authorization
@@ -32,12 +32,13 @@ to the exact current manifest digest and must be accompanied by a separate trust
 local rollback authorization. A signed manifest alone cannot authorize rollback. This
 prevents a compromised or stale feed from silently downgrading a healthy installation.
 
-The verifier returns a candidate state but does not persist it. A future installer must
-verify the complete bundle, atomically install it, persist the candidate state only
-after successful installation, and retain the previous verified state for recovery.
-Power loss, partial replacement, failed verification, and failed startup must leave
-the previously accepted state active or leave execution unavailable; there is no
-fallback to an unverified directory.
+The verifier returns a candidate state but does not persist it. The storage boundary
+in [the bundle installation ADR](0001-phase2-bundle-installation.md) verifies the
+complete bundle, atomically installs an immutable digest generation, persists the
+candidate state only after successful installation, and retains the previous verified
+state for explicit recovery. Power loss, partial replacement, failed verification,
+and failed startup leave the previously accepted state active or leave execution
+unavailable; there is no fallback to an unverified directory.
 
 ## Bundle byte verification
 
@@ -58,11 +59,12 @@ detail is returned to a model or API response.
 
 ## Explicit non-goals
 
-This gate does not provide key rotation, remote update transport, persistent state
-storage, atomic installation, SBOM validation, Authenticode verification, image
-decoding, production broker IPC, OS sandboxing, or runtime/provider enablement. Each
-requires its own implementation evidence and review. The application lifecycle remains
-disabled for production execution.
+This manifest verifier does not provide remote update transport, SBOM validation,
+Authenticode verification, image decoding, production broker IPC, OS sandboxing, or
+runtime/provider enablement. Key rotation, persistent state, atomic installation,
+and explicit recovery are implemented separately in
+[the bundle installation ADR](0001-phase2-bundle-installation.md). The application
+lifecycle remains disabled for production execution.
 
 ## Verification
 
diff --git a/tests/test_phase2_bundle_installer.py b/tests/test_phase2_bundle_installer.py
new file mode 100644
index 0000000..a183fb2
--- /dev/null
+++ b/tests/test_phase2_bundle_installer.py
@@ -0,0 +1,268 @@
+"""Signed bundle staging, activation, key rotation, and recovery tests."""
+
+from __future__ import annotations
+
+import base64
+from hashlib import sha256
+import json
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+import cortex_backend.execution.bundle_installer as installer_module
+from cortex_backend.execution.bundle_installer import (
+    BundleInstallError,
+    KeyringUpdate,
+    SignedBundleInstaller,
+)
+from cortex_backend.execution.manifest import (
+    ManifestVerificationError,
+    TrustedRecipeKeys,
+    verify_manifest_signature,
+)
+
+
+def _canonical(payload: dict) -> bytes:
+    return json.dumps(
+        payload,
+        ensure_ascii=True,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("ascii")
+
+
+def _public(private_key: Ed25519PrivateKey) -> bytes:
+    return private_key.public_key().public_bytes(
+        serialization.Encoding.Raw,
+        serialization.PublicFormat.Raw,
+    )
+
+
+def _keyring(private_key: Ed25519PrivateKey, *, key_id: str = "release-1") -> TrustedRecipeKeys:
+    return TrustedRecipeKeys({key_id: _public(private_key)})
+
+
+def _payload(
+    private_key: Ed25519PrivateKey,
+    *,
+    key_id: str = "release-1",
+    sequence: int = 1,
+    bundle_version: str = "1.0.0",
+    rollback_of: str | None = None,
+    content: bytes = b"signed recipe bytes",
+) -> dict:
+    unsigned = {
+        "schema_version": "recipe.manifest.v1",
+        "key_id": key_id,
+        "sequence": sequence,
+        "bundle_version": bundle_version,
+        "rollback_of": rollback_of,
+        "entries": [
+            {
+                "recipe_id": "image-transform",
+                "bundle_path": "recipes/image.bin",
+                "entrypoint": "image_transform",
+                "version": bundle_version,
+                "size": len(content),
+                "sha256": sha256(content).hexdigest(),
+            }
+        ],
+    }
+    signature = base64.urlsafe_b64encode(private_key.sign(_canonical(unsigned))).decode("ascii")
+    return {**unsigned, "signature": signature.rstrip("=")}
+
+
+def _source(root: Path, content: bytes = b"signed recipe bytes") -> None:
+    target = root / "recipes" / "image.bin"
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_bytes(content)
+
+
+def _keyring_update(
+    installer: SignedBundleInstaller,
+    signer: Ed25519PrivateKey,
+    new_key: Ed25519PrivateKey,
+) -> dict:
+    keyring = TrustedRecipeKeys(
+        {"release-1": _public(signer), "release-2": _public(new_key)},
+        revoked=frozenset({"release-1"}),
+    )
+    unsigned = KeyringUpdate(
+        sequence=1,
+        signer_key_id="release-1",
+        previous_digest=installer._bootstrap_digest,
+        keyring=keyring,
+        signature="placeholder",
+    )
+    signature = base64.urlsafe_b64encode(signer.sign(unsigned.signed_payload())).decode("ascii")
+    return {**unsigned.as_payload(), "signature": signature.rstrip("=")}
+
+
+def test_install_is_verified_copied_and_restart_stable(tmp_path: Path):
+    signer = Ed25519PrivateKey.generate()
+    source = tmp_path / "incoming"
+    _source(source)
+    installer = SignedBundleInstaller(tmp_path / "store", _keyring(signer))
+
+    installed = installer.install(_payload(signer), source)
+
+    assert installed.bundle_root.name == f"bundle-{installed.state.digest}"
+    assert (installed.bundle_root / "recipes/image.bin").read_bytes() == b"signed recipe bytes"
+    assert (installed.bundle_root / "manifest.json").is_file()
+    (source / "unlisted.txt").write_text("not trusted", encoding="utf-8")
+    restarted = SignedBundleInstaller(tmp_path / "store", _keyring(signer))
+    assert restarted.status() == installed
+    assert not (installed.bundle_root / "unlisted.txt").exists()
+
+
+def test_failed_state_commit_preserves_previous_active_generation(tmp_path: Path, monkeypatch):
+    signer = Ed25519PrivateKey.generate()
+    keys = _keyring(signer)
+    source = tmp_path / "incoming"
+    _source(source)
+    installer = SignedBundleInstaller(tmp_path / "store", keys)
+    first = installer.install(_payload(signer), source)
+
+    updated = b"updated signed recipe bytes"
+    (source / "recipes/image.bin").write_bytes(updated)
+    payload = _payload(signer, sequence=2, bundle_version="1.1.0", content=updated)
+
+    def fail_state(*_args, **_kwargs):
+        raise BundleInstallError("bundle_install_io_failed")
+
+    monkeypatch.setattr(installer_module, "_atomic_write", fail_state)
+    with pytest.raises(BundleInstallError) as error:
+        installer.install(payload, source)
+    assert error.value.code == "bundle_install_io_failed"
+    assert installer.status() == first
+    assert not any(path.name.startswith(".staging-") for path in installer.bundle_root.iterdir())
+
+
+def test_rollback_requires_explicit_authorization_and_recovery_is_explicit(tmp_path: Path):
+    signer = Ed25519PrivateKey.generate()
+    keys = _keyring(signer)
+    source = tmp_path / "incoming"
+    installer = SignedBundleInstaller(tmp_path / "store", keys)
+    _source(source, b"version one")
+    first = installer.install(
+        _payload(signer, content=b"version one"),
+        source,
+    )
+    _source(source, b"version two")
+    second = installer.install(
+        _payload(signer, sequence=2, bundle_version="2.0.0", content=b"version two"),
+        source,
+    )
+    rollback = _payload(
+        signer,
+        sequence=3,
+        bundle_version="1.0.0",
+        rollback_of=second.state.digest,
+        content=b"version one",
+    )
+    _source(source, b"version one")
+    with pytest.raises(ManifestVerificationError) as unauthorized:
+        installer.install(rollback, source)
+    assert getattr(unauthorized.value, "code", "") == "manifest_rollback_not_authorized"
+
+    rolled_back = installer.install(
+        rollback,
+        source,
+        rollback_authorizer=lambda current, candidate: current == second.state.digest
+        and candidate == verify_manifest_signature(rollback, keys).digest,
+    )
+    assert rolled_back.rollback is True
+    assert rolled_back.state.bundle_version == "1.0.0"
+
+    # Corruption never auto-falls back; the retained previous generation needs a
+    # separate recovery decision.
+    (rolled_back.bundle_root / "recipes/image.bin").write_bytes(b"corrupted")
+    with pytest.raises(BundleInstallError) as corrupted:
+        installer.status()
+    assert corrupted.value.code == "bundle_install_recovery_failed"
+    with pytest.raises(BundleInstallError) as recovery_denied:
+        installer.recover_previous()
+    assert recovery_denied.value.code == "bundle_recovery_not_authorized"
+    recovered = installer.recover_previous(
+        lambda current, candidate: current == rolled_back.state.digest
+        and candidate == second.state.digest
+    )
+    assert recovered.state.bundle_version == "2.0.0"
+
+
+def test_key_rotation_is_signed_chained_and_persistent(tmp_path: Path):
+    signer = Ed25519PrivateKey.generate()
+    next_signer = Ed25519PrivateKey.generate()
+    installer = SignedBundleInstaller(tmp_path / "store", _keyring(signer))
+    update = _keyring_update(installer, signer, next_signer)
+
+    installer.rotate_keyring(update)
+    source = tmp_path / "incoming"
+    _source(source, b"new key bundle")
+    installed = installer.install(
+        _payload(
+            next_signer,
+            key_id="release-2",
+            content=b"new key bundle",
+        ),
+        source,
+    )
+    restarted = SignedBundleInstaller(tmp_path / "store", _keyring(signer))
+    assert restarted.status() == installed
+
+    with pytest.raises(BundleInstallError) as replay:
+        restarted.rotate_keyring(update)
+    assert replay.value.code == "keyring_replay"
+
+
+def test_key_rotation_cannot_revoke_active_signer(tmp_path: Path):
+    signer = Ed25519PrivateKey.generate()
+    next_signer = Ed25519PrivateKey.generate()
+    source = tmp_path / "incoming"
+    _source(source)
+    installer = SignedBundleInstaller(tmp_path / "store", _keyring(signer))
+    installer.install(_payload(signer), source)
+    update = _keyring_update(installer, signer, next_signer)
+    with pytest.raises(BundleInstallError) as error:
+        installer.rotate_keyring(update)
+    assert error.value.code == "keyring_active_key_revoked"
+
+
+def test_hardlinked_source_is_rejected_when_platform_supports_it(tmp_path: Path):
+    signer = Ed25519PrivateKey.generate()
+    source = tmp_path / "incoming"
+    _source(source)
+    external = tmp_path / "external.bin"
+    external.write_bytes(b"signed recipe bytes")
+    target = source / "recipes/image.bin"
+    target.unlink()
+    try:
+        target.hardlink_to(external)
+    except (OSError, NotImplementedError):
+        pytest.skip("hard links are unavailable on this platform")
+    installer = SignedBundleInstaller(tmp_path / "store", _keyring(signer))
+    with pytest.raises(BundleInstallError) as error:
+        installer.install(_payload(signer), source)
+    assert error.value.code == "bundle_hardlink_rejected"
+
+
+def test_state_tampering_with_keyring_history_fails_closed(tmp_path: Path):
+    signer = Ed25519PrivateKey.generate()
+    next_signer = Ed25519PrivateKey.generate()
+    installer = SignedBundleInstaller(tmp_path / "store", _keyring(signer))
+    installer.rotate_keyring(_keyring_update(installer, signer, next_signer))
+    source = tmp_path / "incoming"
+    _source(source, b"new key bundle")
+    installer.install(
+        _payload(next_signer, key_id="release-2", content=b"new key bundle"),
+        source,
+    )
+    state = installer.state_path
+    payload = json.loads(state.read_text(encoding="ascii"))
+    payload["keyring_history"] = []
+    state.write_text(json.dumps(payload), encoding="ascii")
+    with pytest.raises(BundleInstallError) as error:
+        SignedBundleInstaller(tmp_path / "store", _keyring(signer)).status()
+    assert error.value.code == "bundle_install_recovery_failed"