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

The recovery path deliberately avoids in-place repair:

* the supplied source database is never opened by SQLite;
* the source file and any WAL/SHM/rollback-journal sidecars are copied into a
  disposable working directory first;
* canonical rows are copied into a newly initialized current-schema database;
* derived FTS tables and migration bookkeeping are rebuilt, not copied; and
* the recovered database is never installed over the active database.
"""

from __future__ import annotations

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

from hermes_state import (
    FTS_STORAGE_VERSION,
    SCHEMA_VERSION,
    SessionDB,
    _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",
)

# These values describe derived indexes or the schema that owns an optional
# table. A fresh destination must generate them from its own current schema.
_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",
})

_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."""

    parent = path.expanduser().parent.resolve(strict=True)
    return parent / 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 not path.exists():
            continue
        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:
        # Existing directories were already required by _validate_paths. This
        # fallback is defensive for platforms with 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(
        _sidecar_path(source, suffix).stat().st_size
        for suffix in _SIDECAR_SUFFIXES
        if _sidecar_path(source, suffix).exists()
    )
    # The v23 external-content rebuild is normally substantially smaller than
    # a legacy database, but using the complete source bundle as the estimate
    # avoids betting the user's disk on that expectation.
    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["shared_filesystem"] = True
        report["work_dir_required_bytes"] = required
        if work_free < required:
            raise SessionRecoverySafetyError(
                "Not enough free disk space for a safe recovery copy: "
                f"{_format_bytes(work_free)} available at {work_root}, "
                f"{_format_bytes(required)} required "
                f"({_format_bytes(bundle_bytes)} source bundle + "
                f"{_format_bytes(output_allowance)} output allowance + "
                f"{_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: list[str] = []
    if work_free < work_required:
        shortages.append(
            f"{work_root}: {_format_bytes(work_free)} available, "
            f"{_format_bytes(work_required)} required"
        )
    if output_free < output_required:
        shortages.append(
            f"{output_parent}: {_format_bytes(output_free)} available, "
            f"{_format_bytes(output_required)} required"
        )
    if shortages:
        raise SessionRecoverySafetyError(
            "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``, which holds the
    connection-lifecycle lock for its duration. 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.

    Recovery normally runs as its own short-lived CLI process against an
    offline/quarantined file, so the refusal should never fire; the guard
    keeps this path consistent with ``hermes_state._backup_db_file``.
    """
    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 not source_part.exists():
                    continue
                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 _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["available"] = True
        result["columns"] = columns
        result["rows"] = int(
            conn.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0]
        )
    except sqlite3.DatabaseError as exc:
        result["error"] = str(exc)
    return result


def _inspect_connection(conn: sqlite3.Connection) -> dict[str, Any]:
    conn.execute("PRAGMA writable_schema=ON")
    report: dict[str, Any] = {"tables": {}, "errors": [], "warnings": []}
    try:
        row = conn.execute("PRAGMA journal_mode").fetchone()
        report["journal_mode"] = str(row[0]).lower() if row else None
    except sqlite3.DatabaseError as exc:
        report["journal_mode"] = None
        # Journal metadata is useful context but not canonical session data.
        # A damaged journal pragma must not block rows that are still readable.
        report["warnings"].append(f"journal mode: {exc}")

    for table in (*_CANONICAL_TABLES, "state_meta", *_TOPIC_TABLES):
        report["tables"][table] = _table_inventory(conn, table)

    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),
    )
    snapshot_dir = Path(temp_dir.name)
    try:
        snapshot_source, copied = _copy_source_bundle(source, snapshot_dir)
        after = _source_fingerprint(source)
        if before != after:
            raise SessionRecoverySafetyError(
                "The source database bundle changed while it was being copied. "
                "Stop every Hermes process using this profile and retry."
            )

        conn = sqlite3.connect(
            str(snapshot_source),
            isolation_level=None,
            timeout=1.0,
        )
        try:
            inspection = _inspect_connection(conn)
        finally:
            conn.close()
        inspection["source_bundle"] = copied
        inspection["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 _copy_table(
    source: sqlite3.Connection,
    destination: sqlite3.Connection,
    table: str,
    *,
    chunk_size: int,
    progress_cb: Optional[ProgressCallback],
    source_rows: Optional[int],
) -> dict[str, Any]:
    source_columns = _table_columns(source, table)
    destination_columns = _table_columns(destination, table)
    columns = [column for column in destination_columns if column in source_columns]
    result: dict[str, Any] = {
        "source_rows": source_rows,
        "copied_rows": 0,
        "columns": columns,
    }
    if not source_columns:
        result["status"] = "missing"
        return result
    if not columns:
        result["status"] = "failed"
        result["error"] = "source and destination have no compatible columns"
        return result

    quoted = ", ".join(f'"{column}"' for column in columns)
    placeholders = ", ".join("?" for _ in columns)
    select_sql = f'SELECT {quoted} FROM "{table}"'
    insert_prefix = "INSERT OR REPLACE" if table == "state_meta" else "INSERT"
    insert_sql = f'{insert_prefix} INTO "{table}" ({quoted}) VALUES ({placeholders})'

    try:
        cursor = source.execute(select_sql)
        while True:
            rows = cursor.fetchmany(chunk_size)
            if not rows:
                break
            destination.execute("BEGIN IMMEDIATE")
            try:
                destination.executemany(insert_sql, rows)
                destination.execute("COMMIT")
            except BaseException:
                destination.execute("ROLLBACK")
                raise
            result["copied_rows"] += len(rows)
            if progress_cb is not None:
                progress_cb({
                    "table": table,
                    "copied_rows": result["copied_rows"],
                    "source_rows": source_rows,
                })
    except sqlite3.DatabaseError as exc:
        result["status"] = "partial" if result["copied_rows"] else "failed"
        result["error"] = str(exc)
        return result

    result["status"] = (
        "complete"
        if source_rows is None or result["copied_rows"] == source_rows
        else "partial"
    )
    if result["status"] == "partial":
        result["error"] = (
            f"copied {result['copied_rows']} of {source_rows} readable rows"
        )
    return result


def _append_skipped_range(
    ranges: list[dict[str, Any]],
    low: int,
    high: int,
    error: str,
) -> None:
    """Record skipped rowid ranges without producing 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}
    directions = (("low", "ASC"), ("high", "DESC"))
    for edge, direction in directions:
        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 and not result["errors"]:
        result["empty"] = True
        return result
    if rows["low"] is None and rows["high"] is None:
        result["unavailable"] = True
        return result

    # A damaged edge can prevent one of the ordered probes from completing.
    # Keep the other readable edge and bound the missing side by SQLite's
    # rowid domain. Range bisection can then approach the surviving data
    # without assuming that user-created databases contain 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["low"] = rows["low"]
    result["high"] = rows["high"]
    return result


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."""

    source_columns = _table_columns(source, table)
    destination_columns = _table_columns(destination, table)
    columns = [column for column in destination_columns if column in source_columns]
    result: dict[str, Any] = {
        "mode": "rowid_range_salvage",
        "source_rows": source_rows,
        "copied_rows": 0,
        "excluded_rows": 0,
        "columns": columns,
        "range_queries": 0,
        "skipped_rowid_ranges": [],
    }
    if not source_columns:
        result["status"] = "missing"
        return result
    if not columns:
        result["status"] = "failed"
        result["error"] = "source and destination have no compatible columns"
        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:
        result["status"] = "failed"
        details = "; ".join(bounds.get("errors") or [])
        result["error"] = "could not determine a rowid range for salvage"
        if details:
            result["error"] += f": {details}"
        return result

    quoted = ", ".join(f'"{column}"' for column in columns)
    placeholders = ", ".join("?" for _ in columns)
    select_sql = (
        f'SELECT rowid, {quoted} FROM "{table}" '
        "WHERE rowid BETWEEN ? AND ? ORDER BY rowid"
    )
    insert_sql = (
        f'{insert_prefix} INTO "{table}" ({quoted}) VALUES ({placeholders})'
    )
    column_names = tuple(columns)
    stopped_at_query_limit = False

    def copy_range(low: int, high: int) -> None:
        nonlocal stopped_at_query_limit
        if low > high:
            return
        if result["range_queries"] >= _MAX_SALVAGE_RANGE_QUERIES:
            stopped_at_query_limit = True
            _append_skipped_range(
                result["skipped_rowid_ranges"],
                low,
                high,
                "salvage range query limit reached",
            )
            return

        result["range_queries"] += 1
        last_committed_rowid: Optional[int] = None
        try:
            cursor = source.execute(select_sql, (low, high))
            while True:
                fetched = cursor.fetchmany(chunk_size)
                if not fetched:
                    return

                values = [tuple(row[1:]) for row in fetched]
                if row_filter is not None:
                    included = [
                        row for row in values if row_filter(row, column_names)
                    ]
                    excluded_count = len(values) - len(included)
                else:
                    included = values
                    excluded_count = 0

                if included:
                    destination.execute("BEGIN IMMEDIATE")
                    try:
                        destination.executemany(insert_sql, included)
                        destination.execute("COMMIT")
                    except BaseException:
                        destination.execute("ROLLBACK")
                        raise

                result["copied_rows"] += len(included)
                result["excluded_rows"] += excluded_count
                last_committed_rowid = int(fetched[-1][0])
                if progress_cb is not None:
                    progress_cb({
                        "table": table,
                        "copied_rows": result["copied_rows"],
                        "source_rows": 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:
                _append_skipped_range(
                    result["skipped_rowid_ranges"],
                    retry_low,
                    high,
                    str(exc),
                )
                return
            midpoint = retry_low + (high - retry_low) // 2
            copy_range(retry_low, midpoint)
            copy_range(midpoint + 1, high)

    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"] = 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"
        result["error"] = (
            f"copied {result['copied_rows']} and excluded "
            f"{result['excluded_rows']} of {source_rows} source rows"
        )
    else:
        result["status"] = "complete"
    return result


def _copy_state_meta(
    source: sqlite3.Connection,
    destination: sqlite3.Connection,
    *,
    chunk_size: int,
    progress_cb: Optional[ProgressCallback],
    source_rows: Optional[int],
) -> dict[str, Any]:
    source_columns = _table_columns(source, "state_meta")
    destination_columns = _table_columns(destination, "state_meta")
    result: dict[str, Any] = {
        "source_meta_rows": source_rows,
        "copied_rows": 0,
        "columns": ["key", "value"],
        "excluded_keys": sorted(_GENERATED_META_KEYS),
    }
    if not {"key", "value"}.issubset(source_columns):
        result["status"] = "missing"
        return result
    if not {"key", "value"}.issubset(destination_columns):
        result["status"] = "failed"
        result["error"] = "destination state_meta schema is incomplete"
        return result

    placeholders = ", ".join("?" for _ in _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})",
                tuple(_GENERATED_META_KEYS),
            ).fetchone()[0]
        )
    except sqlite3.DatabaseError:
        # The copy loop below will return the concrete read error.
        pass

    try:
        cursor = source.execute(
            f"SELECT key, value FROM state_meta WHERE key NOT IN ({placeholders})",
            tuple(_GENERATED_META_KEYS),
        )
        while True:
            rows = cursor.fetchmany(chunk_size)
            if not rows:
                break
            destination.execute("BEGIN IMMEDIATE")
            try:
                destination.executemany(
                    "INSERT OR REPLACE INTO state_meta(key, value) VALUES (?, ?)",
                    rows,
                )
                destination.execute("COMMIT")
            except BaseException:
                destination.execute("ROLLBACK")
                raise
            result["copied_rows"] += len(rows)
            if progress_cb is not None:
                progress_cb({
                    "table": "state_meta",
                    "copied_rows": result["copied_rows"],
                    "source_rows": filtered_source_rows,
                })
    except sqlite3.DatabaseError as exc:
        result["status"] = "partial" if result["copied_rows"] else "failed"
        result["error"] = str(exc)
        return result

    result["status"] = (
        "complete"
        if filtered_source_rows is None or result["copied_rows"] == filtered_source_rows
        else "partial"
    )
    if result["status"] == "partial":
        result["error"] = (
            f"copied {result['copied_rows']} of {filtered_source_rows} readable rows"
        )
    return result


def _copy_state_meta_salvage(
    source: sqlite3.Connection,
    destination: sqlite3.Connection,
    *,
    chunk_size: int,
    progress_cb: Optional[ProgressCallback],
    source_rows: Optional[int],
) -> dict[str, Any]:
    """Salvage readable user metadata while regenerating derived FTS state.

    Requires both ``key`` and ``value``, matching the non-partial
    :func:`_copy_state_meta`. A damaged ``state_meta`` can retain one column
    and lose the other; without this check a missing ``key`` raised
    ``ValueError`` from ``columns.index("key")`` and aborted the entire
    partial recovery, and a missing ``value`` would have copied key-only rows
    while reporting the table complete.

    Status matters here. An unusable-but-PRESENT table reports ``failed``, not
    ``missing``: verification only escalates ``failed``/``partial`` into a
    warning + ``loss_detected``, so reporting ``missing`` would silently drop
    real metadata and still claim ``complete=True``. ``missing`` is reserved
    for a table that genuinely is not there. Either way ``state_meta`` is
    optional, so ``--allow-partial`` records the loss and carries on
    recovering sessions and messages.
    """
    source_columns = _table_columns(source, "state_meta")
    destination_columns = _table_columns(destination, "state_meta")
    if not source_columns:
        # Genuinely absent from the source — nothing was lost.
        return {
            "mode": "rowid_range_salvage",
            "source_meta_rows": source_rows,
            "copied_rows": 0,
            "columns": ["key", "value"],
            "excluded_keys": sorted(_GENERATED_META_KEYS),
            "status": "missing",
        }
    if not {"key", "value"}.issubset(source_columns):
        # Present but unusable: this IS data loss and must be reported.
        return {
            "mode": "rowid_range_salvage",
            "source_meta_rows": source_rows,
            "copied_rows": 0,
            "columns": ["key", "value"],
            "excluded_keys": sorted(_GENERATED_META_KEYS),
            "status": "failed",
            "error": (
                "source state_meta exists but is missing the key/value "
                f"columns (found: {', '.join(source_columns) or 'none'})"
            ),
        }
    if not {"key", "value"}.issubset(destination_columns):
        return {
            "mode": "rowid_range_salvage",
            "source_meta_rows": source_rows,
            "copied_rows": 0,
            "columns": ["key", "value"],
            "excluded_keys": sorted(_GENERATED_META_KEYS),
            "status": "failed",
            "error": "destination state_meta schema is incomplete",
        }

    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


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

    When the ``sessions`` b-tree is damaged worse than ``messages``, salvage
    can recover the conversation text while recovering few or none of the
    session rows that own it. Deleting those messages as "orphans" throws away
    the only readable copy of the user's data — the exact opposite of what
    ``--allow-partial`` is for. A real report (July 2026) copied 20,817 of
    20,824 messages and then removed every one of them, producing an output
    with 0 sessions and 0 messages.

    Instead, synthesize a minimal session row per orphaned ``session_id``
    (only ``id``/``source``/``started_at`` are NOT NULL) so the messages stay
    reachable and foreign keys hold. ``started_at`` is taken from the earliest
    surviving message so ordering stays sane. Rows are marked with
    ``source='recovered'`` and a ``title`` that says so, because a fabricated
    session must never be mistaken for an original.
    """
    result: dict[str, Any] = {"sessions_reconstructed": 0, "messages_retained": 0}
    if not _table_columns(destination, "sessions"):
        return result
    if not _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

    title_sequence = 1
    for session_id, first_timestamp, message_count in orphaned:
        started_at = float(first_timestamp) if first_timestamp is not None else 0.0
        while True:
            title = (
                f"[recovered {title_sequence}] "
                "session metadata was unreadable"
            )
            title_sequence += 1
            if (
                destination.execute(
                    "SELECT 1 FROM sessions WHERE title = ? LIMIT 1",
                    (title,),
                ).fetchone()
                is None
            ):
                break

        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 _cleanup_partial_orphans(
    destination: sqlite3.Connection,
) -> dict[str, Any]:
    """Reconcile references to sessions that could not be salvaged.

    Messages are never discarded for lack of a session row: their owning
    session is reconstructed as a placeholder first (see
    :func:`_reconstruct_missing_sessions`). Only rows that remain orphaned
    after that — and rows in tables carrying no recoverable user content —
    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,
        "messages_removed": 0,
        "session_model_usage_removed": 0,
        "compression_locks_removed": 0,
        "telegram_dm_topic_bindings_removed": 0,
    }
    destination.execute("BEGIN IMMEDIATE")
    try:
        # Rebuild owners BEFORE any orphan deletion so salvaged conversation
        # text is never dropped for want of a session row.
        rebuilt = _reconstruct_missing_sessions(destination)
        result["sessions_reconstructed"] = rebuilt["sessions_reconstructed"]
        result["messages_retained"] = rebuilt["messages_retained"]

        parent_count = int(
            destination.execute(
                "SELECT COUNT(*) FROM sessions AS child "
                "WHERE child.parent_session_id IS NOT NULL "
                "AND NOT EXISTS ("
                "SELECT 1 FROM sessions AS parent "
                "WHERE parent.id = child.parent_session_id)"
            ).fetchone()[0]
        )
        if parent_count:
            destination.execute(
                "UPDATE sessions SET parent_session_id = NULL "
                "WHERE parent_session_id IS NOT NULL "
                "AND NOT EXISTS ("
                "SELECT 1 FROM sessions AS parent "
                "WHERE parent.id = sessions.parent_session_id)"
            )
        result["sessions_parent_cleared"] = parent_count

        prompt_ref_count = int(
            destination.execute(
                "SELECT COUNT(*) FROM sessions "
                "WHERE system_prompt_hash IS NOT NULL "
                "AND NOT EXISTS ("
                "SELECT 1 FROM system_prompts "
                "WHERE system_prompts.hash = sessions.system_prompt_hash)"
            ).fetchone()[0]
        )
        if prompt_ref_count:
            destination.execute(
                "UPDATE sessions SET system_prompt_hash = NULL "
                "WHERE system_prompt_hash IS NOT NULL "
                "AND NOT EXISTS ("
                "SELECT 1 FROM system_prompts "
                "WHERE system_prompts.hash = sessions.system_prompt_hash)"
            )
        result["session_prompt_refs_cleared"] = prompt_ref_count

        unreferenced_prompt_count = int(
            destination.execute(
                "SELECT COUNT(*) FROM system_prompts "
                "WHERE NOT EXISTS ("
                "SELECT 1 FROM sessions "
                "WHERE sessions.system_prompt_hash = system_prompts.hash)"
            ).fetchone()[0]
        )
        if unreferenced_prompt_count:
            destination.execute(
                "DELETE FROM system_prompts "
                "WHERE NOT EXISTS ("
                "SELECT 1 FROM sessions "
                "WHERE sessions.system_prompt_hash = system_prompts.hash)"
            )
        result["system_prompts_removed"] = unreferenced_prompt_count

        dependent_tables = (
            ("messages", "messages_removed"),
            ("session_model_usage", "session_model_usage_removed"),
            ("compression_locks", "compression_locks_removed"),
            (
                "telegram_dm_topic_bindings",
                "telegram_dm_topic_bindings_removed",
            ),
        )
        for table, report_key in dependent_tables:
            if not _table_columns(destination, table):
                continue
            orphan_count = int(
                destination.execute(
                    f'SELECT COUNT(*) FROM "{table}" AS dependent '
                    "WHERE NOT EXISTS ("
                    "SELECT 1 FROM sessions "
                    "WHERE sessions.id = dependent.session_id)"
                ).fetchone()[0]
            )
            if orphan_count:
                destination.execute(
                    f'DELETE FROM "{table}" '
                    "WHERE NOT EXISTS ("
                    "SELECT 1 FROM sessions "
                    f'WHERE sessions.id = "{table}".session_id)'
                )
            result[report_key] = orphan_count
        destination.execute("COMMIT")
    except BaseException:
        destination.execute("ROLLBACK")
        raise
    # Only destructive/relinking actions belong in this total. The
    # reconstruction counters describe data RETAINED, so summing them here
    # would report saving the user's messages as if it were losing them.
    result["total_removed_or_relinked"] = (
        int(result["session_prompt_refs_cleared"])
        + int(result["sessions_parent_cleared"])
        + int(result["messages_removed"])
        + int(result["session_model_usage_removed"])
        + int(result["compression_locks_removed"])
        + int(result["telegram_dm_topic_bindings_removed"])
    )
    return result


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:
        integrity_rows = [
            str(row[0]) for row in conn.execute("PRAGMA integrity_check").fetchall()
        ]
        verification["integrity_check"] = integrity_rows
        if integrity_rows != ["ok"]:
            verification["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:
            verification["errors"].append("foreign key violations remain")

        journal_row = conn.execute("PRAGMA journal_mode").fetchone()
        verification["journal_mode"] = (
            str(journal_row[0]).lower() if journal_row else None
        )

        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:
            verification["errors"].append(
                f"schema version is {verification['schema_version']}, "
                f"expected {SCHEMA_VERSION}"
            )

        meta = {
            str(row[0]): row[1]
            for row in conn.execute(
                "SELECT key, value FROM state_meta WHERE key LIKE 'fts_%'"
            ).fetchall()
        }
        verification["fts_meta"] = meta
        if meta.get("fts_storage_version") != str(FTS_STORAGE_VERSION):
            verification["errors"].append(
                "fresh FTS storage version was not established"
            )
        pending_keys = sorted(
            key
            for key in (
                "fts_optimize_available",
                "fts_rebuild_high_water",
                "fts_rebuild_progress",
                "fts_cjk_stale",
                "fts_cjk_rebuild_high_water",
                "fts_cjk_rebuild_progress",
            )
            if key in meta
        )
        verification["pending_fts_keys"] = pending_keys
        if pending_keys:
            verification["errors"].append(
                "derived FTS transition markers remain in the recovered database"
            )

        counts: dict[str, int] = {}
        for table in (*_CANONICAL_TABLES, "state_meta", *_TOPIC_TABLES):
            columns = _table_columns(conn, table)
            if columns:
                counts[table] = int(
                    conn.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0]
                )
        verification["table_counts"] = counts

        for table in ("sessions", "messages"):
            expected = expected_counts.get(table)
            if expected is not None and counts.get(table) != expected:
                message = (
                    f"{table} count is {counts.get(table)}, expected {expected}"
                )
                if allow_partial:
                    verification["warnings"].append(message)
                    verification["loss_detected"] = True
                else:
                    verification["errors"].append(message)

        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.
        # This is still data loss, but it is not structural verification 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
            message = f"{table} copy status is {status}"
            if allow_partial and (
                status == "partial"
                or table not in {"sessions", "messages"}
                or (
                    table == "sessions"
                    and status == "failed"
                    and sessions_fully_reconstructed
                )
            ):
                verification["warnings"].append(message)
                verification["loss_detected"] = True
            else:
                verification["errors"].append(message)

        if orphan_cleanup:
            orphan_count = int(
                orphan_cleanup.get("total_removed_or_relinked") or 0
            )
            if orphan_count:
                verification["warnings"].append(
                    f"{orphan_count} orphaned reference(s) were removed or relinked"
                )
                verification["loss_detected"] = True
            rebuilt_sessions = int(
                orphan_cleanup.get("sessions_reconstructed") or 0
            )
            if rebuilt_sessions:
                retained = int(orphan_cleanup.get("messages_retained") or 0)
                # Not a clean recovery: the conversation text survived but its
                # session metadata did not, so these rows are placeholders.
                verification["warnings"].append(
                    f"{rebuilt_sessions} session(s) could not be salvaged and "
                    f"were reconstructed as placeholders to retain "
                    f"{retained} message(s); their metadata (title, model, "
                    "timestamps, cost) is lost"
                )
                verification["loss_detected"] = True

        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
    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 _finalize_derived_metadata(destination: sqlite3.Connection) -> dict[str, Any]:
    """Stamp only metadata that the newly created destination actually owns."""

    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}
    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)
    destination.execute("BEGIN IMMEDIATE")
    try:
        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)),
        )
        destination.execute("COMMIT")
    except BaseException:
        destination.execute("ROLLBACK")
        raise
    result["finalized"] = True
    return result


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 path 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}"
            )
        if allow_partial:
            missing_required = [
                table
                for table in ("sessions", "messages")
                if not inspection["tables"][table].get("available")
            ]
            if missing_required:
                raise SessionRecoverySourceError(
                    "Partial recovery still requires readable table schemas for: "
                    + ", ".join(missing_required)
                )

        source_conn = sqlite3.connect(
            str(snapshot_source),
            isolation_level=None,
            timeout=1.0,
        )
        source_conn.execute("PRAGMA writable_schema=ON")
        destination_db: Optional[SessionDB] = None
        destination_conn: Optional[sqlite3.Connection] = None
        try:
            has_topic_tables = any(
                inspection["tables"][table].get("available") for table in _TOPIC_TABLES
            )

            destination_db = SessionDB(db_path=output)
            if has_topic_tables:
                destination_db.apply_telegram_topic_migration()
            destination_db.close()
            destination_db = None

            destination_conn = sqlite3.connect(
                str(output),
                isolation_level=None,
                timeout=1.0,
            )
            destination_conn.execute("PRAGMA foreign_keys=OFF")

            copy_report: dict[str, dict[str, Any]] = {}
            for table in _CANONICAL_TABLES:
                table_inspection = inspection["tables"][table]
                copy_function = (
                    _copy_table_salvage if allow_partial else _copy_table
                )
                copy_report[table] = copy_function(
                    source_conn,
                    destination_conn,
                    table,
                    chunk_size=chunk_size,
                    progress_cb=progress_cb,
                    source_rows=table_inspection.get("rows"),
                )

            state_meta_inspection = inspection["tables"]["state_meta"]
            if state_meta_inspection.get("available"):
                state_meta_copy_function = (
                    _copy_state_meta_salvage
                    if allow_partial
                    else _copy_state_meta
                )
                copy_report["state_meta"] = state_meta_copy_function(
                    source_conn,
                    destination_conn,
                    chunk_size=chunk_size,
                    progress_cb=progress_cb,
                    source_rows=state_meta_inspection.get("rows"),
                )
            else:
                copy_report["state_meta"] = {"status": "missing", "copied_rows": 0}

            for table in _TOPIC_TABLES:
                table_inspection = inspection["tables"][table]
                if not table_inspection.get("available"):
                    copy_report[table] = {
                        "status": "missing",
                        "copied_rows": 0,
                    }
                    continue
                copy_function = (
                    _copy_table_salvage if allow_partial else _copy_table
                )
                copy_report[table] = copy_function(
                    source_conn,
                    destination_conn,
                    table,
                    chunk_size=chunk_size,
                    progress_cb=progress_cb,
                    source_rows=table_inspection.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()
            if destination_db is not None:
                destination_db.close()

        verification = _verify_recovered_database(
            output,
            expected_counts={
                table: inspection["tables"][table].get("rows")
                for table in _CANONICAL_TABLES
            },
            copy_report=copy_report,
            allow_partial=allow_partial,
            orphan_cleanup=orphan_cleanup,
        )
        source_unchanged = (
            _source_fingerprint(source) == inspection["source_fingerprint"]
        )
        if not source_unchanged:
            verification["errors"].append(
                "the source database bundle changed during recovery"
            )
            verification["complete"] = False

        return {
            "operation": "recover",
            "allow_partial": allow_partial,
            "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"],
            },
            "copy": copy_report,
            "orphan_cleanup": orphan_cleanup,
            "derived_metadata": derived_metadata,
            "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,
        }
    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