"""Offline, non-destructive recovery for a damaged Hermes session database.

The source is never opened by SQLite: it and its WAL/SHM/journal sidecars are copied to a disposable
work dir first. Canonical rows are copied into a fresh current-schema database; derived FTS tables and
migration bookkeeping are rebuilt, not copied; the result is never installed over the active database.
"""

from __future__ import annotations

import json
import os
import shutil
import sqlite3
import tempfile
from contextlib import contextmanager
from pathlib import Path
from typing import Any, Callable, Iterator, Optional

from hermes_cli.timefmt import EPOCH_MAX, EPOCH_MIN
from hermes_state import SessionDB
from hermes_state_common import FTS_STORAGE_VERSION, SCHEMA_VERSION
from hermes_state_repair import _db_opens_cleanly


ProgressCallback = Callable[[dict[str, Any]], None]
_CANONICAL_TABLES = (
    "system_prompts", "sessions", "messages", "session_model_usage", "compression_locks", "gateway_routing",
    "async_delegations",
)
_TOPIC_TABLES = ("telegram_dm_topic_mode", "telegram_dm_topic_bindings")


def _init_delivery_ledger_schema(conn: sqlite3.Connection) -> None:
    from gateway.delivery_ledger import _initialize_schema
    _initialize_schema(conn)


# state.db tables created lazily by a gateway module (base ``SessionDB`` never creates them on a fresh
# destination) -> the initializer owning their DDL. Recovery creates them before copying so owed rows
# don't silently vanish from a "complete" salvage. Register new lazy tables HERE, not as ``if table ==``.
# See #100313, #86236.
_AUXILIARY_TABLE_SCHEMAS: dict[str, Callable[[sqlite3.Connection], None]] = {
    "delivery_obligations": _init_delivery_ledger_schema,
}
_AUXILIARY_TABLES = tuple(_AUXILIARY_TABLE_SCHEMAS)
_INVENTORY_TABLES = (*_CANONICAL_TABLES, "state_meta", *_TOPIC_TABLES, *_AUXILIARY_TABLES)

# Derived-index / optional-schema markers: a fresh destination regenerates these, never copies them.
_GENERATED_META_KEYS = frozenset({
    "fts_storage_version", "fts_optimize_available", "fts_rebuild_high_water", "fts_rebuild_progress",
    "fts_cjk_stale", "fts_cjk_rebuild_high_water", "fts_cjk_rebuild_progress", "telegram_dm_topic_schema_version",
    "fts_tool_full_content_high_water",  # retired marker; never copied into a recovered store
})
_SIDECAR_SUFFIXES = ("", "-wal", "-shm", "-journal")
_MINIMUM_SPACE_HEADROOM = 256 * 1024 * 1024
_MAX_SALVAGE_RANGE_QUERIES = 10_000
_MIN_SQLITE_ROWID = -(2**63)
_MAX_SQLITE_ROWID = 2**63 - 1


class SessionRecoveryError(RuntimeError):
    """Base error for offline session recovery."""


class SessionRecoverySafetyError(SessionRecoveryError):
    """Raised before recovery when a path or overwrite guard fails."""


class SessionRecoverySourceError(SessionRecoveryError):
    """Raised when the source cannot provide the required canonical tables."""


def _sidecar_path(db_path: Path, suffix: str) -> Path:
    return db_path if not suffix else db_path.with_name(db_path.name + suffix)


def _resolved_output_path(path: Path) -> Path:
    """Resolve a not-yet-created output path without requiring it to exist."""
    return path.expanduser().parent.resolve(strict=True) / path.name


def _validate_paths(
    source_path: Path, output_path: Optional[Path] = None, work_dir: Optional[Path] = None,
) -> tuple[Path, Optional[Path], Path]:
    source = source_path.expanduser().resolve(strict=True)
    if not source.is_file():
        raise SessionRecoverySafetyError(f"Source is not a file: {source}")
    output: Optional[Path] = None
    if output_path is not None:
        output = _resolved_output_path(output_path)
        protected = {_sidecar_path(source, suffix).resolve(strict=False) for suffix in _SIDECAR_SUFFIXES}
        if output.resolve(strict=False) in protected:
            raise SessionRecoverySafetyError(
                "The recovery output must not be the source database or one of its journal sidecars."
            )
        for suffix in _SIDECAR_SUFFIXES:
            candidate = _sidecar_path(output, suffix)
            if os.path.lexists(candidate):
                raise SessionRecoverySafetyError(f"Refusing to overwrite existing recovery output: {candidate}")
    work_root = (
        work_dir.expanduser().resolve(strict=True)
        if work_dir is not None
        else (output.parent if output is not None else source.parent)
    )
    if not work_root.is_dir():
        raise SessionRecoverySafetyError(f"Recovery work directory is not a directory: {work_root}")
    return source, output, work_root


def _source_fingerprint(source: Path) -> dict[str, dict[str, int]]:
    fingerprint: dict[str, dict[str, int]] = {}
    for suffix in _SIDECAR_SUFFIXES:
        path = _sidecar_path(source, suffix)
        if path.exists():
            stat = path.stat()
            fingerprint[suffix or "main"] = {"size": stat.st_size, "mtime_ns": stat.st_mtime_ns}
    return fingerprint


def _format_bytes(value: int) -> str:
    units = ("B", "KiB", "MiB", "GiB", "TiB")
    amount = float(value)
    for unit in units:
        if amount < 1024 or unit == units[-1]:
            return f"{amount:.1f} {unit}"
        amount /= 1024
    return f"{value} B"


def _same_filesystem(left: Path, right: Path) -> bool:
    try:
        return os.stat(left).st_dev == os.stat(right).st_dev
    except OSError:  # both dirs exist (_validate_paths); fallback for incomplete st_dev support
        return left.anchor.casefold() == right.anchor.casefold()


def _disk_space_preflight(source: Path, work_root: Path, output_parent: Optional[Path]) -> dict[str, Any]:
    """Require space for the disposable bundle, output, and safety headroom."""
    bundle_bytes = sum(info["size"] for info in _source_fingerprint(source).values())
    # The v23 external-content rebuild is usually much smaller than a legacy database, but estimating
    # with the complete bundle avoids betting the user's disk on that.
    output_allowance = bundle_bytes if output_parent is not None else 0
    headroom = max(_MINIMUM_SPACE_HEADROOM, int((bundle_bytes + output_allowance) * 0.05))
    work_free = int(shutil.disk_usage(work_root).free)
    report: dict[str, Any] = {
        "source_bundle_bytes": bundle_bytes, "estimated_output_bytes": output_allowance, "headroom_bytes": headroom,
        "work_dir": str(work_root), "work_dir_free_bytes": work_free,
    }
    if output_parent is None or _same_filesystem(work_root, output_parent):
        required = bundle_bytes + output_allowance + headroom
        report.update(shared_filesystem=True, work_dir_required_bytes=required)
        if work_free < required:
            raise SessionRecoverySafetyError(
                f"Not enough free disk space for a safe recovery copy: {_format_bytes(work_free)} available at "
                f"{work_root}, {_format_bytes(required)} required ({_format_bytes(bundle_bytes)} source bundle + "
                f"{_format_bytes(output_allowance)} output allowance + {_format_bytes(headroom)} headroom). "
                "Use --work-dir or --output on a filesystem with more free space."
            )
        return report
    output_free = int(shutil.disk_usage(output_parent).free)
    work_required = bundle_bytes + headroom
    output_required = output_allowance + headroom
    report.update({
        "shared_filesystem": False, "work_dir_required_bytes": work_required, "output_dir": str(output_parent),
        "output_dir_free_bytes": output_free, "output_dir_required_bytes": output_required,
    })
    shortages = [
        f"{where}: {_format_bytes(free)} available, {_format_bytes(required)} required"
        for where, free, required in (
            (work_root, work_free, work_required), (output_parent, output_free, output_required),
        )
        if free < required
    ]
    if shortages:
        raise SessionRecoverySafetyError(
            f"Not enough free disk space for safe recovery: {'; '.join(shortages)}. "
            "Choose work/output filesystems with more free space."
        )
    return report


def _copy_source_bundle(source: Path, snapshot_dir: Path) -> tuple[Path, list[str]]:
    """Copy the source DB bundle aside so SQLite never opens the original.

    The whole copy runs inside ``offline_file_access`` (holds the connection-lifecycle lock). Recovery
    normally runs as its own CLI process against an offline file, so the refusal should never fire; the
    guard keeps this path consistent with ``hermes_state_repair._backup_db_file``.

    Checking for a live connection and *then* copying would be a check/use race: a connection could open in
    that window, and the copy's ``close()`` would cancel its POSIX advisory locks -- the failure class
    ``hermes_cli.sqlite_safe_read`` exists to prevent (see #71724). Holding the lock means no connection can
    appear mid-copy, across the main file and every sidecar.
    """
    from hermes_cli.sqlite_safe_read import LiveConnectionError, offline_file_access
    snapshot_source = snapshot_dir / source.name
    copied: list[str] = []
    try:
        with offline_file_access(source, what="snapshot"):
            for suffix in _SIDECAR_SUFFIXES:
                source_part = _sidecar_path(source, suffix)
                if source_part.exists():
                    destination_part = _sidecar_path(snapshot_source, suffix)
                    shutil.copy2(source_part, destination_part)
                    copied.append(destination_part.name)
    except LiveConnectionError as exc:
        raise SessionRecoverySafetyError(str(exc)) from exc
    return snapshot_source, copied


def _table_columns(conn: sqlite3.Connection, table: str) -> list[str]:
    return [str(row[1]) for row in conn.execute(f'PRAGMA table_info("{table}")')]


def _count_rows(conn: sqlite3.Connection, table: str) -> int:
    return int(conn.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0])


def _connect(path: Path) -> sqlite3.Connection:
    """Autocommit connection with a short busy timeout (source snapshot or fresh output)."""
    return sqlite3.connect(str(path), isolation_level=None, timeout=1.0)


@contextmanager
def _immediate_transaction(conn: sqlite3.Connection) -> Iterator[None]:
    """``BEGIN IMMEDIATE`` ... ``COMMIT``, rolling back on any exception."""
    conn.execute("BEGIN IMMEDIATE")
    try:
        yield
        conn.execute("COMMIT")
    except BaseException:
        conn.execute("ROLLBACK")
        raise


def _compatible_columns(
    source: sqlite3.Connection, destination: sqlite3.Connection, table: str, result: dict[str, Any],
) -> Optional[list[str]]:
    """Columns shared by source and destination; sets a terminal status and returns None otherwise."""
    source_columns = _table_columns(source, table)
    columns = [column for column in _table_columns(destination, table) if column in source_columns]
    result["columns"] = columns
    if not source_columns:
        result["status"] = "missing"
        return None
    if not columns:
        result["status"] = "failed"
        result["error"] = "source and destination have no compatible columns"
        return None
    return columns


def _quoted_columns(columns: list[str]) -> tuple[str, str]:
    """``("a", "b")`` column list and matching ``?, ?`` placeholders."""
    return ", ".join(f'"{column}"' for column in columns), ", ".join("?" for _ in columns)


def _copy_rows(
    source: sqlite3.Connection, destination: sqlite3.Connection, select_sql: str, params: tuple[Any, ...],
    insert_sql: str, *, table: str, chunk_size: int, progress_cb: Optional[ProgressCallback],
    expected_rows: Optional[int], result: dict[str, Any],
) -> dict[str, Any]:
    """Chunked straight copy; fills ``status``/``error`` on ``result``."""
    try:
        cursor = source.execute(select_sql, params)
        while True:
            rows = cursor.fetchmany(chunk_size)
            if not rows:
                break
            with _immediate_transaction(destination):
                destination.executemany(insert_sql, rows)
            result["copied_rows"] += len(rows)
            if progress_cb is not None:
                progress_cb({"table": table, "copied_rows": result["copied_rows"], "source_rows": expected_rows})
    except sqlite3.DatabaseError as exc:
        result["status"] = "partial" if result["copied_rows"] else "failed"
        result["error"] = str(exc)
        return result
    if expected_rows is None or result["copied_rows"] == expected_rows:
        result["status"] = "complete"
    else:
        result["status"] = "partial"
        result["error"] = f"copied {result['copied_rows']} of {expected_rows} readable rows"
    return result


def _table_inventory(conn: sqlite3.Connection, table: str) -> dict[str, Any]:
    result: dict[str, Any] = {"available": False, "columns": [], "rows": None}
    try:
        columns = _table_columns(conn, table)
        if not columns:
            return result
        result.update(available=True, columns=columns)
        result["rows"] = _count_rows(conn, table)
    except sqlite3.DatabaseError as exc:
        result["error"] = str(exc)
    return result


def _journal_mode(conn: sqlite3.Connection) -> Optional[str]:
    row = conn.execute("PRAGMA journal_mode").fetchone()
    return str(row[0]).lower() if row else None


def _inspect_connection(conn: sqlite3.Connection) -> dict[str, Any]:
    conn.execute("PRAGMA writable_schema=ON")
    report: dict[str, Any] = {"tables": {}, "errors": [], "warnings": []}
    try:
        report["journal_mode"] = _journal_mode(conn)
    except sqlite3.DatabaseError as exc:
        report["journal_mode"] = None
        # Journal metadata is context, not canonical data: a damaged pragma must not block readable rows.
        report["warnings"].append(f"journal mode: {exc}")
    report["tables"] = {table: _table_inventory(conn, table) for table in _INVENTORY_TABLES}
    for required in ("sessions", "messages"):
        table_report = report["tables"][required]
        if not table_report.get("available") or table_report.get("rows") is None:
            report["errors"].append(f"required table {required} is not completely readable")
    report["recoverable"] = not report["errors"]
    return report


def _snapshot_and_inspect(
    source: Path, work_root: Path,
) -> tuple[tempfile.TemporaryDirectory[str], Path, dict[str, Any]]:
    before = _source_fingerprint(source)
    temp_dir = tempfile.TemporaryDirectory(prefix="hermes-session-recovery-", dir=str(work_root))
    try:
        snapshot_source, copied = _copy_source_bundle(source, Path(temp_dir.name))
        if _source_fingerprint(source) != before:
            raise SessionRecoverySafetyError(
                "The source database bundle changed while it was being copied. Stop every Hermes process using this "
                "profile and retry. This includes the interactive `hermes` CLI session this command may have been "
                "launched from: a running parent CLI writes session bookkeeping (compression ticks, context "
                "tracking) to state.db in the background and counts as a Hermes process even after the gateway is "
                "stopped. Run the recovery from a fresh shell with no `hermes` session open, or point --source at an "
                "immutable snapshot copy of the database."
            )
        conn = _connect(snapshot_source)
        try:
            inspection = _inspect_connection(conn)
        finally:
            conn.close()
        inspection.update(source_bundle=copied, source_fingerprint=before)
        return temp_dir, snapshot_source, inspection
    except BaseException:
        temp_dir.cleanup()
        raise


def inspect_session_database(source_path: Path, *, work_dir: Optional[Path] = None) -> dict[str, Any]:
    """Inspect canonical table readability without opening the source itself."""
    source, _, work_root = _validate_paths(source_path, work_dir=work_dir)
    disk_space = _disk_space_preflight(source, work_root, None)
    temp_dir, _, inspection = _snapshot_and_inspect(source, work_root)
    try:
        return {
            "operation": "inspect", "source": str(source), "disk_space": disk_space, **inspection,
            "source_unchanged": _source_fingerprint(source) == inspection["source_fingerprint"],
        }
    finally:
        temp_dir.cleanup()


def _fresh_destination(output: Path, *, topic_tables: bool = False) -> sqlite3.Connection:
    """Initialize a current-schema database at ``output`` and open it with foreign keys off."""
    with SessionDB(db_path=output) as destination_db:
        if topic_tables:
            destination_db.apply_telegram_topic_migration()
    conn = _connect(output)
    conn.execute("PRAGMA foreign_keys=OFF")
    return conn


def _copy_table(
    source: sqlite3.Connection, destination: sqlite3.Connection, table: str, *, salvage: bool, chunk_size: int,
    progress_cb: Optional[ProgressCallback], source_rows: Optional[int],
) -> dict[str, Any]:
    """Copy one canonical table: straight chunked copy, or rowid-range salvage when ``salvage``."""
    copy_kwargs = dict(chunk_size=chunk_size, progress_cb=progress_cb, source_rows=source_rows)
    if table == "state_meta":
        return _copy_state_meta(source, destination, salvage=salvage, **copy_kwargs)
    if salvage:
        return _copy_table_salvage(source, destination, table, **copy_kwargs)
    result: dict[str, Any] = {"source_rows": source_rows, "copied_rows": 0}
    columns = _compatible_columns(source, destination, table, result)
    if columns is None:
        return result
    quoted, placeholders = _quoted_columns(columns)
    return _copy_rows(
        source, destination, f'SELECT {quoted} FROM "{table}"', (),
        f'INSERT INTO "{table}" ({quoted}) VALUES ({placeholders})', table=table, chunk_size=chunk_size,
        progress_cb=progress_cb, expected_rows=source_rows, result=result,
    )


def _append_skipped_range(ranges: list[dict[str, Any]], low: int, high: int, error: str) -> None:
    """Record skipped rowid ranges, merging adjacent same-error ranges (not one entry per row)."""
    if ranges and ranges[-1]["high"] + 1 == low and ranges[-1]["error"] == error:
        ranges[-1]["high"] = high
        return
    ranges.append({"low": low, "high": high, "error": error})


def _salvage_rowid_bounds(source: sqlite3.Connection, table: str) -> dict[str, Any]:
    """Find the readable rowid edges without scanning the complete table."""
    result: dict[str, Any] = {"errors": [], "fallback_edges": []}
    rows: dict[str, Optional[int]] = {"low": None, "high": None}
    for edge, direction in (("low", "ASC"), ("high", "DESC")):
        try:
            row = source.execute(f'SELECT rowid FROM "{table}" ORDER BY rowid {direction} LIMIT 1').fetchone()
            if row is not None:
                rows[edge] = int(row[0])
        except sqlite3.DatabaseError as exc:
            result["errors"].append(f"{edge} rowid: {exc}")
    if rows["low"] is None and rows["high"] is None:
        result["empty" if not result["errors"] else "unavailable"] = True
        return result

    # An ordered LIMIT 1 walks the table b-tree and dies on a damaged edge leaf, while the
    # aggregate lets the planner answer from any covering index (every Hermes table has at
    # least a PRIMARY KEY autoindex). Ask it before falling back to the synthetic domain:
    # bisecting from INT64_MIN burned the whole query budget on a 4-row table (#98050).
    missing = [edge for edge in ("low", "high") if rows[edge] is None]
    if missing:
        try:
            aggregate = source.execute(f'SELECT min(rowid), max(rowid) FROM "{table}"').fetchone()
        except sqlite3.DatabaseError as exc:
            result["errors"].append(f"aggregate rowid bounds: {exc}")
        else:
            for edge, value in zip(("low", "high"), aggregate):
                if rows[edge] is None and value is not None:
                    rows[edge] = int(value)
                    result.setdefault("aggregate_edges", []).append(edge)

    # A damaged edge can stop one ordered probe. Keep the readable edge and bound the other side by the
    # SQLite rowid domain, so bisection never assumes user databases hold only positive ids.
    if rows["low"] is None:
        rows["low"] = _MIN_SQLITE_ROWID
        result["fallback_edges"].append("low")
    if rows["high"] is None:
        rows["high"] = _MAX_SQLITE_ROWID
        result["fallback_edges"].append("high")
    result.update(rows)
    # Bisecting the whole synthetic domain tail used to exhaust the range-query budget before readable
    # tail rows were copied (#80205); gallop outward from the surviving edge for a finite bound first.
    if result["fallback_edges"]:
        result["edge_probes"] = []
        for edge, anchor_edge in (("high", "low"), ("low", "high")):
            if edge in result["fallback_edges"]:
                probe = _probe_populated_edge(source, table, edge=edge, anchor=int(result[anchor_edge]))
                result["edge_probes"].append(probe)
                if probe["capped"]:
                    result[edge] = int(probe["bound"])
    return result


def _probe_populated_edge(source: sqlite3.Connection, table: str, *, edge: str, anchor: int) -> dict[str, Any]:
    """Finite bound for a damaged rowid edge: gallop outward from the readable ``anchor`` with doubling
    offsets. A clean "no rows beyond X" caps the domain; an error or a hit keeps growing (~64 probes max).
    """
    ascending = edge == "high"
    comparison = ">" if ascending else "<"
    probe_sql = (
        f'SELECT rowid FROM "{table}" WHERE rowid {comparison} ? '
        f'ORDER BY rowid {"ASC" if ascending else "DESC"} LIMIT 1'
    )
    domain_limit = _MAX_SQLITE_ROWID if ascending else _MIN_SQLITE_ROWID
    result: dict[str, Any] = {"edge": edge, "probes": 0, "capped": False}
    position = anchor
    span = 1
    while True:
        candidate = position + span if ascending else position - span
        if (ascending and candidate >= domain_limit) or (not ascending and candidate <= domain_limit):
            result["bound"] = domain_limit  # no clean empty-tail answer: keep the domain fallback
            return result
        result["probes"] += 1
        try:
            row = source.execute(probe_sql, (candidate,)).fetchone()
        except sqlite3.DatabaseError:  # damage on the probe path: inconclusive, widen further
            span *= 2
            continue
        if row is None:  # nothing beyond candidate: the synthetic domain tail is provably empty
            result.update(bound=candidate, capped=True)
            return result
        position = int(row[0])  # rows exist beyond; advance. Span never resets -> O(log range)
        span *= 2


class _RowidRangeSalvage:
    """Bisecting rowid-range copy for one table; counters live on the shared ``result`` dict."""

    def __init__(
        self, source: sqlite3.Connection, destination: sqlite3.Connection, table: str, columns: list[str], *,
        chunk_size: int, progress_cb: Optional[ProgressCallback], source_rows: Optional[int], insert_prefix: str,
        row_filter: Optional[Callable[[tuple[Any, ...], tuple[str, ...]], bool]], result: dict[str, Any],
    ) -> None:
        self.source, self.destination, self.table = source, destination, table
        self.chunk_size, self.progress_cb, self.source_rows = chunk_size, progress_cb, source_rows
        self.row_filter, self.result = row_filter, result
        self.column_names = tuple(columns)
        quoted, placeholders = _quoted_columns(columns)
        self.select_sql = f'SELECT rowid, {quoted} FROM "{table}" WHERE rowid BETWEEN ? AND ? ORDER BY rowid'
        self.insert_sql = f'{insert_prefix} INTO "{table}" ({quoted}) VALUES ({placeholders})'
        self.exact_sql = f'SELECT {quoted} FROM "{table}" WHERE rowid = ?'
        self.stopped_at_query_limit = False

    def _keep(self, values: list[tuple[Any, ...]]) -> list[tuple[Any, ...]]:
        return values if self.row_filter is None else [r for r in values if self.row_filter(r, self.column_names)]

    def _skip(self, low: int, high: int, error: str) -> None:
        _append_skipped_range(self.result["skipped_rowid_ranges"], low, high, error)

    def recover_exact_rowid(self, rowid: int) -> bool:
        """Salvage one row by exact-key lookup (issue #80205).

        A singleton range scan (``rowid BETWEEN x AND x ORDER BY rowid``) must advance the cursor
        past ``x`` to prove the range is exhausted; when the *next* cell or page is damaged that
        advance raises AFTER the row was produced, and the driver discards the already-fetched row.
        """
        result = self.result
        result["range_queries"] += 1
        try:
            row = self.source.execute(self.exact_sql, (rowid,)).fetchone()
        except sqlite3.DatabaseError:
            return False
        if row is None:
            return True  # genuinely absent: nothing to skip
        value = tuple(row)
        if not self._keep([value]):
            result["excluded_rows"] += 1
            return True
        try:
            with _immediate_transaction(self.destination):
                self.destination.execute(self.insert_sql, value)
        except sqlite3.IntegrityError as exc:
            # A phantom row from a damaged page (NULL in a NOT NULL column, FK to nothing) is rejected
            # by the destination schema; report it as a skipped singleton instead of aborting the table.
            result["destination_rejected_rows"] += 1
            self._skip(rowid, rowid, f"destination constraint rejected row: {exc}")
            return True
        result["copied_rows"] += 1
        result["exact_lookup_recovered"] += 1
        return True

    def copy_range(self, low: int, high: int) -> None:
        """Copy ``[low, high]``; on a read error bisect the unread remainder, exact-lookup singletons."""
        result = self.result
        if low > high:
            return
        if result["range_queries"] >= _MAX_SALVAGE_RANGE_QUERIES:
            self.stopped_at_query_limit = True
            self._skip(low, high, "salvage range query limit reached")
            return
        result["range_queries"] += 1
        last_committed_rowid: Optional[int] = None
        try:
            cursor = self.source.execute(self.select_sql, (low, high))
            while True:
                fetched = cursor.fetchmany(self.chunk_size)
                if not fetched:
                    return
                values = [tuple(row[1:]) for row in fetched]
                included = self._keep(values)
                if included:
                    with _immediate_transaction(self.destination):
                        self.destination.executemany(self.insert_sql, included)
                result["copied_rows"] += len(included)
                result["excluded_rows"] += len(values) - len(included)
                last_committed_rowid = int(fetched[-1][0])
                if self.progress_cb is not None:
                    self.progress_cb({
                        "table": self.table, "copied_rows": result["copied_rows"], "source_rows": self.source_rows,
                        "skipped_ranges": len(result["skipped_rowid_ranges"]),
                    })
        except sqlite3.DatabaseError as exc:
            retry_low = last_committed_rowid + 1 if last_committed_rowid is not None else low
            if retry_low > high:
                return
            if retry_low == high:
                if not self.recover_exact_rowid(retry_low):
                    self._skip(retry_low, high, str(exc))
                return
            midpoint = retry_low + (high - retry_low) // 2
            self.copy_range(retry_low, midpoint)
            self.copy_range(midpoint + 1, high)


def _copy_table_salvage(
    source: sqlite3.Connection, destination: sqlite3.Connection, table: str, *, chunk_size: int,
    progress_cb: Optional[ProgressCallback], source_rows: Optional[int], insert_prefix: str = "INSERT",
    row_filter: Optional[Callable[[tuple[Any, ...], tuple[str, ...]], bool]] = None,
) -> dict[str, Any]:
    """Best-effort rowid-range copy that continues past damaged source pages."""
    result: dict[str, Any] = {
        "mode": "rowid_range_salvage", "source_rows": source_rows, "copied_rows": 0, "excluded_rows": 0,
        "columns": [], "range_queries": 0, "exact_lookup_recovered": 0, "destination_rejected_rows": 0,
        "skipped_rowid_ranges": [],
    }
    columns = _compatible_columns(source, destination, table, result)
    if columns is None:
        return result
    bounds = _salvage_rowid_bounds(source, table)
    result["rowid_bounds"] = bounds
    if bounds.get("empty"):
        result["status"] = "complete"
        return result
    if bounds.get("low") is None or bounds.get("high") is None:
        details = "; ".join(bounds.get("errors") or [])
        result["status"] = "failed"
        result["error"] = "could not determine a rowid range for salvage" + (f": {details}" if details else "")
        return result
    salvage = _RowidRangeSalvage(
        source, destination, table, columns, chunk_size=chunk_size, progress_cb=progress_cb, source_rows=source_rows,
        insert_prefix=insert_prefix, row_filter=row_filter, result=result,
    )
    salvage.copy_range(int(bounds["low"]), int(bounds["high"]))
    skipped_ranges = result["skipped_rowid_ranges"]
    result["skipped_rowid_span"] = sum(item["high"] - item["low"] + 1 for item in skipped_ranges)
    result["query_limit_reached"] = salvage.stopped_at_query_limit
    if skipped_ranges:
        result["status"] = "partial" if result["copied_rows"] else "failed"
        result["error"] = f"{len(skipped_ranges)} rowid range(s) skipped"
    elif source_rows is not None and result["copied_rows"] + result["excluded_rows"] != source_rows:
        result["status"] = "partial"
        copied, excluded = result["copied_rows"], result["excluded_rows"]
        result["error"] = f"copied {copied} and excluded {excluded} of {source_rows} source rows"
    else:
        result["status"] = "complete"
    return result


def _state_meta_result(source_rows: Optional[int], **extra: Any) -> dict[str, Any]:
    return {
        "source_meta_rows": source_rows, "copied_rows": 0, "columns": ["key", "value"],
        "excluded_keys": sorted(_GENERATED_META_KEYS), **extra,
    }


def _state_meta_precheck(
    source: sqlite3.Connection, destination: sqlite3.Connection, source_rows: Optional[int], *, salvage: bool,
) -> Optional[dict[str, Any]]:
    """Terminal ``state_meta`` result when the key/value schema is unusable, else ``None``.

    In salvage mode an unusable-but-PRESENT table reports ``failed``, not ``missing``: verification
    only escalates ``failed``/``partial`` into a warning + ``loss_detected``, so ``missing`` would
    silently drop real metadata and still claim ``complete=True``.
    """
    extra = {"mode": "rowid_range_salvage"} if salvage else {}
    source_columns = _table_columns(source, "state_meta")
    if not {"key", "value"}.issubset(source_columns):
        if salvage and source_columns:  # present but unusable: real data loss
            found = ", ".join(source_columns) or "none"
            error = f"source state_meta exists but is missing the key/value columns (found: {found})"
            return _state_meta_result(source_rows, **extra, status="failed", error=error)
        return _state_meta_result(source_rows, **extra, status="missing")  # genuinely absent: nothing lost
    if not {"key", "value"}.issubset(_table_columns(destination, "state_meta")):
        error = "destination state_meta schema is incomplete"
        return _state_meta_result(source_rows, **extra, status="failed", error=error)
    return None


def _copy_state_meta(
    source: sqlite3.Connection, destination: sqlite3.Connection, *, salvage: bool, chunk_size: int,
    progress_cb: Optional[ProgressCallback], source_rows: Optional[int],
) -> dict[str, Any]:
    """Copy user metadata rows; derived FTS/topic keys (``_GENERATED_META_KEYS``) are regenerated, not copied."""
    problem = _state_meta_precheck(source, destination, source_rows, salvage=salvage)
    if problem is not None:
        return problem
    if salvage:
        def keep_user_meta(row: tuple[Any, ...], columns: tuple[str, ...]) -> bool:
            return str(row[columns.index("key")]) not in _GENERATED_META_KEYS
        result = _copy_table_salvage(
            source, destination, "state_meta", chunk_size=chunk_size, progress_cb=progress_cb,
            source_rows=source_rows, insert_prefix="INSERT OR REPLACE", row_filter=keep_user_meta,
        )
        result["source_meta_rows"] = result.pop("source_rows")
        result["excluded_keys"] = sorted(_GENERATED_META_KEYS)
        return result
    placeholders = ", ".join("?" for _ in _GENERATED_META_KEYS)
    params = tuple(_GENERATED_META_KEYS)
    filtered_source_rows: Optional[int] = None
    try:
        filtered_source_rows = int(
            source.execute(f"SELECT COUNT(*) FROM state_meta WHERE key NOT IN ({placeholders})", params).fetchone()[0]
        )
    except sqlite3.DatabaseError:
        pass  # the copy loop below will return the concrete read error
    return _copy_rows(
        source, destination, f"SELECT key, value FROM state_meta WHERE key NOT IN ({placeholders})", params,
        "INSERT OR REPLACE INTO state_meta(key, value) VALUES (?, ?)", table="state_meta", chunk_size=chunk_size,
        progress_cb=progress_cb, expected_rows=filtered_source_rows, result=_state_meta_result(source_rows),
    )


def _placeholder_titles(destination: sqlite3.Connection, prefix: str) -> Iterator[str]:
    """Yield ``[<prefix> N] session metadata was unreadable`` titles not yet present in ``sessions``."""
    sequence = 1
    while True:
        title = f"[{prefix} {sequence}] session metadata was unreadable"
        sequence += 1
        if destination.execute("SELECT 1 FROM sessions WHERE title = ? LIMIT 1", (title,)).fetchone() is None:
            yield title


def _reconstruct_missing_sessions(destination: sqlite3.Connection) -> dict[str, Any]:
    """Recreate placeholder session rows for salvaged orphaned messages.

    When ``sessions`` is damaged worse than ``messages``, deleting the recovered messages as orphans
    would discard the only readable copy of the user's data. Instead synthesize a minimal session per
    orphaned ``session_id`` (``started_at`` = earliest surviving message) so FKs hold.
    """
    result: dict[str, Any] = {"sessions_reconstructed": 0, "messages_retained": 0}
    if not (_table_columns(destination, "sessions") and _table_columns(destination, "messages")):
        return result
    orphaned = destination.execute(
        "SELECT m.session_id, MIN(m.timestamp), COUNT(*) FROM messages AS m WHERE m.session_id IS NOT NULL AND NOT "
        "EXISTS (SELECT 1 FROM sessions WHERE sessions.id = m.session_id) GROUP BY m.session_id"
    ).fetchall()
    if not orphaned:
        return result
    titles = _placeholder_titles(destination, "recovered")
    for session_id, first_timestamp, message_count in orphaned:
        started_at = float(first_timestamp) if first_timestamp is not None else 0.0
        title = next(titles)
        cursor = destination.execute(
            "INSERT INTO sessions (id, source, started_at, title, message_count) VALUES (?, 'recovered', ?, ?, ?)",
            (session_id, started_at, title, int(message_count)),
        )
        if cursor.rowcount != 1:
            raise sqlite3.IntegrityError(f"failed to reconstruct missing session {session_id!r}")
        result["sessions_reconstructed"] += 1
        result["messages_retained"] += int(message_count)
    return result


def _reconcile(destination: sqlite3.Connection, table: str, where: str, mutation: str) -> int:
    """Count rows of ``table`` matching ``where``; run ``mutation WHERE where`` only when there are any."""
    count = int(destination.execute(f'SELECT COUNT(*) FROM "{table}" WHERE {where}').fetchone()[0])
    if count:
        destination.execute(f"{mutation} WHERE {where}")
    return count


_DEPENDENT_TABLES = ("messages", "session_model_usage", "compression_locks", "telegram_dm_topic_bindings")
_DANGLING_TOOL_PIN = (
    "length(tool_names) = 64 AND NOT EXISTS (SELECT 1 FROM system_prompts WHERE system_prompts.hash = sessions.tool_names)")
_RELINK_COUNTERS = ("session_prompt_refs_cleared", "sessions_parent_cleared")


def _cleanup_partial_orphans(destination: sqlite3.Connection) -> dict[str, Any]:
    """Reconcile references to unsalvageable sessions. Messages are never dropped for lack of a
    session row (placeholders are rebuilt first); only rows still orphaned after that are removed."""
    result: dict[str, Any] = {
        "session_prompt_refs_cleared": 0, "system_prompts_removed": 0, "sessions_parent_cleared": 0,
        "sessions_reconstructed": 0, "messages_retained": 0,
        **{f"{table}_removed": 0 for table in _DEPENDENT_TABLES},
    }
    with _immediate_transaction(destination):
        # Rebuild owners BEFORE any orphan deletion.
        result.update(_reconstruct_missing_sessions(destination))
        result["sessions_parent_cleared"] = _reconcile(
            destination, "sessions",
            "parent_session_id IS NOT NULL AND NOT EXISTS ("
            "SELECT 1 FROM sessions AS parent WHERE parent.id = sessions.parent_session_id)",
            "UPDATE sessions SET parent_session_id = NULL",
        )
        result["session_prompt_refs_cleared"] = _reconcile(
            destination, "sessions",
            "system_prompt_hash IS NOT NULL AND NOT EXISTS ("
            "SELECT 1 FROM system_prompts WHERE system_prompts.hash = sessions.system_prompt_hash)",
            "UPDATE sessions SET system_prompt_hash = NULL",
        )
        # A tools[] pin is a system_prompts row too, referenced by hash from sessions.tool_names
        # (legacy rows hold an inline JSON list, never 64 chars of hex).
        result["session_prompt_refs_cleared"] += _reconcile(
            destination, "sessions",
            _DANGLING_TOOL_PIN,
            "UPDATE sessions SET tool_names = NULL",
        )
        result["system_prompts_removed"] = _reconcile(
            destination, "system_prompts",
            "NOT EXISTS (SELECT 1 FROM sessions WHERE sessions.system_prompt_hash = system_prompts.hash) "
            "AND NOT EXISTS (SELECT 1 FROM sessions WHERE sessions.tool_names = system_prompts.hash)",
            "DELETE FROM system_prompts",
        )
        for table in _DEPENDENT_TABLES:
            if _table_columns(destination, table):
                result[f"{table}_removed"] = _reconcile(
                    destination, table,
                    f'NOT EXISTS (SELECT 1 FROM sessions WHERE sessions.id = "{table}".session_id)',
                    f'DELETE FROM "{table}"',
                )
    # Only destructive/relinking actions count: reconstruction counters describe RETAINED data.
    result["total_removed_or_relinked"] = sum(
        int(result[key]) for key in (*_RELINK_COUNTERS, *(f"{table}_removed" for table in _DEPENDENT_TABLES))
    )
    return result


def _verify_structure(conn: sqlite3.Connection, verification: dict[str, Any]) -> None:
    """Integrity, foreign keys, journal mode, schema version and FTS meta of the recovered file."""
    errors = verification["errors"]
    integrity_rows = [str(row[0]) for row in conn.execute("PRAGMA integrity_check").fetchall()]
    verification["integrity_check"] = integrity_rows
    if integrity_rows != ["ok"]:
        errors.append("PRAGMA integrity_check did not return exactly 'ok'")
    foreign_key_rows = [list(row) for row in conn.execute("PRAGMA foreign_key_check").fetchall()]
    verification["foreign_key_check"] = foreign_key_rows
    if foreign_key_rows:
        errors.append("foreign key violations remain")
    verification["journal_mode"] = _journal_mode(conn)
    schema_row = conn.execute("SELECT version FROM schema_version LIMIT 1").fetchone()
    verification["schema_version"] = int(schema_row[0]) if schema_row else None
    if verification["schema_version"] != SCHEMA_VERSION:
        errors.append(f"schema version is {verification['schema_version']}, expected {SCHEMA_VERSION}")
    meta = dict(conn.execute("SELECT key, value FROM state_meta WHERE key LIKE 'fts_%'").fetchall())
    meta = {str(key): value for key, value in meta.items()}
    verification["fts_meta"] = meta
    if meta.get("fts_storage_version") != str(FTS_STORAGE_VERSION):
        errors.append("fresh FTS storage version was not established")
    pending_keys = sorted(
        key for key in _GENERATED_META_KEYS if key.startswith("fts_") and key != "fts_storage_version" and key in meta
    )
    verification["pending_fts_keys"] = pending_keys
    if pending_keys:
        errors.append("derived FTS transition markers remain in the recovered database")


def _verify_fts_indexes(conn: sqlite3.Connection, verification: dict[str, Any]) -> None:
    fts_checks: dict[str, str] = {}
    for table in ("messages_fts", "messages_fts_trigram", "messages_fts_cjk"):
        if not _table_columns(conn, table):
            continue
        try:
            conn.execute(f'INSERT INTO "{table}" ("{table}") VALUES (\'integrity-check\')')
            conn.execute(f'SELECT 1 FROM "{table}" WHERE "{table}" MATCH \'""\' LIMIT 1').fetchone()
            fts_checks[table] = "ok"
        except sqlite3.DatabaseError as exc:
            fts_checks[table] = str(exc)
            verification["errors"].append(f"{table} integrity check failed: {exc}")
    verification["fts_checks"] = fts_checks


def _verify_row_counts(
    conn: sqlite3.Connection, verification: dict[str, Any], *, expected_counts: dict[str, Optional[int]],
    copy_report: dict[str, dict[str, Any]], allow_partial: bool, orphan_cleanup: Optional[dict[str, Any]],
) -> None:
    """Compare recovered counts and copy statuses against the source; classify shortfalls as loss."""

    def flag(message: str, *, soft: bool) -> None:
        """Record data loss as a warning (``soft``) or as a verification error."""
        if soft:
            verification["warnings"].append(message)
            verification["loss_detected"] = True
        else:
            verification["errors"].append(message)
    counts = {table: _count_rows(conn, table) for table in _INVENTORY_TABLES if _table_columns(conn, table)}
    verification["table_counts"] = counts
    for table in ("sessions", "messages", *_AUXILIARY_TABLES):
        expected = expected_counts.get(table)
        if expected is not None and counts.get(table) != expected:
            flag(f"{table} count is {counts.get(table)}, expected {expected}", soft=allow_partial)
    cleanup = orphan_cleanup or {}
    rebuilt_sessions = int(cleanup.get("sessions_reconstructed") or 0)
    retained_messages = int(cleanup.get("messages_retained") or 0)
    removed_messages = int(cleanup.get("messages_removed") or 0)
    # A wholly unreadable sessions b-tree is recoverable when every output parent was rebuilt from
    # the surviving messages and none were dropped: data loss, but not structural failure.
    sessions_fully_reconstructed = bool(
        rebuilt_sessions > 0 and counts.get("sessions") == rebuilt_sessions
        and counts.get("messages") == retained_messages and removed_messages == 0
    )
    for table, table_report in copy_report.items():
        status = table_report.get("status")
        if status not in {"failed", "partial"}:
            continue
        tolerable = (
            status == "partial" or table not in {"sessions", "messages"}
            or (table == "sessions" and sessions_fully_reconstructed)
        )
        flag(f"{table} copy status is {status}", soft=allow_partial and tolerable)
    if orphan_cleanup:
        orphan_count = int(orphan_cleanup.get("total_removed_or_relinked") or 0)
        if orphan_count:
            flag(f"{orphan_count} orphaned reference(s) were removed or relinked", soft=True)
        if rebuilt_sessions:
            # Not a clean recovery: the conversation text survived but its session metadata did not.
            flag(
                f"{rebuilt_sessions} session(s) could not be salvaged and were reconstructed as placeholders to "
                f"retain {retained_messages} message(s); their metadata (title, model, timestamps, cost) is lost",
                soft=True,
            )


def _verify_recovered_database(
    output: Path, *, expected_counts: dict[str, Optional[int]], copy_report: dict[str, dict[str, Any]],
    allow_partial: bool = False, orphan_cleanup: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    verification: dict[str, Any] = {"errors": [], "warnings": [], "loss_detected": False}
    open_error = _db_opens_cleanly(output)
    verification["opens_cleanly"] = open_error is None
    if open_error is not None:
        verification["errors"].append(f"database health probe: {open_error}")
    conn = sqlite3.connect(str(output), isolation_level=None)
    try:
        _verify_structure(conn, verification)
        _verify_row_counts(
            conn, verification, expected_counts=expected_counts, copy_report=copy_report, allow_partial=allow_partial,
            orphan_cleanup=orphan_cleanup,
        )
        _verify_fts_indexes(conn, verification)
    except sqlite3.DatabaseError as exc:
        verification["errors"].append(f"verification query failed: {exc}")
    finally:
        conn.close()
    verification["healthy"] = not verification["errors"]
    verification["complete"] = bool(verification["healthy"] and not verification["loss_detected"])
    return verification


def _sanitize_session_model_config(destination: sqlite3.Connection) -> int:
    """Rewrite unparseable ``sessions.model_config`` blobs to ``'{}'``; returns the row count.

    ``integrity_check`` validates b-tree structure, never column *contents*: a row whose
    JSON was truncated by the damage verifies clean, and the recovered store then raises
    ``OperationalError: malformed JSON`` the first time ``reopen_session`` rewrites the
    reset-child markers with ``json_set`` (``hermes_state_sessions.py::reopen_session``) —
    i.e. on the first resume of a parent session. Read paths are already guarded
    (``_sql_json_extract`` wraps every extract in ``CASE WHEN json_valid``), so this is
    about the write path. The blob is unrecoverable either way, so neutralise it at the
    copy boundary both lanes pass through rather than shipping a store that breaks on
    the first resume.
    """
    if "model_config" not in _table_columns(destination, "sessions"):
        return 0
    with _immediate_transaction(destination):
        return _reconcile(
            destination, "sessions",
            "model_config IS NOT NULL AND json_valid(model_config) = 0",
            "UPDATE sessions SET model_config = '{}'",
        )


def _repair_out_of_window_timestamps(destination: sqlite3.Connection) -> int:
    """Rewrite ``messages``/``sessions`` timestamp cells outside the ``coerce_epoch`` window; returns the count.

    A damaged cell can decode as a valid-looking garbage double (``5.49e+246``) that the salvage copies
    verbatim, and one such row pins the session's recency and breaks every renderer that turns it into
    a date (#91536). A message takes its nearest valid neighbour's time in the same session (keeps the
    order), then its session's start, then now; ``started_at`` takes the earliest valid message; the
    nullable ``ended_at``/``last_activity_at`` become NULL.
    """
    message_columns, session_columns = _table_columns(destination, "messages"), _table_columns(destination, "sessions")
    ok = f"BETWEEN {EPOCH_MIN!r} AND {EPOCH_MAX!r}"
    now = "CAST(strftime('%s', 'now') AS REAL)"
    repaired = 0
    with _immediate_transaction(destination):
        if "timestamp" in message_columns:
            repaired += _reconcile(
                destination, "messages", f"NOT (timestamp {ok})",
                "UPDATE messages SET timestamp = COALESCE("
                f"(SELECT p.timestamp FROM messages p WHERE p.session_id = messages.session_id AND p.id < messages.id"
                f" AND p.timestamp {ok} ORDER BY p.id DESC LIMIT 1),"
                f"(SELECT n.timestamp FROM messages n WHERE n.session_id = messages.session_id AND n.id > messages.id"
                f" AND n.timestamp {ok} ORDER BY n.id LIMIT 1),"
                f"(SELECT s.started_at FROM sessions s WHERE s.id = messages.session_id AND s.started_at {ok}), {now})",
            )
        if "started_at" in session_columns:
            repaired += _reconcile(
                destination, "sessions", f"NOT (started_at {ok})",
                "UPDATE sessions SET started_at = COALESCE((SELECT MIN(m.timestamp) FROM messages m"
                f" WHERE m.session_id = sessions.id AND m.timestamp {ok}), {now})",
            )
        for column in ("ended_at", "last_activity_at"):
            if column in session_columns:
                repaired += _reconcile(
                    destination, "sessions", f"NOT ({column} {ok})", f"UPDATE sessions SET {column} = NULL")
    return repaired


def _finalize_derived_metadata(destination: sqlite3.Connection) -> dict[str, Any]:
    """Sanitize copied JSON columns and timestamps, and stamp metadata the new destination actually owns."""
    model_config_reset = _sanitize_session_model_config(destination)
    timestamps_repaired = _repair_out_of_window_timestamps(destination)
    fts_tables = {
        str(row[0])
        for row in destination.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name IN ('messages_fts', 'messages_fts_trigram')"
        ).fetchall()
    }
    result: dict[str, Any] = {
        "fts_tables": sorted(fts_tables), "finalized": False, "model_config_reset": model_config_reset,
        "timestamps_repaired": timestamps_repaired}
    if fts_tables != {"messages_fts", "messages_fts_trigram"}:
        result["error"] = "fresh destination is missing required FTS tables"
        return result
    fts_keys = tuple(key for key in _GENERATED_META_KEYS if key.startswith("fts_"))
    placeholders = ", ".join("?" for _ in fts_keys)
    with _immediate_transaction(destination):
        destination.execute(f"DELETE FROM state_meta WHERE key IN ({placeholders})", fts_keys)
        destination.execute(
            "INSERT INTO state_meta(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
            ("fts_storage_version", str(FTS_STORAGE_VERSION)),
        )
    result["finalized"] = True
    return result


def _lost_and_found_plausibility_errors(
    conn: sqlite3.Connection,
) -> list[str]:
    """Flag systematic timestamp mis-mapping in a salvaged database.

    Structural checks (integrity, FK, FTS, row counts) pass on mis-mapped
    salvage because every row still inserts. Only semantics give it away:
    the physical column order of a source upgraded via ALTER TABLE differs
    from the destination template's declared order, so positional cell
    mapping lands counters/strings where ``started_at``/``timestamp``
    belong — and the NOT NULL substitutes turn gaps into 0.0. When every
    mapped row violates the epoch floor, the mapping was wrong.

    Stub rows written by ``stub_missing_parent_sessions`` legitimately carry
    ``started_at = 0.0`` when no timestamped message survived, so they are
    excluded from the denominator.
    """
    from hermes_cli.session_lost_and_found import _EPOCH_LOW, STUB_TITLE_PREFIX

    errors: list[str] = []
    checks = (
        ("sessions", "started_at", f"WHERE COALESCE(title, '') NOT LIKE '{STUB_TITLE_PREFIX}%'"),
        ("messages", "timestamp", ""),
    )
    for table, column, mapped_filter in checks:
        (total,) = conn.execute(f"SELECT COUNT(*) FROM {table} {mapped_filter}").fetchone()
        if not total:
            continue
        (implausible,) = conn.execute(
            f"SELECT COUNT(*) FROM {table} {mapped_filter} "
            f"{'AND' if mapped_filter else 'WHERE'} ({column} IS NULL OR {column} < ?)",
            (_EPOCH_LOW,),
        ).fetchone()
        if implausible == total:
            errors.append(
                f"{table}.{column} is implausible in all {total} salvaged row(s) "
                "(NULL or before 2001-09): the source's physical column order "
                "did not match the destination template, so cells were mapped "
                "onto the wrong columns"
            )
    return errors


def _recover_via_lost_and_found(
    *, source: Path, snapshot_source: Path, snapshot_dir: Path, output: Path, inspection: dict[str, Any],
    disk_space: dict[str, Any], missing_required: list[str],
) -> dict[str, Any]:
    """Best-effort page-level salvage when table schemas are unreadable: the sqlite3 CLI's ``.recover``
    (shell-only, not in Python's ``sqlite3``) rebuilds rows into a scratch lost_and_found database which
    is then heuristically mapped into a fresh current-schema database."""
    from hermes_cli.session_lost_and_found import (
        SQLITE3_CLI_GUIDANCE, LostAndFoundError, find_sqlite3_cli, find_sqlite3_cli_refusal, map_lost_and_found_rows,
        rebuild_fts_indexes, run_cli_lost_and_found_recover, stub_missing_parent_sessions,
    )
    missing = ", ".join(missing_required)
    sqlite3_bin = find_sqlite3_cli()
    if sqlite3_bin is None:
        refusal = find_sqlite3_cli_refusal()
        if refusal.get("reason") == "wal_reset_vulnerable":
            raise SessionRecoverySourceError(
                "Partial recovery requires a page-level salvage shell, but "
                "the only sqlite3 CLI on PATH is not safe to use for it: it "
                + refusal["detail"]
                + ". The readable table schemas for: "
                + ", ".join(missing_required)
                + " are still required."
            )
        raise SessionRecoverySourceError(
            f"Partial recovery still requires readable table schemas for: {missing}. {SQLITE3_CLI_GUIDANCE}"
        )
    lf_path = snapshot_dir / "lost_and_found.db"
    try:
        cli_report = run_cli_lost_and_found_recover(snapshot_source, lf_path, sqlite3_bin)
    except (LostAndFoundError, OSError) as exc:
        raise SessionRecoverySourceError(
            f"Partial recovery could not read the table schemas for: {missing}, "
            f"and page-level .recover salvage failed: {exc}"
        ) from exc
    lf_conn = sqlite3.connect(str(lf_path), isolation_level=None)
    destination_conn = _fresh_destination(output)
    try:
        mapping = map_lost_and_found_rows(lf_conn, destination_conn)
        stubbing = stub_missing_parent_sessions(destination_conn)
        fts = rebuild_fts_indexes(destination_conn)
        derived_metadata = _finalize_derived_metadata(destination_conn)
    finally:
        lf_conn.close()
        destination_conn.close()
    copy_report: dict[str, dict[str, Any]] = {
        table: {
            "mode": "lost_and_found_salvage", "status": "partial",
            "copied_rows": int(mapping["direct_table_rows"].get(table) or 0) + int(mapping["mapped"].get(table) or 0),
            "error": "recovered via page-level lost_and_found salvage; "
            "row completeness cannot be verified against the source",
        }
        for table in ("sessions", "messages", "session_model_usage")
    }
    orphan_cleanup = {
        "sessions_reconstructed": stubbing["sessions_stubbed"], "messages_retained": stubbing["messages_retained"],
        "messages_removed": 0, "total_removed_or_relinked": 0,
    }
    verification = _verify_recovered_database(
        output, expected_counts={"sessions": None, "messages": None}, copy_report=copy_report, allow_partial=True,
        orphan_cleanup=orphan_cleanup,
    )
    verification["warnings"].append(
        "BEST-EFFORT page-level salvage: the source table schemas were unreadable, so rows were rebuilt from raw "
        "pages and mapped heuristically. Review every count before trusting this output."
    )
    if cli_report.get("header_zeroed"):
        verification["warnings"].append(
            "header salvage: SQLite refused the source outright (page-1 header damaged, 'file is not a "
            "database'); the header of the private snapshot copy was zeroed so .recover could walk the "
            "surviving pages. Rows written only to a -wal after the last checkpoint are not included."
        )
    verification.update(loss_detected=True, complete=False)
    # Structural checks cannot see a positional mis-mapping: every row still inserts, so integrity/FK/FTS
    # stay green. A systematic timestamp violation is the semantic tell — never report such a salvage as verified.
    # See #101409.
    plausibility_conn = sqlite3.connect(str(output), isolation_level=None)
    try:
        plausibility_errors = _lost_and_found_plausibility_errors(plausibility_conn)
    finally:
        plausibility_conn.close()
    # Records whose width matched no known physical layout were inserted
    # positionally as a guess. Once the recognised rows map correctly the
    # all-rows timestamp gate above no longer sees them, so they need their
    # own tell: the guessed rows may be shifted while the report otherwise
    # looks clean.
    guessed = int(mapping.get("unrecognized_layout_rows") or 0)
    if guessed:
        widths = ", ".join(
            f"{kind} rows with {'/'.join(map(str, sorted(ws)))} fields"
            for kind, ws in sorted((mapping.get("unrecognized_layout_widths") or {}).items())
        )
        plausibility_errors.append(
            f"{guessed} salvaged row(s) matched no known physical column layout and were mapped "
            f"positionally ({widths}); their fields may be shifted. Inspect those sessions before trusting them."
        )
    if plausibility_errors:
        verification["errors"].extend(plausibility_errors)
        verification["healthy"] = False
    return _recovery_report(
        source, output, inspection, disk_space, verification, on_source_change="healthy",
        allow_partial=True, mode="lost_and_found_salvage", best_effort=True, unreadable_schemas=missing_required,
        sqlite3_cli=cli_report, lost_and_found=mapping, session_stubs=stubbing, fts_rebuild=fts, copy=copy_report,
        orphan_cleanup=orphan_cleanup, derived_metadata=derived_metadata,
    )


def _recovery_report(
    source: Path, output: Path, inspection: dict[str, Any], disk_space: dict[str, Any], verification: dict[str, Any],
    *, on_source_change: str, **fields: Any,
) -> dict[str, Any]:
    """The ``recover`` report: shared header, mode-specific ``fields``, then the verdict. A source bundle
    that changed during recovery is a verification error and also clears ``verification[on_source_change]``.
    """
    source_unchanged = _source_fingerprint(source) == inspection["source_fingerprint"]
    if not source_unchanged:
        verification["errors"].append("the source database bundle changed during recovery")
        verification[on_source_change] = False
    return {
        "operation": "recover", "source": str(source), "output": str(output),
        "source_bundle": inspection["source_bundle"], "source_fingerprint": inspection["source_fingerprint"],
        "source_unchanged": source_unchanged, "disk_space": disk_space,
        "inspection": {
            "journal_mode": inspection.get("journal_mode"), "tables": inspection["tables"],
            "errors": inspection["errors"], "warnings": inspection["warnings"],
        },
        **fields,
        "verification": verification,
        "complete": bool(verification.get("complete") and source_unchanged),
        "partial": bool(verification.get("loss_detected")),
        "verified": bool(verification.get("healthy") and source_unchanged),
        "installed": False,
    }


def recover_session_database(
    source_path: Path, output_path: Path, *, work_dir: Optional[Path] = None, chunk_size: int = 1_000,
    progress_cb: Optional[ProgressCallback] = None, allow_partial: bool = False,
) -> dict[str, Any]:
    """Recover canonical rows into a separate current-schema database. The source and its sidecars are
    copied before SQLite opens anything; ``output_path`` must not exist and is never swapped into place."""
    if chunk_size <= 0:
        raise SessionRecoverySafetyError("chunk_size must be greater than zero")
    source, output, work_root = _validate_paths(source_path, output_path=output_path, work_dir=work_dir)
    assert output is not None
    disk_space = _disk_space_preflight(source, work_root, output.parent)
    temp_dir, snapshot_source, inspection = _snapshot_and_inspect(source, work_root)
    try:
        if not inspection.get("recoverable") and not allow_partial:
            reasons = "; ".join(inspection.get("errors") or ["unknown source error"])
            raise SessionRecoverySourceError(
                f"Required canonical tables are not readable: {reasons}. "
                "Re-run with --allow-partial to salvage every readable row "
                "into a new database (the source is never modified)."
            )
        missing_required = [t for t in ("sessions", "messages") if not inspection["tables"][t].get("available")]
        if allow_partial and missing_required:  # no readable schema -> page-level lost_and_found salvage
            return _recover_via_lost_and_found(
                source=source, snapshot_source=snapshot_source, snapshot_dir=Path(temp_dir.name), output=output,
                inspection=inspection, disk_space=disk_space, missing_required=missing_required,
            )
        source_conn = _connect(snapshot_source)
        source_conn.execute("PRAGMA writable_schema=ON")
        destination_conn: Optional[sqlite3.Connection] = None
        try:
            destination_conn = _fresh_destination(
                output, topic_tables=any(inspection["tables"][table].get("available") for table in _TOPIC_TABLES),
            )
            copy_report: dict[str, dict[str, Any]] = {}
            for table in (*_CANONICAL_TABLES, "state_meta", *_TOPIC_TABLES, *_AUXILIARY_TABLES):
                if table not in _CANONICAL_TABLES and not inspection["tables"][table].get("available"):
                    copy_report[table] = {"status": "missing", "copied_rows": 0}
                    continue
                if table in _AUXILIARY_TABLES:  # lazy gateway table: create it or the copy reports "missing"
                    _AUXILIARY_TABLE_SCHEMAS[table](destination_conn)
                copy_report[table] = _copy_table(
                    source_conn, destination_conn, table, salvage=allow_partial, chunk_size=chunk_size,
                    progress_cb=progress_cb, source_rows=inspection["tables"][table].get("rows"),
                )
            orphan_cleanup = _cleanup_partial_orphans(destination_conn) if allow_partial else None
            derived_metadata = _finalize_derived_metadata(destination_conn)
        finally:
            source_conn.close()
            if destination_conn is not None:
                destination_conn.close()
        expected_counts = {
            table: inspection["tables"][table].get("rows")
            for table in (*_CANONICAL_TABLES, *_AUXILIARY_TABLES)
            if table in _CANONICAL_TABLES or inspection["tables"].get(table, {}).get("available")
        }
        verification = _verify_recovered_database(
            output, expected_counts=expected_counts, copy_report=copy_report, allow_partial=allow_partial,
            orphan_cleanup=orphan_cleanup,
        )
        return _recovery_report(
            source, output, inspection, disk_space, verification, on_source_change="complete",
            allow_partial=allow_partial, copy=copy_report, orphan_cleanup=orphan_cleanup,
            derived_metadata=derived_metadata,
        )
    finally:
        temp_dir.cleanup()


def write_recovery_report(path: Path, report: dict[str, Any]) -> Path:
    """Write a JSON report without overwriting an existing file."""
    destination = _resolved_output_path(path)
    with destination.open("x", encoding="utf-8") as handle:
        json.dump(report, handle, indent=2, sort_keys=True)
        handle.write("\n")
    return destination