RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.
This commit is contained in:
Teknium 2026-08-12 16:39:00 -07:00
parent 7d0b5a332c
commit 6dad74596e
4 changed files with 1411 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

543
hermes_cli/session_lost_and_found.py Normal file
View File

@ -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 <source> .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

307
hermes_cli/session_recovery.py
View File

@ -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(

10
hermes_cli/sessions_cmd.py
View File

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

563
tests/hermes_cli/test_session_recovery_lost_and_found.py Normal file
View File

@ -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