"""
Backup and import commands for hermes CLI.

`hermes backup` creates a zip archive of the entire ~/.hermes/ directory
(excluding the hermes-agent repo and transient files).

`hermes import` restores from a backup zip, overlaying onto the current
HERMES_HOME root.
"""

import json
import logging
import os
import shutil
import sqlite3
import sys
import tempfile
import threading
import time
import zipfile
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

from hermes_constants import get_default_hermes_root, get_hermes_home, display_hermes_home

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# Exclusion rules
# ---------------------------------------------------------------------------

# Directory names to skip entirely (matched against each path component)
# ``hermes-agent`` is special-cased to root level only in ``_should_exclude``
# so that skill directories like ``skills/autonomous-ai-agents/hermes-agent/``
# are not accidentally excluded.
#
# The dependency/cache entries below matter for more than tidiness: without
# them a single plugin venv, MCP-server install, or pip/uv cache living under
# HERMES_HOME gets walked file-by-file, ballooning a backup to hundreds of
# thousands of entries that crawl for hours — the exact "backup stuck for
# days / 426543 files" symptom users hit. The dependency/test-env names mostly
# mirror ``agent.skill_utils.EXCLUDED_SKILL_DIRS`` (the project's canonical
# "regeneratable dir" set); ``.cache`` is an additional backup-only entry, as
# it names a broad regeneratable cache convention (pip/uv/etc.) that the skill
# scanner doesn't need to prune but a backup walk does. We deliberately do NOT
# exclude ``.archive`` here because the curator's ``skills/.archive/`` holds
# restorable user skills that must survive a backup.
_EXCLUDED_DIRS = {
    "hermes-agent",     # the codebase repo — re-clone instead
    "__pycache__",      # bytecode caches — regenerated on import
    ".git",             # nested git dirs (profiles shouldn't have these, but safety)
    "node_modules",     # js deps — reinstalled on demand
    "backups",          # prior auto-backups — don't nest backups exponentially
    "checkpoints",      # session-local trajectory caches — regenerated per-session,
                        # session-hash-keyed so they don't port to another machine anyway
    # Python dependency trees (plugin / MCP-server venvs under HERMES_HOME) —
    # regenerated by reinstalling; never irreplaceable state.
    ".venv",
    "venv",
    "site-packages",
    # Tool / build caches — all regeneratable.
    ".cache",
    ".tox",
    ".nox",
    ".pytest_cache",
    ".mypy_cache",
    ".ruff_cache",
}

# File-name suffixes to skip
_EXCLUDED_SUFFIXES = (
    ".pyc",
    ".pyo",
    # SQLite sidecar files — the backup takes a consistent snapshot of ``*.db``
    # via ``sqlite3.backup()``, so shipping the live WAL / shared-memory /
    # rollback-journal alongside would pair a fresh snapshot with stale sidecar
    # state and produce a torn restore on the next open. They're transient and
    # regenerated on first connection anyway.
    ".db-wal",
    ".db-shm",
    ".db-journal",
)

# File names to skip (runtime state that's meaningless on another machine)
_EXCLUDED_NAMES = {
    ".backup.lock",
    "gateway.pid",
    "cron.pid",
}

# File names that ``hermes import`` must never overwrite, matched by basename so
# they're caught for the root profile (``gateway_state.json``) and for named
# profiles alike (``profiles/<name>/gateway_state.json``).
#
# These hold *volatile gateway/process runtime state that is namespaced to the
# machine or container the backup was taken on* — PIDs in a dead process
# namespace, a runtime lock, the process registry, and the gateway's last
# recorded run/desired state. Restoring them onto a different host (or a hosted
# container) is at best meaningless and at worst actively harmful:
#
#   - ``gateway_state.json`` drives the container-boot reconciler
#     (``container_boot._read_desired_state``), which only auto-starts a
#     gateway whose recorded state is ``running``. A backup taken from a
#     machine where the gateway was stopped (or carrying a stale/foreign
#     value) overwrites the container's own state and leaves the gateway
#     stuck "starting"/"cooking", disconnecting it from the Nous portal
#     (NS-508 / the second half of NS-501).
#   - ``gateway.pid`` / ``cron.pid`` / ``gateway.lock`` / ``processes.json``
#     reference PIDs and locks in the *source* machine's process namespace; a
#     numerically-equal PID in the new environment is a different process.
#     These mirror exactly what ``container_boot._STALE_RUNTIME_FILES`` already
#     sweeps on every container boot.
#
# Older backups predate the backup-side exclusions, so we filter on import too
# rather than trusting the archive's contents.
_IMPORT_SKIP_NAMES = {
    "gateway_state.json",
    "gateway.pid",
    "cron.pid",
    "gateway.lock",
    "processes.json",
}

# zipfile.open() drops Unix mode bits on extract; restore tightens these to 0600.
_SECRET_FILE_NAMES = {".env", "auth.json", "state.db"}

# Reserved archive subtree for provider state that lives OUTSIDE HERMES_HOME
# (e.g. ~/.honcho, ~/.hindsight). The active memory provider declares these via
# MemoryProvider.backup_paths(); they're stored under this prefix encoded
# relative to the user's home directory, and restored to their original
# home-relative location on import. Anything not under home is skipped.
_EXTERNAL_PREFIX = "_external/"


class BackupInProgressError(RuntimeError):
    """Raised when another process already owns the Hermes backup slot."""


class _SQLiteSnapshotError(RuntimeError):
    pass


@contextmanager
def _backup_operation_lock(hermes_home: Path, timeout_seconds: float = 0.25):
    """Acquire one cross-process backup slot for full and quick snapshots."""
    lock_path = hermes_home / ".backup.lock"
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    handle = lock_path.open("a+b")
    acquired = False
    deadline = time.monotonic() + max(0.0, timeout_seconds)
    try:
        if os.name == "nt":
            import msvcrt

            if lock_path.stat().st_size == 0:
                handle.write(b" ")
                handle.flush()
            while True:
                try:
                    handle.seek(0)
                    msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
                    acquired = True
                    break
                except (OSError, PermissionError):
                    if time.monotonic() >= deadline:
                        raise BackupInProgressError("another Hermes backup is already running")
                    time.sleep(0.05)
        else:
            import fcntl

            while True:
                try:
                    fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                    acquired = True
                    break
                except (BlockingIOError, OSError):
                    if time.monotonic() >= deadline:
                        raise BackupInProgressError("another Hermes backup is already running")
                    time.sleep(0.05)

        yield
    finally:
        if acquired:
            try:
                if os.name == "nt":
                    import msvcrt

                    handle.seek(0)
                    msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
                else:
                    import fcntl

                    fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
            except (OSError, PermissionError):
                pass
        handle.close()


@contextmanager
def _atomic_output_path(final_path: Path):
    """Yield a hidden sibling path and publish it only after a clean close."""
    partial_path = final_path.with_name(
        f".{final_path.name}.{os.getpid()}-{threading.get_ident()}.partial"
    )
    partial_path.unlink(missing_ok=True)
    try:
        yield partial_path
        os.replace(partial_path, final_path)
    except BaseException:
        partial_path.unlink(missing_ok=True)
        raise


def _collect_memory_provider_external_paths() -> List[Path]:
    """Return existing absolute paths the active memory provider stores
    outside HERMES_HOME, resolved from config only (no network, no init).

    Reads ``memory.provider`` from config, loads just that provider, and asks
    it for ``backup_paths()``. Returns an empty list when no external provider
    is active or the provider can't be loaded — backup must never fail because
    of a flaky plugin.
    """
    try:
        from plugins.memory import _get_active_memory_provider, load_memory_provider
    except Exception:
        return []

    try:
        active = _get_active_memory_provider()
    except Exception:
        active = None
    if not active:
        return []

    try:
        provider = load_memory_provider(active)
    except Exception:
        provider = None
    if provider is None:
        return []

    try:
        declared = provider.backup_paths() or []
    except Exception as exc:
        logger.warning("backup_paths() failed for memory provider %r: %s", active, exc)
        return []

    out: List[Path] = []
    seen: set = set()
    for raw in declared:
        try:
            p = Path(raw).expanduser()
        except Exception:
            continue
        if not p.exists():
            continue
        try:
            resolved = p.resolve()
        except (OSError, ValueError):
            continue
        if resolved in seen:
            continue
        seen.add(resolved)
        out.append(p)
    return out


def _iter_external_files(base: Path) -> List[Path]:
    """Yield regular files under *base* (a file or a directory), skipping
    symlinks, caches, and pyc files. *base* itself may be a file."""
    files: List[Path] = []
    if base.is_file() and not base.is_symlink():
        files.append(base)
        return files
    if not base.is_dir():
        return files
    for dirpath, dirnames, filenames in os.walk(base, followlinks=False):
        dp = Path(dirpath)
        dirnames[:] = [d for d in dirnames if d not in _EXCLUDED_DIRS]
        for fname in filenames:
            fpath = dp / fname
            if fpath.is_symlink():
                continue
            if fpath.name in _EXCLUDED_NAMES or fpath.name.endswith(_EXCLUDED_SUFFIXES):
                continue
            files.append(fpath)
    return files


def _should_exclude(rel_path: Path) -> bool:
    """Return True if *rel_path* (relative to hermes root) should be skipped."""
    parts = rel_path.parts

    for part in parts:
        if part not in _EXCLUDED_DIRS:
            continue
        # ``hermes-agent`` only matches at the root level (first component).
        # Nested directories with the same name — e.g.
        # ``skills/autonomous-ai-agents/hermes-agent/`` — must be preserved.
        if part == "hermes-agent" and part != parts[0]:
            continue
        return True

    name = rel_path.name

    if name in _EXCLUDED_NAMES:
        return True

    if name.endswith(_EXCLUDED_SUFFIXES):
        return True

    return False


def _should_skip_backup_file(abs_path: Path, rel_path: Path, out_path: Path) -> bool:
    """Return True when a candidate file should not be written to a backup zip."""
    if _should_exclude(rel_path):
        return True

    # zipfile.write() follows file symlinks, so skip links before any archive
    # write can copy data from outside HERMES_HOME.
    if abs_path.is_symlink():
        return True

    try:
        return abs_path.resolve() == out_path.resolve()
    except (OSError, ValueError):
        return False


# ---------------------------------------------------------------------------
# SQLite safe copy
# ---------------------------------------------------------------------------

def _safe_copy_db(src: Path, dst: Path) -> bool:
    """Copy a SQLite database safely using the backup() API.

    Handles WAL mode — produces a consistent snapshot even while
    the DB is being written to. Fail closed if a consistent snapshot cannot
    be created: copying only the live main file can omit committed WAL data.
    """
    conn = None
    backup_conn = None
    try:
        conn = sqlite3.connect(f"file:{src}?mode=ro", uri=True)
        backup_conn = sqlite3.connect(str(dst))
        conn.backup(backup_conn)
        return True
    except Exception as exc:
        logger.warning("SQLite safe copy failed for %s: %s", src, exc)
        try:
            dst.unlink(missing_ok=True)
        except OSError:
            pass
        return False
    finally:
        for connection in (backup_conn, conn):
            if connection is not None:
                try:
                    connection.close()
                except Exception:
                    pass


def is_zeroed_sqlite_file(
    path: Path, *, probe_bytes: int = 100, force: bool = False
) -> bool:
    """True when *path* looks like the #68474 zeroed-state.db signature.

    Signature: size > 0, first *probe_bytes* are all NUL (no ``SQLite format 3``
    header). Used at SessionDB open and for snapshot diagnostics so a silent
    all-zero file becomes a guided recovery instead of a generic failure.
    """
    try:
        size = path.stat().st_size
    except OSError:
        return False
    if size <= 0:
        return False
    from hermes_cli.sqlite_safe_read import read_header_bytes_preopen

    head = read_header_bytes_preopen(
        path, length=max(16, probe_bytes), force=force
    )
    if not head:
        return False
    if head.startswith(b"SQLite format 3"):
        return False
    return all(byte == 0 for byte in head)



# ---------------------------------------------------------------------------
# SQLite integrity verification
# ---------------------------------------------------------------------------

_SQLITE_HEADER = b"SQLite format 3\0"

# Default ceiling above which ``PRAGMA integrity_check`` is skipped in favour
# of the (O(1)) header + structural probe. ``integrity_check`` walks every
# b-tree page in the file, so its cost scales with database size: on a 30 GB
# state.db it runs for many minutes of pegged CPU with no output, which reads
# to the user as a hung `hermes update` (#70553 follow-up). Sessions databases
# in the tens of GB are normal for heavy users, so the size-unbounded check is
# never an acceptable default on the update path.
DEFAULT_INTEGRITY_CHECK_MAX_BYTES = 2 << 30  # 2 GiB


def verify_sqlite_integrity(
    path: Path,
    *,
    check_header: bool = True,
    run_pragma: bool = True,
    max_bytes: int = DEFAULT_INTEGRITY_CHECK_MAX_BYTES,
) -> dict:
    """Verify that a SQLite database at *path* is intact.

    Checks, in order:
      1. File exists and has an expected minimum size.
      2. SQLite header magic bytes are present.
      3. For files at or under ``max_bytes``, a read-only
         ``PRAGMA integrity_check``. For larger files, a cheap structural
         probe (schema read) instead — see ``max_bytes``.

    Args:
        path: Path to the database file.
        check_header: When true (default), verify the SQLite header magic.
        run_pragma: When true (default), run ``PRAGMA integrity_check`` via
            a read-only connection and verify the result is ``"ok"``.
        max_bytes: Size ceiling for the full ``PRAGMA integrity_check``.
            Files larger than this fall back to the header check plus a
            cheap structural probe, because ``integrity_check`` pages
            through the ENTIRE file — minutes of silent pegged CPU on a
            multi-GB database. Defaults to
            :data:`DEFAULT_INTEGRITY_CHECK_MAX_BYTES` (2 GiB); pass ``0``
            to force the full check regardless of size.

    Returns:
        A dict with keys:
          - ``valid`` (bool): true when all requested checks passed.
          - ``message`` (str): human-readable outcome or error detail.
          - ``size`` (int | None): file size in bytes, or None if stat failed.
    """
    result: dict = {"valid": False, "message": "", "size": None}

    try:
        st = path.stat()
    except FileNotFoundError:
        result["message"] = f"not found: {path}"
        return result
    except OSError as exc:
        result["message"] = f"cannot stat: {exc}"
        return result

    result["size"] = st.st_size

    if st.st_size < 100:  # SQLite minimum viable size (header + 1 page)
        result["message"] = f"too small ({st.st_size} bytes) to be a valid SQLite database"
        return result

    oversized = max_bytes > 0 and st.st_size > max_bytes

    if check_header:
        # Byte-level read: refused when a live connection exists, because
        # close() would cancel this process's POSIX locks on the file (see
        # hermes_cli.sqlite_safe_read). Verification targets snapshots and
        # backup artifacts, which are offline by construction.
        from hermes_cli.sqlite_safe_read import read_header_bytes_preopen

        head = read_header_bytes_preopen(path, length=len(_SQLITE_HEADER))
        if head is None:
            result["valid"] = False
            result["message"] = "cannot read header"
            return result
        if head != _SQLITE_HEADER:
            result["valid"] = False
            result["message"] = (
                f"missing SQLite header magic (got {head[:16].hex()!r})"
            )
            return result

    if oversized:
        # Too large to page through PRAGMA integrity_check (which is O(file
        # size) and would peg a CPU for minutes on a multi-GB state.db).
        # Fall back to a cheap O(1) structural probe: the header check above
        # catches the #68474 zeroed signature, and opening the DB read-only
        # plus reading sqlite_master + the page geometry catches the
        # malformed-schema and truncated-header-page classes. Both are
        # constant-time — they parse the schema, they do not walk the data.
        run_pragma = False
        probe = None
        try:
            probe = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=1.0)
            probe.execute("PRAGMA schema_version").fetchone()
            probe.execute("SELECT count(*) FROM sqlite_master").fetchone()
            result["valid"] = True
            result["message"] = (
                f"size {st.st_size:,} bytes exceeds max_bytes {max_bytes:,}; "
                "skipped PRAGMA integrity_check (header + schema probe passed)"
            )
        except sqlite3.DatabaseError as exc:
            result["valid"] = False
            result["message"] = f"schema probe failed: {exc}"
            return result
        except Exception as exc:
            result["valid"] = False
            result["message"] = f"schema probe error: {exc}"
            return result
        finally:
            if probe is not None:
                try:
                    probe.close()
                except Exception:
                    pass

    if run_pragma:
        conn = None
        try:
            conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=1.0)
            cursor = conn.execute("PRAGMA integrity_check")
            rows = cursor.fetchall()
            if len(rows) == 1 and rows[0][0] == "ok":
                result["valid"] = True
                result["message"] = "integrity check passed"
                return result
            errors = [str(r[0]) for r in rows]
            result["message"] = f"integrity check failed: {'; '.join(errors[:5])}"
            return result
        except sqlite3.DatabaseError as exc:
            result["message"] = f"cannot open database: {exc}"
            return result
        except Exception as exc:
            result["message"] = f"integrity check error: {exc}"
            return result
        finally:
            if conn is not None:
                try:
                    conn.close()
                except Exception:
                    pass

    result["valid"] = True
    if not result["message"]:
        result["message"] = "header check passed"
    return result


def copy_db_and_verify(src: Path, dst: Path) -> bool:
    """Like :func:`_safe_copy_db` but verifies the destination after copy.

    Returns True only when the copy succeeded AND the destination is valid
    SQLite (header + integrity check). Verification honours the default
    size ceiling — a multi-GB destination gets the header + schema probe
    rather than a full ``PRAGMA integrity_check`` that would page through
    the whole file.
    """
    if not _safe_copy_db(src, dst):
        return False
    integrity = verify_sqlite_integrity(dst, run_pragma=True)
    if not integrity.get("valid"):
        try:
            dst.unlink(missing_ok=True)
        except OSError:
            pass
        logger.warning("Backup of %s failed integrity verification: %s", src, integrity.get("message"))
        return False
    return True


# ---------------------------------------------------------------------------
# Backup
# ---------------------------------------------------------------------------

def _format_size(nbytes: int) -> str:
    """Human-readable file size."""
    for unit in ("B", "KB", "MB", "GB"):
        if nbytes < 1024:
            return f"{nbytes:.1f} {unit}" if unit != "B" else f"{nbytes} {unit}"
        nbytes /= 1024
    return f"{nbytes:.1f} TB"


def run_backup(args) -> None:
    """Create a zip backup of the Hermes home directory."""
    hermes_root = get_default_hermes_root()

    if not hermes_root.is_dir():
        print(f"Error: Hermes home directory not found at {hermes_root}")
        sys.exit(1)

    try:
        with _backup_operation_lock(hermes_root):
            _run_backup_locked(args, hermes_root)
    except BackupInProgressError as exc:
        print(f"Error: {exc}")
        raise SystemExit(2) from exc


def _run_backup_locked(args, hermes_root: Path) -> None:
    """Write a full backup while the cross-process backup slot is held."""

    # Determine output path
    if args.output:
        out_path = Path(args.output).expanduser().resolve()
        # If user gave a directory, put the zip inside it
        if out_path.is_dir():
            stamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
            out_path = out_path / f"hermes-backup-{stamp}.zip"
    else:
        stamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
        out_path = Path.home() / f"hermes-backup-{stamp}.zip"

    # Ensure the suffix is .zip
    if out_path.suffix.lower() != ".zip":
        out_path = out_path.with_suffix(out_path.suffix + ".zip")

    # Ensure parent directory exists
    out_path.parent.mkdir(parents=True, exist_ok=True)

    # Collect files
    scan_started = time.monotonic()
    logger.info("backup phase=scan status=started")
    print(f"Scanning {display_hermes_home()} ...")
    files_to_add: list[tuple[Path, Path]] = []  # (absolute, relative)
    skipped_dirs = set()

    for dirpath, dirnames, filenames in os.walk(hermes_root, followlinks=False):
        dp = Path(dirpath)
        rel_dir = dp.relative_to(hermes_root)

        # Prune excluded directories in-place so os.walk doesn't descend
        # ``hermes-agent`` is only pruned at the root level; nested dirs
        # with the same name (e.g. in skills/) must be preserved.
        is_root = rel_dir == Path(".")
        orig_dirnames = dirnames[:]
        dirnames[:] = [
            d for d in dirnames
            if d not in _EXCLUDED_DIRS or (d == "hermes-agent" and not is_root)
        ]
        for removed in set(orig_dirnames) - set(dirnames):
            skipped_dirs.add(str(rel_dir / removed))

        for fname in filenames:
            fpath = dp / fname
            rel = fpath.relative_to(hermes_root)

            if _should_skip_backup_file(fpath, rel, out_path):
                continue

            files_to_add.append((fpath, rel))

    # External memory-provider state (e.g. ~/.honcho, ~/.hindsight) lives
    # outside HERMES_HOME, so the walk above never sees it. Ask the active
    # provider for its declared paths and stage them under the reserved
    # ``_external/`` arc prefix, encoded relative to the user's home dir.
    # Only paths under home are captured (security + portability); anything
    # else is skipped with a note.
    home_dir = Path.home().resolve()
    external_to_add: list[tuple[Path, str]] = []  # (absolute, arcname)
    skipped_external: list[str] = []
    for base in _collect_memory_provider_external_paths():
        try:
            base_resolved = base.resolve()
            base_resolved.relative_to(home_dir)
        except (ValueError, OSError):
            skipped_external.append(str(base))
            continue
        for fpath in _iter_external_files(base):
            try:
                rel_to_home = fpath.resolve().relative_to(home_dir)
            except (ValueError, OSError):
                continue
            arcname = _EXTERNAL_PREFIX + rel_to_home.as_posix()
            external_to_add.append((fpath, arcname))

    if not files_to_add and not external_to_add:
        logger.info(
            "backup phase=scan status=empty duration_ms=%.1f",
            (time.monotonic() - scan_started) * 1000,
        )
        print("No files to back up.")
        return

    # Create the zip
    file_count = len(files_to_add) + len(external_to_add)
    logger.info(
        "backup phase=scan status=complete duration_ms=%.1f files=%d",
        (time.monotonic() - scan_started) * 1000,
        file_count,
    )
    logger.info("backup phase=archive status=started files=%d", file_count)
    print(f"Backing up {file_count} files ...")

    total_bytes = 0
    errors = []
    t0 = time.monotonic()

    with _atomic_output_path(out_path) as archive_path, zipfile.ZipFile(
        archive_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6
    ) as zf:
        for i, (abs_path, rel_path) in enumerate(files_to_add, 1):
            try:
                # Safe copy for SQLite databases (handles WAL mode)
                if abs_path.suffix == ".db":
                    # Stage the snapshot alongside the output zip so that the
                    # temp file lives on the same filesystem.  The system
                    # default (/tmp) may be a small tmpfs that cannot hold
                    # large databases, causing silent backup incompleteness.
                    with tempfile.NamedTemporaryFile(
                        suffix=".db", delete=False, dir=str(out_path.parent)
                    ) as tmp:
                        tmp_db = Path(tmp.name)
                    if _safe_copy_db(abs_path, tmp_db):
                        zf.write(tmp_db, arcname=str(rel_path))
                        total_bytes += tmp_db.stat().st_size
                        tmp_db.unlink(missing_ok=True)
                    else:
                        tmp_db.unlink(missing_ok=True)
                        errors.append(f"  {rel_path}: SQLite safe copy failed")
                        continue
                else:
                    zf.write(abs_path, arcname=str(rel_path))
                    total_bytes += abs_path.stat().st_size
            except (PermissionError, OSError, ValueError) as exc:
                errors.append(f"  {rel_path}: {exc}")
                continue

            # Progress every 500 files
            if i % 500 == 0:
                print(f"  {i}/{file_count} files ...")
                logger.info(
                    "backup phase=archive status=progress completed=%d total=%d",
                    i,
                    file_count,
                )

        # External memory-provider state, stored under the ``_external/`` arc
        # prefix. These never include ``.db`` files in practice (config/env
        # blobs), so a straight zf.write is fine.
        for abs_path, arcname in external_to_add:
            try:
                zf.write(abs_path, arcname=arcname)
                total_bytes += abs_path.stat().st_size
            except (PermissionError, OSError, ValueError) as exc:
                errors.append(f"  {arcname}: {exc}")
                continue

    elapsed = time.monotonic() - t0
    zip_size = out_path.stat().st_size
    logger.info(
        "backup phase=archive status=complete duration_ms=%.1f files=%d errors=%d bytes=%d",
        elapsed * 1000,
        file_count,
        len(errors),
        zip_size,
    )

    # Summary
    print()
    if errors:
        print(f"Backup incomplete: {out_path}")
    else:
        print(f"Backup complete: {out_path}")
    print(f"  Files:       {file_count}")
    print(f"  Original:    {_format_size(total_bytes)}")
    print(f"  Compressed:  {_format_size(zip_size)}")
    print(f"  Time:        {elapsed:.1f}s")

    if external_to_add:
        print(
            f"\n  Included {len(external_to_add)} memory-provider file(s) "
            f"stored outside {display_hermes_home()}."
        )

    if skipped_external:
        print(
            f"\n  Skipped {len(skipped_external)} memory-provider path(s) "
            f"outside your home directory (not portable):"
        )
        for p in sorted(skipped_external)[:10]:
            print(f"    {p}")

    if skipped_dirs:
        print("\n  Excluded directories:")
        for d in sorted(skipped_dirs):
            print(f"    {d}/")

    if errors:
        print(f"\n  Warnings ({len(errors)} files skipped):")
        for e in errors[:10]:
            print(e)
        if len(errors) > 10:
            print(f"  ... and {len(errors) - 10} more")

    if not errors:
        print(f"\nRestore with: hermes import {out_path.name}")


# ---------------------------------------------------------------------------
# Import
# ---------------------------------------------------------------------------

def _validate_backup_zip(zf: zipfile.ZipFile) -> tuple[bool, str]:
    """Check that a zip looks like a Hermes backup.

    Returns (ok, reason).
    """
    names = zf.namelist()
    if not names:
        return False, "zip archive is empty"

    # Look for telltale files that a hermes home would have
    markers = {"config.yaml", ".env", "state.db"}
    found = set()
    for n in names:
        # Could be at the root or one level deep (if someone zipped the directory)
        basename = Path(n).name
        if basename in markers:
            found.add(basename)

    if not found:
        return False, (
            "zip does not appear to be a Hermes backup "
            "(no config.yaml, .env, or state databases found)"
        )

    return True, ""


def _detect_prefix(zf: zipfile.ZipFile) -> str:
    """Detect if the zip has a common directory prefix wrapping all entries.

    Some tools zip as `.hermes/config.yaml` instead of `config.yaml`.
    Returns the prefix to strip (empty string if none).
    """
    names = [n for n in zf.namelist() if not n.endswith("/")]
    if not names:
        return ""

    # Find common prefix
    parts_list = [Path(n).parts for n in names]

    # Check if all entries share a common first directory
    first_parts = {p[0] for p in parts_list if len(p) > 1}
    if len(first_parts) == 1:
        prefix = first_parts.pop()
        # Only strip if it looks like a hermes dir name
        if prefix in {".hermes", "hermes"}:
            return prefix + "/"

    return ""


def run_import(args) -> None:
    """Restore a Hermes backup from a zip file."""
    zip_path = Path(args.zipfile).expanduser().resolve()

    if not zip_path.is_file():
        print(f"Error: File not found: {zip_path}")
        sys.exit(1)

    if not zipfile.is_zipfile(zip_path):
        print(f"Error: Not a valid zip file: {zip_path}")
        sys.exit(1)

    hermes_root = get_default_hermes_root()

    with zipfile.ZipFile(zip_path, "r") as zf:
        # Validate
        ok, reason = _validate_backup_zip(zf)
        if not ok:
            print(f"Error: {reason}")
            sys.exit(1)

        prefix = _detect_prefix(zf)
        members = [n for n in zf.namelist() if not n.endswith("/")]
        file_count = len(members)

        print(f"Backup contains {file_count} files")
        print(f"Target: {display_hermes_home()}")

        if prefix:
            print(f"Detected archive prefix: {prefix!r} (will be stripped)")

        # Check for existing installation
        has_config = (hermes_root / "config.yaml").exists()
        has_env = (hermes_root / ".env").exists()

        if (has_config or has_env) and not args.force:
            print()
            print("Warning: Target directory already has Hermes configuration.")
            print("Importing will overwrite existing files with backup contents.")
            print()
            try:
                answer = input("Continue? [y/N] ").strip().lower()
            except (EOFError, KeyboardInterrupt):
                print("\nAborted.")
                sys.exit(1)
            if answer not in {"y", "yes"}:
                print("Aborted.")
                return

        # Extract
        print(f"\nImporting {file_count} files ...")
        hermes_root.mkdir(parents=True, exist_ok=True)

        errors = []
        restored = 0
        restored_external = 0
        skipped_runtime: list[str] = []
        home_dir = Path.home().resolve()
        t0 = time.monotonic()

        for member in members:
            # External memory-provider state captured under the reserved
            # ``_external/`` arc prefix restores to its original home-relative
            # location (e.g. ~/.honcho/config.json), NOT under HERMES_HOME.
            if member.startswith(_EXTERNAL_PREFIX):
                ext_rel = member[len(_EXTERNAL_PREFIX):]
                if not ext_rel:
                    continue
                target = home_dir / ext_rel
                # Security: the resolved target must stay under the home dir.
                try:
                    target.resolve().relative_to(home_dir)
                except ValueError:
                    errors.append(f"  {member}: path traversal blocked")
                    continue
                try:
                    target.parent.mkdir(parents=True, exist_ok=True)
                    with zf.open(member) as src, open(target, "wb") as dst:
                        dst.write(src.read())
                    # External provider configs commonly hold credentials.
                    if target.suffix in {".json", ".env", ".conf"} or target.name in _SECRET_FILE_NAMES:
                        try:
                            os.chmod(target, 0o600)
                        except OSError:
                            pass
                    restored += 1
                    restored_external += 1
                except (PermissionError, OSError) as exc:
                    errors.append(f"  {member}: {exc}")
                if restored % 500 == 0:
                    print(f"  {restored}/{file_count} files ...")
                continue

            # Strip prefix if detected
            if prefix and member.startswith(prefix):
                rel = member[len(prefix):]
            else:
                rel = member

            if not rel:
                continue

            # Never overwrite volatile gateway/process runtime state. These are
            # namespaced to the machine/container the backup was taken on;
            # clobbering them (especially gateway_state.json) breaks the gateway
            # reconciler on the target and disconnects hosted instances from the
            # Nous portal. Matched by basename so both the root profile and
            # named profiles (profiles/<name>/gateway_state.json) are covered.
            if Path(rel).name in _IMPORT_SKIP_NAMES:
                skipped_runtime.append(rel)
                continue

            target = hermes_root / rel

            # Security: reject absolute paths and traversals
            try:
                target.resolve().relative_to(hermes_root.resolve())
            except ValueError:
                errors.append(f"  {rel}: path traversal blocked")
                continue

            try:
                target.parent.mkdir(parents=True, exist_ok=True)
                with zf.open(member) as src, open(target, "wb") as dst:
                    dst.write(src.read())
                if target.name in _SECRET_FILE_NAMES:
                    os.chmod(target, 0o600)
                restored += 1
            except (PermissionError, OSError) as exc:
                errors.append(f"  {rel}: {exc}")

            if restored % 500 == 0:
                print(f"  {restored}/{file_count} files ...")

        elapsed = time.monotonic() - t0

        # Summary
        print()
        print(f"Import complete: {restored} files restored in {elapsed:.1f}s")
        print(f"  Target: {display_hermes_home()}")

        if restored_external:
            print(
                f"\n  Restored {restored_external} memory-provider file(s) to "
                f"their original location(s) outside {display_hermes_home()}."
            )

        if errors:
            print(f"\n  Warnings ({len(errors)} files skipped):")
            for e in errors[:10]:
                print(e)
            if len(errors) > 10:
                print(f"  ... and {len(errors) - 10} more")

        if skipped_runtime:
            print(
                f"\n  Preserved {len(skipped_runtime)} runtime state "
                f"file(s) (kept this machine's, not the backup's):"
            )
            for rel in sorted(skipped_runtime)[:10]:
                print(f"    {rel}")
            if len(skipped_runtime) > 10:
                print(f"    ... and {len(skipped_runtime) - 10} more")

        # Post-import: restore profile wrapper scripts
        profiles_dir = hermes_root / "profiles"
        restored_profiles = []
        if profiles_dir.is_dir():
            try:
                from hermes_cli.profiles import (
                    create_wrapper_script, check_alias_collision,
                    _is_wrapper_dir_in_path, _get_wrapper_dir,
                )
                for entry in sorted(profiles_dir.iterdir()):
                    if not entry.is_dir():
                        continue
                    profile_name = entry.name
                    # Only create wrappers for directories with config
                    if not (entry / "config.yaml").exists() and not (entry / ".env").exists():
                        continue
                    collision = check_alias_collision(profile_name)
                    if collision:
                        print(f"  Skipped alias '{profile_name}': {collision}")
                        restored_profiles.append((profile_name, False))
                    else:
                        wrapper = create_wrapper_script(profile_name)
                        restored_profiles.append((profile_name, wrapper is not None))

                if restored_profiles:
                    created = [n for n, ok in restored_profiles if ok]
                    skipped = [n for n, ok in restored_profiles if not ok]
                    if created:
                        print(f"\n  Profile aliases restored: {', '.join(created)}")
                    if skipped:
                        print(f"  Profile aliases skipped:  {', '.join(skipped)}")
                    if not _is_wrapper_dir_in_path():
                        print(f"\n  Note: {_get_wrapper_dir()} is not in your PATH.")
                        print('  Add to your shell config (~/.bashrc or ~/.zshrc):')
                        print('    export PATH="$HOME/.local/bin:$PATH"')
            except ImportError:
                # hermes_cli.profiles might not be available (fresh install)
                if any(profiles_dir.iterdir()):
                    print("\n  Profiles detected but aliases could not be created.")
                    print("  Run: hermes profile list  (after installing hermes)")

        # Guidance
        print()
        if not (hermes_root / "hermes-agent").is_dir():
            print("Note: The hermes-agent codebase was not included in the backup.")
            print("  If this is a fresh install, run: hermes update")

        if restored_profiles:
            gw_profiles = [n for n, _ in restored_profiles]
            print("\nTo re-enable gateway services for profiles:")
            for pname in gw_profiles:
                print(f"  hermes -p {pname} gateway install")

        print("Done. Your Hermes configuration has been restored.")


# ---------------------------------------------------------------------------
# Quick state snapshots (used by /snapshot slash command and hermes backup --quick)
# ---------------------------------------------------------------------------

# Critical state files to include in quick snapshots (relative to HERMES_HOME).
# Everything else is either regeneratable (logs, cache) or managed separately
# (skills, repo, sessions/).
#
# Entries may be individual files OR directories.  Directories are captured
# recursively; missing entries are silently skipped.  Pairing data lives in
# platform-specific JSON blobs outside state.db, so it's listed here explicitly
# — `hermes update` snapshots this set before pulling so approved-user lists
# are recoverable if anything goes wrong (issue #15733).
_QUICK_STATE_FILES = (
    "state.db",
    "config.yaml",
    ".env",
    "auth.json",
    "cron/jobs.json",
    "cron/executions.db",
    "gateway_state.json",
    "channel_directory.json",
    "channel_aliases.json",
    "processes.json",
    "gateway/discord_message_recovery.db",  # Discord reconnect replay ledger
    # Per-profile user-created stores that live outside the git checkout and
    # are therefore destroyed if the update flow removes/replaces the file and
    # the post-update schema-init re-creates an empty one (issue #52889). All
    # are at $HERMES_HOME/<name> for the default/root profile; on non-root
    # profiles the real path is outside HERMES_HOME and the entry is silently
    # skipped (best-effort, same as the pairing stores). SQLite DBs are copied
    # WAL-safely via _safe_copy_db.
    "projects.db",                      # per-profile project store
    "response_store.db",                # gateway conversation history / tool payloads
    "memory_store.db",                  # holographic memory facts/entities
    "verification_evidence.db",         # agent verification audit trail
    "kanban.db",                        # default board (back-compat <root>/kanban.db)
    "kanban/boards",                    # non-default boards: each <slug>/kanban.db + board metadata (workspaces/ + attachments/ are skipped as regenerable)
    # Pairing stores (generic + per-platform JSONs outside state.db)
    "pairing",                          # legacy location (gateway/pairing.py)
    "platforms/pairing",                # new location (gateway/pairing.py)
    "feishu_comment_pairing.json",      # Feishu comment subscription pairings
)

_QUICK_SNAPSHOTS_DIR = "state-snapshots"
_QUICK_DEFAULT_KEEP = 20


def _quick_snapshot_root(hermes_home: Optional[Path] = None) -> Path:
    home = hermes_home or get_hermes_home()
    return home / _QUICK_SNAPSHOTS_DIR


def create_quick_snapshot(
    label: Optional[str] = None,
    hermes_home: Optional[Path] = None,
    keep: Optional[int] = None,
    max_file_size: Optional[int] = None,
) -> Optional[str]:
    """Create one atomic quick snapshot while holding the shared backup slot."""
    home = hermes_home or get_hermes_home()
    with _backup_operation_lock(home):
        return _create_quick_snapshot_locked(
            label=label,
            hermes_home=home,
            keep=keep,
            max_file_size=max_file_size,
        )


def _create_quick_snapshot_locked(
    label: Optional[str] = None,
    hermes_home: Optional[Path] = None,
    keep: Optional[int] = None,
    max_file_size: Optional[int] = None,
) -> Optional[str]:
    """Create a quick state snapshot of critical files.

    Copies STATE_FILES to a timestamped directory under state-snapshots/.
    Auto-prunes old snapshots beyond the keep limit.

    Args:
        max_file_size: When set, individual files larger than this many bytes
            are skipped (with a printed warning) instead of copied. Used by
            the pre-update safety snapshot so a multi-GB ``state.db`` can
            never stall ``hermes update`` or silently eat disk — the small
            pairing/cron/config files the snapshot exists to protect are
            always captured. ``None`` (default) copies everything, which
            preserves manual ``/snapshot`` and ``hermes backup --quick``
            behavior.

    Returns:
        Snapshot ID (timestamp-based), or None if no files found.
    """
    home = hermes_home or get_hermes_home()
    root = _quick_snapshot_root(home)

    def _too_large(path: Path, rel_name: str) -> bool:
        """True (and warn) when ``path`` exceeds the max_file_size cap."""
        if max_file_size is None:
            return False
        try:
            size = path.stat().st_size
        except OSError:
            return False
        if size <= max_file_size:
            return False
        print(
            f"  ⚠ Snapshot: skipping {rel_name} "
            f"({_format_size(size)} exceeds {_format_size(max_file_size)} limit)"
        )
        logger.warning(
            "Quick snapshot skipped %s: %d bytes exceeds %d byte limit",
            rel_name,
            size,
            max_file_size,
        )
        return True

    ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
    base_snap_id = f"{ts}-{label}" if label else ts
    snap_id = base_snap_id
    suffix = 2
    while (root / snap_id).exists():
        snap_id = f"{base_snap_id}-{suffix}"
        suffix += 1
    snap_dir = root / snap_id
    staging_dir = root / f".{snap_id}.{os.getpid()}.partial"
    shutil.rmtree(staging_dir, ignore_errors=True)
    staging_dir.mkdir(parents=True, exist_ok=False)
    logger.info("quick snapshot phase=copy status=started id=%s", snap_id)

    manifest: Dict[str, int] = {}  # rel_path -> file size
    failed_dbs: list[str] = []  # present *.db that could not be snapshotted
    # #68805: track protected DB files skipped for size — they are snapshot
    # incompleteness just like a failed copy, so pruning must be suppressed
    # to preserve the older complete snapshot that may contain the only
    # recoverable database.
    oversized_skipped: list[str] = []

    for rel in _QUICK_STATE_FILES:
        src = home / rel
        if not src.exists():
            continue

        if src.is_dir():
            # Walk the directory and record each file individually in the
            # manifest so restore can treat them uniformly.  Empty dirs are
            # skipped (nothing to snapshot).
            for sub in src.rglob("*"):
                if not sub.is_file():
                    continue
                sub_rel = sub.relative_to(home).as_posix()
                # Skip heavy, regenerable per-board subtrees (scratch
                # workspaces and task attachments can be large); we only need
                # the board databases + their metadata to restore a board.
                if "/workspaces/" in f"/{sub_rel}/" or "/attachments/" in f"/{sub_rel}/":
                    continue
                if _too_large(sub, sub_rel):
                    if sub.suffix == ".db":
                        oversized_skipped.append(sub_rel)
                    continue
                dst = staging_dir / sub_rel
                dst.parent.mkdir(parents=True, exist_ok=True)
                try:
                    # Route SQLite DBs through the WAL-safe backup() path so a
                    # board DB with an open WAL (the gateway may hold it at
                    # snapshot time) is captured consistently.
                    if sub.suffix == ".db":
                        if not _safe_copy_db(sub, dst):
                            failed_dbs.append(sub_rel)
                            print(
                                f"  ⚠ Snapshot: SQLite safe copy FAILED for {sub_rel} "
                                f"— file may be locked or corrupted"
                            )
                            if is_zeroed_sqlite_file(sub):
                                print(
                                    f"  ⚠ Snapshot: {sub_rel} looks ZEROED "
                                    f"(no SQLite header; {sub.stat().st_size} bytes of NULs?)"
                                )
                            continue
                    else:
                        shutil.copy2(sub, dst)
                    manifest[sub_rel] = dst.stat().st_size
                except (OSError, PermissionError) as exc:
                    logger.warning("Could not snapshot %s: %s", sub_rel, exc)
            continue

        if not src.is_file():
            continue

        if _too_large(src, rel):
            if src.suffix == ".db":
                oversized_skipped.append(rel)
            continue

        dst = staging_dir / rel
        dst.parent.mkdir(parents=True, exist_ok=True)

        try:
            if src.suffix == ".db":
                if not _safe_copy_db(src, dst):
                    failed_dbs.append(rel)
                    print(
                        f"  ⚠ Snapshot: SQLite safe copy FAILED for {rel} "
                        f"— file may be locked or corrupted"
                    )
                    if is_zeroed_sqlite_file(src):
                        print(
                            f"  ⚠ Snapshot: {rel} looks ZEROED "
                            f"(no SQLite header; {src.stat().st_size} bytes)"
                        )
                    continue
            else:
                shutil.copy2(src, dst)
            manifest[rel] = dst.stat().st_size
        except (OSError, PermissionError) as exc:
            logger.warning("Could not snapshot %s: %s", rel, exc)

    if failed_dbs:
        # Critical: update path used to log-and-continue with exit 0, so a
        # missing state.db backup looked like a successful pre-update snapshot
        # (#68474). Surface this on stdout where operators actually look.
        print(
            "  ⚠ CRITICAL: could not snapshot DB file(s): "
            + ", ".join(failed_dbs)
        )
        print(
            "  ⚠ If sessions disappear after update, check "
            f"{root} and run: hermes snapshot list"
        )
        logger.error(
            "Quick snapshot failed to capture DB file(s): %s",
            ", ".join(failed_dbs),
        )

    if not manifest:
        shutil.rmtree(staging_dir, ignore_errors=True)
        if failed_dbs:
            # Distinguish "nothing to snapshot" from "state.db present but unreadable"
            print(
                "  ⚠ Snapshot aborted: no files captured "
                f"(failed DBs: {', '.join(failed_dbs)})"
            )
        return None

    # Write manifest
    meta = {
        "id": snap_id,
        "timestamp": ts,
        "label": label,
        "file_count": len(manifest),
        "total_size": sum(manifest.values()),
        "files": manifest,
        "failed_dbs": failed_dbs,
        "oversized_skipped": oversized_skipped,
    }
    with open(staging_dir / "manifest.json", "w", encoding="utf-8") as f:
        json.dump(meta, f, indent=2)

    os.replace(staging_dir, snap_dir)

    # Auto-prune. Defaults preserve historical manual /snapshot behavior; callers
    # with known high-churn safety snapshots (for example pre-update) can pass a
    # smaller keep value so large state.db copies do not accumulate indefinitely.
    # #68805 review: skip pruning when a present DB failed to capture OR was
    # skipped for size — either way the snapshot is incomplete and the older
    # snapshot may contain the only recoverable database.
    incomplete = failed_dbs or oversized_skipped
    if not incomplete:
        _prune_quick_snapshots(root, keep=_QUICK_DEFAULT_KEEP if keep is None else keep)
    else:
        if oversized_skipped:
            print(
                "  ⚠ Skipping snapshot prune: DB file(s) skipped for size: "
                + ", ".join(oversized_skipped)
            )
            logger.warning(
                "Quick snapshot skipped oversized DB file(s): %s",
                ", ".join(oversized_skipped),
            )
        logger.warning(
            "Skipping snapshot prune because %d DB(s) failed to capture "
            "and/or %d were oversized — preserving older snapshots as "
            "recovery source",
            len(failed_dbs), len(oversized_skipped),
        )

    logger.info(
        "quick snapshot phase=copy status=complete id=%s files=%d bytes=%d",
        snap_id,
        len(manifest),
        sum(manifest.values()),
    )
    return snap_id


def list_quick_snapshots(
    limit: int = 20,
    hermes_home: Optional[Path] = None,
) -> List[Dict[str, Any]]:
    """List existing quick state snapshots, most recent first."""
    root = _quick_snapshot_root(hermes_home)
    if not root.exists():
        return []

    results = []
    for d in sorted(root.iterdir(), reverse=True):
        if not d.is_dir() or d.name.startswith(".") or d.name.endswith(".partial"):
            continue
        manifest_path = d / "manifest.json"
        if manifest_path.exists():
            try:
                with open(manifest_path, encoding="utf-8") as f:
                    results.append(json.load(f))
            except (json.JSONDecodeError, OSError):
                results.append({"id": d.name, "file_count": 0, "total_size": 0})
        if len(results) >= limit:
            break

    return results


def restore_quick_snapshot(
    snapshot_id: str,
    hermes_home: Optional[Path] = None,
) -> bool:
    """Restore state from a quick snapshot.

    Overwrites current state files with the snapshot's copies.
    Returns True if at least one file was restored.
    """
    home = hermes_home or get_hermes_home()
    root = _quick_snapshot_root(home)

    # Security: reject snapshot_id values that contain path separators or
    # traversal sequences so that `root / snapshot_id` stays inside root.
    if not snapshot_id or "/" in snapshot_id or "\\" in snapshot_id or snapshot_id in (".", ".."):
        logger.error("Invalid snapshot_id: %s", snapshot_id)
        return False

    snap_dir = root / snapshot_id

    # Confirm the resolved path is still inside root (handles symlinks etc.)
    try:
        snap_dir.resolve().relative_to(root.resolve())
    except ValueError:
        logger.error("Snapshot path traversal blocked for id: %s", snapshot_id)
        return False

    if not snap_dir.is_dir():
        return False

    manifest_path = snap_dir / "manifest.json"
    if not manifest_path.exists():
        return False

    with open(manifest_path, encoding="utf-8") as f:
        meta = json.load(f)

    restored = 0
    for rel in meta.get("files", {}):
        # Security: reject absolute paths and traversals in manifest entries
        src = snap_dir / rel
        try:
            src.resolve().relative_to(snap_dir.resolve())
        except ValueError:
            logger.error("Manifest path traversal blocked: %s", rel)
            continue

        dst = home / rel
        try:
            dst.resolve().relative_to(home.resolve())
        except ValueError:
            logger.error("Manifest path traversal blocked: %s", rel)
            continue

        if not src.exists():
            continue

        dst.parent.mkdir(parents=True, exist_ok=True)

        try:
            if dst.suffix == ".db":
                # Atomic-ish replace for databases
                tmp = dst.parent / f".{dst.name}.snap_restore"
                shutil.copy2(src, tmp)
                dst.unlink(missing_ok=True)
                shutil.move(str(tmp), str(dst))
            else:
                shutil.copy2(src, dst)
            restored += 1
        except (OSError, PermissionError) as exc:
            logger.error("Failed to restore %s: %s", rel, exc)

    logger.info("Restored %d files from snapshot %s", restored, snapshot_id)
    return restored > 0


# Relative path of the cron job database inside HERMES_HOME. Kept in sync with
# the entry in ``_QUICK_STATE_FILES`` and with ``cron/jobs.py``'s ``JOBS_FILE``.
_CRON_JOBS_REL = "cron/jobs.json"


def _count_cron_jobs(path: Path) -> Optional[int]:
    """Return the number of cron jobs stored in ``path``.

    The canonical on-disk shape is ``{"jobs": [...]}`` (see ``cron/jobs.py``).
    A legacy bare-list shape (``[...]``) is also honoured.

    Returns:
        The job count for any *valid, readable* JSON document, or ``None`` if
        the file is missing or cannot be parsed. ``None`` means "unknown" —
        callers must not treat it as "zero jobs", because acting on an
        unreadable file could mask a real corruption the user needs to see.
    """
    if not path.is_file():
        return None
    try:
        # utf-8-sig: same dialect as cron/jobs.load_jobs — Windows editors
        # may leave a UTF-8 BOM that plain utf-8 json.load rejects. Without
        # it a BOM'd jobs.json counts as "unreadable" (None) and the
        # post-update cron-loss auto-restore safety net silently disables.
        with open(path, "r", encoding="utf-8-sig") as f:
            data = json.load(f)
    except (OSError, json.JSONDecodeError):
        return None
    if isinstance(data, dict):
        jobs = data.get("jobs", [])
        return len(jobs) if isinstance(jobs, list) else None
    if isinstance(data, list):
        return len(data)
    return None


def restore_cron_jobs_if_emptied(
    snapshot_id: str,
    hermes_home: Optional[Path] = None,
) -> Optional[Dict[str, Any]]:
    """Safety net for silent cron-job loss across ``hermes update``.

    Config-version migrations have been observed to leave ``cron/jobs.json``
    valid-but-empty after an update, silently dropping every scheduled job
    (issue #34600). The desktop scheduler can also overwrite the file with its
    own small set of internally-tracked crons, causing partial loss (issue
    #52144).

    This compares the *current* job count against the pre-update snapshot. If
    the live file now has **fewer** jobs than the snapshot, the snapshot copy
    of ``cron/jobs.json`` is restored in place.

    The check is deliberately conservative — it only ever restores when there
    is unambiguous evidence of loss (snapshot had more jobs than live file),
    so a user who genuinely deleted jobs during/after the update is never
    second-guessed, and an unreadable live file (count ``None``) is left
    untouched so real corruption still surfaces.

    Args:
        snapshot_id: The pre-update quick-snapshot id (from
            :func:`create_quick_snapshot`).
        hermes_home: Override for the Hermes home directory (tests).

    Returns:
        ``None`` when no action was taken (the common, healthy path). On a
        successful restore, a dict ``{"restored": True, "job_count": N,
        "snapshot_id": ...}`` so the caller can warn the user.
    """
    if not snapshot_id:
        return None

    home = hermes_home or get_hermes_home()
    live_path = home / _CRON_JOBS_REL

    live_count = _count_cron_jobs(live_path)
    # ``None`` (missing or unparseable) is intentionally left alone — that's a
    # different failure mode the user should see rather than have papered over.
    if live_count is None:
        return None

    snap_path = _quick_snapshot_root(home) / snapshot_id / _CRON_JOBS_REL
    snap_count = _count_cron_jobs(snap_path)
    if not snap_count:  # None or 0 — nothing worth restoring
        return None

    # Restore when live has FEWER jobs than the pre-update snapshot.
    # Catches both total loss (0 vs N) and partial loss (1 vs 19) — the
    # desktop scheduler can overwrite jobs.json with its own small set of
    # internally-tracked crons after an update/restart.
    if live_count >= snap_count:
        return None

    try:
        live_path.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(snap_path, live_path)
    except (OSError, PermissionError) as exc:
        logger.error(
            "Cron jobs were emptied during update but auto-restore failed: %s", exc
        )
        return None

    logger.warning(
        "Restored %d cron job(s) from pre-update snapshot %s "
        "(live file had %d job(s), snapshot had %d — jobs were lost during migration)",
        snap_count,
        snapshot_id,
        live_count,
        snap_count,
    )
    return {"restored": True, "job_count": snap_count, "snapshot_id": snapshot_id}


def _prune_quick_snapshots(root: Path, keep: int = _QUICK_DEFAULT_KEEP) -> int:
    """Remove oldest quick snapshots beyond the keep limit. Returns count deleted."""
    if not root.exists():
        return 0

    dirs = sorted(
        (
            d
            for d in root.iterdir()
            if d.is_dir() and not d.name.startswith(".") and not d.name.endswith(".partial")
        ),
        key=lambda d: d.name,
        reverse=True,
    )

    deleted = 0
    for d in dirs[keep:]:
        try:
            shutil.rmtree(d)
            deleted += 1
        except OSError as exc:
            logger.warning("Failed to prune snapshot %s: %s", d.name, exc)

    return deleted


def prune_quick_snapshots(
    keep: int = _QUICK_DEFAULT_KEEP,
    hermes_home: Optional[Path] = None,
) -> int:
    """Manually prune quick snapshots. Returns count deleted."""
    return _prune_quick_snapshots(_quick_snapshot_root(hermes_home), keep=keep)


def run_quick_backup(args) -> None:
    """CLI entry point for hermes backup --quick."""
    label = getattr(args, "label", None)
    snap_id = create_quick_snapshot(label=label)
    if snap_id:
        print(f"State snapshot created: {snap_id}")
        snaps = list_quick_snapshots()
        print(f"  {len(snaps)} snapshot(s) stored in {display_hermes_home()}/state-snapshots/")
        print(f"  Restore with: /snapshot restore {snap_id}")
    else:
        print("No state files found to snapshot.")


# ---------------------------------------------------------------------------
# Shared full-zip backup helper
# ---------------------------------------------------------------------------

def _write_full_zip_backup(out_path: Path, hermes_root: Path) -> Optional[Path]:
    """Single-flight wrapper for automatic full zip backups."""
    try:
        with _backup_operation_lock(hermes_root):
            return _write_full_zip_backup_locked(out_path, hermes_root)
    except BackupInProgressError as exc:
        logger.warning("Full-zip backup skipped: %s", exc)
        return None


def _write_full_zip_backup_locked(out_path: Path, hermes_root: Path) -> Optional[Path]:
    """Write a full zip snapshot of ``hermes_root`` to ``out_path``.

    Uses the same exclusion rules and SQLite safe-copy as :func:`run_backup`.
    Returns the output path on success, None on failure (nothing to back up,
    or write error — caller should surface the outcome but not raise).
    """
    scan_started = time.monotonic()
    logger.info("automatic backup phase=scan status=started")
    files_to_add: list[tuple[Path, Path]] = []
    try:
        for dirpath, dirnames, filenames in os.walk(hermes_root, followlinks=False):
            dp = Path(dirpath)
            # Prune excluded directories in-place so os.walk doesn't descend
            dirnames[:] = [d for d in dirnames if d not in _EXCLUDED_DIRS]

            for fname in filenames:
                fpath = dp / fname
                try:
                    rel = fpath.relative_to(hermes_root)
                except ValueError:
                    continue

                if _should_skip_backup_file(fpath, rel, out_path):
                    continue

                files_to_add.append((fpath, rel))
    except OSError as exc:
        logger.warning("Full-zip backup: walk failed: %s", exc)
        return None

    if not files_to_add:
        return None

    logger.info(
        "automatic backup phase=scan status=complete duration_ms=%.1f files=%d",
        (time.monotonic() - scan_started) * 1000,
        len(files_to_add),
    )

    archive_started = time.monotonic()
    try:
        with _atomic_output_path(out_path) as archive_path, zipfile.ZipFile(
            archive_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6
        ) as zf:
            for index, (abs_path, rel_path) in enumerate(files_to_add, 1):
                try:
                    if abs_path.suffix == ".db":
                        # Stage the snapshot alongside the output zip so that the
                        # temp file lives on the same filesystem.  The system
                        # default (/tmp) may be a small tmpfs that cannot hold
                        # large databases, causing silent backup incompleteness.
                        with tempfile.NamedTemporaryFile(
                            suffix=".db", delete=False, dir=str(out_path.parent)
                        ) as tmp:
                            tmp_db = Path(tmp.name)
                        try:
                            if not _safe_copy_db(abs_path, tmp_db):
                                logger.warning(
                                    "Full-zip backup aborted: SQLite snapshot failed for %s",
                                    rel_path,
                                )
                                raise _SQLiteSnapshotError(str(rel_path))
                            zf.write(tmp_db, arcname=str(rel_path))
                        finally:
                            tmp_db.unlink(missing_ok=True)
                    else:
                        zf.write(abs_path, arcname=str(rel_path))
                except (PermissionError, OSError, ValueError) as exc:
                    logger.debug("Skipping %s in zip backup: %s", rel_path, exc)
                    continue
                if index % 500 == 0:
                    logger.info(
                        "automatic backup phase=archive status=progress completed=%d total=%d",
                        index,
                        len(files_to_add),
                    )
    except (OSError, _SQLiteSnapshotError) as exc:
        logger.warning("Full-zip backup: zip write failed: %s", exc)
        # ``_atomic_output_path`` already removed the hidden partial.  Do not
        # unlink ``out_path`` here: it may be a previous valid backup that the
        # atomic publisher deliberately preserved.
        return None

    logger.info(
        "automatic backup phase=archive status=complete duration_ms=%.1f files=%d bytes=%d",
        (time.monotonic() - archive_started) * 1000,
        len(files_to_add),
        out_path.stat().st_size,
    )

    return out_path


# ---------------------------------------------------------------------------
# Pre-update auto-backup
# ---------------------------------------------------------------------------

_PRE_UPDATE_BACKUPS_DIR = "backups"
_PRE_UPDATE_PREFIX = "pre-update-"
_PRE_UPDATE_DEFAULT_KEEP = 5


def _pre_update_backup_dir(hermes_home: Optional[Path] = None) -> Path:
    home = hermes_home or get_hermes_home()
    return home / _PRE_UPDATE_BACKUPS_DIR


def _prune_pre_update_backups(backup_dir: Path, keep: int) -> int:
    """Remove oldest pre-update backups beyond the keep limit.

    Returns the number of files deleted.  Only touches files matching
    ``pre-update-*.zip`` so hand-made zips dropped in the same directory
    are never touched.

    ``keep`` is floored to 1 because this helper is only called immediately
    after a fresh backup is written: deleting that backup right after the
    user paid the disk/CPU cost to create it would leave them worse off
    than no backup at all (and the wrapper in ``main.py`` would still print
    a misleading ``Saved: <path>`` line for a file that no longer exists).
    Operators who genuinely don't want a backup should set
    ``updates.pre_update_backup: off`` in config — that gates creation.
    """
    keep = max(keep, 1)
    if not backup_dir.exists():
        return 0

    backups = sorted(
        (p for p in backup_dir.iterdir()
         if p.is_file() and p.name.startswith(_PRE_UPDATE_PREFIX) and p.suffix.lower() == ".zip"),
        key=lambda p: p.name,
        reverse=True,
    )

    deleted = 0
    for p in backups[keep:]:
        try:
            p.unlink()
            deleted += 1
        except OSError as exc:
            logger.warning("Failed to prune backup %s: %s", p.name, exc)

    return deleted


def create_pre_update_backup(
    hermes_home: Optional[Path] = None,
    keep: int = _PRE_UPDATE_DEFAULT_KEEP,
) -> Optional[Path]:
    """Create a full zip backup of HERMES_HOME under ``backups/``.

    Mirrors :func:`run_backup` (same exclusion rules, same SQLite safe-copy)
    but writes to ``<HERMES_HOME>/backups/pre-update-<timestamp>.zip`` and
    auto-prunes old pre-update backups.

    Returns the path to the created zip, or ``None`` if no files were
    found or the backup could not be created.  Never raises — the caller
    (``hermes update``) should continue even if the backup fails.
    """
    hermes_root = hermes_home or get_default_hermes_root()
    if not hermes_root.is_dir():
        return None

    backup_dir = _pre_update_backup_dir(hermes_root)
    try:
        backup_dir.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        logger.warning("Could not create pre-update backup dir %s: %s", backup_dir, exc)
        return None

    stamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
    out_path = backup_dir / f"{_PRE_UPDATE_PREFIX}{stamp}.zip"

    result = _write_full_zip_backup(out_path, hermes_root)
    if result is None:
        return None

    _prune_pre_update_backups(backup_dir, keep=keep)
    return out_path


# ---------------------------------------------------------------------------
# Pre-migration auto-backup (used by `hermes claw migrate`)
# ---------------------------------------------------------------------------

_PRE_MIGRATION_PREFIX = "pre-migration-"
_PRE_MIGRATION_DEFAULT_KEEP = 5


def _prune_pre_migration_backups(backup_dir: Path, keep: int) -> int:
    """Remove oldest pre-migration backups beyond the keep limit.

    Only touches files matching ``pre-migration-*.zip`` so other backups in
    the same directory are never touched.
    """
    keep = max(keep, 0)
    if not backup_dir.exists():
        return 0

    backups = sorted(
        (p for p in backup_dir.iterdir()
         if p.is_file() and p.name.startswith(_PRE_MIGRATION_PREFIX) and p.suffix.lower() == ".zip"),
        key=lambda p: p.name,
        reverse=True,
    )

    deleted = 0
    for p in backups[keep:]:
        try:
            p.unlink()
            deleted += 1
        except OSError as exc:
            logger.warning("Failed to prune pre-migration backup %s: %s", p.name, exc)

    return deleted


def create_pre_migration_backup(
    hermes_home: Optional[Path] = None,
    keep: int = _PRE_MIGRATION_DEFAULT_KEEP,
) -> Optional[Path]:
    """Create a full zip backup of HERMES_HOME under ``backups/`` before a
    ``hermes claw migrate`` apply.

    Shares implementation with :func:`create_pre_update_backup` via
    ``_write_full_zip_backup`` — same exclusions, same SQLite safe-copy,
    restorable with ``hermes import <archive>``.  Writes to
    ``<HERMES_HOME>/backups/pre-migration-<timestamp>.zip`` and auto-prunes
    old pre-migration backups.

    Returns the path to the created zip, or ``None`` if nothing was found
    to back up (fresh install) or the write failed.  Never raises — the
    caller decides whether to abort or proceed.
    """
    hermes_root = hermes_home or get_default_hermes_root()
    if not hermes_root.is_dir():
        return None

    # Reuses the shared backups/ directory so `hermes import` and the
    # update-backup listing pick up pre-migration archives too.
    backup_dir = _pre_update_backup_dir(hermes_root)
    try:
        backup_dir.mkdir(parents=True, exist_ok=True)
    except OSError as exc:
        logger.warning("Could not create pre-migration backup dir %s: %s", backup_dir, exc)
        return None

    stamp = datetime.now().strftime("%Y-%m-%d-%H%M%S")
    out_path = backup_dir / f"{_PRE_MIGRATION_PREFIX}{stamp}.zip"

    result = _write_full_zip_backup(out_path, hermes_root)
    if result is None:
        return None

    _prune_pre_migration_backups(backup_dir, keep=keep)
    return out_path
