From 6dad74596e468c1f1abc2cb6f745992aad3d96d1 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 12 Aug 2026 16:39:00 -0700 Subject: [PATCH] fix(sessions): recover budget exhaustion + lost_and_found last-resort lane Fixes #80205: when one ordered rowid-edge probe failed, _salvage_rowid_bounds() substituted the whole SQLite rowid domain and _copy_table_salvage() burned the 10,000-query budget bisecting a synthetic tail that could not contain rows, silently omitting readable boundary rows (field case: message 76882 of 76882). Two-part fix: * _probe_populated_edge(): gallop outward from the surviving edge with doubling offsets; a clean 'no rows beyond X' probe caps the domain in O(log range) queries instead of exhausting the budget on it. * exact-key singleton salvage: a one-row range scan must advance the cursor past the hit into the damaged sibling page to prove exhaustion, which discards the already-produced row; 'WHERE rowid = ?' stops at the hit, recovering the boundary row exactly like sqlite3 .recover. * the strict-path refusal now points users at --allow-partial. New last-resort lane for --allow-partial when the sessions/messages table schemas themselves are unreadable (previously a hard refusal even though page-level salvage recovers the rows fine). If a sqlite3 CLI is on PATH, shell out to '.recover --ignore-freelist' into a scratch lost_and_found DB, then heuristically map rows back into a fresh SessionDB-schema database (hermes_cli/session_lost_and_found.py): classification keyed on nfield counts + sentinel columns (session ids matching ^\d{8}_\d{6}_, roles in user/assistant/tool/system, known source strings), covering the current 54-col sessions layout, the 52-col historical layout, a 14-col legacy identity-only salvage, rowid-alias messages rows and 18-col session_model_usage rows. Missing parent sessions are stubbed (children are never deleted for FK cleanup), FTS is rebuilt at the end, and output is labeled BEST-EFFORT everywhere. Without the CLI the error names the sqlite3 requirement with actionable guidance. Mirrors a successful manual recovery of a real corrupt state.db (2026-08-12), and this lane was validated against that preserved file: 32 sessions / 7 messages / 4 usage rows mapped, integrity_check ok, opens via SessionDB. Also fixes #72291: the source-fingerprint 'bundle changed while it was being copied' error now enumerates that the parent interactive CLI session itself counts as a Hermes process and suggests a fresh shell or an immutable snapshot. Tests use real physical page corruption (flipped b-tree/schema header bytes), skip the CLI-dependent path cleanly when sqlite3 is absent, and keep the mapper unit tests binary-independent via a synthetic lost_and_found DB. Sabotage-verified: reverting the fixes makes the regression tests fail with the exact field failure shape. --- hermes_cli/session_lost_and_found.py | 543 +++++++++++++++++ hermes_cli/session_recovery.py | 307 +++++++++- hermes_cli/sessions_cmd.py | 10 +- .../test_session_recovery_lost_and_found.py | 563 ++++++++++++++++++ 4 files changed, 1411 insertions(+), 12 deletions(-) create mode 100644 hermes_cli/session_lost_and_found.py create mode 100644 tests/hermes_cli/test_session_recovery_lost_and_found.py diff --git a/hermes_cli/session_lost_and_found.py b/hermes_cli/session_lost_and_found.py new file mode 100644 index 0000000000000..e505cf5843171 --- /dev/null +++ b/hermes_cli/session_lost_and_found.py @@ -0,0 +1,543 @@ +"""Last-resort page-level salvage for an unreadable session database schema. + +``hermes sessions recover --allow-partial`` normally copies rows through SQL, +which requires the ``sessions`` and ``messages`` table *schemas* to be +readable. When the schema page itself is damaged, SQL-level salvage is +impossible — but the row payloads frequently survive on their b-tree pages. + +The SQLite command-line shell ships a page-level ``.recover`` command that +walks raw pages and rebuilds rows it cannot attribute to a schema into +``lost_and_found`` tables of the shape:: + + lost_and_found(rootpgno, pgno, nfield, id, c0, c1, ..., cN) + +This module shells out to that CLI (it is a shell feature, NOT available via +the Python ``sqlite3`` module) and then heuristically maps ``lost_and_found`` +rows back into a fresh current-schema Hermes session database. + +Everything produced through this lane is explicitly **best effort**: column +mapping is heuristic (field counts plus sentinel values), fabricated parent +sessions are stubbed for orphaned child rows rather than deleting salvaged +data, and derived FTS indexes are rebuilt from scratch. +""" + +from __future__ import annotations + +import re +import shutil +import sqlite3 +import subprocess +from pathlib import Path +from typing import Any, Optional + +# Hermes session ids are timestamps: 20260812_135332_ab12cd. This is the +# strongest sentinel available for classifying schema-less rows. +SESSION_ID_PATTERN = re.compile(r"^\d{8}_\d{6}_") + +MESSAGE_ROLES = frozenset({"user", "assistant", "tool", "system"}) + +# Values observed in sessions.source across gateway platforms and tooling. +KNOWN_SOURCES = frozenset({ + "cli", "telegram", "discord", "slack", "whatsapp", "signal", "matrix", + "irc", "email", "x", "twitter", "api", "gateway", "web", "dashboard", + "tool", "subagent", "cron", "recovered", "imported", "acp", +}) + +# Historical physical layouts of the sessions table. Columns are only ever +# appended (ALTER TABLE ADD COLUMN), so an older record is a strict prefix of +# the current column order. +SESSIONS_LAYOUT_NFIELDS = frozenset({54, 52}) +SESSIONS_LEGACY_MINIMAL_NFIELD = 14 +SESSION_MODEL_USAGE_NFIELD = 18 + +# Plausible unix-epoch window for started_at heuristics on legacy layouts. +_EPOCH_LOW = 1_000_000_000.0 # 2001 +_EPOCH_HIGH = 4_000_000_000.0 # 2096 + +SQLITE3_CLI_GUIDANCE = ( + "A last-resort page-level salvage is available when the `sqlite3` " + "command-line shell is installed: its `.recover` command can rebuild " + "rows into lost_and_found tables even when the table schemas are " + "unreadable (this is a CLI-only feature, not part of Python's sqlite3 " + "module). Install the sqlite3 CLI (e.g. `apt install sqlite3` or " + "`brew install sqlite`) so it is on PATH, then re-run with " + "--allow-partial." +) + + +class LostAndFoundError(RuntimeError): + """Raised when the CLI .recover pass cannot produce a usable database.""" + + +def find_sqlite3_cli() -> Optional[str]: + """Return the sqlite3 CLI path, or None when page-level salvage is out.""" + + return shutil.which("sqlite3") + + +def run_cli_lost_and_found_recover( + source: Path, + lf_path: Path, + sqlite3_bin: str, + *, + timeout: float = 3600.0, +) -> dict[str, Any]: + """Run ``sqlite3 .recover`` streamed into a fresh scratch DB. + + ``--ignore-freelist`` avoids resurrecting deleted rows; older shells + without that option fall back to a plain ``.recover``. + """ + + attempts: list[dict[str, Any]] = [] + for command in (".recover --ignore-freelist", ".recover"): + if lf_path.exists(): + lf_path.unlink() + dump = subprocess.Popen( + [sqlite3_bin, "-readonly", str(source), command], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + load = subprocess.Popen( + [sqlite3_bin, str(lf_path)], + stdin=dump.stdout, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + ) + assert dump.stdout is not None + dump.stdout.close() # let dump receive SIGPIPE if load dies + try: + _, load_err = load.communicate(timeout=timeout) + dump_err = dump.stderr.read() if dump.stderr is not None else b"" + dump.wait(timeout=60) + except subprocess.TimeoutExpired: + dump.kill() + load.kill() + raise LostAndFoundError( + f"sqlite3 .recover timed out after {timeout:.0f}s" + ) + attempt = { + "command": command, + "dump_returncode": dump.returncode, + "load_returncode": load.returncode, + "dump_stderr_tail": dump_err.decode("utf-8", "replace")[-2000:], + "load_stderr_tail": load_err.decode("utf-8", "replace")[-2000:], + } + attempts.append(attempt) + if _lost_and_found_db_usable(lf_path): + attempt["usable"] = True + return {"binary": sqlite3_bin, "attempts": attempts} + attempt["usable"] = False + + raise LostAndFoundError( + "sqlite3 .recover did not produce a usable lost_and_found database: " + + "; ".join( + f"[{a['command']}] dump rc={a['dump_returncode']} " + f"load rc={a['load_returncode']} " + f"{a['dump_stderr_tail'] or a['load_stderr_tail']}".strip() + for a in attempts + ) + ) + + +def _lost_and_found_db_usable(lf_path: Path) -> bool: + if not lf_path.exists() or lf_path.stat().st_size == 0: + return False + try: + conn = sqlite3.connect(str(lf_path)) + try: + tables = [ + str(row[0]) + for row in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + ] + finally: + conn.close() + except sqlite3.DatabaseError: + return False + return bool(tables) + + +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 _notnull_defaults(conn: sqlite3.Connection, table: str) -> dict[int, Any]: + """Map column index -> substitute value for NOT NULL columns. + + Page-level salvage can hand back records with NULL in positions that the + live schema declares NOT NULL (torn cells, historical rows). Dropping the + whole row over one damaged optional counter would defeat the lane, so + NULLs in such positions are replaced by the schema default (or '' / 0 + when no default is declared). + """ + + substitutes: dict[int, Any] = {} + for index, row in enumerate(conn.execute(f'PRAGMA table_info("{table}")')): + if not row[3]: # notnull flag + continue + default = row[4] + if default is None: + declared = str(row[2] or "").upper() + substitutes[index] = 0 if ("INT" in declared or "REAL" in declared) else "" + continue + text = str(default) + if text.startswith("'") and text.endswith("'"): + substitutes[index] = text[1:-1] + else: + try: + substitutes[index] = int(text) + except ValueError: + try: + substitutes[index] = float(text) + except ValueError: + substitutes[index] = text + return substitutes + + +def _is_session_id(value: Any) -> bool: + return isinstance(value, str) and bool(SESSION_ID_PATTERN.match(value)) + + +def _looks_like_source(value: Any) -> bool: + if not isinstance(value, str) or not value: + return False + return value in KNOWN_SOURCES or bool(re.fullmatch(r"[a-z][a-z0-9_-]{0,31}", value)) + + +def classify_lost_and_found_row( + nfield: int, + cells: tuple[Any, ...], +) -> Optional[str]: + """Classify one lost_and_found record by field count + sentinel values. + + Returns 'sessions', 'messages', 'session_model_usage', or None. + """ + + if len(cells) >= 3 and cells[0] is None: + # Rowid-alias tables store their INTEGER PRIMARY KEY as NULL in the + # record; messages is the only canonical table shaped like that with + # a session id second and a role third. + if ( + isinstance(cells[1], str) + and cells[1] + and isinstance(cells[2], str) + and cells[2] in MESSAGE_ROLES + ): + return "messages" + return None + + if not _is_session_id(cells[0] if cells else None): + return None + + if nfield == SESSION_MODEL_USAGE_NFIELD: + # 18 fields, session id first, model string second. + if len(cells) > 1 and isinstance(cells[1], str) and cells[1]: + return "session_model_usage" + return None + + if nfield in SESSIONS_LAYOUT_NFIELDS or nfield == SESSIONS_LEGACY_MINIMAL_NFIELD: + if len(cells) > 1 and _looks_like_source(cells[1]): + return "sessions" + return None + + # Unknown historical sessions layout: session-id first cell plus a + # recognizable source string is still strong enough for a prefix map. + if nfield >= 30 and len(cells) > 1 and _looks_like_source(cells[1]): + return "sessions" + return None + + +def _heuristic_started_at(cells: tuple[Any, ...]) -> float: + for value in cells: + if isinstance(value, (int, float)) and _EPOCH_LOW <= float(value) <= _EPOCH_HIGH: + return float(value) + return 0.0 + + +def _insert_prefix_row( + dest: sqlite3.Connection, + table: str, + dest_columns: list[str], + values: list[Any], + notnull_substitutes: Optional[dict[int, Any]] = None, +) -> bool: + if notnull_substitutes: + values = [ + notnull_substitutes[index] + if value is None and index in notnull_substitutes + else value + for index, value in enumerate(values) + ] + columns = dest_columns[: len(values)] + quoted = ", ".join(f'"{column}"' for column in columns) + placeholders = ", ".join("?" for _ in columns) + cursor = dest.execute( + f'INSERT OR IGNORE INTO "{table}" ({quoted}) VALUES ({placeholders})', + values, + ) + return cursor.rowcount == 1 + + +def _copy_direct_tables( + lf_conn: sqlite3.Connection, + dest: sqlite3.Connection, +) -> dict[str, int]: + """Copy rows .recover managed to attribute to real canonical tables.""" + + copied: dict[str, int] = {} + for table in ( + "system_prompts", + "sessions", + "messages", + "session_model_usage", + "compression_locks", + "gateway_routing", + "async_delegations", + ): + source_columns = _table_columns(lf_conn, table) + if not source_columns: + continue + dest_columns = _table_columns(dest, table) + columns = [c for c in dest_columns if c in source_columns] + if not columns: + continue + quoted = ", ".join(f'"{c}"' for c in columns) + placeholders = ", ".join("?" for _ in columns) + rows = lf_conn.execute(f'SELECT {quoted} FROM "{table}"').fetchall() + if not rows: + copied[table] = 0 + continue + before = int(dest.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0]) + dest.executemany( + f'INSERT OR IGNORE INTO "{table}" ({quoted}) VALUES ({placeholders})', + rows, + ) + after = int(dest.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0]) + copied[table] = after - before + return copied + + +def map_lost_and_found_rows( + lf_conn: sqlite3.Connection, + dest: sqlite3.Connection, +) -> dict[str, Any]: + """Best-effort mapping of a .recover output DB into a fresh SessionDB. + + Handles both rows .recover attributed to real tables and unattributed + ``lost_and_found`` rows classified by field count + sentinel columns. + """ + + report: dict[str, Any] = { + "direct_table_rows": {}, + "mapped": {"sessions": 0, "messages": 0, "session_model_usage": 0}, + "legacy_minimal_sessions": 0, + "unmapped_rows": 0, + "insert_conflicts": 0, + "lost_and_found_tables": [], + } + + dest.execute("BEGIN IMMEDIATE") + try: + report["direct_table_rows"] = _copy_direct_tables(lf_conn, dest) + + sessions_columns = _table_columns(dest, "sessions") + messages_columns = _table_columns(dest, "messages") + usage_columns = _table_columns(dest, "session_model_usage") + sessions_defaults = _notnull_defaults(dest, "sessions") + messages_defaults = _notnull_defaults(dest, "messages") + usage_defaults = _notnull_defaults(dest, "session_model_usage") + # Never fabricate identity fields: a row whose session id / role / + # source cell is genuinely NULL was already rejected by + # classify_lost_and_found_row, so these substitutions only fill + # NOT NULL bookkeeping counters and flag columns. + for defaults, protected in ( + (sessions_defaults, (0, 1)), + (messages_defaults, (1, 2)), + (usage_defaults, (0, 1)), + ): + for index in protected: + defaults.pop(index, None) + + lf_tables = [ + str(row[0]) + for row in lf_conn.execute( + "SELECT name FROM sqlite_master " + "WHERE type='table' AND name LIKE 'lost_and_found%'" + ) + ] + report["lost_and_found_tables"] = lf_tables + + for lf_table in lf_tables: + lf_columns = _table_columns(lf_conn, lf_table) + if lf_columns[:3] != ["rootpgno", "pgno", "nfield"]: + continue + for row in lf_conn.execute(f'SELECT * FROM "{lf_table}"'): + try: + nfield = int(row[2]) if row[2] is not None else 0 + except (TypeError, ValueError): + report["unmapped_rows"] += 1 + continue + lf_rowid = row[3] + cells = tuple(row[4 : 4 + max(nfield, 0)]) + kind = classify_lost_and_found_row(nfield, cells) + if kind is None: + report["unmapped_rows"] += 1 + continue + try: + if kind == "messages": + values = [lf_rowid, *cells[1 : min(nfield, len(messages_columns))]] + inserted = _insert_prefix_row( + dest, "messages", messages_columns, values, + messages_defaults, + ) + elif kind == "session_model_usage": + values = list(cells[: len(usage_columns)]) + inserted = _insert_prefix_row( + dest, "session_model_usage", usage_columns, values, + usage_defaults, + ) + elif nfield == SESSIONS_LEGACY_MINIMAL_NFIELD: + # A pre-modern layout whose column order is unknown: + # salvage identity + timing rather than guessing 14 + # positional meanings. + inserted = bool( + dest.execute( + "INSERT OR IGNORE INTO sessions " + "(id, source, started_at, title) " + "VALUES (?, ?, ?, ?)", + ( + cells[0], + cells[1] if _looks_like_source(cells[1]) + else "recovered", + _heuristic_started_at(cells), + "[best-effort recovered] legacy session " + "row (layout unknown)", + ), + ).rowcount + == 1 + ) + if inserted: + report["legacy_minimal_sessions"] += 1 + else: + values = list(cells[: min(nfield, len(sessions_columns))]) + inserted = _insert_prefix_row( + dest, "sessions", sessions_columns, values, + sessions_defaults, + ) + except sqlite3.DatabaseError: + report["unmapped_rows"] += 1 + continue + if inserted: + report["mapped"][ + "sessions" if kind == "sessions" else kind + ] += 1 + else: + report["insert_conflicts"] += 1 + dest.execute("COMMIT") + except BaseException: + dest.execute("ROLLBACK") + raise + return report + + +def stub_missing_parent_sessions(dest: sqlite3.Connection) -> dict[str, Any]: + """Fabricate placeholder parents for salvaged child rows. + + Salvaged children (messages, model-usage rows) are NEVER deleted for + foreign-key cleanup — a fabricated parent is cheaper than losing the only + surviving copy of the user's data. Stubs are clearly marked. + """ + + result: dict[str, Any] = { + "sessions_stubbed": 0, + "messages_retained": 0, + "usage_rows_retained": 0, + } + dest.execute("BEGIN IMMEDIATE") + try: + orphan_ids: dict[str, dict[str, Any]] = {} + for session_id, first_ts, count in dest.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" + ): + orphan_ids[str(session_id)] = { + "started_at": float(first_ts) if first_ts is not None else 0.0, + "message_count": int(count), + } + for (session_id,) in dest.execute( + "SELECT DISTINCT u.session_id FROM session_model_usage AS u " + "WHERE u.session_id IS NOT NULL AND NOT EXISTS " + "(SELECT 1 FROM sessions WHERE sessions.id = u.session_id)" + ): + orphan_ids.setdefault( + str(session_id), {"started_at": 0.0, "message_count": 0} + ) + + sequence = 1 + for session_id, info in sorted(orphan_ids.items()): + while True: + title = ( + f"[best-effort recovered {sequence}] session metadata " + "was unreadable" + ) + sequence += 1 + if ( + dest.execute( + "SELECT 1 FROM sessions WHERE title = ? LIMIT 1", + (title,), + ).fetchone() + is None + ): + break + dest.execute( + "INSERT INTO sessions (id, source, started_at, title, " + "message_count) VALUES (?, 'recovered', ?, ?, ?)", + ( + session_id, + info["started_at"], + title, + info["message_count"], + ), + ) + result["sessions_stubbed"] += 1 + result["messages_retained"] += info["message_count"] + + result["usage_rows_retained"] = int( + dest.execute("SELECT COUNT(*) FROM session_model_usage").fetchone()[0] + ) + + # Repair dangling intra-sessions references without deleting rows. + dest.execute( + "UPDATE sessions SET parent_session_id = NULL " + "WHERE parent_session_id IS NOT NULL AND NOT EXISTS " + "(SELECT 1 FROM sessions AS p WHERE p.id = sessions.parent_session_id)" + ) + dest.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)" + ) + dest.execute("COMMIT") + except BaseException: + dest.execute("ROLLBACK") + raise + return result + + +def rebuild_fts_indexes(dest: sqlite3.Connection) -> dict[str, str]: + """Rebuild derived FTS indexes from the salvaged canonical rows.""" + + results: dict[str, str] = {} + for table in ("messages_fts", "messages_fts_trigram", "messages_fts_cjk"): + if not _table_columns(dest, table): + continue + try: + dest.execute(f'INSERT INTO "{table}" ("{table}") VALUES (\'rebuild\')') + results[table] = "rebuilt" + except sqlite3.DatabaseError as exc: + results[table] = f"rebuild failed: {exc}" + return results diff --git a/hermes_cli/session_recovery.py b/hermes_cli/session_recovery.py index 39787c1bfb8da..6d5f5e8bf812d 100644 --- a/hermes_cli/session_recovery.py +++ b/hermes_cli/session_recovery.py @@ -334,7 +334,15 @@ def _snapshot_and_inspect( if before != after: raise SessionRecoverySafetyError( "The source database bundle changed while it was being copied. " - "Stop every Hermes process using this profile and retry." + "Stop every Hermes process using this profile and retry. " + "This includes the interactive `hermes` CLI session this " + "command may have been launched from: a running parent CLI " + "writes session bookkeeping (compression ticks, context " + "tracking) to state.db in the background and counts as a " + "Hermes process even after the gateway is stopped. Run the " + "recovery from a fresh shell with no `hermes` session open, " + "or point --source at an immutable snapshot copy of the " + "database." ) conn = sqlite3.connect( @@ -505,6 +513,68 @@ def _salvage_rowid_bounds( return result +def _probe_populated_edge( + source: sqlite3.Connection, + table: str, + *, + edge: str, + anchor: int, +) -> dict[str, Any]: + """Find a finite bound for a damaged rowid edge (issue #80205). + + When an ordered edge probe fails, :func:`_salvage_rowid_bounds` used to + substitute the whole SQLite rowid domain. Range bisection then burned the + entire ``_MAX_SALVAGE_RANGE_QUERIES`` budget subdividing an enormous + synthetic tail that could not contain real rows — and once the budget was + gone, rows that were still readable were silently recorded as skipped. + + This gallops outward from the readable ``anchor`` edge with exponentially + growing offsets. A probe that cleanly reports "no rows beyond X" caps the + domain at X; a probe that errors (its b-tree path crosses the damage) or + finds a row keeps growing. At most ~64 probes per edge, so the cap costs + a bounded, tiny slice of the salvage budget instead of all of it. + """ + + ascending = edge == "high" + comparison = ">" if ascending else "<" + probe_sql = ( + f'SELECT rowid FROM "{table}" WHERE rowid {comparison} ? ' + f'ORDER BY rowid {"ASC" if ascending else "DESC"} LIMIT 1' + ) + domain_limit = _MAX_SQLITE_ROWID if ascending else _MIN_SQLITE_ROWID + result: dict[str, Any] = {"edge": edge, "probes": 0, "capped": False} + + position = anchor + span = 1 + while True: + candidate = position + span if ascending else position - span + if (ascending and candidate >= domain_limit) or ( + not ascending and candidate <= domain_limit + ): + # No clean empty-tail answer before the domain edge; keep the + # domain fallback rather than inventing a bound. + result["bound"] = domain_limit + return result + result["probes"] += 1 + try: + row = source.execute(probe_sql, (candidate,)).fetchone() + except sqlite3.DatabaseError: + # Damage on the probe path — inconclusive, widen further. + span *= 2 + continue + if row is None: + # Clean answer: nothing beyond ``candidate``. The salvageable + # data ends at or before it, so the synthetic domain tail is + # provably empty and need not be bisected at all. + result["bound"] = candidate + result["capped"] = True + return result + # Rows exist beyond the candidate; advance the anchor. The span keeps + # doubling (never resets) so the whole gallop stays O(log range). + position = int(row[0]) + span *= 2 + + def _copy_table_salvage( source: sqlite3.Connection, destination: sqlite3.Connection, @@ -530,6 +600,7 @@ def _copy_table_salvage( "excluded_rows": 0, "columns": columns, "range_queries": 0, + "exact_lookup_recovered": 0, "skipped_rowid_ranges": [], } if not source_columns: @@ -553,6 +624,28 @@ def _copy_table_salvage( result["error"] += f": {details}" return result + # Issue #80205: a damaged ordered edge probe used to substitute the whole + # SQLite rowid domain, and bisecting that synthetic tail exhausted the + # range-query budget while readable tail rows were still waiting to be + # copied. Gallop outward from the surviving edge for a finite bound first. + fallback_edges = bounds.get("fallback_edges") or [] + if fallback_edges: + bounds["edge_probes"] = [] + if "high" in fallback_edges and bounds.get("low") is not None: + probe = _probe_populated_edge( + source, table, edge="high", anchor=int(bounds["low"]) + ) + bounds["edge_probes"].append(probe) + if probe["capped"]: + bounds["high"] = int(probe["bound"]) + if "low" in fallback_edges and bounds.get("high") is not None: + probe = _probe_populated_edge( + source, table, edge="low", anchor=int(bounds["high"]) + ) + bounds["edge_probes"].append(probe) + if probe["capped"]: + bounds["low"] = int(probe["bound"]) + quoted = ", ".join(f'"{column}"' for column in columns) placeholders = ", ".join("?" for _ in columns) select_sql = ( @@ -564,6 +657,41 @@ def _copy_table_salvage( ) column_names = tuple(columns) stopped_at_query_limit = False + exact_sql = f'SELECT {quoted} FROM "{table}" WHERE rowid = ?' + + def recover_exact_rowid(rowid: int) -> bool: + """Issue #80205: salvage one row by exact-key lookup. + + A singleton range scan (``rowid BETWEEN x AND x ORDER BY rowid``) + must advance the cursor past ``x`` to prove the range is exhausted; + when the *next* cell or page is damaged that advance raises AFTER the + row was produced, and the driver discards the already-fetched row. An + equality lookup on the rowid stops at the hit, so the boundary row + directly before a damaged page — readable in the field case and by + SQLite's page-level ``.recover`` — is recovered instead of being + recorded as skipped. + """ + result["range_queries"] += 1 + try: + row = source.execute(exact_sql, (rowid,)).fetchone() + except sqlite3.DatabaseError: + return False + if row is None: + return True # genuinely absent: nothing to skip + value = tuple(row) + if row_filter is not None and not row_filter(value, column_names): + result["excluded_rows"] += 1 + return True + destination.execute("BEGIN IMMEDIATE") + try: + destination.execute(insert_sql, value) + destination.execute("COMMIT") + except BaseException: + destination.execute("ROLLBACK") + raise + result["copied_rows"] += 1 + result["exact_lookup_recovered"] += 1 + return True def copy_range(low: int, high: int) -> None: nonlocal stopped_at_query_limit @@ -626,12 +754,13 @@ def _copy_table_salvage( if retry_low > high: return if retry_low == high: - _append_skipped_range( - result["skipped_rowid_ranges"], - retry_low, - high, - str(exc), - ) + if not recover_exact_rowid(retry_low): + _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) @@ -1255,6 +1384,152 @@ def _finalize_derived_metadata(destination: sqlite3.Connection) -> dict[str, Any return result +def _recover_via_lost_and_found( + *, + source: Path, + snapshot_source: Path, + snapshot_dir: Path, + output: Path, + inspection: dict[str, Any], + disk_space: dict[str, Any], + missing_required: list[str], +) -> dict[str, Any]: + """Best-effort page-level salvage when table schemas are unreadable. + + Shells out to the sqlite3 CLI's ``.recover`` (a shell-only feature, not + part of Python's ``sqlite3`` module) to rebuild rows into a scratch + lost_and_found database, then heuristically maps them into a fresh + current-schema database. The result is explicitly labeled best-effort. + """ + + from hermes_cli.session_lost_and_found import ( + SQLITE3_CLI_GUIDANCE, + LostAndFoundError, + find_sqlite3_cli, + map_lost_and_found_rows, + rebuild_fts_indexes, + run_cli_lost_and_found_recover, + stub_missing_parent_sessions, + ) + + sqlite3_bin = find_sqlite3_cli() + if sqlite3_bin is None: + raise SessionRecoverySourceError( + "Partial recovery still requires readable table schemas for: " + + ", ".join(missing_required) + + ". " + SQLITE3_CLI_GUIDANCE + ) + + lf_path = snapshot_dir / "lost_and_found.db" + try: + cli_report = run_cli_lost_and_found_recover( + snapshot_source, lf_path, sqlite3_bin + ) + except (LostAndFoundError, OSError) as exc: + raise SessionRecoverySourceError( + "Partial recovery could not read the table schemas for: " + + ", ".join(missing_required) + + f", and page-level .recover salvage failed: {exc}" + ) from exc + + destination_db = SessionDB(db_path=output) + destination_db.close() + + lf_conn = sqlite3.connect(str(lf_path), isolation_level=None) + destination_conn = sqlite3.connect( + str(output), isolation_level=None, timeout=1.0 + ) + try: + destination_conn.execute("PRAGMA foreign_keys=OFF") + mapping = map_lost_and_found_rows(lf_conn, destination_conn) + stubbing = stub_missing_parent_sessions(destination_conn) + fts = rebuild_fts_indexes(destination_conn) + derived_metadata = _finalize_derived_metadata(destination_conn) + finally: + lf_conn.close() + destination_conn.close() + + copy_report: dict[str, dict[str, Any]] = { + table: { + "mode": "lost_and_found_salvage", + "status": "partial", + "copied_rows": ( + int(mapping["direct_table_rows"].get(table) or 0) + + int(mapping["mapped"].get(table) or 0) + ), + "error": "recovered via page-level lost_and_found salvage; " + "row completeness cannot be verified against the source", + } + for table in ("sessions", "messages", "session_model_usage") + } + + verification = _verify_recovered_database( + output, + expected_counts={"sessions": None, "messages": None}, + copy_report=copy_report, + allow_partial=True, + orphan_cleanup={ + "sessions_reconstructed": stubbing["sessions_stubbed"], + "messages_retained": stubbing["messages_retained"], + "messages_removed": 0, + "total_removed_or_relinked": 0, + }, + ) + verification["loss_detected"] = True + verification["warnings"].append( + "BEST-EFFORT page-level salvage: the source table schemas were " + "unreadable, so rows were rebuilt from raw pages and mapped " + "heuristically. Review every count before trusting this output." + ) + verification["complete"] = False + + source_unchanged = ( + _source_fingerprint(source) == inspection["source_fingerprint"] + ) + if not source_unchanged: + verification["errors"].append( + "the source database bundle changed during recovery" + ) + verification["healthy"] = False + + return { + "operation": "recover", + "allow_partial": True, + "mode": "lost_and_found_salvage", + "best_effort": True, + "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"], + }, + "unreadable_schemas": missing_required, + "sqlite3_cli": cli_report, + "lost_and_found": mapping, + "session_stubs": stubbing, + "fts_rebuild": fts, + "copy": copy_report, + "orphan_cleanup": { + "sessions_reconstructed": stubbing["sessions_stubbed"], + "messages_retained": stubbing["messages_retained"], + "messages_removed": 0, + "total_removed_or_relinked": 0, + }, + "derived_metadata": derived_metadata, + "verification": verification, + "complete": False, + "partial": True, + "verified": bool(verification.get("healthy") and source_unchanged), + "installed": False, + } + + def recover_session_database( source_path: Path, output_path: Path, @@ -1286,7 +1561,9 @@ def recover_session_database( 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}" + f"Required canonical tables are not readable: {reasons}. " + "Re-run with --allow-partial to salvage every readable row " + "into a new database (the source is never modified)." ) if allow_partial: missing_required = [ @@ -1295,9 +1572,17 @@ def recover_session_database( if not inspection["tables"][table].get("available") ] if missing_required: - raise SessionRecoverySourceError( - "Partial recovery still requires readable table schemas for: " - + ", ".join(missing_required) + # SQL-level salvage is impossible without readable table + # schemas. Fall back to page-level lost_and_found salvage via + # the sqlite3 CLI's .recover (a shell-only feature). + return _recover_via_lost_and_found( + source=source, + snapshot_source=snapshot_source, + snapshot_dir=Path(temp_dir.name), + output=output, + inspection=inspection, + disk_space=disk_space, + missing_required=missing_required, ) source_conn = sqlite3.connect( diff --git a/hermes_cli/sessions_cmd.py b/hermes_cli/sessions_cmd.py index 095ad5f9e7e2a..b0afd01e01567 100644 --- a/hermes_cli/sessions_cmd.py +++ b/hermes_cli/sessions_cmd.py @@ -208,7 +208,15 @@ def cmd_sessions(args, sessions_parser=None): return 0 if allow_partial and report.get("verified"): counts = report.get("verification", {}).get("table_counts", {}) - print(f"✓ Partial recovery output verified at: {output}") + if report.get("best_effort"): + print(f"✓ BEST-EFFORT page-level salvage verified at: {output}") + print( + " The source table schemas were unreadable; rows were " + "rebuilt from raw pages via sqlite3 .recover and mapped " + "heuristically." + ) + else: + print(f"✓ Partial recovery output verified at: {output}") print( " Recovered " f"{int(counts.get('sessions') or 0):,} sessions and " diff --git a/tests/hermes_cli/test_session_recovery_lost_and_found.py b/tests/hermes_cli/test_session_recovery_lost_and_found.py new file mode 100644 index 0000000000000..74bcbf3fe83ab --- /dev/null +++ b/tests/hermes_cli/test_session_recovery_lost_and_found.py @@ -0,0 +1,563 @@ +"""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.session_lost_and_found import ( + 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, +) + + +HAVE_SQLITE3_CLI = shutil.which("sqlite3") 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 and fell back. + assert any("high rowid" in error for error in bounds["errors"]), bounds + assert "high" in bounds["fallback_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() + + +# ── 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() + assert len(sessions_columns) == 54 + assert len(usage_columns) == 18 + + max_fields = 54 + 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 54-column layout and historical 52-column layout. + insert(54, 1, session_row("20260101_010101_aaa001", 54)) + 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(54, 301, ["not-a-session-id", "cli"] + [None] * 52) + 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( + 54, ("20260101_010101_aaa001", "cli") + (None,) * 52 + ) + == "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(54, ("not-a-session-id", "cli") + (None,) * 52) + 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