"""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 import sys import time 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, _cli_recover_attempts, 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"]) # ── .recover stderr pipe must be drained while the child runs ────────────── @pytest.mark.platforms("posix") # stub sqlite3 is a /bin/sh script def test_recover_attempts_survive_dump_stderr_beyond_pipe_buffer(tmp_path: Path) -> None: """A heavily damaged source makes ``.recover`` emit per-page diagnostics on stderr. Past the OS pipe buffer (~64KB) an undrained stderr blocks the dump child, its stdout never reaches EOF, and ``load.communicate()`` burns the whole salvage timeout instead of finishing in milliseconds.""" stub = tmp_path / "sqlite3" stub.write_text( "#!/bin/sh\n" 'if [ "$1" = "-readonly" ]; then\n' " i=0\n" " while [ $i -lt 3000 ]; do\n" ' echo "lost and found page $i: orphan btree cell, unable to reconstruct row" >&2\n' " i=$((i+1))\n" " done\n" ' echo "Error: near line 4000: file is not a database" >&2\n' ' echo "CREATE TABLE t(x); INSERT INTO t VALUES(1);"\n' "else\n" " cat >/dev/null\n" "fi\n", encoding="utf-8", ) stub.chmod(0o755) source = tmp_path / "source.db" source.write_bytes(b"corrupt") lf_path = tmp_path / "lost_and_found.db" started = time.monotonic() attempts = _cli_recover_attempts(source, lf_path, str(stub), timeout=30.0) assert time.monotonic() - started < 30.0 assert attempts[-1]["dump_returncode"] == 0 # The stderr tail is load-bearing: the caller keys the header-zeroing retry # on "not a database" appearing in it, so the drain must preserve it. assert "file is not a database" in attempts[-1]["dump_stderr_tail"]