hermes-agent/tests/hermes_cli/test_session_recovery_lost_and_found.py

"""Tests for recovery-tooling gaps: issue #80205 (range-query budget can
omit a recoverable tail row) and the lost_and_found last-resort lane for
sources whose table schemas are unreadable.

The corrupted fixtures here are REAL physical SQLite page damage (flipped
b-tree/schema header bytes), not mocked cursor exceptions.
"""

from __future__ import annotations

import shutil
import sqlite3
from pathlib import Path

import pytest

from hermes_state import SessionDB
from hermes_cli import session_recovery
from hermes_cli import session_schema_history
from hermes_cli.session_lost_and_found import (
    STUB_TITLE_PREFIX,
    classify_lost_and_found_row,
    map_lost_and_found_rows,
    rebuild_fts_indexes,
    stub_missing_parent_sessions,
)
from hermes_cli.session_recovery import (
    SessionRecoverySafetyError,
    SessionRecoverySourceError,
    _probe_populated_edge,
    recover_session_database,
)

from tests.hermes_cli.test_session_recovery import (
    _btree_leaf_pages,
    _make_page_spanning_source,
)


from hermes_cli.session_lost_and_found import find_sqlite3_cli

# .recover needs a sqlite3 shell built with sqlite_dbpage — PATH presence
# alone is not enough (Ubuntu CI ships a build without it).
HAVE_SQLITE3_CLI = find_sqlite3_cli() is not None


# ── physical corruption helpers ─────────────────────────────────────────────


def _page_size(data: bytes) -> int:
    size = int.from_bytes(data[16:18], "big")
    return 65_536 if size == 1 else size


def _leaf_cell_count(path: Path, page_number: int) -> int:
    data = path.read_bytes()
    page_size = _page_size(data)
    header = (page_number - 1) * page_size + (100 if page_number == 1 else 0)
    assert data[header] in {0x0A, 0x0D}
    return int.from_bytes(data[header + 3 : header + 5], "big")


def _corrupt_leaf(path: Path, page_number: int) -> None:
    data = bytearray(path.read_bytes())
    page_size = _page_size(bytes(data))
    header = (page_number - 1) * page_size + (100 if page_number == 1 else 0)
    assert data[header] in {0x0A, 0x0D}
    data[header + 3 : header + 5] = b"\xff\xff"
    path.write_bytes(data)


def _corrupt_schema_page(path: Path) -> None:
    """Damage the sqlite_master b-tree so no table schema is readable.

    Page 1 holds the schema table root. An impossible cell count in its
    header makes every ``PRAGMA table_info`` / schema read raise
    'database disk image is malformed' while the file still opens and the
    data pages of every table remain physically intact.
    """
    data = bytearray(path.read_bytes())
    assert data[:16] == b"SQLite format 3\x00"
    header = 100
    assert data[header] in {0x02, 0x05, 0x0A, 0x0D}
    data[header + 3 : header + 5] = b"\xff\xff"
    path.write_bytes(data)


def _make_schema_unreadable_source(path: Path) -> dict[str, int]:
    db = SessionDB(db_path=path)
    try:
        for session_number in range(3):
            session_id = f"20260812_1353{session_number:02d}_abc{session_number:03x}"
            db.create_session(session_id, "cli", cwd=f"/tmp/laf-{session_number}")
            db.set_session_title(session_id, f"LAF {session_number}")
            for message_number in range(9):
                db.append_message(
                    session_id,
                    "user" if message_number % 2 == 0 else "assistant",
                    f"lost-and-found payload {session_number} {message_number}",
                )
    finally:
        db.close()
    conn = sqlite3.connect(str(path), isolation_level=None)
    try:
        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
        conn.execute("PRAGMA journal_mode=DELETE")
        conn.execute("VACUUM")
    finally:
        conn.close()
    _corrupt_schema_page(path)
    return {"sessions": 3, "messages": 27}


# ── issue #80205: recoverable tail row next to a damaged rowid edge ─────────


def test_exact_lookup_recovers_tail_row_next_to_damaged_high_edge(
    tmp_path: Path,
) -> None:
    """Regression for #80205: a readable boundary row must not be omitted.

    Damaging the RIGHTMOST messages leaf makes the ordered high-edge probe
    fail, so salvage falls back to the full rowid domain. The last readable
    row (the final cell of the last healthy leaf) can only be reached through
    a singleton range once bisection narrows down — and a singleton *range*
    scan must advance past the row into the damaged sibling page to prove the
    range is exhausted, discarding the already-produced row. The fix performs
    an exact ``rowid = ?`` lookup for singleton ranges, which stops at the
    hit and recovers the row exactly as SQLite's page-level ``.recover``
    does.
    """
    source = tmp_path / "tail-damaged.db"
    output = tmp_path / "tail-recovered.db"
    message_count = 320
    messages_root, count_index_root = _make_page_spanning_source(
        source, message_count
    )

    _, leaf_pages = _btree_leaf_pages(source, messages_root)
    assert len(leaf_pages) >= 3
    rightmost_leaf = leaf_pages[-1]
    lost_rows = _leaf_cell_count(source, rightmost_leaf)
    assert 0 < lost_rows < message_count
    boundary_rowid = message_count - lost_rows
    _corrupt_leaf(source, rightmost_leaf)
    if count_index_root is not None:
        _, index_leaves = _btree_leaf_pages(source, count_index_root)
        _corrupt_leaf(source, index_leaves[-1])

    report = recover_session_database(
        source,
        output,
        work_dir=tmp_path,
        chunk_size=8,
        allow_partial=True,
    )

    copied = report["copy"]["messages"]
    bounds = copied["rowid_bounds"]
    # Premise check: the high edge probe really failed; the bound came from the aggregate
    # (#98050) or, when that fails too, the synthetic-domain fallback.
    assert any("high rowid" in error for error in bounds["errors"]), bounds
    assert "high" in bounds["fallback_edges"] or "high" in bounds.get("aggregate_edges", ())

    conn = sqlite3.connect(str(output))
    try:
        recovered_ids = {
            int(row[0]) for row in conn.execute("SELECT id FROM messages")
        }
    finally:
        conn.close()

    assert 1 in recovered_ids
    # The headline regression: the last readable row before the damage.
    assert boundary_rowid in recovered_ids, (
        f"boundary row {boundary_rowid} was omitted; max recovered "
        f"{max(recovered_ids)}; exact_lookup_recovered="
        f"{copied.get('exact_lookup_recovered')}"
    )
    assert copied["exact_lookup_recovered"] >= 1
    assert recovered_ids == set(range(1, boundary_rowid + 1))
    assert report["verification"]["integrity_check"] == ["ok"]
    assert report["verified"] is True


def test_probe_populated_edge_caps_synthetic_domain(tmp_path: Path) -> None:
    """The gallop converges on a finite bound in O(log) probes when the
    region beyond the data is cleanly seekable."""
    db_path = tmp_path / "clean.db"
    conn = sqlite3.connect(str(db_path), isolation_level=None)
    try:
        conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)")
        conn.executemany(
            "INSERT INTO t (id, v) VALUES (?, ?)",
            [(i, f"value {i}") for i in range(1, 101)],
        )
        probe = _probe_populated_edge(conn, "t", edge="high", anchor=1)
        assert probe["capped"] is True
        assert probe["bound"] >= 100
        assert probe["bound"] < 10_000
        assert probe["probes"] <= 64

        probe_low = _probe_populated_edge(conn, "t", edge="low", anchor=100)
        assert probe_low["capped"] is True
        assert probe_low["bound"] <= 1
    finally:
        conn.close()


# ── lost_and_found lane: unreadable table schemas ───────────────────────────


def test_unreadable_schema_without_cli_names_the_sqlite3_requirement(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Without a sqlite3 CLI the refusal must say exactly what to install."""
    source = tmp_path / "schemaless.db"
    output = tmp_path / "schemaless-recovered.db"
    _make_schema_unreadable_source(source)

    import hermes_cli.session_lost_and_found as laf

    monkeypatch.setattr(laf, "find_sqlite3_cli", lambda: None)
    with pytest.raises(SessionRecoverySourceError) as excinfo:
        recover_session_database(
            source,
            output,
            work_dir=tmp_path,
            allow_partial=True,
        )
    message = str(excinfo.value)
    assert "sessions" in message and "messages" in message
    assert "sqlite3" in message
    assert ".recover" in message
    assert not output.exists()


@pytest.mark.skipif(
    not HAVE_SQLITE3_CLI,
    reason="sqlite3 CLI not on PATH; .recover is a shell-only feature",
)
def test_lost_and_found_lane_recovers_schema_unreadable_source(
    tmp_path: Path,
) -> None:
    """The last-resort lane must salvage rows SQL-level recovery cannot."""
    source = tmp_path / "schemaless.db"
    output = tmp_path / "schemaless-recovered.db"
    expected = _make_schema_unreadable_source(source)

    # Premise: the schema really is unreadable at the SQL level.
    probe = sqlite3.connect(str(source))
    try:
        with pytest.raises(sqlite3.DatabaseError):
            probe.execute("SELECT COUNT(*) FROM messages").fetchone()
    finally:
        probe.close()

    report = recover_session_database(
        source,
        output,
        work_dir=tmp_path,
        allow_partial=True,
    )

    assert report["mode"] == "lost_and_found_salvage"
    assert report["best_effort"] is True
    assert report["partial"] is True
    assert report["complete"] is False
    assert report["installed"] is False
    assert report["unreadable_schemas"] == ["sessions", "messages"]
    assert any(
        "BEST-EFFORT" in warning
        for warning in report["verification"]["warnings"]
    )

    conn = sqlite3.connect(str(output))
    try:
        assert conn.execute("PRAGMA integrity_check").fetchall() == [("ok",)]
        assert conn.execute("PRAGMA foreign_key_check").fetchall() == []
        session_count = conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]
        message_count = conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
        orphans = conn.execute(
            "SELECT COUNT(*) FROM messages WHERE session_id NOT IN "
            "(SELECT id FROM sessions)"
        ).fetchone()[0]
        fts_matches = conn.execute(
            "SELECT COUNT(*) FROM messages_fts WHERE messages_fts MATCH ?",
            ("payload",),
        ).fetchone()[0]
    finally:
        conn.close()

    assert session_count == expected["sessions"]
    assert message_count == expected["messages"]
    assert orphans == 0
    assert fts_matches == expected["messages"]

    # The output must open as a regular current-schema session database.
    recovered_db = SessionDB(db_path=output)
    try:
        sessions = recovered_db.list_sessions_rich(limit=10)
        assert len(sessions) == expected["sessions"]
    finally:
        recovered_db.close()



@pytest.mark.skipif(
    not HAVE_SQLITE3_CLI,
    reason="sqlite3 CLI not on PATH; .recover is a shell-only feature",
)
def test_lost_and_found_lane_recovers_page1_header_damaged_source(tmp_path: Path) -> None:
    """#106667: a garbage page-1 header makes SQLite (and the shell's .recover) refuse the file
    with 'file is not a database' although every data page survives. The lane must still
    salvage the rows, and must do it on its snapshot — the user's file stays byte-identical."""
    source = tmp_path / "header-damaged.db"
    output = tmp_path / "header-recovered.db"
    db = SessionDB(db_path=source)
    try:
        for session_number in range(3):
            session_id = f"hdr-session-{session_number}"
            db.create_session(session_id, "cli", cwd="/tmp/hdr")
            for message_number in range(9):
                db.append_message(session_id, "user", f"payload {session_number} {message_number}")
    finally:
        db.close()
    conn = sqlite3.connect(str(source), isolation_level=None)
    try:
        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
        conn.execute("PRAGMA journal_mode=DELETE")
    finally:
        conn.close()
    data = bytearray(source.read_bytes())
    data[0:100] = bytes(range(1, 101))  # not the magic, not zeroes: the incident shape
    source.write_bytes(data)
    damaged_bytes = source.read_bytes()

    with pytest.raises(sqlite3.DatabaseError, match="not a database"):
        sqlite3.connect(str(source)).execute("SELECT count(*) FROM sqlite_master").fetchone()

    report = recover_session_database(source, output, work_dir=tmp_path, allow_partial=True)

    assert report["mode"] == "lost_and_found_salvage"
    assert report["sqlite3_cli"]["header_zeroed"] is True
    assert any("header salvage" in warning for warning in report["verification"]["warnings"])
    assert source.read_bytes() == damaged_bytes
    conn = sqlite3.connect(str(output))
    try:
        assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] == 3
        assert conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0] == 27
    finally:
        conn.close()


# ── mapper unit tests (no sqlite3 CLI required) ─────────────────────────────


def _make_synthetic_lost_and_found(
    path: Path,
    dest_schema_db: Path,
) -> dict[str, int]:
    """Build a .recover-shaped lost_and_found DB directly, no CLI needed."""
    schema = sqlite3.connect(str(dest_schema_db))
    try:
        sessions_columns = [
            str(row[1]) for row in schema.execute("PRAGMA table_info(sessions)")
        ]
        messages_columns = [
            str(row[1]) for row in schema.execute("PRAGMA table_info(messages)")
        ]
        usage_columns = [
            str(row[1])
            for row in schema.execute("PRAGMA table_info(session_model_usage)")
        ]
    finally:
        schema.close()
    # Width is derived from the live schema so ordinary column additions
    # don't break this test (it pinned 54, then 55, then 56 in one week).
    # The floor guards against accidentally reading an empty/old schema.
    current_width = len(sessions_columns)
    assert current_width >= 55
    assert len(usage_columns) == 18

    max_fields = current_width
    conn = sqlite3.connect(str(path), isolation_level=None)
    try:
        cells = ", ".join(f"c{i}" for i in range(max_fields))
        conn.execute(
            f"CREATE TABLE lost_and_found (rootpgno INTEGER, pgno INTEGER, "
            f"nfield INTEGER, id INTEGER, {cells})"
        )

        def insert(nfield: int, rowid, values: list) -> None:
            padded = list(values) + [None] * (max_fields - len(values))
            placeholders = ", ".join("?" for _ in range(4 + max_fields))
            conn.execute(
                f"INSERT INTO lost_and_found VALUES ({placeholders})",
                [2, 5, nfield, rowid, *padded],
            )

        def session_row(session_id: str, ncols: int) -> list:
            base = {
                "id": session_id,
                "source": "telegram",
                "started_at": 1_754_000_000.0,
                "message_count": 2,
                "title": f"synthetic {session_id}",
            }
            return [base.get(column) for column in sessions_columns[:ncols]]

        # Current layout (dynamic width) and historical 52-column layout.
        insert(max_fields, 1, session_row("20260101_010101_aaa001", max_fields))
        insert(52, 2, session_row("20260202_020202_bbb002", 52))
        # 14-column legacy layout: identity + a plausible epoch timestamp.
        legacy = ["20250303_030303_ccc003", "cli", 1_741_000_000.0] + [None] * 11
        insert(14, 3, legacy)

        # messages rows: NULL first cell (rowid alias), session id second,
        # role third.
        for index, (session_id, role, content) in enumerate(
            [
                ("20260101_010101_aaa001", "user", "hello from user"),
                ("20260101_010101_aaa001", "assistant", "hello from assistant"),
                ("20261111_111111_ddd004", "user", "orphaned message payload"),
                ("20261111_111111_ddd004", "tool", "orphaned tool payload"),
            ]
        ):
            row = {
                "id": None,
                "session_id": session_id,
                "role": role,
                "content": content,
                "timestamp": 1_754_000_100.0 + index,
            }
            insert(
                23,
                100 + index,
                [row.get(column) for column in messages_columns[:23]],
            )

        # session_model_usage: 18 columns, orphaned session id on purpose.
        usage = {
            "session_id": "20261212_121212_eee005",
            "model": "test/model",
            "billing_provider": "",
            "billing_base_url": "",
            "billing_mode": "",
            "task": "",
            "api_call_count": 4,
            "input_tokens": 100,
            "output_tokens": 50,
            "cache_read_tokens": 0,
            "cache_write_tokens": 0,
            "reasoning_tokens": 0,
            "estimated_cost_usd": 0.01,
            "actual_cost_usd": 0.01,
            "first_seen": 1_754_000_000.0,
            "last_seen": 1_754_000_500.0,
        }
        insert(18, 200, [usage.get(column) for column in usage_columns])

        # Junk that must NOT be classified into canonical tables.
        insert(3, 300, ["random", "noise", 42])
        insert(max_fields, 301, ["not-a-session-id", "cli"] + [None] * (max_fields - 2))
        insert(23, 302, [None, "sess-x", "not-a-role", "junk"])
    finally:
        conn.close()
    return {
        "sessions": 3,
        "messages": 4,
        "session_model_usage": 1,
        "junk": 3,
    }


def test_classify_lost_and_found_row_sentinels() -> None:
    assert (
        classify_lost_and_found_row(
            23, (None, "20260101_010101_aaa001", "user", "hi")
        )
        == "messages"
    )
    assert (
        classify_lost_and_found_row(
            55, ("20260101_010101_aaa001", "cli") + (None,) * 53
        )
        == "sessions"
    )
    assert (
        classify_lost_and_found_row(
            52, ("20260101_010101_aaa001", "discord") + (None,) * 50
        )
        == "sessions"
    )
    assert (
        classify_lost_and_found_row(
            14, ("20250101_010101_zzz999", "cli") + (None,) * 12
        )
        == "sessions"
    )
    assert (
        classify_lost_and_found_row(
            18, ("20260101_010101_aaa001", "gpt-x") + (None,) * 16
        )
        == "session_model_usage"
    )
    # Junk shapes.
    assert classify_lost_and_found_row(3, ("random", "noise", 42)) is None
    assert (
        classify_lost_and_found_row(55, ("not-a-session-id", "cli") + (None,) * 53)
        is None
    )
    assert (
        classify_lost_and_found_row(23, (None, "sess", "not-a-role", "x")) is None
    )
    assert classify_lost_and_found_row(0, ()) is None


def test_mapper_rebuilds_sessiondb_from_synthetic_lost_and_found(
    tmp_path: Path,
) -> None:
    """Binary-independent: mapper + stubbing + FTS rebuild end to end."""
    schema_ref = tmp_path / "schema-ref.db"
    SessionDB(db_path=schema_ref).close()

    lf_path = tmp_path / "lost_and_found.db"
    expected = _make_synthetic_lost_and_found(lf_path, schema_ref)

    output = tmp_path / "mapped.db"
    SessionDB(db_path=output).close()

    lf_conn = sqlite3.connect(str(lf_path), isolation_level=None)
    dest = sqlite3.connect(str(output), isolation_level=None)
    try:
        dest.execute("PRAGMA foreign_keys=OFF")
        mapping = map_lost_and_found_rows(lf_conn, dest)
        stubbing = stub_missing_parent_sessions(dest)
        fts = rebuild_fts_indexes(dest)

        assert mapping["mapped"]["sessions"] == expected["sessions"]
        assert mapping["mapped"]["messages"] == expected["messages"]
        assert (
            mapping["mapped"]["session_model_usage"]
            == expected["session_model_usage"]
        )
        assert mapping["legacy_minimal_sessions"] == 1
        assert mapping["unmapped_rows"] == expected["junk"]

        # Orphaned children got stub parents — never deleted.
        assert stubbing["sessions_stubbed"] == 2  # ddd004 + eee005
        assert stubbing["messages_retained"] == 2
        message_count = dest.execute("SELECT COUNT(*) FROM messages").fetchone()[0]
        assert message_count == expected["messages"]
        usage_count = dest.execute(
            "SELECT COUNT(*) FROM session_model_usage"
        ).fetchone()[0]
        assert usage_count == expected["session_model_usage"]

        stub_titles = [
            str(row[0])
            for row in dest.execute(
                "SELECT title FROM sessions WHERE source = 'recovered'"
            )
        ]
        assert len(stub_titles) == 2
        assert all(title.startswith("[best-effort recovered") for title in stub_titles)

        # The 52-col row landed with its real metadata preserved.
        row = dest.execute(
            "SELECT source, title FROM sessions WHERE id = ?",
            ("20260202_020202_bbb002",),
        ).fetchone()
        assert row == ("telegram", "synthetic 20260202_020202_bbb002")

        assert fts.get("messages_fts") == "rebuilt"
        fts_hits = dest.execute(
            "SELECT COUNT(*) FROM messages_fts WHERE messages_fts MATCH ?",
            ("payload OR hello",),
        ).fetchone()[0]
        assert fts_hits == expected["messages"]

        assert dest.execute("PRAGMA integrity_check").fetchall() == [("ok",)]
        assert dest.execute("PRAGMA foreign_key_check").fetchall() == []
    finally:
        lf_conn.close()
        dest.close()

    # And the mapped output opens through the normal SessionDB path.
    db = SessionDB(db_path=output)
    try:
        assert len(db.list_sessions_rich(limit=20)) == 5
    finally:
        db.close()


# ── issue #72291: source-fingerprint error must name the parent CLI ─────────


def test_fingerprint_error_enumerates_parent_cli_session(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    source = tmp_path / "busy.db"
    SessionDB(db_path=source).close()

    fingerprints = iter([{"main": {"size": 1, "mtime_ns": 1}},
                         {"main": {"size": 2, "mtime_ns": 2}},
                         {"main": {"size": 3, "mtime_ns": 3}}])
    monkeypatch.setattr(
        session_recovery,
        "_source_fingerprint",
        lambda _source: next(fingerprints),
    )
    with pytest.raises(SessionRecoverySafetyError) as excinfo:
        session_recovery.inspect_session_database(source, work_dir=tmp_path)
    message = str(excinfo.value)
    assert "Stop every Hermes process" in message
    # The gap from #72291: the parent CLI session itself must be enumerated.
    assert "CLI session" in message
    assert "fresh shell" in message
    assert "snapshot" in message


# ── issue #101409: mis-mapped salvage must not be reported verified ─────────


def _map_salvage_rows(
    tmp_path: Path,
    *,
    blank_session_started_at: bool,
    blank_message_timestamp: bool,
) -> sqlite3.Connection:
    """Map synthetic lost_and_found cells into a fresh template DB.

    With either ``blank_*`` flag the cells mimic an upgraded source's
    *physical* column order (#101409): whatever lands on the declared
    ``started_at``/``timestamp`` position is not an epoch timestamp, so
    the NOT NULL substitute turns it into 0.0 on every row.
    """

    schema_ref = tmp_path / "schema-ref.db"
    SessionDB(db_path=schema_ref).close()
    schema = sqlite3.connect(str(schema_ref))
    try:
        sessions_columns = [
            str(row[1]) for row in schema.execute("PRAGMA table_info(sessions)")
        ]
        messages_columns = [
            str(row[1]) for row in schema.execute("PRAGMA table_info(messages)")
        ]
    finally:
        schema.close()
    current_width = len(sessions_columns)

    lf_path = tmp_path / "lost_and_found.db"
    lf_conn = sqlite3.connect(str(lf_path), isolation_level=None)
    try:
        lf_cells = ", ".join(f"c{i}" for i in range(current_width))
        lf_conn.execute(
            "CREATE TABLE lost_and_found (rootpgno INTEGER, pgno INTEGER, "
            "nfield INTEGER, id INTEGER, " + lf_cells + ")"
        )

        def insert(nfield: int, rowid: int, values: list) -> None:
            padded = list(values) + [None] * (current_width - len(values))
            placeholders = ", ".join("?" for _ in range(4 + current_width))
            lf_conn.execute(
                "INSERT INTO lost_and_found VALUES (" + placeholders + ")",
                [2, 5, nfield, rowid, *padded],
            )

        def session_row(session_id: str) -> list:
            # title is UNIQUE (idx_sessions_title_unique) — keep it distinct
            # per row so the probe isolates timestamp mis-mapping.
            # Five populated cells cannot pin a 58-column layout by
            # themselves; the discriminating cells a real store carries
            # (session_key, model_config, cwd, ...) make the declared order
            # the only surviving layout, so the probe below isolates the
            # timestamp gate rather than layout inference.
            row = {
                "id": session_id,
                "source": "telegram",
                "session_key": f"agent:main:telegram:dm:{session_id}",
                "chat_type": "dm",
                "model": "gpt-4.1",
                "model_config": "{}",
                "started_at": None
                if blank_session_started_at
                else 1_754_000_000.0,
                "ended_at": 1_754_000_600.0,
                "end_reason": "completed",
                "message_count": 2,
                "cwd": "/home/user/project",
                "title": f"mis-mapped probe {session_id}",
                "title_source": "llm",
                "api_call_count": 1,
            }
            return [row.get(column) for column in sessions_columns]

        for index in range(3):
            insert(
                current_width,
                index + 1,
                session_row(f"20260101_01010{index}_aaa00{index}"),
            )

        for index in range(2):
            message = {
                "id": None,
                "session_id": "20260101_010100_aaa000",
                "role": "user",
                "content": "payload",
                "timestamp": None
                if blank_message_timestamp
                else 1_754_000_100.0 + index,
            }
            insert(
                23,
                100 + index,
                [message.get(column) for column in messages_columns[:23]],
            )
    finally:
        lf_conn.close()

    output = tmp_path / "mapped.db"
    SessionDB(db_path=output).close()
    lf_conn = sqlite3.connect(str(lf_path), isolation_level=None)
    dest = sqlite3.connect(str(output), isolation_level=None)
    try:
        dest.execute("PRAGMA foreign_keys=OFF")
        map_lost_and_found_rows(lf_conn, dest)
    finally:
        lf_conn.close()
        dest.close()
    return sqlite3.connect(str(output), isolation_level=None)


def test_plausibility_gate_flags_positional_mis_mapping(
    tmp_path: Path,
) -> None:
    """A salvage whose timestamps all landed below the epoch floor was
    mapped onto the wrong columns and must be flagged, not verified
    (#101409)."""

    conn = _map_salvage_rows(
        tmp_path,
        blank_session_started_at=True,
        blank_message_timestamp=False,
    )
    try:
        # The mapper happily inserted every row; structural checks pass.
        assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] == 3
        assert conn.execute(
            "SELECT COUNT(*) FROM sessions WHERE started_at = 0.0"
        ).fetchone()[0] == 3

        errors = session_recovery._lost_and_found_plausibility_errors(conn)
        assert len(errors) == 1
        assert "sessions.started_at" in errors[0]
    finally:
        conn.close()


def test_plausibility_gate_flags_mis_mapped_message_timestamps(
    tmp_path: Path,
) -> None:
    conn = _map_salvage_rows(
        tmp_path,
        blank_session_started_at=False,
        blank_message_timestamp=True,
    )
    try:
        errors = session_recovery._lost_and_found_plausibility_errors(conn)
        assert len(errors) == 1
        assert "messages.timestamp" in errors[0]
    finally:
        conn.close()


def test_plausibility_gate_passes_correctly_mapped_salvage(
    tmp_path: Path,
) -> None:
    """Well-mapped rows — and partially damaged ones (a torn cell on some
    rows is expected salvage noise) — must not trip the gate: it fires
    only on a *systematic* violation."""

    conn = _map_salvage_rows(
        tmp_path,
        blank_session_started_at=False,
        blank_message_timestamp=False,
    )
    try:
        # Damage one of three sessions the way a torn cell would.
        conn.execute(
            "UPDATE sessions SET started_at = 0.0 WHERE id = ?",
            ("20260101_010101_aaa001",),
        )
        conn.commit()

        assert session_recovery._lost_and_found_plausibility_errors(conn) == []
    finally:
        conn.close()


def _rebuild_with_started_at_appended(conn: sqlite3.Connection) -> None:
    """Give ``sessions`` the physical layout of an upgraded DB: ``started_at``
    lands at the END (as ``ALTER TABLE ADD COLUMN`` would place a column that
    the current template declares mid-definition). Data is preserved."""
    info = list(conn.execute("PRAGMA table_info(sessions)"))
    declared = [row[1] for row in info]

    def coldef(row):
        _, name, ctype, notnull, dflt, pk = row
        parts = [f'"{name}" {ctype}']
        if pk:
            parts.append("PRIMARY KEY")
        if notnull:
            parts.append("NOT NULL")
        if dflt is not None:
            parts.append(f"DEFAULT {dflt}")
        return " ".join(parts)

    reordered = [r for r in info if r[1] != "started_at"] + [r for r in info if r[1] == "started_at"]
    cols = ", ".join(f'"{c}"' for c in declared)
    # ``messages_fts_trigram_src`` joins sessions; SQLite refuses to rename a
    # table a view references unless legacy_alter_table is on (the view's
    # body is text, so it re-resolves ``sessions`` after the swap).
    conn.executescript("PRAGMA foreign_keys=OFF; PRAGMA legacy_alter_table=ON;")
    conn.execute("CREATE TABLE sessions_new (" + ", ".join(coldef(r) for r in reordered) + ")")
    conn.execute(f"INSERT INTO sessions_new({cols}) SELECT {cols} FROM sessions")
    conn.executescript("DROP TABLE sessions; ALTER TABLE sessions_new RENAME TO sessions;")


@pytest.mark.skipif(
    not HAVE_SQLITE3_CLI,
    reason="sqlite3 CLI not on PATH; .recover is a shell-only feature",
)
def test_lost_and_found_lane_refuses_to_verify_a_physically_shifted_source(
    tmp_path: Path,
) -> None:
    """#101409 end to end: a source whose physical column order differs from
    the template's declared order maps every cell onto the wrong column. The
    output still passes integrity/FK/FTS, so only the plausibility gate can
    stop the report from claiming ``verified``."""
    source = tmp_path / "upgraded.db"
    output = tmp_path / "upgraded-recovered.db"
    db = SessionDB(db_path=source)
    try:
        for n in range(3):
            sid = f"20260812_1400{n:02d}_def{n:03x}"
            db.create_session(sid, "cli", cwd=f"/tmp/shift-{n}")
            db.set_session_title(sid, f"shift {n}")
            for m in range(4):
                db.append_message(sid, "user" if m % 2 == 0 else "assistant", f"payload {n} {m}")
    finally:
        db.close()
    conn = sqlite3.connect(str(source), isolation_level=None)
    try:
        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
        conn.execute("PRAGMA journal_mode=DELETE")
        _rebuild_with_started_at_appended(conn)
        conn.execute("VACUUM")
        physical = [r[1] for r in conn.execute("PRAGMA table_info(sessions)")]
        assert physical[-1] == "started_at"
    finally:
        conn.close()
    # The reporter's damage: page 1 (header + sqlite_master) overwritten, so
    # ``.recover`` cannot name any table and every row lands in
    # lost_and_found, to be mapped positionally onto the template.
    with open(source, "r+b") as fh:
        fh.write(b"\0" * _page_size(source.read_bytes()))

    report = recover_session_database(source, output, work_dir=tmp_path, allow_partial=True)

    assert report["mode"] == "lost_and_found_salvage"
    # Mis-mapped rows that trip a NOT NULL / type constraint are stubbed, not
    # mapped (the reporter saw 190 of 1,875) — at least one lands positionally.
    assert report["lost_and_found"]["mapped"]["sessions"] >= 1
    assert report["verification"]["healthy"] is False
    assert report["verified"] is False
    assert any("sessions.started_at is implausible" in e for e in report["verification"]["errors"])
    out = sqlite3.connect(str(output))
    try:
        # The mis-mapping the gate caught: every mapped (non-stub) session got
        # the NOT NULL substitute where its real start time should be.
        mapped = out.execute(
            f"SELECT started_at FROM sessions WHERE COALESCE(title, '') NOT LIKE '{STUB_TITLE_PREFIX}%'"
        ).fetchall()
        assert mapped and all(row[0] == 0.0 for row in mapped)
    finally:
        out.close()


# The physical column order of the reporter's upgraded store in #101409:
# every column ALTER TABLE ADD COLUMN appended, in add order, which is NOT
# the order SCHEMA_SQL declares them in. Written out literally (rather than
# read back from the production layout table) so the regression pins the
# reported layout, not whatever the mapper believes today.
_UPGRADED_SESSIONS_PHYSICAL = (
    "id", "source", "user_id", "model", "model_config", "system_prompt",
    "parent_session_id", "started_at", "ended_at", "end_reason",
    "message_count", "tool_call_count", "input_tokens", "output_tokens",
    "cache_read_tokens", "cache_write_tokens", "reasoning_tokens",
    "billing_provider", "billing_base_url", "billing_mode",
    "estimated_cost_usd", "actual_cost_usd", "cost_status", "cost_source",
    "pricing_version", "title", "api_call_count", "handoff_state",
    "handoff_platform", "handoff_error", "cwd", "rewind_count", "archived",
    "session_key", "chat_id", "chat_type", "thread_id", "git_branch",
    "git_repo_root", "compression_failure_cooldown_until",
    "compression_failure_error", "display_name", "origin_json",
    "expiry_finalized", "compression_fallback_streak", "profile_name",
    "compression_ineffective_count", "pinned", "system_prompt_hash",
    "last_activity_at", "last_activity_description",
    "last_activity_provenance", "git_metadata_generation", "title_source",
    "hidden", "last_read_at",
)

_UPGRADED_MESSAGES_PHYSICAL = (
    "id", "session_id", "role", "content", "tool_call_id", "tool_calls",
    "tool_name", "timestamp", "token_count", "finish_reason", "reasoning",
    "reasoning_content", "reasoning_details", "codex_reasoning_items",
    "codex_message_items", "platform_message_id", "observed", "active",
    "compacted", "effect_disposition", "api_content", "display_kind",
    "display_metadata", "_compressed_summary",
)

_UPGRADED_USAGE_PHYSICAL = (
    "session_id", "model", "billing_provider", "billing_base_url",
    "api_call_count", "input_tokens", "output_tokens", "cache_read_tokens",
    "cache_write_tokens", "reasoning_tokens", "estimated_cost_usd",
    "first_seen", "last_seen", "billing_mode", "actual_cost_usd",
    "cost_status", "cost_source", "task",
)


def test_upgraded_physical_layout_maps_cells_by_name(tmp_path: Path) -> None:
    """#101409: salvaged cells from an ALTER-TABLE-upgraded store must land on
    the columns they came from, not on the destination's declared order.

    The reporter's store has ``model`` where the template declares
    ``message_count``, ``started_at`` eight columns earlier than declared, and
    ``timestamp`` where ``effect_disposition`` is declared. Mapping by
    position writes the message count into ``model``, 0.0 into ``started_at``
    and blanks every title.
    """

    output = tmp_path / "mapped.db"
    SessionDB(db_path=output).close()

    declared = sqlite3.connect(str(output))
    try:
        sessions_declared = [
            str(row[1]) for row in declared.execute("PRAGMA table_info(sessions)")
        ]
        messages_declared = [
            str(row[1]) for row in declared.execute("PRAGMA table_info(messages)")
        ]
    finally:
        declared.close()
    # Premise: declared order really does differ from the physical order, so
    # a positional map cannot be correct.
    assert sessions_declared[:56] != list(_UPGRADED_SESSIONS_PHYSICAL)
    assert messages_declared[:24] != list(_UPGRADED_MESSAGES_PHYSICAL)

    session_id = "20260701_101010_abc001"
    started_at = 1_754_000_000.0
    message_timestamp = 1_754_000_321.0
    session_cells = {
        "id": session_id,
        "source": "telegram",
        "model": "gpt-4.1",
        "started_at": started_at,
        "message_count": 42,
        "tool_call_count": 7,
        "input_tokens": 1_234,
        "output_tokens": 567,
        "title": "quarterly planning notes",
        "title_source": "llm",
        "cwd": "/home/user/project",
        "git_branch": "main",
        "last_activity_at": started_at + 900.0,
        "archived": 0,
        "pinned": 0,
    }
    message_cells = {
        "id": None,  # rowid alias: NULL in the record
        "session_id": session_id,
        "role": "assistant",
        "content": "salvaged assistant payload",
        "timestamp": message_timestamp,
        "token_count": 55,
        "finish_reason": "stop",
        "observed": 1,
        "active": 1,
    }
    usage_cells = {
        "session_id": session_id,
        "model": "gpt-4.1",
        "billing_provider": "openai",
        "billing_base_url": "https://api.openai.com/v1",
        "api_call_count": 4,
        "input_tokens": 1_234,
        "output_tokens": 567,
        "estimated_cost_usd": 0.25,
        "billing_mode": "api",
        "task": "",
        "first_seen": started_at,
        "last_seen": started_at + 900.0,
    }

    lf_path = tmp_path / "lost_and_found.db"
    lf_conn = sqlite3.connect(str(lf_path), isolation_level=None)
    try:
        width = len(_UPGRADED_SESSIONS_PHYSICAL)
        columns = ", ".join(f"c{index}" for index in range(width))
        lf_conn.execute(
            "CREATE TABLE lost_and_found (rootpgno INTEGER, pgno INTEGER, "
            "nfield INTEGER, id INTEGER, " + columns + ")"
        )

        def insert(layout: tuple[str, ...], rowid: int, values: dict) -> None:
            cells = [values.get(name) for name in layout]
            padded = cells + [None] * (width - len(cells))
            placeholders = ", ".join("?" for _ in range(4 + width))
            lf_conn.execute(
                "INSERT INTO lost_and_found VALUES (" + placeholders + ")",
                [2, 5, len(layout), rowid, *padded],
            )

        insert(_UPGRADED_SESSIONS_PHYSICAL, 1, session_cells)
        insert(_UPGRADED_MESSAGES_PHYSICAL, 100, message_cells)
        insert(_UPGRADED_USAGE_PHYSICAL, 200, usage_cells)
    finally:
        lf_conn.close()

    lf_conn = sqlite3.connect(str(lf_path), isolation_level=None)
    dest = sqlite3.connect(str(output), isolation_level=None)
    try:
        dest.execute("PRAGMA foreign_keys=OFF")
        report = map_lost_and_found_rows(lf_conn, dest)
        assert report["mapped"] == {
            "sessions": 1, "messages": 1, "session_model_usage": 1,
        }
        assert report["unrecognized_layout_rows"] == 0

        session = dict(
            zip(
                ("started_at", "title", "model", "message_count", "source",
                 "cwd", "title_source"),
                dest.execute(
                    "SELECT started_at, title, model, message_count, source, "
                    "cwd, title_source FROM sessions WHERE id = ?",
                    (session_id,),
                ).fetchone(),
            )
        )
        # Each cell landed on the column it was written from — not shifted.
        assert session["started_at"] == started_at
        assert session["title"] == session_cells["title"]
        assert session["model"] == session_cells["model"]
        assert session["message_count"] == session_cells["message_count"]
        assert session["source"] == session_cells["source"]
        assert session["cwd"] == session_cells["cwd"]
        assert session["title_source"] == session_cells["title_source"]

        timestamp, role, content, disposition, token_count = dest.execute(
            "SELECT timestamp, role, content, effect_disposition, token_count "
            "FROM messages WHERE session_id = ?",
            (session_id,),
        ).fetchone()
        assert timestamp == message_timestamp
        assert role == message_cells["role"]
        assert content == message_cells["content"]
        assert token_count == message_cells["token_count"]
        # The declared-order collision the issue names: timestamp must not
        # have been written into effect_disposition.
        assert disposition is None

        usage_model, task, calls, mode = dest.execute(
            "SELECT model, task, api_call_count, billing_mode "
            "FROM session_model_usage WHERE session_id = ?",
            (session_id,),
        ).fetchone()
        assert usage_model == usage_cells["model"]
        assert task == usage_cells["task"]
        assert calls == usage_cells["api_call_count"]
        assert mode == usage_cells["billing_mode"]

        # Correctly mapped salvage has nothing for the gate to flag.
        assert session_recovery._lost_and_found_plausibility_errors(dest) == []
    finally:
        lf_conn.close()
        dest.close()


def test_plausibility_gate_ignores_stub_only_sessions(tmp_path: Path) -> None:
    """Stub rows from ``stub_missing_parent_sessions`` legitimately carry
    ``started_at = 0.0``; a salvage where only stubs survived is depleted,
    not mis-mapped, and must not be flagged."""
    output = tmp_path / "stubs.db"
    SessionDB(db_path=output).close()
    conn = sqlite3.connect(str(output))
    try:
        now = 1_750_000_000.0
        conn.execute(
            "INSERT INTO sessions (id, source, started_at, title) VALUES (?, ?, ?, ?)",
            ("20260812_140000_aaa000", "recovered", 0.0, "[best-effort recovered 1] session metadata was unreadable"),
        )
        conn.execute(
            "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
            ("20260812_140000_aaa000", "user", "hi", now),
        )
        conn.commit()
        assert session_recovery._lost_and_found_plausibility_errors(conn) == []
        # One genuinely mapped row with a real timestamp keeps it clean too...
        conn.execute(
            "INSERT INTO sessions (id, source, started_at, title) VALUES (?, ?, ?, ?)",
            ("20260812_140001_aaa001", "cli", now, None),
        )
        conn.commit()
        assert session_recovery._lost_and_found_plausibility_errors(conn) == []
        # ...and a mapped row at 0.0 with a NULL title (the mis-mapped shape:
        # blank titles) is still counted as mapped, not as a stub.
        conn.execute("UPDATE sessions SET started_at = 0.0 WHERE id = '20260812_140001_aaa001'")
        conn.commit()
        errors = session_recovery._lost_and_found_plausibility_errors(conn)
        assert len(errors) == 1 and "sessions.started_at" in errors[0]
    finally:
        conn.close()


@pytest.mark.skipif(
    not HAVE_SQLITE3_CLI,
    reason="sqlite3 CLI not on PATH; .recover is a shell-only feature",
)
def test_recovery_lane_refuses_to_verify_when_rows_matched_no_layout(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Wiring: ``unrecognized_layout_rows`` from the mapper must reach the
    verifier — once the recognised rows map correctly, the all-rows
    timestamp gate cannot see a few positionally guessed ones."""
    import hermes_cli.session_lost_and_found as lf_module

    real = lf_module.map_lost_and_found_rows

    def counting(lf_conn, dest):
        report = real(lf_conn, dest)
        report["unrecognized_layout_rows"] += 2
        return report

    monkeypatch.setattr(lf_module, "map_lost_and_found_rows", counting)

    source = tmp_path / "source.db"
    output = tmp_path / "recovered.db"
    db = SessionDB(db_path=source)
    try:
        sid = "20260812_140000_def000"
        db.create_session(sid, "cli", cwd="/tmp/x")
        db.append_message(sid, "user", "payload")
    finally:
        db.close()
    conn = sqlite3.connect(str(source), isolation_level=None)
    try:
        conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
        conn.execute("PRAGMA journal_mode=DELETE")
        conn.execute("VACUUM")
    finally:
        conn.close()
    with open(source, "r+b") as fh:
        fh.write(b"\0" * _page_size(source.read_bytes()))

    report = recover_session_database(source, output, work_dir=tmp_path, allow_partial=True)
    assert report["mode"] == "lost_and_found_salvage"
    assert report["lost_and_found"]["unrecognized_layout_rows"] == 2
    assert report["verified"] is False
    assert any("matched no known physical column layout" in e for e in report["verification"]["errors"])