From f0b339e1960f1449fbf5c017237a3e440f08e00e Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:29:40 -0700 Subject: [PATCH] refactor(hermes_cli): compact session_recovery docstrings/comments and result literals --- hermes_cli/session_recovery.py | 340 +++++++++------------------------ 1 file changed, 94 insertions(+), 246 deletions(-) diff --git a/hermes_cli/session_recovery.py b/hermes_cli/session_recovery.py index e67b1174ad..f74d072df6 100644 --- a/hermes_cli/session_recovery.py +++ b/hermes_cli/session_recovery.py @@ -1,10 +1,8 @@ """Offline, non-destructive recovery for a damaged Hermes session database. -* the supplied source database is never opened by SQLite; * the source file and any -WAL/SHM/rollback-journal sidecars are copied into a disposable working directory first; * canonical -rows are copied into a newly initialized current-schema database; * derived FTS tables and migration -bookkeeping are rebuilt, not copied; and * the recovered database is never installed over the active -database. +The source is never opened by SQLite: it and its WAL/SHM/journal sidecars are copied to a disposable +work dir first. Canonical rows are copied into a fresh current-schema database; derived FTS tables and +migration bookkeeping are rebuilt, not copied; the result is never installed over the active database. """ from __future__ import annotations @@ -24,12 +22,7 @@ from hermes_state import (FTS_STORAGE_VERSION, SCHEMA_VERSION, SessionDB, _db_op ProgressCallback = Callable[[dict[str, Any]], None] _CANONICAL_TABLES = ( - "system_prompts", - "sessions", - "messages", - "session_model_usage", - "compression_locks", - "gateway_routing", + "system_prompts", "sessions", "messages", "session_model_usage", "compression_locks", "gateway_routing", "async_delegations", ) @@ -42,13 +35,9 @@ def _init_delivery_ledger_schema(conn: sqlite3.Connection) -> None: _initialize_schema(conn) -# Tables that live in state.db but are created lazily by a gateway module on -# first use, so base ``SessionDB`` never creates them on a fresh destination. -# Every entry maps the table to the initializer that owns its DDL; recovery -# creates the table on the destination before copying, so owed rows survive -# instead of silently vanishing from a "complete" salvage (#100313, #86236). -# Add new lazily-created state.db tables HERE, never as one-off ``if table ==`` -# branches. +# state.db tables created lazily by a gateway module (base ``SessionDB`` never creates them on a fresh +# destination) -> the initializer owning their DDL. Recovery creates them before copying so owed rows +# don't silently vanish from a "complete" salvage. Register new lazy tables HERE, not as ``if table ==``. _AUXILIARY_TABLE_SCHEMAS: dict[str, Callable[[sqlite3.Connection], None]] = { "delivery_obligations": _init_delivery_ledger_schema, } @@ -57,17 +46,10 @@ _AUXILIARY_TABLES = tuple(_AUXILIARY_TABLE_SCHEMAS) _INVENTORY_TABLES = (*_CANONICAL_TABLES, "state_meta", *_TOPIC_TABLES, *_AUXILIARY_TABLES) -# These values describe derived indexes or the schema that owns an optional -# table. A fresh destination must generate them from its own current schema. +# Derived-index / optional-schema markers: a fresh destination regenerates these, never copies them. _GENERATED_META_KEYS = frozenset({ - "fts_storage_version", - "fts_optimize_available", - "fts_rebuild_high_water", - "fts_rebuild_progress", - "fts_cjk_stale", - "fts_cjk_rebuild_high_water", - "fts_cjk_rebuild_progress", - "telegram_dm_topic_schema_version", + "fts_storage_version", "fts_optimize_available", "fts_rebuild_high_water", "fts_rebuild_progress", + "fts_cjk_stale", "fts_cjk_rebuild_high_water", "fts_cjk_rebuild_progress", "telegram_dm_topic_schema_version", }) _SIDECAR_SUFFIXES = ("", "-wal", "-shm", "-journal") @@ -95,9 +77,7 @@ def _sidecar_path(db_path: Path, suffix: str) -> Path: def _resolved_output_path(path: Path) -> Path: """Resolve a not-yet-created output path without requiring it to exist.""" - - parent = path.expanduser().parent.resolve(strict=True) - return parent / path.name + return path.expanduser().parent.resolve(strict=True) / path.name def _validate_paths( @@ -157,33 +137,26 @@ def _format_bytes(value: int) -> str: def _same_filesystem(left: Path, right: Path) -> bool: try: return os.stat(left).st_dev == os.stat(right).st_dev - except OSError: - # Existing directories were already required by _validate_paths. This - # fallback is defensive for platforms with incomplete st_dev support. + except OSError: # both dirs exist (_validate_paths); fallback for incomplete st_dev support return left.anchor.casefold() == right.anchor.casefold() def _disk_space_preflight(source: Path, work_root: Path, output_parent: Optional[Path]) -> dict[str, Any]: """Require space for the disposable bundle, output, and safety headroom.""" - bundle_bytes = sum( _sidecar_path(source, suffix).stat().st_size for suffix in _SIDECAR_SUFFIXES if _sidecar_path(source, suffix).exists() ) - # The v23 external-content rebuild is normally substantially smaller than - # a legacy database, but using the complete source bundle as the estimate - # avoids betting the user's disk on that expectation. + # The v23 external-content rebuild is usually much smaller than a legacy database, but estimating + # with the complete bundle avoids betting the user's disk on that. output_allowance = bundle_bytes if output_parent is not None else 0 headroom = max(_MINIMUM_SPACE_HEADROOM, int((bundle_bytes + output_allowance) * 0.05)) work_free = int(shutil.disk_usage(work_root).free) report: dict[str, Any] = { - "source_bundle_bytes": bundle_bytes, - "estimated_output_bytes": output_allowance, - "headroom_bytes": headroom, - "work_dir": str(work_root), - "work_dir_free_bytes": work_free, + "source_bundle_bytes": bundle_bytes, "estimated_output_bytes": output_allowance, "headroom_bytes": headroom, + "work_dir": str(work_root), "work_dir_free_bytes": work_free, } if output_parent is None or _same_filesystem(work_root, output_parent): @@ -206,23 +179,16 @@ def _disk_space_preflight(source: Path, work_root: Path, output_parent: Optional work_required = bundle_bytes + headroom output_required = output_allowance + headroom report.update({ - "shared_filesystem": False, - "work_dir_required_bytes": work_required, - "output_dir": str(output_parent), - "output_dir_free_bytes": output_free, - "output_dir_required_bytes": output_required, + "shared_filesystem": False, "work_dir_required_bytes": work_required, "output_dir": str(output_parent), + "output_dir_free_bytes": output_free, "output_dir_required_bytes": output_required, }) - shortages: list[str] = [] - if work_free < work_required: - shortages.append( - f"{work_root}: {_format_bytes(work_free)} available, " - f"{_format_bytes(work_required)} required" - ) - if output_free < output_required: - shortages.append( - f"{output_parent}: {_format_bytes(output_free)} available, " - f"{_format_bytes(output_required)} required" + shortages = [ + f"{where}: {_format_bytes(free)} available, {_format_bytes(required)} required" + for where, free, required in ( + (work_root, work_free, work_required), (output_parent, output_free, output_required), ) + if free < required + ] if shortages: raise SessionRecoverySafetyError( "Not enough free disk space for safe recovery: " @@ -235,12 +201,9 @@ def _disk_space_preflight(source: Path, work_root: Path, output_parent: Optional def _copy_source_bundle(source: Path, snapshot_dir: Path) -> tuple[Path, list[str]]: """Copy the source DB bundle aside so SQLite never opens the original. - The whole copy runs inside ``offline_file_access``, which holds the connection-lifecycle lock - for its duration. - - Recovery normally runs as its own short-lived CLI process against an offline/quarantined file, - so the refusal should never fire; the guard keeps this path consistent with - ``hermes_state._backup_db_file``. + The whole copy runs inside ``offline_file_access`` (holds the connection-lifecycle lock). Recovery + normally runs as its own CLI process against an offline file, so the refusal should never fire; the + guard keeps this path consistent with ``hermes_state._backup_db_file``. """ from hermes_cli.sqlite_safe_read import LiveConnectionError, offline_file_access @@ -250,11 +213,10 @@ def _copy_source_bundle(source: Path, snapshot_dir: Path) -> tuple[Path, list[st with offline_file_access(source, what="snapshot"): for suffix in _SIDECAR_SUFFIXES: source_part = _sidecar_path(source, suffix) - if not source_part.exists(): - continue - destination_part = _sidecar_path(snapshot_source, suffix) - shutil.copy2(source_part, destination_part) - copied.append(destination_part.name) + if source_part.exists(): + destination_part = _sidecar_path(snapshot_source, suffix) + shutil.copy2(source_part, destination_part) + copied.append(destination_part.name) except LiveConnectionError as exc: raise SessionRecoverySafetyError(str(exc)) from exc return snapshot_source, copied @@ -340,9 +302,10 @@ def _copy_rows( result["error"] = str(exc) return result - complete = expected_rows is None or result["copied_rows"] == expected_rows - result["status"] = "complete" if complete else "partial" - if result["status"] == "partial": + if expected_rows is None or result["copied_rows"] == expected_rows: + result["status"] = "complete" + else: + result["status"] = "partial" result["error"] = f"copied {result['copied_rows']} of {expected_rows} readable rows" return result @@ -373,8 +336,7 @@ def _inspect_connection(conn: sqlite3.Connection) -> dict[str, Any]: report["journal_mode"] = _journal_mode(conn) except sqlite3.DatabaseError as exc: report["journal_mode"] = None - # Journal metadata is useful context but not canonical session data. - # A damaged journal pragma must not block rows that are still readable. + # Journal metadata is context, not canonical data: a damaged pragma must not block readable rows. report["warnings"].append(f"journal mode: {exc}") for table in _INVENTORY_TABLES: @@ -394,11 +356,9 @@ def _snapshot_and_inspect( ) -> tuple[tempfile.TemporaryDirectory[str], Path, dict[str, Any]]: before = _source_fingerprint(source) temp_dir = tempfile.TemporaryDirectory(prefix="hermes-session-recovery-", dir=str(work_root)) - snapshot_dir = Path(temp_dir.name) try: - snapshot_source, copied = _copy_source_bundle(source, snapshot_dir) - after = _source_fingerprint(source) - if before != after: + snapshot_source, copied = _copy_source_bundle(source, Path(temp_dir.name)) + if _source_fingerprint(source) != before: raise SessionRecoverySafetyError( "The source database bundle changed while it was being copied. " "Stop every Hermes process using this profile and retry. " @@ -427,7 +387,6 @@ def _snapshot_and_inspect( def inspect_session_database(source_path: Path, *, work_dir: Optional[Path] = None) -> dict[str, Any]: """Inspect canonical table readability without opening the source itself.""" - source, _, work_root = _validate_paths(source_path, work_dir=work_dir) disk_space = _disk_space_preflight(source, work_root, None) temp_dir, _, inspection = _snapshot_and_inspect(source, work_root) @@ -444,13 +403,8 @@ def inspect_session_database(source_path: Path, *, work_dir: Optional[Path] = No def _ensure_auxiliary_destination_schema(destination: sqlite3.Connection, table: str) -> None: - """Create a lazy auxiliary table on the recovered destination. - - Recovery initializes the destination via base ``SessionDB``, which does not create gateway-owned - tables; copying into a missing table would report ``missing``/``no compatible columns`` and drop - the rows. - """ - + """Create a lazy gateway-owned table on the destination; without it the copy would report + ``missing``/``no compatible columns`` and drop the rows.""" initialize = _AUXILIARY_TABLE_SCHEMAS.get(table) if initialize is None: raise SessionRecoverySafetyError(f"no destination schema initializer registered for table {table!r}") @@ -506,8 +460,7 @@ def _copy_table( def _append_skipped_range(ranges: list[dict[str, Any]], low: int, high: int, error: str) -> None: - """Record skipped rowid ranges without producing one entry per row.""" - + """Record skipped rowid ranges, merging adjacent same-error ranges (not one entry per row).""" if ranges and ranges[-1]["high"] + 1 == low and ranges[-1]["error"] == error: ranges[-1]["high"] = high return @@ -516,11 +469,9 @@ def _append_skipped_range(ranges: list[dict[str, Any]], low: int, high: int, err def _salvage_rowid_bounds(source: sqlite3.Connection, table: str) -> dict[str, Any]: """Find the readable rowid edges without scanning the complete table.""" - result: dict[str, Any] = {"errors": [], "fallback_edges": []} rows: dict[str, Optional[int]] = {"low": None, "high": None} - directions = (("low", "ASC"), ("high", "DESC")) - for edge, direction in directions: + for edge, direction in (("low", "ASC"), ("high", "DESC")): try: row = source.execute(f'SELECT rowid FROM "{table}" ORDER BY rowid {direction} LIMIT 1').fetchone() if row is not None: @@ -532,10 +483,8 @@ def _salvage_rowid_bounds(source: sqlite3.Connection, table: str) -> dict[str, A result["empty" if not result["errors"] else "unavailable"] = True return result - # A damaged edge can prevent one of the ordered probes from completing. - # Keep the other readable edge and bound the missing side by SQLite's - # rowid domain. Range bisection can then approach the surviving data - # without assuming that user-created databases contain only positive IDs. + # A damaged edge can stop one ordered probe. Keep the readable edge and bound the other side by the + # SQLite rowid domain, so bisection never assumes user databases hold only positive ids. if rows["low"] is None: rows["low"] = _MIN_SQLITE_ROWID result["fallback_edges"].append("low") @@ -544,10 +493,8 @@ def _salvage_rowid_bounds(source: sqlite3.Connection, table: str) -> dict[str, A result["fallback_edges"].append("high") result.update(rows) - # 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. + # Bisecting the whole synthetic domain tail used to exhaust the range-query budget before readable + # tail rows were copied (#80205); gallop outward from the surviving edge for a finite bound first. if result["fallback_edges"]: result["edge_probes"] = [] for edge, anchor_edge in (("high", "low"), ("low", "high")): @@ -560,14 +507,9 @@ def _salvage_rowid_bounds(source: sqlite3.Connection, table: str) -> dict[str, A 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). - - Substituting the whole rowid domain for a failed edge probe made range bisection burn the entire - salvage budget on an empty synthetic tail, after which readable rows were silently skipped. - Instead gallop outward from the readable ``anchor`` with exponential offsets: a clean "no rows - beyond X" caps the domain, an error or hit keeps growing. At most ~64 probes per edge. + """Finite bound for a damaged rowid edge: gallop outward from the readable ``anchor`` with doubling + offsets. A clean "no rows beyond X" caps the domain; an error or a hit keeps growing (~64 probes max). """ - ascending = edge == "high" comparison = ">" if ascending else "<" probe_sql = ( @@ -582,27 +524,19 @@ def _probe_populated_edge(source: sqlite3.Connection, table: str, *, edge: str, 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 + result["bound"] = domain_limit # no clean empty-tail answer: keep the domain fallback return result result["probes"] += 1 try: row = source.execute(probe_sql, (candidate,)).fetchone() - except sqlite3.DatabaseError: - # Damage on the probe path — inconclusive, widen further. + 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. + if row is None: # nothing beyond candidate: the synthetic domain tail is provably empty 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]) + position = int(row[0]) # rows exist beyond; advance. Span never resets -> O(log range) span *= 2 @@ -724,16 +658,9 @@ def _copy_table_salvage( row_filter: Optional[Callable[[tuple[Any, ...], tuple[str, ...]], bool]] = None, ) -> dict[str, Any]: """Best-effort rowid-range copy that continues past damaged source pages.""" - result: dict[str, Any] = { - "mode": "rowid_range_salvage", - "source_rows": source_rows, - "copied_rows": 0, - "excluded_rows": 0, - "columns": [], - "range_queries": 0, - "exact_lookup_recovered": 0, - "skipped_rowid_ranges": [], + "mode": "rowid_range_salvage", "source_rows": source_rows, "copied_rows": 0, "excluded_rows": 0, + "columns": [], "range_queries": 0, "exact_lookup_recovered": 0, "skipped_rowid_ranges": [], } columns = _compatible_columns(source, destination, table, result) if columns is None: @@ -802,16 +729,14 @@ def _state_meta_precheck( ) -> Optional[dict[str, Any]]: """Terminal ``state_meta`` result when the key/value schema is unusable, else ``None``. - Status matters here. In salvage mode an unusable-but-PRESENT table reports ``failed``, not - ``missing``: verification only escalates ``failed``/``partial`` into a warning + - ``loss_detected``, so reporting ``missing`` would silently drop real metadata and still claim - ``complete=True``. + In salvage mode an unusable-but-PRESENT table reports ``failed``, not ``missing``: verification + only escalates ``failed``/``partial`` into a warning + ``loss_detected``, so ``missing`` would + silently drop real metadata and still claim ``complete=True``. """ extra = {"mode": "rowid_range_salvage"} if salvage else {} source_columns = _table_columns(source, "state_meta") if not {"key", "value"}.issubset(source_columns): - if salvage and source_columns: - # Present but unusable: this IS data loss and must be reported. + if salvage and source_columns: # present but unusable: real data loss return _state_meta_result( source_rows, **extra, @@ -821,8 +746,7 @@ def _state_meta_precheck( f"columns (found: {', '.join(source_columns) or 'none'})" ), ) - # Genuinely absent from the source — nothing was lost. - return _state_meta_result(source_rows, **extra, status="missing") + return _state_meta_result(source_rows, **extra, status="missing") # genuinely absent: nothing lost if not {"key", "value"}.issubset(_table_columns(destination, "state_meta")): return _state_meta_result( source_rows, **extra, status="failed", error="destination state_meta schema is incomplete" @@ -889,14 +813,9 @@ def _copy_state_meta( def _reconstruct_missing_sessions(destination: sqlite3.Connection) -> dict[str, Any]: """Recreate placeholder session rows for salvaged orphaned messages. - When the ``sessions`` b-tree is damaged worse than ``messages``, salvage can recover the - conversation text while recovering few or none of the session rows that own it. Deleting those - messages as "orphans" throws away the only readable copy of the user's data — the exact opposite - of what ``--allow-partial`` is for. - - Instead, synthesize a minimal session row per orphaned ``session_id`` (only - ``id``/``source``/``started_at`` are NOT NULL) so the messages stay reachable and foreign keys - hold. ``started_at`` is taken from the earliest surviving message so ordering stays sane. + When ``sessions`` is damaged worse than ``messages``, deleting the recovered messages as orphans + would discard the only readable copy of the user's data. Instead synthesize a minimal session per + orphaned ``session_id`` (``started_at`` = earliest surviving message) so FKs hold. """ result: dict[str, Any] = {"sessions_reconstructed": 0, "messages_retained": 0} if not (_table_columns(destination, "sessions") and _table_columns(destination, "messages")): @@ -948,24 +867,15 @@ _RELINK_COUNTERS = ("session_prompt_refs_cleared", "sessions_parent_cleared") def _cleanup_partial_orphans(destination: sqlite3.Connection) -> dict[str, Any]: - """Reconcile references to sessions that could not be salvaged. - - Messages are never discarded for lack of a session row: their owning session is reconstructed as - a placeholder first (see :func:`_reconstruct_missing_sessions`). Only rows that remain orphaned - after that — and rows in tables carrying no recoverable user content — are removed. - """ - + """Reconcile references to unsalvageable sessions. Messages are never dropped for lack of a + session row (placeholders are rebuilt first); only rows still orphaned after that are removed.""" result: dict[str, Any] = { - "session_prompt_refs_cleared": 0, - "system_prompts_removed": 0, - "sessions_parent_cleared": 0, - "sessions_reconstructed": 0, - "messages_retained": 0, + "session_prompt_refs_cleared": 0, "system_prompts_removed": 0, "sessions_parent_cleared": 0, + "sessions_reconstructed": 0, "messages_retained": 0, **{f"{table}_removed": 0 for table in _DEPENDENT_TABLES}, } with _immediate_transaction(destination): - # Rebuild owners BEFORE any orphan deletion so salvaged conversation - # text is never dropped for want of a session row. + # Rebuild owners BEFORE any orphan deletion. result.update(_reconstruct_missing_sessions(destination)) result["sessions_parent_cleared"] = _reconcile( @@ -997,9 +907,7 @@ def _cleanup_partial_orphans(destination: sqlite3.Connection) -> dict[str, Any]: f'NOT EXISTS (SELECT 1 FROM sessions WHERE sessions.id = "{table}".session_id)', f'DELETE FROM "{table}"', ) - # Only destructive/relinking actions belong in this total. The - # reconstruction counters describe data RETAINED, so summing them here - # would report saving the user's messages as if it were losing them. + # Only destructive/relinking actions count: reconstruction counters describe RETAINED data. result["total_removed_or_relinked"] = sum( int(result[key]) for key in (*_RELINK_COUNTERS, *(f"{table}_removed" for table in _DEPENDENT_TABLES)) ) @@ -1169,7 +1077,6 @@ def _verify_recovered_database( def _finalize_derived_metadata(destination: sqlite3.Connection) -> dict[str, Any]: """Stamp only metadata that the newly created destination actually owns.""" - fts_tables = { str(row[0]) for row in destination.execute( @@ -1205,21 +1112,12 @@ def _recover_via_lost_and_found( 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. - """ - + """Best-effort page-level salvage when table schemas are unreadable: the sqlite3 CLI's ``.recover`` + (shell-only, not in Python's ``sqlite3``) rebuilds rows into a scratch lost_and_found database which + is then heuristically mapped into a fresh current-schema database.""" 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_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() @@ -1255,21 +1153,15 @@ def _recover_via_lost_and_found( 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) - ), + "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") } - orphan_cleanup = { - "sessions_reconstructed": stubbing["sessions_stubbed"], - "messages_retained": stubbing["messages_retained"], - "messages_removed": 0, - "total_removed_or_relinked": 0, + "sessions_reconstructed": stubbing["sessions_stubbed"], "messages_retained": stubbing["messages_retained"], + "messages_removed": 0, "total_removed_or_relinked": 0, } verification = _verify_recovered_database( output, @@ -1287,23 +1179,10 @@ def _recover_via_lost_and_found( verification["complete"] = False return _recovery_report( - source, - output, - inspection, - disk_space, - verification, - on_source_change="healthy", - allow_partial=True, - mode="lost_and_found_salvage", - best_effort=True, - unreadable_schemas=missing_required, - sqlite3_cli=cli_report, - lost_and_found=mapping, - session_stubs=stubbing, - fts_rebuild=fts, - copy=copy_report, - orphan_cleanup=orphan_cleanup, - derived_metadata=derived_metadata, + source, output, inspection, disk_space, verification, on_source_change="healthy", + allow_partial=True, mode="lost_and_found_salvage", best_effort=True, unreadable_schemas=missing_required, + sqlite3_cli=cli_report, lost_and_found=mapping, session_stubs=stubbing, fts_rebuild=fts, copy=copy_report, + orphan_cleanup=orphan_cleanup, derived_metadata=derived_metadata, ) @@ -1317,10 +1196,8 @@ def _recovery_report( on_source_change: str, **fields: Any, ) -> dict[str, Any]: - """The ``recover`` report: shared header, mode-specific ``fields``, then the verdict. - - A source bundle that changed during recovery is a verification error and additionally clears - ``verification[on_source_change]``. + """The ``recover`` report: shared header, mode-specific ``fields``, then the verdict. A source bundle + that changed during recovery is a verification error and also clears ``verification[on_source_change]``. """ source_unchanged = _source_fingerprint(source) == inspection["source_fingerprint"] if not source_unchanged: @@ -1335,10 +1212,8 @@ def _recovery_report( "source_unchanged": source_unchanged, "disk_space": disk_space, "inspection": { - "journal_mode": inspection.get("journal_mode"), - "tables": inspection["tables"], - "errors": inspection["errors"], - "warnings": inspection["warnings"], + "journal_mode": inspection.get("journal_mode"), "tables": inspection["tables"], + "errors": inspection["errors"], "warnings": inspection["warnings"], }, **fields, "verification": verification, @@ -1358,12 +1233,8 @@ def recover_session_database( progress_cb: Optional[ProgressCallback] = None, allow_partial: bool = False, ) -> dict[str, Any]: - """Recover canonical rows into a separate current-schema database. - - The source path and its sidecars are copied before SQLite opens anything. ``output_path`` must - not exist and is never swapped into place. - """ - + """Recover canonical rows into a separate current-schema database. The source and its sidecars are + copied before SQLite opens anything; ``output_path`` must not exist and is never swapped into place.""" if chunk_size <= 0: raise SessionRecoverySafetyError("chunk_size must be greater than zero") @@ -1382,22 +1253,12 @@ def recover_session_database( ) if allow_partial: missing_required = [ - table - for table in ("sessions", "messages") - if not inspection["tables"][table].get("available") + table for table in ("sessions", "messages") if not inspection["tables"][table].get("available") ] - if 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). + if missing_required: # no readable schema -> page-level lost_and_found salvage 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=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 = _connect(snapshot_source) @@ -1417,13 +1278,8 @@ def recover_session_database( if table in _AUXILIARY_TABLES: _ensure_auxiliary_destination_schema(destination_conn, table) copy_report[table] = _copy_table( - source_conn, - destination_conn, - table, - salvage=allow_partial, - chunk_size=chunk_size, - progress_cb=progress_cb, - source_rows=inspection["tables"][table].get("rows"), + source_conn, destination_conn, table, salvage=allow_partial, chunk_size=chunk_size, + progress_cb=progress_cb, source_rows=inspection["tables"][table].get("rows"), ) orphan_cleanup = _cleanup_partial_orphans(destination_conn) if allow_partial else None derived_metadata = _finalize_derived_metadata(destination_conn) @@ -1445,15 +1301,8 @@ def recover_session_database( orphan_cleanup=orphan_cleanup, ) return _recovery_report( - source, - output, - inspection, - disk_space, - verification, - on_source_change="complete", - allow_partial=allow_partial, - copy=copy_report, - orphan_cleanup=orphan_cleanup, + source, output, inspection, disk_space, verification, on_source_change="complete", + allow_partial=allow_partial, copy=copy_report, orphan_cleanup=orphan_cleanup, derived_metadata=derived_metadata, ) finally: @@ -1462,7 +1311,6 @@ def recover_session_database( def write_recovery_report(path: Path, report: dict[str, Any]) -> Path: """Write a JSON report without overwriting an existing file.""" - destination = _resolved_output_path(path) with destination.open("x", encoding="utf-8") as handle: json.dump(report, handle, indent=2, sort_keys=True)