diff --git a/scripts/migrate_graph_layers.py b/scripts/migrate_graph_layers.py
index 5713105..3928147 100755
--- a/scripts/migrate_graph_layers.py
+++ b/scripts/migrate_graph_layers.py
@@ -1,8 +1,17 @@
 #!/usr/bin/env python3
-"""Inventory possible user-to-agent graph-layer migrations.
-
-The inventory command is deliberately read-only. It records evidence for human
-review; it never changes the SQLite database and does not implement apply mode.
+"""Inventory and apply user-to-agent graph-layer migrations.
+
+The ``inventory`` command is deliberately read-only: it records evidence for
+human review and never changes the SQLite database. The ``apply`` command moves
+exactly the approved candidates from the ``user`` layer to the ``agent`` layer
+in one transaction, after a snapshot and a full preflight, and writes an apply
+report next to the manifest (or to ``--report``). Edit manifest ``decision``
+values, inspect the exact manifest bytes, compute the approval using
+``confirm-token --manifest PATH``, then pass that token to ``apply``. The token
+is not stored in the manifest. A fsynced recovery report is created before the
+database commit and atomically moved into place afterward. SQLite and the
+filesystem cannot commit atomically together; if that final move fails, apply
+raises with the path to the complete recovery report.
 """
 
 from __future__ import annotations
@@ -11,12 +20,15 @@
 import hashlib
 import json
 import os
+import re
 import sqlite3
+import stat
 import sys
 import tempfile
-from contextlib import closing
+from contextlib import closing, suppress
+from datetime import datetime, timezone
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, TypedDict
+from typing import TYPE_CHECKING, Any, NoReturn, TypedDict
 from urllib.parse import quote
 
 
@@ -24,8 +36,20 @@
     from collections.abc import Sequence
 
 
+MANIFEST_SCHEMA = 1
 MARKER_VERSION = 1
 EXCLUDE_VERSION = 1
+CONFIDENCE_RANK = {"low": 1, "medium": 2, "high": 3}
+APPROVE_CONFIDENCE_CHOICES = ("high", "medium", "low", "none")
+CANDIDATE_DECISIONS = {"approve", "review", "exclude"}
+CROSS_LAYER_EDGE_SAMPLE_LIMIT = 20
+# Live bases are held open by running MCP servers; a checkpoint or a short write
+# can hold the write lock for a moment. 30 s outlasts a checkpoint flush but
+# still fails a truly wedged base instead of hanging the operator's terminal.
+APPLY_BUSY_TIMEOUT_MS = 30_000
+# SQLite allows 32766 bound variables per statement; stay well below it so one
+# post-move layer lookup never raises inside the transaction.
+LAYER_LOOKUP_CHUNK_SIZE = 500
 MARKER_SOURCES = {"none", "cli", "file", "cli+file"}
 EXCLUDE_SOURCES = {"none", "cli", "file", "cli+file"}
 REQUIRED_NODE_COLUMNS = {"node_id", "layer", "user_id", "content", "node_type", "tags"}
@@ -50,6 +74,33 @@ class SchemaError(RuntimeError):
     """Raised when the database does not expose the expected graph schema."""
 
 
+class ApplyError(RuntimeError):
+    """Raised when a migration apply cannot proceed safely."""
+
+
+def _locked_apply_error(db_path: Path, exc: sqlite3.OperationalError) -> ApplyError:
+    """Translate SQLite lock acquisition failures into the apply contract."""
+    return ApplyError(f"database is locked: could not start the migration transaction on {db_path}")
+
+
+def _fingerprint_mismatch_error(snapshot_path: Path, snapshot_sha: str, expected_sha: str) -> ApplyError:
+    """Describe a lock-time snapshot that differs from the confirmed manifest."""
+    return ApplyError(
+        f"pre-transaction fingerprint {snapshot_sha} does not match the confirmed manifest fingerprint {expected_sha}; "
+        f"the database changed before the lock-time snapshot (snapshot kept at {snapshot_path})"
+    )
+
+
+def _raise_sqlite_error(exc: sqlite3.OperationalError) -> NoReturn:
+    """Re-raise an unrelated SQLite error outside the lock translation branch."""
+    raise exc
+
+
+def _raise_fingerprint_mismatch(snapshot_path: Path, snapshot_sha: str, expected_sha: str) -> NoReturn:
+    """Raise the lock-time fingerprint refusal after the snapshot is kept."""
+    raise _fingerprint_mismatch_error(snapshot_path, snapshot_sha, expected_sha)
+
+
 class Candidate(TypedDict):
     node_id: int
     user_id: str
@@ -104,13 +155,253 @@ def _resolve_output_paths(data_dir: Path, output: Path) -> tuple[Path, Path]:
     return resolved_db, output.absolute()
 
 
-def _fingerprint(path: Path, source_metadata: dict[str, Any]) -> dict[str, Any]:
+def _file_sha256(path: Path) -> str:
     digest = hashlib.sha256()
     with path.open("rb") as source:
         for chunk in iter(lambda: source.read(1024 * 1024), b""):
             digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _fsync_directory(path: Path) -> None:
+    """Persist directory-entry changes, propagating unsupported/error states."""
+    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
+    directory_fd = os.open(path, flags)
+    try:
+        os.fsync(directory_fd)
+    finally:
+        os.close(directory_fd)
+
+
+def _same_file_if_present(first: Path, second: Path) -> bool:
+    """Compare existing paths by device/inode without following missing paths."""
+    if not first.exists() or not second.exists():
+        return False
+    first_stat = first.stat()
+    second_stat = second.stat()
+    return (first_stat.st_dev, first_stat.st_ino) == (second_stat.st_dev, second_stat.st_ino)
+
+
+def _fingerprint(path: Path, source_metadata: dict[str, Any]) -> dict[str, Any]:
     stat = path.stat()
-    return {"sha256": digest.hexdigest(), "size": stat.st_size, "source": source_metadata}
+    return {"sha256": _file_sha256(path), "size": stat.st_size, "source": source_metadata}
+
+
+def _confirm_token(database: str, manifest_sha256: str) -> str:
+    """Return the operator confirmation token for exact manifest bytes."""
+    return f"MIGRATE-{database}-{manifest_sha256[:12]}"
+
+
+def _confirm_token_from_manifest(path: Path) -> str:
+    """Return the token for a reviewed manifest without modifying it."""
+    try:
+        manifest_bytes = path.read_bytes()
+        manifest = json.loads(manifest_bytes)
+    except (OSError, ValueError) as exc:
+        raise ApplyError(f"cannot read manifest {path}: {exc}") from exc
+    if not isinstance(manifest, dict) or not isinstance(manifest.get("database"), str):
+        raise ApplyError("manifest must be a JSON object with a database label")
+    label = str(manifest["database"])
+    _database_label(path.parent, label)
+    return _confirm_token(label, hashlib.sha256(manifest_bytes).hexdigest())
+
+
+def _utc_timestamp() -> str:
+    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+
+
+def _snapshot_to(db_path: Path, destination: Path) -> None:
+    """Write a consistent SQLite online backup, WAL included, to ``destination``."""
+    encoded_db_path = quote(str(db_path.resolve()), safe="/")
+    with (
+        closing(sqlite3.connect(f"file:{encoded_db_path}?mode=ro", uri=True)) as source_conn,
+        sqlite3.connect(destination) as destination_conn,
+    ):
+        source_conn.backup(destination_conn)
+
+
+def _snapshot_connection(source_conn: sqlite3.Connection, destination: Path) -> None:
+    """Back up committed state while ``source_conn`` owns the write lock.
+
+    SQLite's backup API cannot copy from a connection into a file destination
+    while that same connection has an open write transaction on this build: the
+    destination repeatedly waits for the source transaction.  The locked
+    connection remains authoritative; a short read-only handle to the same
+    database supplies the backup API's source while the RESERVED lock prevents
+    any writer from changing the committed state during the copy.
+    """
+    database_list = source_conn.execute("PRAGMA database_list").fetchall()
+    main_database = next((str(row[2]) for row in database_list if str(row[1]) == "main"), "")
+    if not main_database:
+        raise sqlite3.OperationalError("cannot resolve the main database path for the snapshot")
+    encoded_path = quote(main_database, safe="/")
+    with (
+        closing(sqlite3.connect(f"file:{encoded_path}?mode=ro", uri=True)) as backup_source,
+        closing(sqlite3.connect(destination)) as destination_conn,
+    ):
+        backup_source.backup(destination_conn)
+
+
+def _reserve_snapshot(directory: Path) -> tuple[Path, int]:
+    """Create a private staging directory and exclusive temporary snapshot."""
+    staging_dir = Path(tempfile.mkdtemp(prefix=".snapshot-stage-", dir=directory))
+    temp_path = staging_dir / "snapshot.db"
+    fd: int | None = None
+    try:
+        _fsync_directory(directory)
+        fd = os.open(temp_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
+        os.fchmod(fd, 0o600)
+        return temp_path, fd
+    except BaseException:
+        if fd is not None:
+            os.close(fd)
+        _discard_snapshot_staging(temp_path)
+        raise
+
+
+def _before_snapshot_install(path: Path) -> None:
+    """Provide a narrow hook for exercising the no-overwrite installation race."""
+
+
+def _install_snapshot(temp_path: Path, final_path: Path) -> None:
+    """Install a completed snapshot without replacing any existing directory entry."""
+    _before_snapshot_install(final_path)
+    try:
+        os.link(temp_path, final_path)
+    except FileExistsError as exc:
+        raise ApplyError(f"refusing to overwrite existing snapshot {final_path}") from exc
+    temp_path.unlink()
+    _fsync_directory(temp_path.parent)
+    temp_path.parent.rmdir()
+    _fsync_directory(final_path.parent)
+
+
+def _discard_snapshot_staging(temp_path: Path) -> None:
+    """Best-effort cleanup that can never mask the original migration failure."""
+    with suppress(OSError):
+        temp_path.unlink(missing_ok=True)
+    with suppress(OSError):
+        temp_path.parent.rmdir()
+
+
+def _ensure_snapshot_directory(path: Path) -> None:
+    """Create missing snapshot directories and persist each new parent entry."""
+    missing: list[Path] = []
+    current = path
+    while not current.exists():
+        missing.append(current)
+        parent = current.parent
+        if parent == current:
+            raise ApplyError(f"cannot resolve an existing parent for snapshot directory {path}")
+        current = parent
+    if not current.is_dir():
+        raise ApplyError(f"snapshot directory parent is not a directory: {current}")
+    for directory in reversed(missing):
+        try:
+            directory.mkdir()
+        except FileExistsError:
+            if not directory.is_dir():
+                raise ApplyError(f"snapshot directory path is not a directory: {directory}") from None
+        else:
+            _fsync_directory(directory.parent)
+    if not path.is_dir():
+        raise ApplyError(f"snapshot path is not a directory: {path}")
+
+
+def _sync_snapshot(path: Path) -> None:
+    """Set private permissions and durably persist snapshot bytes and its name."""
+    os.chmod(path, 0o600)
+    if stat.S_IMODE(path.stat().st_mode) != 0o600:
+        raise OSError(f"snapshot permissions are not owner-only: {path}")
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    snapshot_fd = os.open(path, flags)
+    try:
+        os.fsync(snapshot_fd)
+    finally:
+        os.close(snapshot_fd)
+    _fsync_directory(path.parent)
+
+
+def _commit_transaction(conn: sqlite3.Connection) -> None:
+    """Commit the migration after its recovery artifacts are durable."""
+    conn.execute("COMMIT")
+
+
+def _resolve_snapshot_path(data_dir: Path, snapshot_dir: Path, name: str) -> Path:
+    """Refuse a snapshot path that would alias the database or one of its sidecars."""
+    db_path = _database_path(data_dir)
+    target = (snapshot_dir / name).absolute()
+    resolved_target = target.resolve()
+    protected = [path.resolve() for path in _protected_database_paths(data_dir)]
+    if resolved_target in protected:
+        raise SchemaError("refusing snapshot path resolving to the database or one of its sidecars")
+    if target.exists():
+        db_stat = db_path.stat()
+        target_stat = target.stat()
+        if (target_stat.st_dev, target_stat.st_ino) == (db_stat.st_dev, db_stat.st_ino):
+            raise SchemaError("refusing snapshot path aliasing database")
+        for sidecar in _protected_database_paths(data_dir)[1:]:
+            if not sidecar.exists():
+                continue
+            sidecar_stat = sidecar.stat()
+            if (target_stat.st_dev, target_stat.st_ino) == (sidecar_stat.st_dev, sidecar_stat.st_ino):
+                raise SchemaError("refusing snapshot path aliasing a database sidecar")
+    return target
+
+
+def _atomic_write_json(path: Path, payload: dict[str, Any]) -> None:
+    temp_path: Path | None = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=path.parent,
+            prefix=f".{path.name}.",
+            suffix=".tmp",
+            delete=False,
+        ) as temp_file:
+            temp_path = Path(temp_file.name)
+            temp_file.write(json.dumps(payload, ensure_ascii=False, indent=2) + "\n")
+            temp_file.flush()
+            os.fsync(temp_file.fileno())
+        os.replace(temp_path, path)
+        temp_path = None
+    finally:
+        if temp_path is not None:
+            temp_path.unlink(missing_ok=True)
+
+
+def _write_report_temp(path: Path, payload: dict[str, Any], manifest_sha256: str) -> Path:
+    """Durably write a unique, deterministic recovery report beside its target."""
+    content = json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
+    stem = f".{path.name}.recovery-{manifest_sha256[:12]}"
+    counter = 0
+    while True:
+        suffix = ".tmp" if counter == 0 else f"-{counter}.tmp"
+        temp_path = path.with_name(stem + suffix)
+        try:
+            fd = os.open(temp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+        except FileExistsError:
+            counter += 1
+            continue
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as temp_file:
+                temp_file.write(content)
+                temp_file.flush()
+                os.fsync(temp_file.fileno())
+        except BaseException:
+            temp_path.unlink(missing_ok=True)
+            raise
+        try:
+            _fsync_directory(path.parent)
+        except BaseException:
+            temp_path.unlink(missing_ok=True)
+            raise
+        return temp_path
+
+
+def _default_report_path(manifest_path: Path) -> Path:
+    return manifest_path.with_name(f"{manifest_path.name}.apply-report.json")
 
 
 def _table_columns(conn: sqlite3.Connection, table: str) -> set[str]:
@@ -360,13 +651,14 @@ def _candidate(
 
 
 def _database_label(data_dir: Path, database: str | None) -> str:
-    if database is not None:
-        if Path(database).is_absolute():
-            raise SchemaError("database label must not be an absolute path")
-        return database
-    prefix = ".mcp-ariel-memory-"
-    name = data_dir.name
-    return name[len(prefix) :] if name.startswith(prefix) and len(name) > len(prefix) else name
+    label = database
+    if label is None:
+        prefix = ".mcp-ariel-memory-"
+        name = data_dir.name
+        label = name[len(prefix) :] if name.startswith(prefix) and len(name) > len(prefix) else name
+    if not label or label in {".", ".."} or ".." in label or re.fullmatch(r"[A-Za-z0-9._-]+", label) is None:
+        raise SchemaError("database label must be one safe filename component (ASCII letters, digits, dot, underscore, hyphen)")
+    return label
 
 
 def _source_metadata(db_path: Path, conn: sqlite3.Connection) -> dict[str, Any]:
@@ -450,7 +742,7 @@ def inventory(
         "user_counts": {str(row[0]): int(row[1]) for row in user_rows},
     }
     return {
-        "schema": 1,
+        "schema": MANIFEST_SCHEMA,
         "database": label,
         "source_layer": "user",
         "target_layer": "agent",
@@ -504,10 +796,376 @@ def write_inventory(
             temp_path.unlink(missing_ok=True)
 
 
+def _should_move(candidate: dict[str, Any], approve_confidence: str) -> bool:
+    """Return whether a manifest candidate belongs to the move set."""
+    decision = str(candidate["decision"])
+    if decision == "exclude":
+        return False
+    if decision == "approve":
+        return True
+    if decision == "review":
+        if approve_confidence == "none":
+            return False
+        return CONFIDENCE_RANK[str(candidate["confidence"])] >= CONFIDENCE_RANK[approve_confidence]
+    raise ApplyError(f"candidate {candidate['node_id']} has unknown decision {decision!r}")
+
+
+def _validated_candidates(conn: sqlite3.Connection, manifest: dict[str, Any]) -> list[dict[str, Any]]:
+    """Validate every manifest candidate against the snapshot before any write."""
+    raw = manifest.get("candidates")
+    if not isinstance(raw, list):
+        raise ApplyError("manifest candidates must be a list")
+    candidates: list[dict[str, Any]] = []
+    seen: set[int] = set()
+    for item in raw:
+        if not isinstance(item, dict):
+            raise ApplyError("manifest candidate entries must be objects")
+        try:
+            node_id = int(item["node_id"])
+            user_id = str(item["user_id"])
+        except (KeyError, TypeError, ValueError) as exc:
+            raise ApplyError("manifest candidate is missing a usable node_id/user_id") from exc
+        decision = str(item.get("decision"))
+        if decision not in CANDIDATE_DECISIONS:
+            raise ApplyError(f"candidate {node_id} has unknown decision {decision!r}")
+        confidence = str(item.get("confidence"))
+        if confidence not in CONFIDENCE_RANK:
+            raise ApplyError(f"candidate {node_id} has unknown confidence {confidence!r}")
+        if node_id in seen:
+            raise ApplyError(f"duplicate candidate node_id {node_id} in manifest")
+        seen.add(node_id)
+        row = conn.execute("SELECT layer, user_id FROM epi_nodes WHERE node_id = ?", (node_id,)).fetchone()
+        if row is None:
+            raise ApplyError(f"candidate node_id {node_id} does not exist in epi_nodes")
+        if str(row["layer"]) != "user":
+            raise ApplyError(f"candidate node_id {node_id} is no longer in layer 'user' (found {row['layer']!r})")
+        if str(row["user_id"]) != user_id:
+            raise ApplyError(f"candidate node_id {node_id} belongs to user {row['user_id']!r}, not {user_id!r}")
+        candidates.append(item)
+    return candidates
+
+
+def _layer_counts(conn: sqlite3.Connection) -> dict[str, int]:
+    rows = conn.execute("SELECT layer, COUNT(*) FROM epi_nodes GROUP BY layer ORDER BY layer").fetchall()
+    return {str(row[0]): int(row[1]) for row in rows}
+
+
+def _layers_for(conn: sqlite3.Connection, node_ids: Sequence[int]) -> dict[int, str]:
+    if not node_ids:
+        return {}
+    ids = list(node_ids)
+    layers: dict[int, str] = {}
+    for start in range(0, len(ids), LAYER_LOOKUP_CHUNK_SIZE):
+        chunk = ids[start : start + LAYER_LOOKUP_CHUNK_SIZE]
+        placeholders = ", ".join("?" for _ in chunk)
+        rows = conn.execute(f"SELECT node_id, layer FROM epi_nodes WHERE node_id IN ({placeholders})", tuple(chunk)).fetchall()
+        layers.update({int(row["node_id"]): str(row["layer"]) for row in rows})
+    return layers
+
+
+def _move_node(conn: sqlite3.Connection, node_id: int, user_id: str) -> int:
+    cursor = conn.execute(
+        "UPDATE epi_nodes SET layer = 'agent' WHERE node_id = ? AND layer = 'user' AND user_id = ?",
+        (node_id, user_id),
+    )
+    return int(cursor.rowcount)
+
+
+def _cross_layer_edges(conn: sqlite3.Connection, edge_pk: list[str]) -> dict[str, Any]:
+    rows = conn.execute(
+        "SELECT e.* FROM epi_edges AS e "
+        "JOIN epi_nodes AS source ON source.node_id = e.source_id "
+        "JOIN epi_nodes AS target ON target.node_id = e.target_id "
+        "WHERE (source.layer = 'user' AND target.layer = 'agent') "
+        "OR (source.layer = 'agent' AND target.layer = 'user') "
+        "ORDER BY e.source_id, e.target_id, e.relation"
+    ).fetchall()
+    keys = [{field: row[field] for field in edge_pk} for row in rows]
+    return {"count": len(keys), "sample": keys[:CROSS_LAYER_EDGE_SAMPLE_LIMIT]}
+
+
+def _run_moves(
+    conn: sqlite3.Connection,
+    edge_pk: list[str],
+    moved: list[dict[str, Any]],
+    unmoved: list[dict[str, Any]],
+    moved_ids: list[int],
+    before_layers: dict[int, str],
+) -> tuple[dict[str, int], dict[str, int], dict[str, bool], dict[str, Any]]:
+    """Move approved rows and verify moved and unapproved node-layer identity.
+
+    Migration changes only ``layer``; the all-unapproved invariant therefore
+    compares the complete node-id-to-layer map outside the moved-id set.
+    """
+    before_counts = _layer_counts(conn)
+    total_before = sum(before_counts.values())
+    for candidate in moved:
+        node_id = int(candidate["node_id"])
+        rowcount = _move_node(conn, node_id, str(candidate["user_id"]))
+        if rowcount != 1:
+            raise ApplyError(f"expected exactly one user row for node_id {node_id}, updated {rowcount}")
+    after_counts = _layer_counts(conn)
+    total_after = sum(after_counts.values())
+    moved_layers = _layers_for(conn, moved_ids)
+    unmoved_layers = _layers_for(conn, [int(candidate["node_id"]) for candidate in unmoved])
+    after_layers = {int(row["node_id"]): str(row["layer"]) for row in conn.execute("SELECT node_id, layer FROM epi_nodes")}
+    before_unapproved = {node_id: layer for node_id, layer in before_layers.items() if node_id not in moved_ids}
+    after_unapproved = {node_id: layer for node_id, layer in after_layers.items() if node_id not in moved_ids}
+    postconditions = {
+        "moved_nodes_in_agent": all(moved_layers.get(node_id) == "agent" for node_id in moved_ids),
+        "moved_nodes_not_in_user": all(moved_layers.get(node_id) != "user" for node_id in moved_ids),
+        "total_count_unchanged": total_after == total_before,
+        "user_delta_exact": before_counts.get("user", 0) - after_counts.get("user", 0) == len(moved_ids),
+        "agent_delta_exact": after_counts.get("agent", 0) - before_counts.get("agent", 0) == len(moved_ids),
+        "unmoved_nodes_unchanged": all(unmoved_layers.get(int(candidate["node_id"])) == str(candidate["current_layer"]) for candidate in unmoved),
+        "all_unapproved_nodes_unchanged": before_unapproved == after_unapproved,
+    }
+    if not all(postconditions.values()):
+        failed = ", ".join(sorted(name for name, passed in postconditions.items() if not passed))
+        raise ApplyError(f"postcondition failed: {failed}")
+    return before_counts, after_counts, postconditions, _cross_layer_edges(conn, edge_pk)
+
+
+def _apply_transaction(
+    db_path: Path,
+    moved: list[dict[str, Any]],
+    unmoved: list[dict[str, Any]],
+    moved_ids: list[int],
+    snapshot_path: Path,
+    expected_sha: str,
+    report_path: Path,
+    report_fields: dict[str, Any],
+    manifest_sha256: str,
+) -> tuple[dict[str, int], dict[str, int], dict[str, bool], dict[str, Any], str, dict[str, Any], Path]:
+    """Move every candidate and verify postconditions in one transaction.
+
+    The write lock is acquired before the authoritative snapshot is created.
+    A busy base therefore fails without a snapshot to keep.
+    """
+    snapshot_sha = ""
+    snapshot_temp: Path | None = None
+    snapshot_durable = False
+    conn = sqlite3.connect(db_path, isolation_level=None)
+    try:
+        conn.row_factory = sqlite3.Row
+        conn.execute(f"PRAGMA busy_timeout = {int(APPLY_BUSY_TIMEOUT_MS)}")
+        edge_pk = _validate_schema(conn)
+        try:
+            conn.execute("BEGIN IMMEDIATE")
+        except sqlite3.OperationalError as exc:
+            if "locked" in str(exc).lower() or "busy" in str(exc).lower():
+                raise _locked_apply_error(db_path, exc) from exc
+            _raise_sqlite_error(exc)
+        try:
+            snapshot_temp, reservation_fd = _reserve_snapshot(snapshot_path.parent)
+            os.close(reservation_fd)
+            before_layers = {int(row["node_id"]): str(row["layer"]) for row in conn.execute("SELECT node_id, layer FROM epi_nodes")}
+            try:
+                _snapshot_connection(conn, snapshot_temp)
+                _sync_snapshot(snapshot_temp)
+                _install_snapshot(snapshot_temp, snapshot_path)
+                snapshot_temp = None
+                snapshot_durable = True
+            except BaseException as exc:
+                if snapshot_temp is not None:
+                    _discard_snapshot_staging(snapshot_temp)
+                    snapshot_temp = None
+                if snapshot_path.is_file():
+                    raise ApplyError(
+                        f"snapshot was installed but final directory durability failed; "
+                        f"the database transaction was rolled back and snapshot remains at {snapshot_path}: {exc}"
+                    ) from exc
+                raise ApplyError(f"snapshot creation or durability failed; partial snapshot discarded: {exc}") from exc
+            snapshot_sha = _file_sha256(snapshot_path)
+            if snapshot_sha != expected_sha:
+                _raise_fingerprint_mismatch(snapshot_path, snapshot_sha, expected_sha)
+            result = _run_moves(conn, edge_pk, moved, unmoved, moved_ids, before_layers)
+            before_counts, after_counts, postconditions, cross_layer_edges = result
+            report_payload = {
+                **report_fields,
+                "snapshot": {"path": str(snapshot_path), "sha256": snapshot_sha},
+                "before": {"layer_counts": before_counts},
+                "after": {"layer_counts": after_counts},
+                "cross_layer_edges": cross_layer_edges,
+                "postconditions": postconditions,
+                "success": all(postconditions.values()),
+            }
+            report_payload["fingerprint"] = {
+                **dict(report_fields["fingerprint"]),
+                "pre_transaction_sha256": snapshot_sha,
+            }
+            report_temp = _write_report_temp(report_path, report_payload, manifest_sha256)
+            _commit_transaction(conn)
+        except BaseException as original_error:
+            rollback_error: BaseException | None = None
+            try:
+                conn.execute("ROLLBACK")
+            except BaseException as exc:
+                rollback_error = exc
+            finally:
+                if snapshot_temp is not None:
+                    _discard_snapshot_staging(snapshot_temp)
+                if "report_temp" in locals():
+                    with suppress(OSError):
+                        report_temp.unlink(missing_ok=True)
+            if rollback_error is not None:
+                raise ApplyError(
+                    f"migration failed ({original_error}); rollback also failed ({rollback_error}); "
+                    f"database state requires recovery from durable snapshot {snapshot_path}"
+                ) from original_error
+            raise
+    finally:
+        conn.close()
+    return (*result, snapshot_sha, report_payload, report_temp)
+
+
+def apply(
+    data_dir: Path,
+    manifest_path: Path,
+    snapshot_dir: Path,
+    *,
+    confirm: str,
+    approve_confidence: str = "none",
+    report: Path | None = None,
+    database: str | None = None,
+) -> dict[str, Any]:
+    """Apply one reviewed inventory manifest to one memory base.
+
+    Every preflight check runs before any write: manifest schema and layer
+    direction, the database label, the exact-manifest confirmation token, the
+    probe-snapshot fingerprint, candidate ids against ``epi_nodes``, and a
+    non-empty move set. Only ``decision="approve"`` candidates move
+    by default; ``--approve-confidence`` additionally promotes
+    ``decision="review"`` candidates at or above that confidence. A candidate
+    with ``decision="exclude"`` never moves.
+
+    An ephemeral read-only probe snapshot is used before the transaction only
+    for fingerprint and candidate preflight. The authoritative SQLite snapshot
+    is written to ``snapshot_dir`` only after the apply connection acquires
+    ``BEGIN IMMEDIATE``; ``snapshot_dir`` must live outside the data directory.
+    Its hash is verified against the confirmed
+    manifest while that same write transaction is held, so a writer between the
+    probe and the transaction cannot slip through unnoticed. All moves happen in
+    that transaction and are rolled back on any rowcount or postcondition
+    failure. A base that cannot be locked within ``APPLY_BUSY_TIMEOUT_MS``
+    raises a clean ``ApplyError`` before a snapshot exists. The report is fsynced
+    to a unique recovery file before commit, then atomically installed at
+    ``<manifest>.apply-report.json`` unless ``report`` overrides it. SQLite and
+    the filesystem cannot commit atomically together: if final replacement fails
+    after commit, the complete recovery report remains at the path named in the
+    raised ``ApplyError``.
+
+    The confirmation token is ``MIGRATE-<database>-<manifest_sha256[:12]>`` over
+    the exact reviewed manifest bytes. Edit decisions, inspect that exact file,
+    then compute its token and apply; no token is embedded in the manifest.
+    """
+    if approve_confidence not in APPROVE_CONFIDENCE_CHOICES:
+        raise ApplyError(f"unknown approve-confidence: {approve_confidence}")
+
+    db_path = _database_path(data_dir).resolve()
+    if not db_path.is_file():
+        raise FileNotFoundError(f"no database at {db_path}")
+
+    report_target = report if report is not None else _default_report_path(manifest_path)
+    _, report_path = _resolve_output_paths(data_dir, report_target)
+    resolved_manifest = manifest_path.resolve()
+    if report_path.resolve() == resolved_manifest or _same_file_if_present(report_path, manifest_path):
+        raise ApplyError(f"report path {report_path} aliases the reviewed manifest {manifest_path}")
+
+    if snapshot_dir.resolve().is_relative_to(data_dir.resolve()):
+        raise ApplyError(f"refusing snapshot directory {snapshot_dir} inside the data directory {data_dir}: snapshots belong outside the live base")
+
+    try:
+        manifest_bytes = manifest_path.read_bytes()
+        manifest = json.loads(manifest_bytes)
+    except ValueError as exc:
+        raise ApplyError(f"manifest {manifest_path} is not valid JSON: {exc}") from exc
+    if not isinstance(manifest, dict):
+        raise ApplyError("manifest must be a JSON object")
+    if manifest.get("schema") != MANIFEST_SCHEMA:
+        raise ApplyError(f"manifest schema {manifest.get('schema')!r} is not {MANIFEST_SCHEMA}")
+    if manifest.get("source_layer") != "user" or manifest.get("target_layer") != "agent":
+        raise ApplyError("manifest must describe a user -> agent migration")
+
+    label = _database_label(data_dir, database)
+    if manifest.get("database") != label:
+        raise ApplyError(f"manifest database {manifest.get('database')!r} does not match resolved database {label!r}")
+
+    fingerprint = manifest.get("db_fingerprint")
+    if not isinstance(fingerprint, dict) or not isinstance(fingerprint.get("sha256"), str):
+        raise ApplyError("manifest is missing db_fingerprint.sha256")
+    expected_sha = str(fingerprint["sha256"])
+
+    manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest()
+    expected_token = _confirm_token(label, manifest_sha256)
+    if confirm != expected_token:
+        raise ApplyError(f"confirmation token does not match the reviewed manifest; expected {expected_token}")
+
+    with tempfile.TemporaryDirectory(prefix="migrate-graph-apply-") as temp_dir:
+        probe_path = Path(temp_dir) / "probe.db"
+        _snapshot_to(db_path, probe_path)
+        with closing(sqlite3.connect(probe_path)) as probe_conn:
+            probe_conn.row_factory = sqlite3.Row
+            _validate_schema(probe_conn)
+            # Candidates are checked before the fingerprint so a re-run reports
+            # the moved nodes rather than only that the database changed.
+            candidates = _validated_candidates(probe_conn, manifest)
+        if _file_sha256(probe_path) != expected_sha:
+            raise ApplyError("current database fingerprint does not match the manifest")
+
+    moved = [candidate for candidate in candidates if _should_move(candidate, approve_confidence)]
+    if not moved:
+        raise ApplyError("no candidates to move: no approve decisions and no review candidates at the requested confidence")
+    moved_by_id = {int(candidate["node_id"]): candidate for candidate in moved}
+    unmoved = [candidate for candidate in candidates if int(candidate["node_id"]) not in moved_by_id]
+    moved_ids = [int(candidate["node_id"]) for candidate in moved]
+
+    report_fields: dict[str, Any] = {
+        "schema": MANIFEST_SCHEMA,
+        "command": "apply",
+        "database": label,
+        "manifest": str(manifest_path),
+        "manifest_sha256": manifest_sha256,
+        "snapshot_taken_under_write_lock": True,
+        "snapshot_verified_under_write_lock": True,
+        "fingerprint_verified_pre_transaction": True,
+        "fingerprint": {
+            "confirmed_sha256": expected_sha,
+            "pre_transaction_sha256": expected_sha,
+        },
+        "approval_policy": {
+            "approve_confidence": approve_confidence,
+            "moved_decisions": sorted({str(candidate["decision"]) for candidate in moved}),
+        },
+        "moved_node_ids": moved_ids,
+        "moved_count": len(moved_ids),
+        "excluded_count": sum(str(candidate["decision"]) == "exclude" for candidate in candidates),
+        "skipped_count": sum(str(candidate["decision"]) == "review" and int(candidate["node_id"]) not in moved_by_id for candidate in candidates),
+    }
+
+    snapshot_name = f"{label}-{_utc_timestamp()}-{expected_sha[:12]}.db"
+    snapshot_path = _resolve_snapshot_path(data_dir, snapshot_dir, snapshot_name)
+    if report_path.resolve() == snapshot_path.resolve() or _same_file_if_present(report_path, snapshot_path):
+        raise ApplyError(f"report path {report_path} aliases the predicted snapshot {snapshot_path}")
+    _ensure_snapshot_directory(snapshot_dir)
+    _, _, _, _, _, report_payload, report_temp = _apply_transaction(
+        db_path, moved, unmoved, moved_ids, snapshot_path, expected_sha, report_path, report_fields, manifest_sha256
+    )
+    try:
+        os.replace(report_temp, report_path)
+        _fsync_directory(report_path.parent)
+    except OSError as exc:
+        recovery_path = report_path if report_path.exists() and not report_temp.exists() else report_temp
+        raise ApplyError(
+            f"database migration committed but final report installation failed; complete recovery report retained at {recovery_path}: {exc}"
+        ) from exc
+    return report_payload
+
+
 def _parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description=__doc__)
     subparsers = parser.add_subparsers(dest="command", required=True)
-    inventory_parser = subparsers.add_parser("inventory", help="write a read-only migration inventory")
+    inventory_parser = subparsers.add_parser("inventory", help="write a read-only migration inventory for review")
     inventory_parser.add_argument("--data-dir", required=True, type=Path)
     inventory_parser.add_argument("--output", required=True, type=Path)
     inventory_parser.add_argument("--database", type=str)
@@ -515,26 +1173,49 @@ def _parser() -> argparse.ArgumentParser:
     inventory_parser.add_argument("--marker", action="append", dest="markers", metavar="TEXT")
     inventory_parser.add_argument("--exclude-file", type=Path)
     inventory_parser.add_argument("--exclude", action="append", dest="excludes", metavar="TEXT")
+
+    apply_parser = subparsers.add_parser("apply", help="apply a reviewed migration manifest")
+    apply_parser.add_argument("--data-dir", required=True, type=Path)
+    apply_parser.add_argument("--manifest", required=True, type=Path)
+    apply_parser.add_argument("--snapshot-dir", required=True, type=Path)
+    apply_parser.add_argument("--confirm", required=True)
+    apply_parser.add_argument("--approve-confidence", choices=list(APPROVE_CONFIDENCE_CHOICES), default="none")
+    apply_parser.add_argument("--database", type=str)
+    apply_parser.add_argument("--report", type=Path)
+
+    token_parser = subparsers.add_parser("confirm-token", help="print the confirmation token for an exact reviewed manifest")
+    token_parser.add_argument("--manifest", required=True, type=Path)
     return parser
 
 
 def main(argv: list[str] | None = None) -> int:
     args = _parser().parse_args(argv)
-    if args.command != "inventory":
-        raise AssertionError(f"unsupported command: {args.command}")
     try:
-        markers, marker_source = _resolve_markers(args.markers_file, args.markers)
-        excludes, exclude_source = _resolve_excludes(args.exclude_file, args.excludes)
-        write_inventory(
-            args.data_dir,
-            args.output,
-            args.database,
-            markers=markers,
-            marker_source=marker_source,
-            excludes=excludes,
-            exclude_source=exclude_source,
-        )
-    except (FileNotFoundError, OSError, SchemaError, sqlite3.Error) as exc:
+        if args.command == "inventory":
+            markers, marker_source = _resolve_markers(args.markers_file, args.markers)
+            excludes, exclude_source = _resolve_excludes(args.exclude_file, args.excludes)
+            write_inventory(
+                args.data_dir,
+                args.output,
+                args.database,
+                markers=markers,
+                marker_source=marker_source,
+                excludes=excludes,
+                exclude_source=exclude_source,
+            )
+        elif args.command == "confirm-token":
+            print(_confirm_token_from_manifest(args.manifest))
+        else:
+            apply(
+                args.data_dir,
+                args.manifest,
+                args.snapshot_dir,
+                confirm=args.confirm,
+                approve_confidence=args.approve_confidence,
+                report=args.report,
+                database=args.database,
+            )
+    except (FileNotFoundError, OSError, SchemaError, ApplyError, sqlite3.Error) as exc:
         print(str(exc), file=sys.stderr)
         return 2
     return 0
diff --git a/tests/test_scripts/test_migrate_graph_layers.py b/tests/test_scripts/test_migrate_graph_layers.py
index 267fc84..594354c 100644
--- a/tests/test_scripts/test_migrate_graph_layers.py
+++ b/tests/test_scripts/test_migrate_graph_layers.py
@@ -4,11 +4,15 @@
 
 import hashlib
 import json
+import os
+import shutil
 import sqlite3
+import stat
 import subprocess
 import sys
+import tempfile
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import pytest
 
@@ -436,7 +440,7 @@ def swap_then_replace(source: str | PathLike[str], destination: str | PathLike[s
     assert _db_hash(db) == before_hash
 
 
-def test_apply_command_is_rejected_without_writing(fixture_base: Path) -> None:
+def test_apply_without_required_arguments_is_rejected_without_writing(fixture_base: Path) -> None:
     db = fixture_base / "memory.db"
     output = fixture_base / "manifest.json"
     before_hash = _db_hash(db)
@@ -448,8 +452,6 @@ def test_apply_command_is_rejected_without_writing(fixture_base: Path) -> None:
             "apply",
             "--data-dir",
             str(fixture_base),
-            "--output",
-            str(output),
         ],
         cwd=REPO_ROOT,
         capture_output=True,
@@ -529,7 +531,37 @@ def test_database_label_strips_memory_base_prefix(tmp_path: Path) -> None:
     assert _inventory(data_dir)["database"] == "demo"
 
 
-def test_inventory_accepts_database_path_with_uri_syntax_characters(tmp_path: Path) -> None:
+@pytest.mark.parametrize("label", ["hermes", "mimocode", "cowagent", "valid.derived_label-1"])
+def test_database_label_accepts_safe_explicit_and_derived_components(tmp_path: Path, label: str) -> None:
+    data_dir = tmp_path / (f".mcp-ariel-memory-{label}" if label != "valid.derived_label-1" else label)
+    data_dir.mkdir()
+    with sqlite3.connect(data_dir / "memory.db") as conn:
+        conn.executescript(NODE_DDL + EDGE_DDL)
+
+    assert migration._database_label(data_dir, None) == label
+    assert migration._database_label(data_dir, label) == label
+
+
+@pytest.mark.parametrize("label", ["../escaped", "/absolute", "nested/name", "", ".", "..", "has..dots"])
+def test_database_label_rejects_path_escape_components(tmp_path: Path, label: str) -> None:
+    data_dir = tmp_path / "label-base"
+    data_dir.mkdir()
+    with sqlite3.connect(data_dir / "memory.db") as conn:
+        conn.executescript(NODE_DDL + EDGE_DDL)
+
+    with pytest.raises(migration.SchemaError, match="database label"):
+        migration._database_label(data_dir, label)
+
+
+def test_derived_parent_directory_label_is_rejected(tmp_path: Path) -> None:
+    data_dir = tmp_path / ".mcp-ariel-memory-.."
+    data_dir.mkdir()
+
+    with pytest.raises(migration.SchemaError, match="database label"):
+        migration._database_label(data_dir, None)
+
+
+def test_inventory_rejects_database_label_with_uri_syntax_characters(tmp_path: Path) -> None:
     data_dir = tmp_path / "inventory?#"
     try:
         data_dir.mkdir()
@@ -544,11 +576,8 @@ def test_inventory_accepts_database_path_with_uri_syntax_characters(tmp_path: Pa
             ("user", "default", "alpha_marker in a special path", "fact", "[]", 0.8, 1.0),
         )
 
-    manifest = _inventory(data_dir)
-
-    assert manifest["summary"]["layer_counts"] == {"user": 1}
-    assert manifest["candidates"][0]["node_id"] == 1
-    assert manifest["candidates"][0]["reasons"] == ["marker:alpha_marker"]
+    with pytest.raises(migration.SchemaError, match="database label"):
+        _inventory(data_dir)
 
 
 def _run_inventory_cli(base: Path, output: Path, *extra: str) -> subprocess.CompletedProcess[str]:
@@ -772,7 +801,7 @@ def test_summary_excluded_counts_excluded_candidates(exclusion_base: Path) -> No
     assert sum(item["decision"] == "exclude" for item in manifest["candidates"]) == 2
 
 
-def test_exclusion_keeps_inventory_read_only_deterministic_and_apply_rejected(exclusion_base: Path) -> None:
+def test_exclusion_keeps_inventory_read_only_deterministic_and_unconfirmed_apply_rejected(exclusion_base: Path) -> None:
     """Kills: exclusion bypassing the snapshot/read-only guarantees or enabling apply."""
     db = exclusion_base / "memory.db"
     before_hash = _db_hash(db)
@@ -803,8 +832,6 @@ def test_exclusion_keeps_inventory_read_only_deterministic_and_apply_rejected(ex
             "apply",
             "--data-dir",
             str(exclusion_base),
-            "--output",
-            str(output),
         ],
         cwd=REPO_ROOT,
         capture_output=True,
@@ -813,3 +840,1428 @@ def test_exclusion_keeps_inventory_read_only_deterministic_and_apply_rejected(ex
     )
     assert apply_result.returncode != 0
     assert _db_hash(db) == before_hash
+
+
+APPLY_MARKERS = ("alpha_marker", "beta_marker")
+
+
+def _write_apply_db(base: Path) -> None:
+    base.mkdir()
+    db = base / "memory.db"
+    with sqlite3.connect(db) as conn:
+        conn.executescript(NODE_DDL + EDGE_DDL)
+        conn.executemany(
+            "INSERT INTO epi_nodes (node_id, layer, user_id, content, node_type, tags, confidence, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            [
+                (1, "user", "default", "alpha_marker note", "fact", json.dumps(["user_preference"]), 0.8, 1.0),
+                (2, "user", "default", "beta_marker rule", "personality_trait", json.dumps(["personality_trait"]), 0.9, 2.0),
+                (3, "agent", "default", "agent persona", "personality_trait", json.dumps(["personality_trait"]), 0.9, 3.0),
+                (4, "user", "other", "ordinary note", "fact", "[]", 0.4, 4.0),
+                (5, "user", "default", "alpha_marker excluded note", "fact", "[]", 0.5, 5.0),
+                (6, "agent", "default", "agent fact", "fact", "[]", 0.5, 6.0),
+                (7, "user", "default", "alpha_marker low note", "fact", "[]", 0.5, 7.0),
+            ],
+        )
+        conn.executemany(
+            "INSERT INTO epi_edges (source_id, target_id, relation, weight, created_at, tags) VALUES (?, ?, ?, ?, ?, ?)",
+            [
+                (1, 4, "mentions", 0.5, 10.0, "[]"),
+                (2, 3, "decided_because", 0.8, 11.0, json.dumps(["agent:decision"])),
+                (4, 1, "mentions", 0.4, 12.0, "[]"),
+                (3, 6, "mentions", 0.5, 13.0, "[]"),
+                (2, 4, "mentions", 0.3, 14.0, "[]"),
+            ],
+        )
+
+
+@pytest.fixture
+def apply_base(tmp_path: Path) -> Generator[Path, None, None]:
+    base = tmp_path / "applybase"
+    _write_apply_db(base)
+    yield base
+
+
+def _manifest_for(
+    base: Path,
+    *,
+    decisions: dict[int, str] | None = None,
+    confidences: dict[int, str] | None = None,
+    path: Path | None = None,
+) -> Path:
+    manifest = migration.inventory(base, markers=list(APPLY_MARKERS))
+    for candidate in manifest["candidates"]:
+        node_id = int(candidate["node_id"])
+        if decisions is not None and node_id in decisions:
+            candidate["decision"] = decisions[node_id]
+        if confidences is not None and node_id in confidences:
+            candidate["confidence"] = confidences[node_id]
+    target = path if path is not None else base / "manifest.json"
+    target.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    return target
+
+
+def _expected_token(manifest_path: Path) -> str:
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest_sha = hashlib.sha256(manifest_path.read_bytes()).hexdigest()
+    return f"MIGRATE-{manifest['database']}-{manifest_sha[:12]}"
+
+
+def _layers(db: Path) -> dict[int, str]:
+    with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as conn:
+        return {int(row[0]): str(row[1]) for row in conn.execute("SELECT node_id, layer FROM epi_nodes ORDER BY node_id")}
+
+
+def _layer_counts(db: Path) -> dict[str, int]:
+    with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as conn:
+        return {str(row[0]): int(row[1]) for row in conn.execute("SELECT layer, COUNT(*) FROM epi_nodes GROUP BY layer")}
+
+
+def _node_rows(db: Path) -> list[tuple[Any, ...]]:
+    with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as conn:
+        rows = conn.execute(
+            "SELECT node_id, layer, user_id, content, node_type, tags, confidence, created_at FROM epi_nodes ORDER BY node_id"
+        ).fetchall()
+    return [tuple(row) for row in rows]
+
+
+def _edge_rows(db: Path) -> list[tuple[Any, ...]]:
+    with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as conn:
+        rows = conn.execute(
+            "SELECT source_id, target_id, relation, weight, created_at, tags FROM epi_edges ORDER BY source_id, target_id, relation"
+        ).fetchall()
+    return [tuple(row) for row in rows]
+
+
+def test_apply_refuses_when_nothing_is_approved(apply_base: Path) -> None:
+    """Kills: moving review candidates without an explicit confidence promotion."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base)
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match="no candidates to move"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+
+
+@pytest.mark.parametrize(
+    ("approve_confidence", "expected_moved"),
+    [("medium", [1, 2]), ("low", [1, 2, 7])],
+    ids=["medium", "low"],
+)
+def test_apply_promotes_review_candidates_at_threshold_and_never_excludes(
+    apply_base: Path, approve_confidence: str, expected_moved: list[int]
+) -> None:
+    """Kills: promoting below-threshold review rows, or ever moving an exclude decision."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(
+        apply_base,
+        decisions={5: "exclude"},
+        confidences={1: "medium", 2: "high", 5: "low", 7: "low"},
+    )
+
+    report = migration.apply(
+        apply_base,
+        manifest,
+        apply_base.parent / "snapshots",
+        confirm=_expected_token(manifest),
+        approve_confidence=approve_confidence,
+    )
+
+    assert report["moved_node_ids"] == expected_moved
+    layers = _layers(db)
+    assert all(layers[node_id] == "agent" for node_id in expected_moved)
+    assert layers[5] == "user"
+    if approve_confidence == "medium":
+        assert layers[7] == "user"
+    assert report["excluded_count"] == 1
+
+
+def test_apply_moves_exactly_approved_ids_with_exact_count_delta(apply_base: Path) -> None:
+    """Kills: moving an extra row, dropping a row, or skipping the postcondition report."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve", 2: "approve"})
+    before_counts = _layer_counts(db)
+
+    report = migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert report["moved_node_ids"] == [1, 2]
+    assert report["moved_count"] == 2
+    assert report["before"]["layer_counts"] == before_counts
+    after_counts = _layer_counts(db)
+    assert after_counts["user"] == before_counts["user"] - 2
+    assert after_counts["agent"] == before_counts["agent"] + 2
+    assert sum(after_counts.values()) == sum(before_counts.values())
+    assert report["postconditions"]
+    assert all(report["postconditions"].values())
+    assert report["success"] is True
+
+
+def test_fingerprint_mismatch_refuses_before_any_write(apply_base: Path) -> None:
+    """Kills: trusting the manifest fingerprint instead of a fresh consistent snapshot."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    with sqlite3.connect(db) as conn:
+        conn.execute(
+            "INSERT INTO epi_nodes (layer, user_id, content, node_type, tags, confidence, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
+            ("user", "default", "late arrival", "fact", "[]", 0.5, 20.0),
+        )
+    before_hash = _db_hash(db)
+    snapshot_dir = apply_base.parent / "snapshots"
+
+    with pytest.raises(migration.ApplyError, match="fingerprint"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert not snapshot_dir.exists() or not any(snapshot_dir.iterdir())
+
+
+@pytest.mark.parametrize("corruption", ["unknown", "duplicate"])
+def test_unknown_or_duplicate_candidate_is_refused(apply_base: Path, corruption: str) -> None:
+    """Kills: applying a manifest whose candidate set drifted or repeats a node."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    data = json.loads(manifest.read_text(encoding="utf-8"))
+    if corruption == "unknown":
+        data["candidates"][0]["node_id"] = 999
+        expected = "does not exist"
+    else:
+        data["candidates"][1]["node_id"] = data["candidates"][0]["node_id"]
+        expected = "duplicate candidate"
+    manifest.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match=expected):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+
+
+def test_confirm_token_is_required_and_exact(apply_base: Path) -> None:
+    """Kills: accepting any token, or deriving it from something other than the manifest."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match="confirmation token"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm="MIGRATE-wrong-token")
+
+    assert _db_hash(db) == before_hash
+
+    report = migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+    assert report["moved_node_ids"] == [1]
+
+
+@pytest.mark.parametrize("change", ["decision", "user_id", "candidate_id"])
+def test_manifest_edits_invalidate_previous_confirmation_token(apply_base: Path, change: str) -> None:
+    """The approval token binds exact manifest bytes, separately from DB freshness."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    old_token = _expected_token(manifest)
+    data = _manifest_data(manifest)
+    candidate = next(candidate for candidate in data["candidates"] if candidate["node_id"] == 1)
+    if change == "decision":
+        candidate["decision"] = "review"
+    elif change == "user_id":
+        candidate["user_id"] = "impostor"
+    else:
+        candidate["node_id"] = 999
+    _rewrite_manifest(manifest, data)
+    before_hash = _db_hash(db)
+    snapshot_dir = apply_base.parent / "snapshots"
+
+    with pytest.raises(migration.ApplyError, match="confirmation token"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=old_token)
+
+    assert _db_hash(db) == before_hash
+    assert not snapshot_dir.exists()
+
+    refreshed_token = _expected_token(manifest)
+    if change != "decision":
+        expected_error = "belongs to user" if change == "user_id" else "does not exist"
+        with pytest.raises(migration.ApplyError, match=expected_error):
+            migration.apply(apply_base, manifest, snapshot_dir, confirm=refreshed_token)
+        assert _db_hash(db) == before_hash
+        assert not snapshot_dir.exists()
+        return
+
+    report = migration.apply(apply_base, manifest, snapshot_dir, confirm=refreshed_token, approve_confidence="low")
+    assert 1 in report["moved_node_ids"]
+
+
+def test_report_temp_write_failure_rolls_back_before_commit(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """A report write/fsync failure must abort the still-open migration transaction."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    before_hash = _db_hash(db)
+    before_layers = _layers(db)
+
+    def fail_report_temp(*args: Any, **kwargs: Any) -> Path:
+        raise OSError("forced report temp write failure")
+
+    monkeypatch.setattr(migration, "_write_report_temp", fail_report_temp, raising=False)
+
+    with pytest.raises(OSError, match="forced report temp write failure"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db) == before_layers
+
+
+def test_final_report_replace_failure_keeps_recovery_report(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """A post-commit replace failure leaves a complete named report with moved ids."""
+    manifest = _manifest_for(apply_base, decisions={1: "approve", 2: "approve"})
+    report_path = manifest.with_name(f"{manifest.name}.apply-report.json")
+    real_replace = migration.os.replace
+
+    def fail_final_replace(source: str | PathLike[str], destination: str | PathLike[str]) -> None:
+        if Path(destination) == report_path:
+            raise OSError("forced final report replace failure")
+        real_replace(source, destination)
+
+    monkeypatch.setattr(migration.os, "replace", fail_final_replace)
+
+    with pytest.raises(migration.ApplyError, match=r"recovery report.*\.tmp") as excinfo:
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    recovery_paths = list(manifest.parent.glob(f".{report_path.name}.recovery-*.tmp"))
+    assert len(recovery_paths) == 1
+    recovery_path = recovery_paths[0]
+    assert str(recovery_path) in str(excinfo.value)
+    assert recovery_path.is_file()
+    assert json.loads(recovery_path.read_text(encoding="utf-8"))["moved_node_ids"] == [1, 2]
+    assert _layers(apply_base / "memory.db")[1] == "agent"
+    assert _layers(apply_base / "memory.db")[2] == "agent"
+
+
+def test_apply_preserves_unmanifested_node_layer(apply_base: Path) -> None:
+    """Normal apply reports and preserves the layer of nodes absent from the manifest."""
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+
+    report = migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert report["postconditions"]["all_unapproved_nodes_unchanged"] is True
+    assert _layers(apply_base / "memory.db")[4] == "user"
+
+
+def test_unmanifested_node_layer_mutation_rolls_back(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """The transaction guard catches layer mutations to nodes absent from candidates."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    before_hash = _db_hash(db)
+    before_layers = _layers(db)
+    real_move = migration._move_node
+
+    def mutate_unmanifested(conn: sqlite3.Connection, node_id: int, user_id: str) -> int:
+        result = real_move(conn, node_id, user_id)
+        conn.execute("UPDATE epi_nodes SET layer = 'agent' WHERE node_id = 4")
+        return result
+
+    monkeypatch.setattr(migration, "_move_node", mutate_unmanifested)
+
+    with pytest.raises(migration.ApplyError, match="all_unapproved_nodes_unchanged"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db) == before_layers
+
+
+def test_snapshot_matches_report_and_restores_pre_apply_state(apply_base: Path) -> None:
+    """Kills: reporting a snapshot that is missing, mis-hashed, or taken after the move."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve", 2: "approve"})
+    before_nodes = _node_rows(db)
+    before_edges = _edge_rows(db)
+
+    report = migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    snapshot = Path(report["snapshot"]["path"])
+    assert snapshot.is_file()
+    assert _db_hash(snapshot) == report["snapshot"]["sha256"]
+    restored = apply_base / "restored.db"
+    shutil.copyfile(snapshot, restored)
+    assert _node_rows(restored) == before_nodes
+    assert _edge_rows(restored) == before_edges
+
+
+def test_forced_mid_move_failure_rolls_back_everything(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Kills: committing a partial move when one UPDATE matches no row."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve", 2: "approve"})
+    before_hash = _db_hash(db)
+    before_nodes = _node_rows(db)
+    before_edges = _edge_rows(db)
+    real_move = migration._move_node
+
+    def flaky_move(conn: sqlite3.Connection, node_id: int, user_id: str) -> int:
+        if node_id == 2:
+            return 0
+        return real_move(conn, node_id, user_id)
+
+    monkeypatch.setattr(migration, "_move_node", flaky_move)
+
+    with pytest.raises(migration.ApplyError, match="exactly one user row"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _node_rows(db) == before_nodes
+    assert _edge_rows(db) == before_edges
+
+
+@pytest.mark.parametrize("alias_kind", ["symlink", "hardlink"])
+def test_snapshot_install_race_cannot_replace_or_write_through_alias(apply_base: Path, monkeypatch: pytest.MonkeyPatch, alias_kind: str) -> None:
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    monkeypatch.setattr(migration, "_utc_timestamp", lambda: "20260101T000000Z")
+    data = json.loads(manifest.read_text(encoding="utf-8"))
+    snapshot_dir = apply_base.parent / "snapshots"
+    final_path = snapshot_dir / f"{data['database']}-20260101T000000Z-{data['db_fingerprint']['sha256'][:12]}.db"
+    target = apply_base.parent / "attacker-target.db"
+    target.write_bytes(b"must remain untouched")
+
+    def replace_at_install(path: Path) -> None:
+        assert path == final_path
+        if alias_kind == "symlink":
+            final_path.symlink_to(target)
+        else:
+            final_path.hardlink_to(target)
+
+    monkeypatch.setattr(migration, "_before_snapshot_install", replace_at_install, raising=False)
+    with pytest.raises(migration.ApplyError, match="existing snapshot"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert target.read_bytes() == b"must remain untouched"
+    assert _layers(db)[1] == "user"
+    assert final_path.is_symlink() if alias_kind == "symlink" else final_path.samefile(target)
+    assert not list(snapshot_dir.glob("*.tmp"))
+
+
+def test_existing_snapshot_is_never_overwritten_on_retry(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    manifest = _manifest_for(apply_base, decisions={1: "approve", 2: "approve"})
+    snapshot_dir = apply_base.parent / "snapshots"
+    monkeypatch.setattr(migration, "_utc_timestamp", lambda: "20260101T000000Z")
+    real_move = migration._move_node
+
+    def fail_second_move(conn: sqlite3.Connection, node_id: int, user_id: str) -> int:
+        if node_id == 2:
+            return 0
+        return real_move(conn, node_id, user_id)
+
+    monkeypatch.setattr(migration, "_move_node", fail_second_move)
+    with pytest.raises(migration.ApplyError, match="exactly one user row"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+    snapshot = next(snapshot_dir.glob("*.db"), None)
+    assert snapshot is not None
+    snapshot_bytes = snapshot.read_bytes()
+    monkeypatch.setattr(migration, "_move_node", real_move)
+
+    with pytest.raises(migration.ApplyError, match="existing snapshot"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert snapshot.read_bytes() == snapshot_bytes
+    assert _db_hash(snapshot) == json.loads(manifest.read_text(encoding="utf-8"))["db_fingerprint"]["sha256"]
+
+
+def test_new_snapshot_directories_are_fsynced_before_commit(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    snapshot_dir = apply_base.parent / "nested" / "snapshots"
+    synced: list[Path] = []
+    real_fsync_directory = migration._fsync_directory
+    real_commit = migration._commit_transaction
+
+    def record_fsync(path: Path) -> None:
+        synced.append(path)
+        real_fsync_directory(path)
+
+    def assert_durable_then_commit(conn: sqlite3.Connection) -> None:
+        assert snapshot_dir in synced
+        assert snapshot_dir.parent in synced
+        assert snapshot_dir.parent.parent in synced
+        real_commit(conn)
+
+    monkeypatch.setattr(migration, "_fsync_directory", record_fsync)
+    monkeypatch.setattr(migration, "_commit_transaction", assert_durable_then_commit)
+
+    migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert snapshot_dir.is_dir()
+    assert snapshot_dir.parent in synced
+
+
+def test_rollback_failure_preserves_original_error_snapshot_and_cleans_report_temp(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve", 2: "approve"})
+    snapshot_dir = apply_base.parent / "snapshots"
+    report_path = manifest.with_name(f"{manifest.name}.apply-report.json")
+    monkeypatch.setattr(migration, "_utc_timestamp", lambda: "20260101T000000Z")
+    real_move = migration._move_node
+
+    def fail_move(conn: sqlite3.Connection, node_id: int, user_id: str) -> int:
+        if node_id == 2:
+            return 0
+        return real_move(conn, node_id, user_id)
+
+    monkeypatch.setattr(migration, "_move_node", fail_move)
+    real_connect = migration.sqlite3.connect
+
+    class RollbackFailConnection:
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            self.connection = real_connect(*args, **kwargs)
+
+        def __enter__(self) -> Any:
+            return self.connection.__enter__()
+
+        def __exit__(self, *args: object) -> Any:
+            return self.connection.__exit__(*args)
+
+        def __getattr__(self, name: str) -> Any:
+            return getattr(self.connection, name)
+
+        @property
+        def row_factory(self) -> Any:
+            return self.connection.row_factory
+
+        @row_factory.setter
+        def row_factory(self, value: Any) -> None:
+            self.connection.row_factory = value
+
+        def execute(self, sql: str, *args: Any, **kwargs: Any) -> Any:
+            if sql == "ROLLBACK":
+                raise sqlite3.OperationalError("forced rollback failure")
+            return self.connection.execute(sql, *args, **kwargs)
+
+    def connect_with_rollback_failure(*args: Any, **kwargs: Any) -> Any:
+        if args and str(db) in str(args[0]):
+            return RollbackFailConnection(*args, **kwargs)
+        return real_connect(*args, **kwargs)
+
+    monkeypatch.setattr(migration.sqlite3, "connect", connect_with_rollback_failure)
+    with pytest.raises(migration.ApplyError) as excinfo:
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    snapshot = next(snapshot_dir.glob("*.db"), None)
+    assert snapshot is not None, str(excinfo.value)
+    message = str(excinfo.value)
+    assert "expected exactly one user row" in message
+    assert "forced rollback failure" in message
+    assert str(snapshot) in message
+    assert _db_hash(snapshot) == json.loads(manifest.read_text(encoding="utf-8"))["db_fingerprint"]["sha256"]
+    assert not list(manifest.parent.glob(f".{report_path.name}.recovery-*.tmp"))
+
+
+def test_cross_layer_edges_are_reported_and_edges_are_untouched(apply_base: Path) -> None:
+    """Kills: deleting or rewriting edges, or omitting the post-move cross-layer view."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve", 2: "approve"})
+    before_edges = _edge_rows(db)
+
+    report = migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _edge_rows(db) == before_edges
+    cross = report["cross_layer_edges"]
+    assert cross["count"] == 3
+    keys = {(item["source_id"], item["target_id"], item["relation"]) for item in cross["sample"]}
+    assert keys == {(1, 4, "mentions"), (4, 1, "mentions"), (2, 4, "mentions")}
+
+
+def test_second_apply_of_same_manifest_refuses(apply_base: Path) -> None:
+    """Kills: half-applying or silently no-op'ing on a re-run."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve", 2: "approve"})
+    migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+    after_hash = _db_hash(db)
+    after_layers = _layers(db)
+
+    with pytest.raises(migration.ApplyError, match="no longer in layer 'user'"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == after_hash
+    assert _layers(db) == after_layers
+
+
+def test_snapshot_path_aliasing_database_is_refused(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Kills: writing the pre-move snapshot through a hard link onto the live database."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    monkeypatch.setattr(migration, "_utc_timestamp", lambda: "20260101T000000Z")
+    data = json.loads(manifest.read_text(encoding="utf-8"))
+    sha = data["db_fingerprint"]["sha256"]
+    snapshot_dir = apply_base.parent / "snapshots"
+    snapshot_dir.mkdir()
+    alias = snapshot_dir / f"{data['database']}-20260101T000000Z-{sha[:12]}.db"
+    alias.hardlink_to(db)
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.SchemaError, match="aliasing database"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _db_hash(alias) == before_hash
+
+
+def test_apply_refuses_report_equal_to_manifest_before_writes(apply_base: Path) -> None:
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    manifest_bytes = manifest.read_bytes()
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match=r"report.*manifest"):
+        migration.apply(
+            apply_base,
+            manifest,
+            apply_base.parent / "snapshots",
+            confirm=_expected_token(manifest),
+            report=manifest,
+        )
+
+    assert manifest.read_bytes() == manifest_bytes
+    assert _db_hash(db) == before_hash
+    assert not (apply_base.parent / "snapshots").exists()
+
+
+@pytest.mark.parametrize("alias_kind", ["hardlink", "symlink"])
+def test_apply_refuses_report_alias_to_manifest(apply_base: Path, alias_kind: str) -> None:
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    report = apply_base / f"{alias_kind}-report.json"
+    try:
+        if alias_kind == "hardlink":
+            os.link(manifest, report)
+        else:
+            report.symlink_to(manifest)
+    except (NotImplementedError, OSError) as exc:
+        if alias_kind == "symlink":
+            pytest.skip(f"symlink creation unsupported: {exc}")
+        raise
+    before_hash = _db_hash(db)
+    manifest_bytes = manifest.read_bytes()
+
+    with pytest.raises(migration.ApplyError, match=r"report.*manifest"):
+        migration.apply(
+            apply_base,
+            manifest,
+            apply_base.parent / "snapshots",
+            confirm=_expected_token(manifest),
+            report=report,
+        )
+
+    assert _db_hash(db) == before_hash
+    assert manifest.read_bytes() == manifest_bytes
+    assert not (apply_base.parent / "snapshots").exists()
+
+
+def test_apply_refuses_report_equal_to_predicted_snapshot_path(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    sha = json.loads(manifest.read_text(encoding="utf-8"))["db_fingerprint"]["sha256"]
+    monkeypatch.setattr(migration, "_utc_timestamp", lambda: "20260101T000000Z")
+    snapshot_dir = apply_base.parent / "snapshots"
+    snapshot_dir.mkdir()
+    snapshot_path = snapshot_dir / f"applybase-20260101T000000Z-{sha[:12]}.db"
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match=r"report.*snapshot"):
+        migration.apply(
+            apply_base,
+            manifest,
+            snapshot_dir,
+            confirm=_expected_token(manifest),
+            report=snapshot_path,
+        )
+
+    assert _db_hash(db) == before_hash
+    assert not snapshot_path.exists()
+
+
+def test_report_temp_syncs_its_parent_directory_before_commit(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    synced: list[Path] = []
+    real_fsync_directory = migration._fsync_directory
+
+    def spy_fsync_directory(path: Path) -> None:
+        synced.append(path)
+        real_fsync_directory(path)
+
+    monkeypatch.setattr(migration, "_fsync_directory", spy_fsync_directory, raising=False)
+
+    migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert manifest.parent in synced
+    assert synced.index(manifest.parent) < len(synced) - 1
+
+
+def test_report_temp_directory_sync_failure_rolls_back(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    before_hash = _db_hash(db)
+
+    def fail_report_directory_sync(path: Path) -> None:
+        if path == manifest.parent:
+            raise OSError("forced report directory fsync failure")
+
+    monkeypatch.setattr(migration, "_fsync_directory", fail_report_directory_sync, raising=False)
+
+    with pytest.raises(OSError, match="forced report directory fsync failure"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+
+
+def test_post_commit_report_directory_sync_failure_keeps_report_and_snapshot(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    report_path = apply_base / "recovery-report.json"
+    real_fsync_directory = migration._fsync_directory
+
+    def fail_only_final_directory_sync(path: Path) -> None:
+        if path == report_path.parent and report_path.exists():
+            raise OSError("forced post-commit directory fsync failure")
+        real_fsync_directory(path)
+
+    monkeypatch.setattr(migration, "_fsync_directory", fail_only_final_directory_sync, raising=False)
+
+    with pytest.raises(migration.ApplyError, match="recovery report") as excinfo:
+        migration.apply(
+            apply_base,
+            manifest,
+            apply_base.parent / "snapshots",
+            confirm=_expected_token(manifest),
+            report=report_path,
+        )
+
+    assert str(report_path) in str(excinfo.value)
+    assert report_path.is_file()
+    assert json.loads(report_path.read_text(encoding="utf-8"))["moved_node_ids"] == [1]
+    assert _layers(db)[1] == "agent"
+    snapshot_paths = list((apply_base.parent / "snapshots").glob("*.db"))
+    assert len(snapshot_paths) == 1
+    assert _db_hash(snapshot_paths[0]) != _db_hash(db)
+
+
+def test_apply_writes_only_the_snapshot_and_report(apply_base: Path) -> None:
+    """Kills: stray writes outside the declared snapshot and report outputs."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    report_path = manifest.with_name(f"{manifest.name}.apply-report.json")
+    before_files = {path for path in apply_base.rglob("*") if path.is_file()}
+
+    migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    after_files = {path for path in apply_base.rglob("*") if path.is_file()}
+    snapshot_files = {path for path in (apply_base.parent / "snapshots").rglob("*") if path.is_file()}
+    assert len(snapshot_files) == 1
+    assert after_files - before_files == {report_path}
+    assert snapshot_files.isdisjoint(after_files)
+    assert _db_hash(db) != _db_hash(next(iter(snapshot_files)))
+
+
+def test_cli_apply_moves_approved_node_and_writes_report(apply_base: Path) -> None:
+    """Kills: a parser or wiring gap between the CLI flags and the apply contract."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    snapshot_dir = apply_base.parent / "snapshots"
+
+    result = subprocess.run(
+        [
+            sys.executable,
+            str(REPO_ROOT / "scripts/migrate_graph_layers.py"),
+            "apply",
+            "--data-dir",
+            str(apply_base),
+            "--manifest",
+            str(manifest),
+            "--snapshot-dir",
+            str(snapshot_dir),
+            "--confirm",
+            _expected_token(manifest),
+        ],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert _layers(db)[1] == "agent"
+    report = manifest.with_name(f"{manifest.name}.apply-report.json")
+    assert report.is_file()
+    assert json.loads(report.read_text(encoding="utf-8"))["moved_node_ids"] == [1]
+
+
+def test_cli_confirm_token_uses_exact_manifest_bytes_and_inventory_has_no_token(apply_base: Path) -> None:
+    """The reviewer computes approval after editing; inventory never stores it."""
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    manifest_bytes = manifest.read_bytes()
+    expected = f"MIGRATE-applybase-{hashlib.sha256(manifest_bytes).hexdigest()[:12]}"
+
+    result = subprocess.run(
+        [
+            sys.executable,
+            str(REPO_ROOT / "scripts/migrate_graph_layers.py"),
+            "confirm-token",
+            "--manifest",
+            str(manifest),
+        ],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert result.stdout.strip() == expected
+    assert "confirm_token" not in json.loads(manifest_bytes)
+
+
+# ---------------------------------------------------------------------------
+# Blind-verifier follow-up: every guard needs a test that goes red when the
+# guard is removed. Each docstring names that guard.
+# ---------------------------------------------------------------------------
+
+
+def _manifest_data(manifest_path: Path) -> dict[str, Any]:
+    return json.loads(manifest_path.read_text(encoding="utf-8"))
+
+
+def _rewrite_manifest(manifest_path: Path, data: dict[str, Any]) -> None:
+    manifest_path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def _candidate_split(manifest_path: Path) -> tuple[list[int], list[int]]:
+    data = _manifest_data(manifest_path)
+    moved = [int(candidate["node_id"]) for candidate in data["candidates"] if candidate["decision"] == "approve"]
+    unmoved = [int(candidate["node_id"]) for candidate in data["candidates"] if candidate["decision"] != "approve"]
+    return moved, unmoved
+
+
+def test_apply_refuses_manifest_schema_mismatch(apply_base: Path) -> None:
+    """Kills: dropping the manifest `schema` check, which would apply a schema-999 manifest."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    data = _manifest_data(manifest)
+    data["schema"] = 999
+    _rewrite_manifest(manifest, data)
+    before_hash = _db_hash(db)
+    snapshot_dir = apply_base.parent / "snapshots"
+
+    with pytest.raises(migration.ApplyError, match="schema"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+    assert not snapshot_dir.exists() or not any(snapshot_dir.iterdir())
+
+
+def test_apply_refuses_manifest_direction(apply_base: Path) -> None:
+    """Kills: dropping the user -> agent direction check (target_layer='user' would move rows)."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    data = _manifest_data(manifest)
+    data["target_layer"] = "user"
+    _rewrite_manifest(manifest, data)
+    before_hash = _db_hash(db)
+    snapshot_dir = apply_base.parent / "snapshots"
+
+    with pytest.raises(migration.ApplyError, match="user -> agent"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+    assert not snapshot_dir.exists() or not any(snapshot_dir.iterdir())
+
+
+def test_apply_refuses_manifest_from_another_base_with_identical_bytes(tmp_path: Path) -> None:
+    """Kills: dropping the database-label check.
+
+    Two bases built from the same bytes share one fingerprint sha, so the sha
+    cannot tell them apart — only the label does. Removing the label guard
+    applies the manifest to the wrong base.
+    """
+    base_a = tmp_path / "baseA"
+    _write_apply_db(base_a)
+    base_b = tmp_path / "baseB"
+    base_b.mkdir()
+    shutil.copyfile(base_a / "memory.db", base_b / "memory.db")
+
+    manifest = _manifest_for(base_a, decisions={1: "approve"})
+    manifest_sha = _manifest_data(manifest)["db_fingerprint"]["sha256"]
+    assert migration.inventory(base_b, markers=list(APPLY_MARKERS))["db_fingerprint"]["sha256"] == manifest_sha
+
+    db_b = base_b / "memory.db"
+    before_hash = _db_hash(db_b)
+    snapshot_dir = tmp_path / "snapshots"
+
+    with pytest.raises(migration.ApplyError, match="does not match resolved database"):
+        migration.apply(base_b, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert _db_hash(db_b) == before_hash
+    assert _layers(db_b)[1] == "user"
+    assert not snapshot_dir.exists() or not any(snapshot_dir.iterdir())
+
+
+def test_existing_snapshot_is_never_overwritten(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Kills: dropping the existing-snapshot guard; the sentinel file would be overwritten by a real snapshot."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    data = _manifest_data(manifest)
+    sha = data["db_fingerprint"]["sha256"]
+    monkeypatch.setattr(migration, "_utc_timestamp", lambda: "20260101T000000Z")
+    snapshot_dir = apply_base.parent / "snapshots"
+    snapshot_dir.mkdir()
+    target = snapshot_dir / f"{data['database']}-20260101T000000Z-{sha[:12]}.db"
+    sentinel = b"not a database, do not overwrite"
+    target.write_bytes(sentinel)
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match="refusing to overwrite"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert target.read_bytes() == sentinel
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+
+
+def test_snapshot_is_created_private(apply_base: Path) -> None:
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    report = migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+    snapshot = Path(report["snapshot"]["path"])
+
+    assert stat.S_IMODE(snapshot.stat().st_mode) == 0o600
+
+
+def test_snapshot_file_and_directory_fsynced_before_commit(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    db = apply_base / "memory.db"
+    events: list[tuple[str, Path | None]] = []
+    real_fsync = os.fsync
+    real_fsync_directory = migration._fsync_directory
+    real_commit = migration._commit_transaction
+
+    def observe_commit(conn: sqlite3.Connection) -> None:
+        events.append(("commit", None))
+        real_commit(conn)
+
+    def observe_fsync(fd: int) -> None:
+        descriptor_path = Path(os.readlink(f"/proc/self/fd/{fd}"))
+        events.append(("file", descriptor_path))
+        real_fsync(fd)
+
+    def observe_directory(path: Path) -> None:
+        events.append(("directory", path))
+        real_fsync_directory(path)
+
+    monkeypatch.setattr(migration.os, "fsync", observe_fsync)
+    monkeypatch.setattr(migration, "_fsync_directory", observe_directory)
+    monkeypatch.setattr(migration, "_commit_transaction", observe_commit)
+
+    report = migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+    snapshot = Path(report["snapshot"]["path"])
+    file_sync = next(index for index, (kind, path) in enumerate(events) if kind == "file" and path is not None and path.name == "snapshot.db")
+    directory_sync = next(index for index, (kind, path) in enumerate(events) if index > file_sync and kind == "directory" and path == snapshot.parent)
+    commit = next(index for index, (kind, _) in enumerate(events) if kind == "commit")
+
+    assert file_sync < directory_sync < commit
+    assert _layers(db)[1] == "agent"
+
+
+def test_concurrent_apply_same_snapshot_name_never_overwrites(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    data = _manifest_data(manifest)
+    monkeypatch.setattr(migration, "_utc_timestamp", lambda: "20260101T000000Z")
+    snapshot_dir = apply_base.parent / "snapshots"
+    snapshot_dir.mkdir()
+    target = snapshot_dir / f"{data['database']}-20260101T000000Z-{data['db_fingerprint']['sha256'][:12]}.db"
+    sentinel = b"created by competing apply"
+
+    def race_at_install(path: Path) -> None:
+        assert path == target
+        target.write_bytes(sentinel)
+
+    monkeypatch.setattr(migration, "_before_snapshot_install", race_at_install)
+
+    before_hash = _db_hash(apply_base / "memory.db")
+    with pytest.raises(migration.ApplyError):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert target.read_bytes() == sentinel
+    assert _db_hash(apply_base / "memory.db") == before_hash
+    assert _layers(apply_base / "memory.db")[1] == "user"
+
+
+def test_output_path_aliasing_nonexistent_wal_sidecar_is_refused(fixture_base: Path) -> None:
+    """Kills: dropping the resolved-path sidecar comparison.
+
+    The inode comparison only fires for an existing sidecar; a `memory.db-wal`
+    that does not exist yet is caught only by the resolved-path comparison.
+    """
+    wal = fixture_base / "memory.db-wal"
+    assert not wal.exists()
+    db = fixture_base / "memory.db"
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.SchemaError, match="sidecar"):
+        migration.write_inventory(fixture_base, wal)
+
+    assert not wal.exists()
+    assert _db_hash(db) == before_hash
+
+
+def test_apply_report_aliasing_nonexistent_wal_sidecar_is_refused(apply_base: Path) -> None:
+    """Kills: dropping the resolved-path sidecar comparison on the apply report path."""
+    db = apply_base / "memory.db"
+    wal = apply_base / "memory.db-wal"
+    assert not wal.exists()
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    before_hash = _db_hash(db)
+    snapshot_dir = apply_base.parent / "snapshots"
+
+    with pytest.raises(migration.SchemaError, match="sidecar"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest), report=wal)
+
+    assert not wal.exists()
+    assert _db_hash(db) == before_hash
+    assert not snapshot_dir.exists()
+
+
+@pytest.mark.parametrize(
+    "broken",
+    [
+        "moved_nodes_in_agent",
+        "moved_nodes_not_in_user",
+        "total_count_unchanged",
+        "user_delta_exact",
+        "agent_delta_exact",
+        "unmoved_nodes_unchanged",
+    ],
+)
+def test_each_postcondition_failure_rolls_back(apply_base: Path, monkeypatch: pytest.MonkeyPatch, broken: str) -> None:
+    """Kills: dropping the named clause from the `_run_moves` postcondition dict.
+
+    Each case perturbs only the input of the named clause; a missing clause
+    would let the (real) move commit and the assertion on the raised message
+    would fail.
+    """
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    moved_ids, unmoved_ids = _candidate_split(manifest)
+    before_hash = _db_hash(db)
+    before_nodes = _node_rows(db)
+    snapshot_dir = apply_base.parent / "snapshots"
+
+    real_counts = migration._layer_counts
+    real_layers = migration._layers_for
+    counts_calls = {"n": 0}
+
+    def fake_counts(conn: sqlite3.Connection) -> dict[str, int]:
+        counts_calls["n"] += 1
+        counts = real_counts(conn)
+        if broken == "total_count_unchanged" and counts_calls["n"] == 2:
+            return {**counts, "quarantine": 1}
+        if broken in {"user_delta_exact", "agent_delta_exact"}:
+            if counts_calls["n"] == 1:
+                return {**counts, "legacy": 1}
+            if broken == "user_delta_exact":
+                return {"user": counts.get("user", 0) + 1, "agent": counts.get("agent", 0), "legacy": 0}
+            return {"user": counts.get("user", 0), "agent": counts.get("agent", 0) - 1, "legacy": 2}
+        return counts
+
+    def fake_layers(conn: sqlite3.Connection, node_ids: list[int]) -> dict[int, str]:
+        result = real_layers(conn, node_ids)
+        ids = set(node_ids)
+        if broken == "moved_nodes_in_agent" and ids == set(moved_ids):
+            result[moved_ids[0]] = "agent_pending"
+        if broken == "moved_nodes_not_in_user" and ids == set(moved_ids):
+            result[moved_ids[0]] = "user"
+        if broken == "unmoved_nodes_unchanged" and ids == set(unmoved_ids):
+            result[unmoved_ids[0]] = "agent"
+        return result
+
+    monkeypatch.setattr(migration, "_layer_counts", fake_counts)
+    monkeypatch.setattr(migration, "_layers_for", fake_layers)
+
+    with pytest.raises(migration.ApplyError, match="postcondition failed") as excinfo:
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert broken in str(excinfo.value)
+    assert _db_hash(db) == before_hash
+    assert _node_rows(db) == before_nodes
+
+
+def test_apply_rejects_manifest_that_is_not_an_object(apply_base: Path) -> None:
+    """Kills: dropping the `isinstance(manifest, dict)` guard; a JSON list would raise AttributeError."""
+    db = apply_base / "memory.db"
+    manifest = apply_base / "manifest.json"
+    manifest.write_text("[]\n", encoding="utf-8")
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match="JSON object"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm="whatever")
+
+    assert _db_hash(db) == before_hash
+
+
+def test_apply_rejects_manifest_without_candidates(apply_base: Path) -> None:
+    """Kills: dropping the candidates-list guard; iterating None would raise TypeError."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    data = _manifest_data(manifest)
+    del data["candidates"]
+    _rewrite_manifest(manifest, data)
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match="candidates must be a list"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+
+
+def test_apply_rejects_unknown_candidate_decision(apply_base: Path) -> None:
+    """Kills: dropping the decision validation in `_validated_candidates`.
+
+    The fingerprint is also wrong on purpose: without the decision guard the
+    fingerprint gate fires first, so the expected message changes. `_should_move`
+    has an equivalent fallback, which is why the fingerprint makes the ordering
+    observable.
+    """
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    data = _manifest_data(manifest)
+    next(candidate for candidate in data["candidates"] if int(candidate["node_id"]) == 1)["decision"] = "maybe"
+    data["db_fingerprint"]["sha256"] = "0" * 64
+    _rewrite_manifest(manifest, data)
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match="unknown decision"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+
+
+def test_apply_rejects_unknown_candidate_confidence(apply_base: Path) -> None:
+    """Kills: dropping the confidence validation in `_validated_candidates`.
+
+    Load-bearing: without it an `approve` row with an unknown confidence moves
+    silently, because `_should_move` never reads the confidence of an approved
+    candidate.
+    """
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    data = _manifest_data(manifest)
+    next(candidate for candidate in data["candidates"] if int(candidate["node_id"]) == 1)["confidence"] = "certain"
+    _rewrite_manifest(manifest, data)
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match="unknown confidence"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+
+
+def test_apply_rejects_unknown_approve_confidence(apply_base: Path) -> None:
+    """Kills: dropping the approve-confidence whitelist; an unknown value would move approved rows."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match="unknown approve-confidence"):
+        migration.apply(
+            apply_base,
+            manifest,
+            apply_base.parent / "snapshots",
+            confirm=_expected_token(manifest),
+            approve_confidence="sometimes",
+        )
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+
+
+def test_apply_refuses_candidate_user_id_mismatch(apply_base: Path) -> None:
+    """Kills: dropping the candidate/DB user_id comparison; the wrong owner would be targeted."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    data = _manifest_data(manifest)
+    next(candidate for candidate in data["candidates"] if int(candidate["node_id"]) == 1)["user_id"] = "impostor"
+    _rewrite_manifest(manifest, data)
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.ApplyError, match="belongs to user"):
+        migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+
+
+def test_concurrent_writer_between_probe_and_snapshot_refuses_before_transaction(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Kills: moving rows after a writer changes the database after preflight."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    before_layers = _layers(db)
+    real_apply_transaction = migration._apply_transaction
+
+    def racing_apply_transaction(*args: Any) -> tuple[dict[str, int], dict[str, int], dict[str, bool], dict[str, Any], str]:
+        db_path = Path(args[0])
+        with sqlite3.connect(db_path) as conn:
+            conn.execute(
+                "INSERT INTO epi_nodes (layer, user_id, content, node_type, tags, confidence, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                ("user", "default", "concurrent writer row", "fact", "[]", 0.5, 99.0),
+            )
+        return real_apply_transaction(*args)
+
+    monkeypatch.setattr(migration, "_apply_transaction", racing_apply_transaction)
+    snapshot_dir = apply_base.parent / "snapshots"
+
+    with pytest.raises(migration.ApplyError, match="pre-transaction fingerprint"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    after_layers = _layers(db)
+    assert {node_id: after_layers[node_id] for node_id in before_layers} == before_layers
+    assert not (apply_base / "manifest.json.apply-report.json").exists()
+    snapshots = [path for path in snapshot_dir.iterdir() if path.is_file()]
+    assert len(snapshots) == 1, "the in-lock snapshot is kept and named in the error"
+
+
+def test_report_records_both_fingerprints_and_pre_transaction_verification(apply_base: Path) -> None:
+    """Kills: dropping the pre-transaction fingerprint fields/boolean from the success report."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    expected_sha = _manifest_data(manifest)["db_fingerprint"]["sha256"]
+
+    report = migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert report["fingerprint_verified_pre_transaction"] is True
+    assert report["snapshot_taken_under_write_lock"] is True
+    assert report["snapshot_verified_under_write_lock"] is True
+    assert report["fingerprint"]["confirmed_sha256"] == expected_sha
+    assert report["fingerprint"]["pre_transaction_sha256"] == expected_sha
+    assert report["fingerprint"]["pre_transaction_sha256"] == report["snapshot"]["sha256"]
+    assert _layers(db)[1] == "agent"
+
+
+def test_writer_after_begin_immediate_is_blocked_during_snapshot(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """The authoritative snapshot owns the same write lock as the apply transaction."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    real_snapshot = migration._snapshot_connection
+    writer_attempted = False
+
+    def probe_writer(source_conn: sqlite3.Connection, destination: Path) -> None:
+        nonlocal writer_attempted
+        writer_attempted = True
+        with sqlite3.connect(db, isolation_level=None) as writer:
+            writer.execute("PRAGMA busy_timeout = 1")
+            with pytest.raises(sqlite3.OperationalError, match=r"locked|busy"):
+                writer.execute(
+                    "INSERT INTO epi_nodes (layer, user_id, content, node_type, tags, confidence, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    ("user", "default", "blocked writer", "fact", "[]", 0.5, 100.0),
+                )
+        real_snapshot(source_conn, destination)
+
+    monkeypatch.setattr(migration, "_snapshot_connection", probe_writer)
+    report = migration.apply(apply_base, manifest, apply_base.parent / "snapshots", confirm=_expected_token(manifest))
+
+    assert writer_attempted
+    assert report["snapshot_taken_under_write_lock"] is True
+    assert "blocked writer" not in {str(row[3]) for row in _node_rows(db)}
+
+
+def test_busy_database_raises_clean_apply_error_without_snapshot(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """A lock failure occurs before the authoritative snapshot is created."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    monkeypatch.setattr(migration, "APPLY_BUSY_TIMEOUT_MS", 100)
+    monkeypatch.setattr(migration, "_utc_timestamp", lambda: "20260101T000000Z")
+    before_hash = _db_hash(db)
+    snapshot_dir = apply_base.parent / "snapshots"
+    blocker = sqlite3.connect(db, isolation_level=None)
+    blocker.execute("BEGIN IMMEDIATE")
+    try:
+        with pytest.raises(migration.ApplyError, match="locked") as excinfo:
+            migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+    finally:
+        blocker.execute("ROLLBACK")
+        blocker.close()
+
+    assert "snapshot" not in str(excinfo.value).lower()
+    assert not snapshot_dir.exists() or not any(snapshot_dir.iterdir())
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+
+
+@pytest.mark.parametrize("database", ["../escaped", "/absolute/path", "nested/name"])
+def test_apply_rejects_unsafe_explicit_database_label_without_external_outputs(apply_base: Path, database: str) -> None:
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    snapshot_dir = apply_base.parent / "snapshots"
+    report_path = apply_base.parent / "escaped-report.json"
+    before_hash = _db_hash(db)
+
+    with pytest.raises(migration.SchemaError, match="database label"):
+        migration.apply(
+            apply_base,
+            manifest,
+            snapshot_dir,
+            confirm=_expected_token(manifest),
+            database=database,
+            report=report_path,
+        )
+
+    assert _db_hash(db) == before_hash
+    assert not snapshot_dir.exists()
+    assert not report_path.exists()
+
+
+def _write_bulk_db(base: Path, count: int) -> Path:
+    base.mkdir()
+    db = base / "memory.db"
+    with sqlite3.connect(db) as conn:
+        conn.executescript(NODE_DDL + EDGE_DDL)
+        conn.executemany(
+            "INSERT INTO epi_nodes (node_id, layer, user_id, content, node_type, tags, confidence, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            ((node_id, "user", "default", f"bulk note {node_id}", "fact", "[]", 0.5, float(node_id)) for node_id in range(1, count + 1)),
+        )
+    return db
+
+
+def test_apply_handles_more_than_the_sqlite_variable_limit(tmp_path: Path) -> None:
+    """Kills: one `IN (...)` list in `_layers_for`; 33000 ids exceed SQLite's 32766 bound variables.
+
+    The fixture is built programmatically. Runtime is measured and reported;
+    the suite's 60 s per-test timeout is the hard ceiling.
+    """
+    count = 33000
+    base = tmp_path / "bulkbase"
+    db = _write_bulk_db(base, count)
+    with tempfile.TemporaryDirectory(prefix="bulk-fingerprint-") as temp_dir:
+        probe = Path(temp_dir) / "probe.db"
+        migration._snapshot_to(db, probe)
+        sha = migration._file_sha256(probe)
+
+    manifest = base / "manifest.json"
+    manifest.write_text(
+        json.dumps(
+            {
+                "schema": migration.MANIFEST_SCHEMA,
+                "database": "bulkbase",
+                "source_layer": "user",
+                "target_layer": "agent",
+                "db_fingerprint": {"sha256": sha},
+                "candidates": [
+                    {
+                        "node_id": node_id,
+                        "user_id": "default",
+                        "current_layer": "user",
+                        "target_layer": "agent",
+                        "confidence": "high",
+                        "decision": "approve",
+                        "reasons": [],
+                    }
+                    for node_id in range(1, count + 1)
+                ],
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+
+    manifest_sha = hashlib.sha256(manifest.read_bytes()).hexdigest()
+    report = migration.apply(base, manifest, tmp_path / "snapshots", confirm=f"MIGRATE-bulkbase-{manifest_sha[:12]}")
+
+    assert report["moved_count"] == count
+    assert all(report["postconditions"].values())
+    with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as conn:
+        assert int(conn.execute("SELECT COUNT(*) FROM epi_nodes WHERE layer = 'agent'").fetchone()[0]) == count
+
+
+@pytest.mark.parametrize("inside", ["subdirectory", "data_dir"])
+def test_apply_refuses_snapshot_dir_inside_data_directory(apply_base: Path, inside: str) -> None:
+    """Kills: allowing `--snapshot-dir` inside the live base, which puts a stray .db next to memory.db."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    before_hash = _db_hash(db)
+    snapshot_dir = apply_base / "snapshots" if inside == "subdirectory" else apply_base
+
+    with pytest.raises(migration.ApplyError, match="inside the data directory"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"
+    if inside == "subdirectory":
+        assert not (apply_base / "snapshots").exists()
+
+
+def test_reserve_snapshot_cleans_staging_dir_when_parent_fsync_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Kills: leaking the just-created staging directory when its parent fsync fails."""
+    snapshot_dir = tmp_path / "snapshots"
+    snapshot_dir.mkdir()
+
+    def fail_parent_fsync(path: Path) -> None:
+        raise OSError(f"forced directory sync failure: {path}")
+
+    monkeypatch.setattr(migration, "_fsync_directory", fail_parent_fsync)
+
+    with pytest.raises(OSError, match="forced directory sync failure"):
+        migration._reserve_snapshot(snapshot_dir)
+
+    assert list(snapshot_dir.glob(".snapshot-stage-*")) == []
+
+
+def test_reserve_snapshot_closes_fd_and_removes_staging_dir_when_chmod_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Kills: leaking the reserved fd/file when fchmod fails during snapshot reservation."""
+    snapshot_dir = tmp_path / "snapshots"
+    snapshot_dir.mkdir()
+    opened: list[int] = []
+    real_open = os.open
+
+    def record_open(path: str | bytes | os.PathLike[str] | os.PathLike[bytes], flags: int, mode: int = 0o777, **kwargs: Any) -> int:
+        fd = real_open(path, flags, mode, **kwargs)
+        if Path(path).name == "snapshot.db":
+            opened.append(fd)
+        return fd
+
+    def fail_fchmod(fd: int, mode: int) -> None:
+        raise OSError("forced fchmod failure")
+
+    monkeypatch.setattr(migration.os, "open", record_open)
+    monkeypatch.setattr(migration.os, "fchmod", fail_fchmod)
+
+    with pytest.raises(OSError, match="forced fchmod failure"):
+        migration._reserve_snapshot(snapshot_dir)
+
+    assert len(opened) == 1
+    with pytest.raises(OSError):
+        os.fstat(opened[0])
+    assert list(snapshot_dir.glob(".snapshot-stage-*")) == []
+
+
+def test_snapshot_install_directory_fsync_failure_preserves_original_error(apply_base: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Kills: cleanup masking the final snapshot-directory fsync failure with FileNotFoundError."""
+    db = apply_base / "memory.db"
+    manifest = _manifest_for(apply_base, decisions={1: "approve"})
+    manifest_data = _manifest_data(manifest)
+    monkeypatch.setattr(migration, "_utc_timestamp", lambda: "20260101T000000Z")
+    before_hash = _db_hash(db)
+    snapshot_dir = apply_base.parent / "snapshots"
+    expected_snapshot = snapshot_dir / f"{manifest_data['database']}-20260101T000000Z-{manifest_data['db_fingerprint']['sha256'][:12]}.db"
+    calls = {"snapshot_parent": 0}
+    real_fsync_directory = migration._fsync_directory
+
+    def fail_after_install(path: Path) -> None:
+        if path == snapshot_dir:
+            calls["snapshot_parent"] += 1
+            if calls["snapshot_parent"] == 2:
+                raise OSError("forced final snapshot directory fsync failure")
+        real_fsync_directory(path)
+
+    monkeypatch.setattr(migration, "_fsync_directory", fail_after_install)
+
+    with pytest.raises(migration.ApplyError, match="forced final snapshot directory fsync failure"):
+        migration.apply(apply_base, manifest, snapshot_dir, confirm=_expected_token(manifest))
+
+    assert expected_snapshot.is_file(), "installed pre-move snapshot must remain available for recovery"
+    assert _db_hash(db) == before_hash
+    assert _layers(db)[1] == "user"