diff --git a/bin/hydra-system b/bin/hydra-system
new file mode 100755
index 0000000..1f9be47
--- /dev/null
+++ b/bin/hydra-system
@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+"""Source checkout entry point; installed copies use the installed Python package."""
+from pathlib import Path
+import sys
+
+checkout = Path(__file__).resolve().parent.parent
+if (checkout / 'hydra/system_recovery.py').is_file():
+    sys.path.insert(0, str(checkout))
+from hydra.system_recovery import main
+
+raise SystemExit(main())
diff --git a/docs/system-recovery.md b/docs/system-recovery.md
new file mode 100644
index 0000000..f535064
--- /dev/null
+++ b/docs/system-recovery.md
@@ -0,0 +1,118 @@
+# Offline Gentoo base recovery
+
+`hydra-system` creates and restores a matched Btrfs root snapshot and the installed
+`EFI/HydraOS` tree. Boot the trusted Hydra live ISO and run it as root with the
+installed root and ESP unmounted. This command does not operate from the installed
+OS. An installed `/etc/hydra/live-mode/active` file alone is insufficient: the
+command also checks the live boot arguments, overlay root and squashfs mount.
+
+The supported installation has `@root`, `@home` and `@snapshots`. `/boot`, kernel
+modules, the Portage package database and world file belong to `@root`. Nested
+subvolumes beneath `@root` are refused. Boot and fstab must select the filesystem
+UUID plus `subvol=@root`, never a numeric root subvolume ID. `@home` and VM disks
+stored there survive restoration; files stored elsewhere in `@root` roll back.
+
+Record the Btrfs filesystem UUID and ESP PARTUUID at installation. From the live
+ISO, inspect the plan before executing it:
+
+```sh
+hydra-system checkpoint --root-uuid ROOT_UUID --esp-partuuid ESP_PARTUUID --name before-update
+hydra-system checkpoint --root-uuid ROOT_UUID --esp-partuuid ESP_PARTUUID --name before-update \
+  --execute --confirm-root-uuid ROOT_UUID
+hydra-system list --root-uuid ROOT_UUID
+```
+
+Commands emit JSON; the completed checkpoint has an `id` field. Boot the installed
+system to perform ordinary Portage updates. To recover, boot the live ISO again:
+
+```sh
+hydra-system restore --root-uuid ROOT_UUID --checkpoint CHECKPOINT_ID
+hydra-system restore --root-uuid ROOT_UUID --checkpoint CHECKPOINT_ID \
+  --execute --confirm-root-uuid ROOT_UUID
+```
+
+Plans and list mount targets read-only; Btrfs inspection disables log replay.
+Execution requires matching confirmation and verifies UUID/PARTUUID uniqueness,
+ordinary partitions, inactive swap, no holders, and a single-device Btrfs layout.
+A private mount namespace and a root-owned lock isolate the command's mounts.
+No installed executable, network access or installed recovery program is needed.
+
+Recovery restores only the recorded `EFI/HydraOS` directory. It does not change
+partition tables, other EFI directories, `EFI/BOOT`, firmware entries or BootOrder.
+Boot the restored system through its HydraOS firmware entry, or select
+`EFI/HydraOS/grubx64.efi` in firmware. A whole-disk installation can also have a
+removable fallback under `EFI/BOOT`; that fallback is deliberately not restored.
+If an updated fallback can no longer load the restored root, select the owned
+HydraOS loader explicitly. Automatic failed-boot recovery is not provided.
+
+## Durable records and interrupted restores
+
+The installer marker `/etc/hydra/storage.json` has exactly these fields:
+
+```json
+{
+  "format": 1,
+  "install_uuid": "installation UUID",
+  "filesystem": "btrfs",
+  "root_fs_uuid": "Btrfs UUID",
+  "root_partuuid": "root GPT partition UUID",
+  "esp_uuid": "FAT UUID (XXXX-XXXX)",
+  "esp_partuuid": "ESP GPT partition UUID",
+  "owned_efi_path": "EFI/HydraOS",
+  "root_subvolume": "@root"
+}
+```
+
+At the Btrfs top level, completed checkpoints are stored as:
+
+```text
+@snapshots/<checkpoint-UUID>/root             read-only Btrfs snapshot
+@snapshots/<checkpoint-UUID>/efi/             owned EFI backup
+@snapshots/<checkpoint-UUID>/checkpoint.json  completed descriptor, published last
+```
+
+The descriptor fields are `format`, `id`, `name`, `created`, `storage`,
+`source_root_uuid`, `snapshot_uuid` and `efi_manifest`. `format` is integer 1;
+`created` is a UTC ISO timestamp; `storage` is the exact marker above. Each manifest
+entry records a relative `path`, `kind` (`dir` or `file`), numeric `mode` and
+`mtime_ns`; files additionally record `size` and SHA-256 `sha256`. Backup paths,
+bytes and metadata are rechecked before every restore. Links, hardlinks, special
+files, ambiguous FAT names and extra checkpoint files are refused. Metadata files
+must be regular, singly linked, owned by root and not group/world writable.
+
+`@snapshots/restore.json` is the durable active transaction. It has exactly
+`format`, `id` (transaction UUID), `checkpoint`, `checkpoint_digest`, `storage`,
+`old_uuid`, `new_uuid`, `prepared_root`, `failed_root` and `phase`. The digest is
+SHA-256 of the descriptor's canonical JSON (sorted keys, compact separators).
+Phases are `preparing`, `prepared`, `root-retained`, `root-installed` and
+`efi-restored`. `new_uuid` starts null; it is recorded once the writable snapshot
+exists. Snapshot parent UUID and marker checks establish the source if power was
+lost before that UUID was journaled. Directory fsync and filesystem sync delimit
+transitions.
+
+A writable copy is prepared as `@restore-<transaction-UUID>`. The previous root is
+retained as `@failed-<transaction-UUID>` before the prepared root becomes `@root`.
+Resume verifies the actual subvolume UUIDs at all three paths and accepts only the
+three valid rename arrangements. The recorded phase alone never decides which
+root to keep. After EFI restoration and verification, the transaction is archived
+as `@snapshots/restored-<transaction-UUID>.json`. No failed roots or checkpoints are
+automatically deleted.
+
+Btrfs root replacement and FAT file copying are not one atomic operation. Power
+loss can leave the system temporarily unbootable. Rerun the **same checkpoint ID**
+from the live ISO with the same explicit confirmation; recovery validates its
+records and resumes. Do not manually rename or delete recovery roots or journal
+files. Unexplained UUID/path combinations are refused for offline inspection.
+An incomplete checkpoint without its completed descriptor is never restorable.
+
+These snapshots remain on the original disk. They are not external backups, and
+there is no promise of online application consistency: checkpoint only while the
+installation is offline. The unit suite uses a persistent fake Btrfs backend and
+fresh processes to exercise journal and rename interruptions. Real Btrfs, FAT,
+UEFI, Portage and disk-only boot acceptance must be verified separately in a fresh
+VM before release; unit tests do not establish those claims.
+
+Btrfs's [subvolume documentation](https://btrfs.readthedocs.io/en/latest/btrfs-subvolume.html)
+explains snapshot boundaries and read-only snapshots; the command checks the
+[read-only property](https://btrfs.readthedocs.io/en/latest/btrfs-property.html)
+explicitly instead of inferring it from mount options.
diff --git a/hydra/system_recovery.py b/hydra/system_recovery.py
new file mode 100644
index 0000000..6fc6c1c
--- /dev/null
+++ b/hydra/system_recovery.py
@@ -0,0 +1,562 @@
+"""Offline, live-ISO-only Btrfs base checkpoints. Never runs installed executables."""
+from __future__ import annotations
+
+import argparse
+import contextlib
+import ctypes
+from datetime import datetime, timezone
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import stat
+import subprocess
+import sys
+import tempfile
+import uuid
+
+
+class RecoveryError(RuntimeError):
+    pass
+
+
+MARKER_KEYS = {'format', 'install_uuid', 'filesystem', 'root_fs_uuid', 'root_partuuid',
+               'esp_uuid', 'esp_partuuid', 'owned_efi_path', 'root_subvolume'}
+ID_RE = re.compile(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\Z')
+
+
+def require(condition, message):
+    if not condition:
+        raise RecoveryError(message)
+
+
+def valid_uuid(value):
+    return isinstance(value, str) and bool(ID_RE.fullmatch(value.lower()))
+
+
+def validate_marker(value):
+    require(isinstance(value, dict) and set(value) == MARKER_KEYS, 'Invalid storage marker fields')
+    require(type(value['format']) is int and value['format'] == 1, 'Unsupported storage marker format')
+    require(all(valid_uuid(value[k]) for k in ('install_uuid', 'root_fs_uuid', 'root_partuuid', 'esp_partuuid')),
+            'Invalid storage UUID')
+    require(isinstance(value['esp_uuid'], str) and re.fullmatch(r'[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}', value['esp_uuid']),
+            'Invalid FAT filesystem UUID')
+    require(value['filesystem'] == 'btrfs' and value['root_subvolume'] == '@root'
+            and value['owned_efi_path'] == 'EFI/HydraOS', 'Unsupported storage layout')
+    return value
+
+
+def safe_path(base, relative, *, missing=False):
+    """Refuse links/special files at every component; caller owns private mount."""
+    base = Path(base)
+    require(not base.is_symlink() and base.is_dir(), 'Unsafe base directory')
+    parts = Path(relative).parts
+    require(parts and not Path(relative).is_absolute() and all(p not in ('.', '..') for p in parts), 'Unsafe path')
+    path = base
+    for part in parts:
+        path = path / part
+        try:
+            mode = path.lstat().st_mode
+        except FileNotFoundError:
+            require(missing, 'Missing path: ' + str(path))
+            continue
+        require(stat.S_ISDIR(mode) or stat.S_ISREG(mode), 'Link or special file refused: ' + str(path))
+    return path
+
+
+def read_json(path):
+    path = Path(path)
+    info = path.lstat()
+    require(stat.S_ISREG(info.st_mode) and info.st_nlink == 1 and info.st_uid == os.geteuid()
+            and not info.st_mode & 0o022, 'Metadata must be a trusted regular file: ' + str(path))
+    require(info.st_size <= 16 * 1024 * 1024, 'Metadata too large')
+    def unique(pairs):
+        out = {}
+        for key, value in pairs:
+            require(key not in out, 'Duplicate JSON field')
+            out[key] = value
+        return out
+    try:
+        return json.loads(path.read_text(), object_pairs_hook=unique)
+    except (ValueError, UnicodeError) as exc:
+        raise RecoveryError('Malformed metadata: ' + str(path)) from exc
+
+
+def fsync_dir(path):
+    fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+    try:
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def durable_json(path, value):
+    path = Path(path)
+    fd, name = tempfile.mkstemp(prefix='.metadata-', dir=path.parent)
+    try:
+        with os.fdopen(fd, 'w') as stream:
+            json.dump(value, stream, sort_keys=True, indent=2)
+            stream.write('\n'); stream.flush(); os.fsync(stream.fileno())
+        os.replace(name, path)
+        fsync_dir(path.parent)
+    finally:
+        if os.path.exists(name): os.unlink(name)
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
+
+
+def file_hash(path):
+    with Path(path).open('rb') as stream:
+        return hashlib.file_digest(stream, 'sha256').hexdigest()
+
+
+def tree_manifest(root):
+    """FAT-compatible names, regular files/directories, exact backup metadata."""
+    root = Path(root)
+    require(root.is_dir() and not root.is_symlink(), 'Owned EFI tree must be a directory')
+    entries = []
+    def walk(directory):
+        folded = set()
+        for path in sorted(directory.iterdir()):
+            name = path.name
+            require(re.fullmatch(r'[A-Za-z0-9_+., -]+', name) and name not in ('.', '..')
+                    and not name.endswith((' ', '.')), 'Unsafe EFI name')
+            require(name.casefold() not in folded, 'Ambiguous FAT filename')
+            folded.add(name.casefold())
+            info = path.lstat()
+            require(stat.S_ISDIR(info.st_mode) or stat.S_ISREG(info.st_mode), 'Unsafe EFI file type')
+            require(not stat.S_ISREG(info.st_mode) or info.st_nlink == 1, 'EFI hardlink refused')
+            entry = dict(path=path.relative_to(root).as_posix(), kind='dir' if path.is_dir() else 'file',
+                         mode=stat.S_IMODE(info.st_mode), mtime_ns=info.st_mtime_ns)
+            if path.is_file():
+                entry.update(size=info.st_size, sha256=file_hash(path))
+            entries.append(entry)
+            if path.is_dir(): walk(path)
+    walk(root)
+    return entries
+
+
+def sync_tree(root):
+    for path in Path(root).rglob('*'):
+        if path.is_file():
+            with path.open('rb') as stream: os.fsync(stream.fileno())
+    for directory, _, _ in os.walk(root, topdown=False): fsync_dir(directory)
+
+
+class Btrfs:
+    def run(self, argv):
+        return subprocess.run(argv, check=True, text=True, stdout=subprocess.PIPE,
+                              stderr=subprocess.PIPE, env=dict(os.environ, LC_ALL='C')).stdout
+    def info(self, path):
+        output = self.run(['btrfs', 'subvolume', 'show', str(path)])
+        fields = dict(line.strip().split(':', 1) for line in output.splitlines() if ':' in line)
+        value = fields.get('UUID', '').strip()
+        parent = fields.get('Parent UUID', '').strip()
+        require(valid_uuid(value), 'Not a Btrfs subvolume: ' + str(path))
+        ro = self.run(['btrfs', 'property', 'get', '-ts', str(path), 'ro']).strip()
+        require(ro in ('ro=true', 'ro=false'), 'Cannot establish snapshot readonly state')
+        return dict(uuid=value, parent_uuid=None if parent == '-' else parent, readonly=ro == 'ro=true')
+    def nested(self, path):
+        return self.run(['btrfs', 'subvolume', 'list', '-o', str(path)]).splitlines()
+    def snapshot(self, source, dest, readonly):
+        self.run(['btrfs', 'subvolume', 'snapshot', *(['-r'] if readonly else []), str(source), str(dest)])
+        return self.info(dest)
+    def sync(self, path):
+        fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+        try:
+            libc = ctypes.CDLL(None, use_errno=True)
+            if libc.syncfs(fd) != 0: raise OSError(ctypes.get_errno(), 'syncfs failed')
+        finally:
+            os.close(fd)
+
+
+class RecoverySession:
+    """Mounted engine; CLI supplies verified identities and private offline mounts."""
+    def __init__(self, top, esp, backend, identities):
+        self.top, self.esp, self.backend = Path(top), Path(esp), backend
+        self.identities = dict(identities)
+        self.snapshots = safe_path(self.top, '@snapshots')
+        self.journal = self.snapshots / 'restore.json'
+        for name in ('@snapshots', '@home'):
+            path = safe_path(self.top, name)
+            require(not self.backend.info(path)['readonly'], 'Layout subvolume is readonly')
+
+    def marker(self, root):
+        marker = validate_marker(read_json(safe_path(root, 'etc/hydra/storage.json')))
+        for key, expected in self.identities.items():
+            require(marker.get(key) == expected, 'Storage identity mismatch: ' + key)
+        require(not self.backend.nested(root), 'Nested subvolumes under root are unsupported')
+        return marker
+
+    def current_root(self):
+        root = safe_path(self.top, '@root')
+        require(not self.backend.info(root)['readonly'], 'Installed root is readonly')
+        return root, self.marker(root)
+
+    def efi_root(self, *, missing=False):
+        # EFI itself must exist. Only the owned directory may be recreated.
+        safe_path(self.esp, 'EFI')
+        root = safe_path(self.esp, 'EFI/HydraOS', missing=missing)
+        for parent, expected in ((self.esp, 'EFI'), (self.esp / 'EFI', 'HydraOS')):
+            require(not any(p.name.casefold() == expected.casefold() and p.name != expected for p in parent.iterdir()),
+                    'Case alias of owned EFI directory refused')
+            names = [p.name.casefold() for p in parent.iterdir()]
+            require(len(names) == len(set(names)), 'Ambiguous EFI directory names')
+        return root
+
+    def checkpoint(self, name, *, dry_run=False):
+        safe_path(self.snapshots, 'restore.json', missing=True)
+        require(not self.journal.exists(), 'Resume pending restore before checkpointing')
+        require(isinstance(name, str) and 1 <= len(name) <= 100 and all(ord(c) >= 32 for c in name), 'Invalid checkpoint name')
+        root, storage = self.current_root()
+        efi = self.efi_root()
+        manifest = tree_manifest(efi)
+        source = self.backend.info(root)
+        if dry_run:
+            return dict(operation='checkpoint', execute=False, name=name, storage=storage)
+        ident = str(uuid.uuid4())
+        directory = self.snapshots / ident
+        directory.mkdir(mode=0o700)
+        fsync_dir(self.snapshots)
+        snapshot = self.backend.snapshot(root, directory / 'root', True)
+        require(snapshot['readonly'] and snapshot['parent_uuid'] == source['uuid'], 'Invalid new snapshot')
+        shutil.copytree(efi, directory / 'efi', copy_function=shutil.copy2)
+        require(tree_manifest(directory / 'efi') == manifest, 'EFI changed during checkpoint')
+        sync_tree(directory / 'efi')
+        self.backend.sync(self.top)
+        descriptor = dict(format=1, id=ident, name=name, created=datetime.now(timezone.utc).isoformat(),
+                          storage=storage, source_root_uuid=source['uuid'], snapshot_uuid=snapshot['uuid'],
+                          efi_manifest=manifest)
+        durable_json(directory / 'checkpoint.json', descriptor)
+        self.backend.sync(self.top)
+        return descriptor
+
+    def descriptor(self, ident):
+        require(valid_uuid(ident), 'Invalid checkpoint ID')
+        directory = safe_path(self.snapshots, ident)
+        require({p.name for p in directory.iterdir()} == {'root', 'efi', 'checkpoint.json'}, 'Unexpected checkpoint files')
+        value = read_json(safe_path(directory, 'checkpoint.json'))
+        require(isinstance(value, dict) and set(value) == {'format', 'id', 'name', 'created', 'storage',
+                'source_root_uuid', 'snapshot_uuid', 'efi_manifest'}, 'Invalid checkpoint fields')
+        require(type(value['format']) is int and value['format'] == 1 and value['id'] == ident,
+                'Unsupported checkpoint descriptor')
+        require(isinstance(value['name'], str) and isinstance(value['created'], str)
+                and valid_uuid(value['source_root_uuid']) and valid_uuid(value['snapshot_uuid']), 'Invalid checkpoint metadata')
+        validate_marker(value['storage'])
+        snapshot = safe_path(directory, 'root')
+        info = self.backend.info(snapshot)
+        require(info == dict(uuid=value['snapshot_uuid'], parent_uuid=value['source_root_uuid'], readonly=True),
+                'Checkpoint snapshot identity or readonly state changed')
+        require(self.marker(snapshot) == value['storage'], 'Checkpoint storage identity changed')
+        require(tree_manifest(safe_path(directory, 'efi')) == value['efi_manifest'], 'EFI backup metadata/hash mismatch')
+        return value
+
+    def list(self):
+        result = []
+        for directory in sorted(self.snapshots.iterdir()):
+            if valid_uuid(directory.name) and not directory.is_symlink() and (directory / 'checkpoint.json').exists():
+                result.append(self.descriptor(directory.name))
+        return result
+
+    def save_journal(self, value, phase):
+        value['phase'] = phase
+        durable_json(self.journal, value)
+        self.backend.sync(self.top)
+
+    def restore(self, ident, *, dry_run=False):
+        safe_path(self.snapshots, 'restore.json', missing=True)
+        descriptor = self.descriptor(ident)  # ALL backup validation precedes any write.
+        source = self.snapshots / ident / 'root'
+        owned = self.efi_root(missing=True)
+        if owned.exists(): tree_manifest(owned)
+        if self.journal.exists():
+            tx = read_json(self.journal)
+            require(isinstance(tx, dict) and set(tx) == {'format', 'id', 'checkpoint', 'checkpoint_digest',
+                    'storage', 'old_uuid', 'new_uuid', 'prepared_root', 'failed_root', 'phase'}, 'Invalid restore journal')
+            require(type(tx['format']) is int and tx['format'] == 1 and valid_uuid(tx['id'])
+                    and tx['checkpoint'] == ident and tx['checkpoint_digest'] == digest(descriptor)
+                    and tx['storage'] == descriptor['storage'] and valid_uuid(tx['old_uuid'])
+                    and (tx['new_uuid'] is None or valid_uuid(tx['new_uuid']))
+                    and tx['phase'] in ('preparing', 'prepared', 'root-retained', 'root-installed', 'efi-restored')
+                    and tx['prepared_root'] == '@restore-' + tx['id'] and tx['failed_root'] == '@failed-' + tx['id'],
+                    'Restore journal does not match requested checkpoint')
+        else:
+            root, marker = self.current_root()
+            require(marker == descriptor['storage'], 'Current root belongs to a different installation')
+            require(not any(p.name.startswith('@restore-') for p in self.top.iterdir()), 'Unexplained prepared root')
+            txid = str(uuid.uuid4())
+            tx = dict(format=1, id=txid, checkpoint=ident, checkpoint_digest=digest(descriptor),
+                      storage=marker, old_uuid=self.backend.info(root)['uuid'], new_uuid=None,
+                      prepared_root='@restore-' + txid, failed_root='@failed-' + txid, phase='preparing')
+        prepared = safe_path(self.top, tx['prepared_root'], missing=True)
+        failed = safe_path(self.top, tx['failed_root'], missing=True)
+        root = safe_path(self.top, '@root', missing=True)
+        def identity(path):
+            if not path.exists(): return None
+            info = self.backend.info(path)
+            require(not info['readonly'] and self.marker(path) == tx['storage'], 'Unexpected root identity or layout')
+            return info
+        actual_root, actual_prepared, actual_failed = identity(root), identity(prepared), identity(failed)
+        # Enumerate the only three legitimate rename arrangements. No phase-based guessing.
+        if tx['new_uuid'] is None:
+            require(tx['phase'] == 'preparing' and actual_root and actual_root['uuid'] == tx['old_uuid']
+                    and actual_failed is None, 'Unexplained initial restore state')
+            if actual_prepared is None:
+                if dry_run:
+                    return dict(operation='restore', execute=False, checkpoint=ident, storage=descriptor['storage'])
+                if not self.journal.exists(): self.save_journal(tx, 'preparing')
+                actual_prepared = self.backend.snapshot(source, prepared, False)
+                fsync_dir(self.top); self.backend.sync(self.top)
+            require(actual_prepared['parent_uuid'] == descriptor['snapshot_uuid'] and not actual_prepared['readonly'],
+                    'Prepared snapshot source mismatch')
+            tx['new_uuid'] = actual_prepared['uuid']
+            if not dry_run: self.save_journal(tx, 'prepared')
+        new_uuid, old_uuid = tx['new_uuid'], tx['old_uuid']
+        state = (actual_root['uuid'] if actual_root else None, actual_prepared['uuid'] if actual_prepared else None,
+                 actual_failed['uuid'] if actual_failed else None)
+        require(state in ((old_uuid, new_uuid, None), (None, new_uuid, old_uuid), (new_uuid, None, old_uuid)),
+                'Unexplained root rename state; preserve all roots and inspect offline')
+        active_new = actual_prepared if actual_prepared else actual_root
+        require(active_new['parent_uuid'] == descriptor['snapshot_uuid'], 'Prepared root parent UUID mismatch')
+        if dry_run:
+            return dict(operation='restore', execute=False, checkpoint=ident, storage=descriptor['storage'])
+        if state == (old_uuid, new_uuid, None):
+            os.rename(root, failed); fsync_dir(self.top); self.backend.sync(self.top)
+            self.save_journal(tx, 'root-retained')
+            state = (None, new_uuid, old_uuid)
+        if state == (None, new_uuid, old_uuid):
+            os.rename(prepared, root); fsync_dir(self.top); self.backend.sync(self.top)
+            self.save_journal(tx, 'root-installed')
+        self.restore_efi(self.snapshots / ident / 'efi', owned)
+        self.backend.sync(self.esp)
+        self.save_journal(tx, 'efi-restored')
+        # Archive rather than discard the identity record and retained root path.
+        os.replace(self.journal, self.snapshots / ('restored-' + tx['id'] + '.json'))
+        fsync_dir(self.snapshots); self.backend.sync(self.top)
+        return dict(checkpoint=ident, failed_root=tx['failed_root'], restored_root_uuid=new_uuid)
+
+    def restore_efi(self, source, dest):
+        # Backup and existing destination have already passed complete path validation.
+        # Repeating this copy is safe after interruption at any individual FAT write.
+        if dest.exists():
+            for child in list(dest.iterdir()):
+                if child.is_dir(): shutil.rmtree(child)
+                else: child.unlink()
+        else:
+            dest.mkdir()
+        shutil.copytree(source, dest, dirs_exist_ok=True, copy_function=shutil.copy2)
+        sync_tree(dest); fsync_dir(dest.parent)
+        expected = [(e['path'], e.get('sha256')) for e in tree_manifest(source)]
+        actual = [(e['path'], e.get('sha256')) for e in tree_manifest(dest)]
+        require(actual == expected, 'Restored EFI verification failed; rerun the same restore')
+
+
+def flatten(nodes):
+    result = {}
+    def visit(node):
+        require(isinstance(node, dict) and isinstance(node.get('path'), str), 'Malformed device topology')
+        path = node['path']
+        require(re.fullmatch(r'/dev/[A-Za-z0-9_./-]+', path) and '..' not in Path(path).parts, 'Invalid device path')
+        require(path not in result or result[path] == node, 'Conflicting device topology')
+        result[path] = node
+        require(isinstance(node.get('children', []), list), 'Malformed device children')
+        for child in node.get('children', []): visit(child)
+    for node in nodes: visit(node)
+    return result
+
+
+def resolve_partition(nodes, key, value, filesystem):
+    devices = flatten(nodes)
+    matches = [n for n in devices.values() if str(n.get(key, '')).lower() == value.lower()]
+    require(len(matches) == 1, 'Missing or ambiguous ' + key + ': ' + value)
+    selected = matches[0]
+    require(selected.get('type') == 'part' and selected.get('fstype') == filesystem
+            and not selected.get('children'), 'Target must be an ordinary partition of type ' + filesystem)
+    require(isinstance(selected.get('mountpoints'), list) and not any(selected['mountpoints']), 'Target is mounted or active swap')
+    for field in ('uuid', 'partuuid'):
+        identity = selected.get(field)
+        require(isinstance(identity, str) and identity and sum(str(n.get(field, '')).lower() == identity.lower()
+                for n in devices.values()) == 1, 'Missing or duplicate target ' + field)
+    parent = selected.get('pkname', '')
+    if not parent.startswith('/dev/'): parent = '/dev/' + parent
+    require(parent in devices and devices[parent].get('type') == 'disk', 'Target parent must be an ordinary disk')
+    require(not any(devices[parent].get('mountpoints', [])), 'Parent disk is mounted')
+    require(re.fullmatch(r'\d+:\d+', selected.get('maj:min', '')), 'Missing target block identity')
+    return selected
+
+
+def validate_live(base=Path('/'), cmdline=None, mountinfo=None):
+    base = Path(base)
+    path = safe_path(base, 'etc/hydra/live-mode/active')
+    info = path.lstat()
+    require(stat.S_ISREG(info.st_mode) and info.st_uid == os.geteuid() and info.st_nlink == 1
+            and not info.st_mode & 0o022, 'Untrusted live marker')
+    for directory in (path.parent, path.parent.parent, path.parent.parent.parent):
+        info = directory.stat()
+        require(info.st_uid == os.geteuid() and not info.st_mode & 0o022, 'Untrusted live marker directory')
+    if cmdline is None: cmdline = Path('/proc/cmdline').read_text()
+    if mountinfo is None: mountinfo = Path('/proc/self/mountinfo').read_text()
+    words = cmdline.split()
+    require('rd.live.image' in words and any(w.startswith('root=live:') for w in words), 'Recovery requires a Hydra live boot')
+    mounts = [line.split() for line in mountinfo.splitlines()]
+    require(any(len(parts) > 8 and parts[4] == '/' and '-' in parts
+                and parts[parts.index('-') + 1] == 'overlay' for parts in mounts)
+            and any(' - squashfs ' in line for line in mountinfo.splitlines()), 'Live root must use overlay and squashfs')
+
+
+class DeviceProbe:
+    def __init__(self, backend, *, proc=Path('/proc'), sysroot=Path('/sys')):
+        self.backend, self.proc, self.sysroot = backend, Path(proc), Path(sysroot)
+    def nodes(self):
+        data = json.loads(self.backend.run(['lsblk', '--json', '--paths', '--output',
+                                          'PATH,TYPE,PKNAME,MAJ:MIN,FSTYPE,UUID,PARTUUID,MOUNTPOINTS']))
+        require(isinstance(data, dict) and isinstance(data.get('blockdevices'), list), 'Malformed lsblk result')
+        nodes = data['blockdevices']
+        # Probe signatures directly; do not trust stale udev UUID caches.
+        for node in flatten(nodes).values():
+            result = subprocess.run(['blkid', '-p', '-o', 'export', node['path']], text=True,
+                                    stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            require(result.returncode in (0, 2), 'Ambiguous or unreadable device signatures')
+            fields = dict(line.split('=', 1) for line in result.stdout.splitlines() if '=' in line)
+            node.update(uuid=fields.get('UUID'), partuuid=fields.get('PART_ENTRY_UUID'), fstype=fields.get('TYPE'))
+        return nodes
+    def check(self, node):
+        path = Path(node['path'])
+        info = path.stat()
+        require(stat.S_ISBLK(info.st_mode) and f'{os.major(info.st_rdev)}:{os.minor(info.st_rdev)}' == node['maj:min'],
+                'Device identity changed')
+        sysnode = self.sysroot / 'dev/block' / node['maj:min']
+        require((sysnode / 'partition').is_file() and not any((sysnode / 'holders').iterdir()), 'Target has holders or is not a partition')
+        parent = sysnode.resolve().parent
+        require(not any((parent / 'holders').iterdir()), 'Parent disk has active holders')
+        for line in (self.proc / 'self/mountinfo').read_text().splitlines():
+            fields = line.split(); separator = fields.index('-')
+            source = fields[separator + 2]
+            require(fields[2] != node['maj:min'], 'Target is mounted')
+            if source.startswith('/dev/'):
+                try: mounted = Path(source).stat()
+                except FileNotFoundError: raise RecoveryError('Cannot resolve mounted block source')
+                require(mounted.st_rdev != info.st_rdev, 'Target backs a mounted filesystem')
+        for line in (self.proc / 'swaps').read_text().splitlines()[1:]:
+            source = Path(line.split()[0])
+            swap = source.stat()
+            require((swap.st_rdev if stat.S_ISBLK(swap.st_mode) else swap.st_dev) != info.st_rdev,
+                    'Target has active swap')
+
+
+@contextlib.contextmanager
+def operation_lock():
+    fd = os.open('/run/hydra-system.lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+    try:
+        info = os.fstat(fd)
+        require(stat.S_ISREG(info.st_mode) and info.st_uid == 0 and info.st_nlink == 1 and not info.st_mode & 0o077,
+                'Untrusted operation lock')
+        try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError: raise RecoveryError('Another hydra-system operation is running')
+        yield
+    finally: os.close(fd)
+
+
+def private_namespace(backend):
+    libc = ctypes.CDLL(None, use_errno=True)
+    if libc.unshare(0x00020000) != 0: raise OSError(ctypes.get_errno(), 'Cannot create private mount namespace')
+    backend.run(['mount', '--make-rprivate', '/'])
+
+
+@contextlib.contextmanager
+def mounted(backend, device, filesystem, writable=False):
+    directory = Path(tempfile.mkdtemp(prefix='hydra-system-', dir='/run'))
+    did_mount = False
+    try:
+        options = ['rw' if writable else 'ro', 'nosuid', 'nodev', 'noexec']
+        if filesystem == 'btrfs':
+            options.append('subvolid=5')
+            if not writable: options.append('nologreplay')
+        backend.run(['mount', '-t', filesystem, '-o', ','.join(options), device, str(directory)])
+        did_mount = True
+        if filesystem == 'btrfs':
+            show = backend.run(['btrfs', 'filesystem', 'show', '--raw', str(directory)])
+            require(re.search(r'Total devices\s+1\b', show) and len(re.findall(r'\bdevid\s+\d+', show)) == 1,
+                    'Recovery requires a single-device Btrfs filesystem')
+        yield directory
+    finally:
+        if did_mount:
+            # A failed unmount deliberately leaves the directory intact; never recurse into a mount.
+            backend.run(['umount', str(directory)])
+        directory.rmdir()
+
+
+def mounted_storage(top):
+    if (top / '@root').exists():
+        return validate_marker(read_json(safe_path(top, '@root/etc/hydra/storage.json')))
+    journal = read_json(safe_path(top, '@snapshots/restore.json'))
+    require(isinstance(journal, dict) and isinstance(journal.get('storage'), dict), 'Missing root without valid recovery journal')
+    return validate_marker(journal['storage'])
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    commands = parser.add_subparsers(dest='command', required=True)
+    for name in ('checkpoint', 'list', 'restore'):
+        command = commands.add_parser(name)
+        command.add_argument('--root-uuid', required=True)
+        if name != 'list':
+            command.add_argument('--execute', action='store_true')
+            command.add_argument('--confirm-root-uuid')
+        if name == 'checkpoint':
+            command.add_argument('--esp-partuuid', required=True)
+            command.add_argument('--name', required=True)
+        if name == 'restore': command.add_argument('--checkpoint', required=True)
+    args = parser.parse_args(argv)
+    try:
+        require(valid_uuid(args.root_uuid), 'Invalid root filesystem UUID')
+        execute = getattr(args, 'execute', False)
+        if execute:
+            require(args.confirm_root_uuid == args.root_uuid, 'Execution requires --confirm-root-uuid matching --root-uuid')
+        require(os.geteuid() == 0, 'Run from the Hydra live ISO as root; no automatic sudo')
+        validate_live()
+        backend = Btrfs(); probe = DeviceProbe(backend)
+        with operation_lock():
+            nodes = probe.nodes()
+            root = resolve_partition(nodes, 'uuid', args.root_uuid, 'btrfs'); probe.check(root)
+            private_namespace(backend)
+            # First inspect read-only, without log replay, before opening either filesystem writable.
+            with mounted(backend, root['path'], 'btrfs') as top:
+                marker = mounted_storage(top)
+                require(marker['root_fs_uuid'] == args.root_uuid and marker['root_partuuid'] == root['partuuid'], 'Root ownership mismatch')
+                if args.command == 'checkpoint': require(args.esp_partuuid == marker['esp_partuuid'], 'ESP differs from installation record')
+                esp = resolve_partition(nodes, 'partuuid', marker['esp_partuuid'], 'vfat'); probe.check(esp)
+                require(esp['uuid'] == marker['esp_uuid'], 'ESP filesystem identity mismatch')
+                with mounted(backend, esp['path'], 'vfat') as efi:
+                    session = RecoverySession(top, efi, backend, marker)
+                    if args.command == 'list': result = session.list()
+                    elif args.command == 'restore':
+                        result = session.restore(args.checkpoint, dry_run=True)
+                    else:
+                        result = session.checkpoint(args.name, dry_run=True)
+            if execute:
+                # Re-resolve and check after releasing inspection mounts, immediately before writes.
+                nodes = probe.nodes()
+                current_root = resolve_partition(nodes, 'uuid', args.root_uuid, 'btrfs')
+                current_esp = resolve_partition(nodes, 'partuuid', marker['esp_partuuid'], 'vfat')
+                for before, after in ((root, current_root), (esp, current_esp)):
+                    require(all(before.get(k) == after.get(k) for k in ('path', 'uuid', 'partuuid', 'maj:min', 'fstype')), 'Target changed during planning')
+                    probe.check(after)
+                with mounted(backend, root['path'], 'btrfs', True) as top:
+                    with mounted(backend, esp['path'], 'vfat', True) as efi:
+                        session = RecoverySession(top, efi, backend, marker)
+                        result = session.checkpoint(args.name) if args.command == 'checkpoint' else session.restore(args.checkpoint)
+            print(json.dumps(result, indent=2, sort_keys=True))
+        return 0
+    except (RecoveryError, OSError, ValueError, subprocess.CalledProcessError) as exc:
+        print('hydra-system: ' + str(exc), file=sys.stderr)
+        return 1
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())
diff --git a/tests/test_system_recovery.py b/tests/test_system_recovery.py
new file mode 100644
index 0000000..950761b
--- /dev/null
+++ b/tests/test_system_recovery.py
@@ -0,0 +1,334 @@
+"""Offline recovery tests; fake Btrfs only, never operate on block devices."""
+import copy
+import io
+import stat
+from types import SimpleNamespace
+import json
+import os
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+import uuid
+from unittest.mock import patch
+
+try:
+    from hydra import system_recovery as r
+except ImportError:
+    r = None
+
+ROOT_UUID = '11111111-2222-3333-4444-555555555555'
+ROOT_PART = '21111111-2222-3333-4444-555555555555'
+ESP_PART = '31111111-2222-3333-4444-555555555555'
+MARKER = dict(format=1, install_uuid='41111111-2222-3333-4444-555555555555',
+              filesystem='btrfs', root_fs_uuid=ROOT_UUID, root_partuuid=ROOT_PART,
+              esp_uuid='1234-ABCD', esp_partuuid=ESP_PART,
+              owned_efi_path='EFI/HydraOS', root_subvolume='@root')
+
+class FakeBtrfs:
+    """Persist identity inside fake subvolumes so a fresh process can recover."""
+    def info(self, path):
+        return json.loads((Path(path) / '.fake-subvolume.json').read_text())
+    def create(self, path, parent=None, readonly=False):
+        Path(path).mkdir(parents=True, exist_ok=True)
+        info = dict(uuid=str(uuid.uuid4()), parent_uuid=parent, readonly=readonly)
+        (Path(path) / '.fake-subvolume.json').write_text(json.dumps(info))
+        return info
+    def snapshot(self, source, dest, readonly):
+        shutil.copytree(source, dest)
+        return self.create(dest, self.info(source)['uuid'], readonly)
+    def nested(self, path):
+        return [str(p) for p in Path(path).glob('**/.fake-subvolume.json') if p.parent != Path(path)]
+    def sync(self, path):
+        pass
+
+class RecoveryTests(unittest.TestCase):
+    def setUp(self):
+        self.assertIsNotNone(r, 'offline recovery module must exist')
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.base = Path(self.temp.name)
+        self.top, self.esp = self.base / 'top', self.base / 'esp'
+        self.top.mkdir(); self.esp.mkdir()
+        self.b = FakeBtrfs()
+        for name in ('@root', '@home', '@snapshots'):
+            self.b.create(self.top / name)
+        marker = self.top / '@root/etc/hydra/storage.json'
+        marker.parent.mkdir(parents=True)
+        marker.write_text(json.dumps(MARKER))
+        (self.top / '@root/kernel').write_text('old kernel')
+        (self.top / '@home/user-data').write_text('user data')
+        (self.esp / 'EFI/HydraOS').mkdir(parents=True)
+        (self.esp / 'EFI/HydraOS/grubx64.efi').write_bytes(b'old EFI')
+        (self.esp / 'EFI/OtherOS').mkdir()
+        (self.esp / 'EFI/OtherOS/sentinel').write_bytes(b'foreign')
+        self.session = r.RecoverySession(self.top, self.esp, self.b, MARKER)
+    def checkpoint(self):
+        return self.session.checkpoint('before-update')
+    def test_exact_marker(self):
+        self.assertEqual(r.validate_marker(MARKER), MARKER)
+        for change in ({'format': True}, {'filesystem': 'ext4'}, {'root_subvolume': '@other'},
+                       {'owned_efi_path': '../EFI'}, {'unexpected': 'field'}, {'root_partuuid': 'bad'}):
+            with self.subTest(change=change), self.assertRaises(r.RecoveryError):
+                r.validate_marker(dict(MARKER, **change))
+    def test_checkpoint_completed_and_readonly(self):
+        result = self.checkpoint()
+        self.assertTrue(self.b.info(self.top / '@snapshots' / result['id'] / 'root')['readonly'])
+        self.assertEqual(self.session.list(), [result])
+        self.assertEqual(result['storage'], MARKER)
+    def test_snapshot_failure_never_publishes_descriptor(self):
+        with patch.object(self.b, 'snapshot', side_effect=OSError('interrupted')):
+            with self.assertRaises(OSError): self.checkpoint()
+        self.assertEqual(self.session.list(), [])
+    def test_nested_subvolume_refused(self):
+        self.b.create(self.top / '@root/nested')
+        with self.assertRaises(r.RecoveryError): self.checkpoint()
+    def test_restore_preserves_home_foreign_and_failed_root(self):
+        cp = self.checkpoint()
+        old_uuid = self.b.info(self.top / '@root')['uuid']
+        (self.top / '@root/kernel').write_text('broken kernel')
+        (self.top / '@home/user-data').write_text('new home')
+        (self.esp / 'EFI/HydraOS/grubx64.efi').write_bytes(b'broken')
+        (self.esp / 'EFI/HydraOS/new-file').write_text('remove')
+        result = self.session.restore(cp['id'])
+        self.assertEqual((self.top / '@root/kernel').read_text(), 'old kernel')
+        self.assertEqual((self.top / '@home/user-data').read_text(), 'new home')
+        self.assertEqual((self.esp / 'EFI/OtherOS/sentinel').read_bytes(), b'foreign')
+        self.assertEqual((self.esp / 'EFI/HydraOS/grubx64.efi').read_bytes(), b'old EFI')
+        self.assertFalse((self.esp / 'EFI/HydraOS/new-file').exists())
+        self.assertEqual(self.b.info(self.top / result['failed_root'])['uuid'], old_uuid)
+    def test_tampered_backup_rejected_before_root_mutation(self):
+        cp = self.checkpoint()
+        original = self.b.info(self.top / '@root')
+        (self.top / '@snapshots' / cp['id'] / 'efi/grubx64.efi').write_bytes(b'bad')
+        with self.assertRaises(r.RecoveryError): self.session.restore(cp['id'])
+        self.assertEqual(self.b.info(self.top / '@root'), original)
+        self.assertFalse((self.top / '@snapshots/restore.json').exists())
+    def test_symlink_efi_and_metadata_refused(self):
+        (self.esp / 'EFI/HydraOS/link').symlink_to('/etc/passwd')
+        with self.assertRaises(r.RecoveryError): self.checkpoint()
+    def test_rw_snapshot_refused(self):
+        cp = self.checkpoint()
+        path = self.top / '@snapshots' / cp['id'] / 'root/.fake-subvolume.json'
+        data = json.loads(path.read_text()); data['readonly'] = False; path.write_text(json.dumps(data))
+        with self.assertRaises(r.RecoveryError): self.session.restore(cp['id'])
+    def test_wrong_installed_identity_refused(self):
+        marker = self.top / '@root/etc/hydra/storage.json'
+        marker.write_text(json.dumps(dict(MARKER, esp_partuuid=ROOT_PART)))
+        with self.assertRaises(r.RecoveryError): self.checkpoint()
+    def test_fresh_process_resume_each_transition(self):
+        cp = self.checkpoint()
+        for phase in ('prepared', 'root-retained', 'root-installed', 'efi-restored'):
+            with self.subTest(phase=phase):
+                (self.top / '@root/kernel').write_text('broken again')
+                script = '''import json, os, sys
+from pathlib import Path
+from hydra import system_recovery as r
+sys.path.insert(0, 'tests')
+from test_system_recovery import FakeBtrfs, MARKER
+s = r.RecoverySession(Path(sys.argv[1]), Path(sys.argv[2]), FakeBtrfs(), MARKER)
+save = s.save_journal
+def crash(value, phase):
+    save(value, phase)
+    if phase == sys.argv[4]: os._exit(77)
+s.save_journal = crash
+s.restore(sys.argv[3])
+'''
+                result = subprocess.run([sys.executable, '-c', script, str(self.top), str(self.esp), cp['id'], phase])
+                self.assertEqual(result.returncode, 77)
+                subprocess.run([sys.executable, '-c', script, str(self.top), str(self.esp), cp['id'], 'none'], check=True)
+                self.assertEqual((self.top / '@root/kernel').read_text(), 'old kernel')
+
+
+class GuardTests(unittest.TestCase):
+    def devices(self):
+        return [dict(path='/dev/vda', type='disk', **{'maj:min': '8:0'}, mountpoints=[], children=[
+            dict(path='/dev/vda1', type='part', fstype='vfat', uuid='1234-ABCD', partuuid=ESP_PART,
+                 pkname='/dev/vda', **{'maj:min': '8:1'}, mountpoints=[], children=[]),
+            dict(path='/dev/vda2', type='part', fstype='btrfs', uuid=ROOT_UUID, partuuid=ROOT_PART,
+                 pkname='/dev/vda', **{'maj:min': '8:2'}, mountpoints=[], children=[])])]
+    def test_unique_ordinary_unmounted_partition(self):
+        nodes = self.devices()
+        self.assertEqual(r.resolve_partition(nodes, 'uuid', ROOT_UUID, 'btrfs')['path'], '/dev/vda2')
+        for change in ('duplicate', 'mounted', 'holder', 'not-partition', 'same-partuuid'):
+            bad = copy.deepcopy(nodes)
+            root = bad[0]['children'][1]
+            if change == 'duplicate':
+                extra = dict(root, path='/dev/vdb2', **{'maj:min': '8:18'})
+                bad.append(extra)
+            elif change == 'mounted': root['mountpoints'] = ['/mnt/target']
+            elif change == 'holder': root['children'] = [dict(path='/dev/dm-0', type='crypt')]
+            elif change == 'not-partition': root['type'] = 'disk'
+            else: bad[0]['children'][0]['partuuid'] = ROOT_PART
+            with self.subTest(change=change), self.assertRaises(r.RecoveryError):
+                r.resolve_partition(bad, 'uuid', ROOT_UUID, 'btrfs')
+    def test_live_marker_must_be_regular_trusted_and_live_boot(self):
+        with tempfile.TemporaryDirectory() as temp:
+            base = Path(temp); marker = base / 'etc/hydra/live-mode/active'
+            marker.parent.mkdir(parents=True); marker.write_text('')
+            cmdline = 'root=live:CDLABEL=HYDRA_OS rd.live.image rd.live.overlay.overlayfs=1'
+            mountinfo = '1 0 0:1 / / rw - overlay overlay rw\n2 1 7:0 / /run/rootfsbase ro - squashfs /dev/loop0 ro\n'
+            r.validate_live(base, cmdline, mountinfo)
+            for command, mounts in (('', mountinfo), (cmdline, '')):
+                with self.assertRaises(r.RecoveryError): r.validate_live(base, command, mounts)
+            marker.unlink(); marker.symlink_to('/etc/passwd')
+            with self.assertRaises(r.RecoveryError): r.validate_live(base, cmdline, mountinfo)
+    def test_untrusted_journal_link_refused_before_write(self):
+        case = RecoveryTests('test_exact_marker'); case.setUp()
+        try:
+            cp = case.checkpoint()
+            (case.top / '@snapshots/restore.json').symlink_to(case.base / 'absent')
+            with self.assertRaises(r.RecoveryError): case.session.restore(cp['id'])
+            self.assertFalse((case.base / 'absent').exists())
+        finally: case.doCleanups()
+
+class HardeningTests(unittest.TestCase):
+    setUp = RecoveryTests.setUp
+    checkpoint = RecoveryTests.checkpoint
+    def test_checkpoint_plan_validates_pending_transaction_without_writes(self):
+        before = sorted(str(p.relative_to(self.base)) for p in self.base.rglob('*'))
+        self.session.checkpoint('planned', dry_run=True)
+        self.assertEqual(before, sorted(str(p.relative_to(self.base)) for p in self.base.rglob('*')))
+        (self.top / '@snapshots/restore.json').write_text('{}')
+        with self.assertRaises(r.RecoveryError): self.session.checkpoint('planned', dry_run=True)
+    def test_dry_restore_verifies_destination_and_writes_nothing(self):
+        cp = self.checkpoint()
+        before = sorted(str(p.relative_to(self.base)) for p in self.base.rglob('*'))
+        self.session.restore(cp['id'], dry_run=True)
+        self.assertEqual(before, sorted(str(p.relative_to(self.base)) for p in self.base.rglob('*')))
+        (self.esp / 'EFI/HydraOS/escape').symlink_to(self.base / 'foreign')
+        with self.assertRaises(r.RecoveryError): self.session.restore(cp['id'], dry_run=True)
+    def test_case_alias_of_owned_directory_refused(self):
+        cp = self.checkpoint()
+        (self.esp / 'EFI/HydraOS').rename(self.esp / 'EFI/hydraos')
+        with self.assertRaises(r.RecoveryError): self.session.restore(cp['id'])
+        self.assertFalse((self.top / '@snapshots/restore.json').exists())
+    def test_unexpected_checkpoint_file_refused(self):
+        cp = self.checkpoint()
+        (self.top / '@snapshots' / cp['id'] / 'unexpected').write_text('bad')
+        with self.assertRaises(r.RecoveryError): self.session.restore(cp['id'])
+    def test_fresh_process_death_between_mutation_and_journal(self):
+        cp = self.checkpoint()
+        for point in ('preparing', 'snapshot', 'first-rename', 'second-rename', 'efi-copy'):
+            script = '''import os, sys
+from pathlib import Path
+from hydra import system_recovery as r
+sys.path.insert(0, 'tests')
+from test_system_recovery import FakeBtrfs, MARKER
+b = FakeBtrfs()
+s = r.RecoverySession(Path(sys.argv[1]), Path(sys.argv[2]), b, MARKER)
+point = sys.argv[4]
+save = s.save_journal
+def crash(value, phase):
+    save(value, phase)
+    if phase == point: os._exit(77)
+s.save_journal = crash
+snapshot = b.snapshot
+def interrupted_snapshot(*args):
+    result = snapshot(*args)
+    if point == 'snapshot': os._exit(77)
+    return result
+b.snapshot = interrupted_snapshot
+rename = r.os.rename
+def interrupted_rename(source, dest):
+    rename(source, dest)
+    if (point == 'first-rename' and Path(source).name == '@root') or (point == 'second-rename' and Path(dest).name == '@root'): os._exit(77)
+r.os.rename = interrupted_rename
+copytree = r.shutil.copytree
+def interrupted_copytree(*args, **kwargs):
+    if point == 'efi-copy' and Path(args[0]).name == 'efi': os._exit(77)
+    return copytree(*args, **kwargs)
+if point == 'efi-copy': r.shutil.copytree = interrupted_copytree
+s.restore(sys.argv[3])
+'''
+            with self.subTest(point=point):
+                result = subprocess.run([sys.executable, '-c', script, str(self.top), str(self.esp), cp['id'], point])
+                self.assertEqual(result.returncode, 77)
+                # A different checkpoint must never take over this transaction.
+                with self.assertRaises(r.RecoveryError): self.session.restore(str(uuid.uuid4()))
+                subprocess.run([sys.executable, '-c', script, str(self.top), str(self.esp), cp['id'], 'none'], check=True)
+                self.assertEqual((self.top / '@root/kernel').read_text(), 'old kernel')
+
+class NativeBoundaryTests(unittest.TestCase):
+    def test_btrfs_show_parser_and_readonly_property(self):
+        backend = r.Btrfs()
+        sample = '''@snapshots/example/root
+        Name: root
+        UUID: 51111111-2222-3333-4444-555555555555
+        Parent UUID: 61111111-2222-3333-4444-555555555555
+        Received UUID: -
+        Creation time: 2026-10-13 10:15:03 +0000
+        Subvolume ID: 259
+        Flags: readonly
+        Snapshot(s):
+'''
+        with patch.object(backend, 'run', side_effect=[sample, 'ro=true\n']):
+            self.assertEqual(backend.info('/mnt/snapshot'), dict(uuid='51111111-2222-3333-4444-555555555555',
+                parent_uuid='61111111-2222-3333-4444-555555555555', readonly=True))
+    def test_failed_mount_never_unmounts_unowned_path(self):
+        backend = r.Btrfs()
+        calls = []
+        def fail(argv):
+            calls.append(argv)
+            raise r.RecoveryError('mount failed')
+        with tempfile.TemporaryDirectory() as temp:
+            own = Path(temp) / 'own'; own.mkdir()
+            with patch.object(r.tempfile, 'mkdtemp', return_value=str(own)), patch.object(backend, 'run', side_effect=fail):
+                with self.assertRaises(r.RecoveryError):
+                    with r.mounted(backend, '/dev/fake', 'btrfs'): pass
+            self.assertEqual([argv[0] for argv in calls], ['mount'])
+            self.assertFalse(own.exists())
+    def test_multidevice_mount_is_rejected_and_owned_mount_cleaned(self):
+        backend = r.Btrfs()
+        calls = []
+        def run(argv):
+            calls.append(argv)
+            return 'Total devices 2\n devid 1 size 100\n devid 2 size 100' if argv[0] == 'btrfs' else ''
+        with tempfile.TemporaryDirectory() as temp:
+            own = Path(temp) / 'own'; own.mkdir()
+            with patch.object(r.tempfile, 'mkdtemp', return_value=str(own)), patch.object(backend, 'run', side_effect=run):
+                with self.assertRaises(r.RecoveryError):
+                    with r.mounted(backend, '/dev/fake', 'btrfs'): pass
+            self.assertEqual([argv[0] for argv in calls], ['mount', 'btrfs', 'umount'])
+            self.assertIn('ro,nosuid,nodev,noexec,subvolid=5,nologreplay', calls[0])
+    def test_cli_requires_matching_confirmation_before_probe(self):
+        with patch.object(r, 'DeviceProbe') as probe, patch.object(sys, 'stderr', new_callable=io.StringIO):
+            self.assertEqual(r.main(['restore', '--root-uuid', ROOT_UUID, '--checkpoint', ROOT_PART, '--execute']), 1)
+            probe.assert_not_called()
+
+
+class DeviceStateTests(unittest.TestCase):
+    def test_fake_proc_sys_mount_swap_and_holders_fail_closed(self):
+        with tempfile.TemporaryDirectory() as temp:
+            base = Path(temp)
+            proc, sysroot = base / 'proc', base / 'sys'
+            proc.mkdir(); (proc / 'self').mkdir()
+            (proc / 'self/mountinfo').write_text('')
+            (proc / 'swaps').write_text('Filename Type Size Used Priority\n')
+            physical = sysroot / 'devices/disk/part'
+            physical.mkdir(parents=True)
+            (physical / 'partition').write_text('2')
+            (physical / 'holders').mkdir(); (physical.parent / 'holders').mkdir()
+            (sysroot / 'dev/block').mkdir(parents=True)
+            (sysroot / 'dev/block/8:2').symlink_to(physical)
+            node = GuardTests().devices()[0]['children'][1]
+            original_stat = Path.stat
+            def device_stat(path, *args, **kwargs):
+                if str(path) in ('/dev/vda2', '/dev/alias'):
+                    return SimpleNamespace(st_mode=stat.S_IFBLK, st_rdev=os.makedev(8, 2))
+                return original_stat(path, *args, **kwargs)
+            with patch.object(Path, 'stat', device_stat):
+                probe = r.DeviceProbe(r.Btrfs(), proc=proc, sysroot=sysroot)
+                probe.check(node)
+                (proc / 'self/mountinfo').write_text('1 0 0:99 / /mnt rw - btrfs /dev/alias rw\n')
+                with self.assertRaises(r.RecoveryError): probe.check(node)
+                (proc / 'self/mountinfo').write_text('')
+                (proc / 'swaps').write_text('Filename Type Size Used Priority\n/dev/vda2 partition 100 0 -2\n')
+                with self.assertRaises(r.RecoveryError): probe.check(node)
+                (proc / 'swaps').write_text('Filename Type Size Used Priority\n')
+                (physical / 'holders/dm-0').touch()
+                with self.assertRaises(r.RecoveryError): probe.check(node)
+
+if __name__ == '__main__': unittest.main()