From eef274150fcd7f0f324211adfc600c09d6f92032 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:51:04 -0700 Subject: [PATCH 1/7] refactor(hermes_cli): unify session_recovery copy_table/copy_state_meta(_salvage) into salvage-flagged helpers --- hermes_cli/session_recovery.py | 84 +++++++++++++++------------------- 1 file changed, 37 insertions(+), 47 deletions(-) diff --git a/hermes_cli/session_recovery.py b/hermes_cli/session_recovery.py index 16e4a257dc..2964503cdf 100644 --- a/hermes_cli/session_recovery.py +++ b/hermes_cli/session_recovery.py @@ -511,26 +511,31 @@ def _copy_table( destination: sqlite3.Connection, table: str, *, + salvage: bool, chunk_size: int, progress_cb: Optional[ProgressCallback], source_rows: Optional[int], ) -> dict[str, Any]: - result: dict[str, Any] = { - "source_rows": source_rows, - "copied_rows": 0, - } + """Copy one canonical table: straight chunked copy, or rowid-range salvage when ``salvage``.""" + if table == "state_meta": + return _copy_state_meta( + source, destination, salvage=salvage, chunk_size=chunk_size, progress_cb=progress_cb, source_rows=source_rows + ) + if salvage: + return _copy_table_salvage( + source, destination, table, chunk_size=chunk_size, progress_cb=progress_cb, source_rows=source_rows + ) + result: dict[str, Any] = {"source_rows": source_rows, "copied_rows": 0} columns = _compatible_columns(source, destination, table, result) if columns is None: return result - quoted, placeholders = _quoted_columns(columns) - insert_prefix = "INSERT OR REPLACE" if table == "state_meta" else "INSERT" return _copy_rows( source, destination, f'SELECT {quoted} FROM "{table}"', (), - f'{insert_prefix} INTO "{table}" ({quoted}) VALUES ({placeholders})', + f'INSERT INTO "{table}" ({quoted}) VALUES ({placeholders})', table=table, chunk_size=chunk_size, progress_cb=progress_cb, @@ -863,14 +868,34 @@ def _copy_state_meta( source: sqlite3.Connection, destination: sqlite3.Connection, *, + salvage: bool, chunk_size: int, progress_cb: Optional[ProgressCallback], source_rows: Optional[int], ) -> dict[str, Any]: - problem = _state_meta_precheck(source, destination, source_rows, salvage=False) + """Copy user metadata rows; derived FTS/topic keys (``_GENERATED_META_KEYS``) are regenerated, not copied.""" + problem = _state_meta_precheck(source, destination, source_rows, salvage=salvage) if problem is not None: return problem + if salvage: + def keep_user_meta(row: tuple[Any, ...], columns: tuple[str, ...]) -> bool: + return str(row[columns.index("key")]) not in _GENERATED_META_KEYS + + result = _copy_table_salvage( + source, + destination, + "state_meta", + chunk_size=chunk_size, + progress_cb=progress_cb, + source_rows=source_rows, + insert_prefix="INSERT OR REPLACE", + row_filter=keep_user_meta, + ) + result["source_meta_rows"] = result.pop("source_rows") + result["excluded_keys"] = sorted(_GENERATED_META_KEYS) + return result + placeholders = ", ".join("?" for _ in _GENERATED_META_KEYS) params = tuple(_GENERATED_META_KEYS) filtered_source_rows: Optional[int] = None @@ -882,8 +907,7 @@ def _copy_state_meta( ).fetchone()[0] ) except sqlite3.DatabaseError: - # The copy loop below will return the concrete read error. - pass + pass # the copy loop below will return the concrete read error return _copy_rows( source, @@ -899,37 +923,6 @@ def _copy_state_meta( ) -def _copy_state_meta_salvage( - source: sqlite3.Connection, - destination: sqlite3.Connection, - *, - chunk_size: int, - progress_cb: Optional[ProgressCallback], - source_rows: Optional[int], -) -> dict[str, Any]: - """Salvage readable user metadata while regenerating derived FTS state.""" - problem = _state_meta_precheck(source, destination, source_rows, salvage=True) - if problem is not None: - return problem - - def keep_user_meta(row: tuple[Any, ...], columns: tuple[str, ...]) -> bool: - return str(row[columns.index("key")]) not in _GENERATED_META_KEYS - - result = _copy_table_salvage( - source, - destination, - "state_meta", - chunk_size=chunk_size, - progress_cb=progress_cb, - source_rows=source_rows, - insert_prefix="INSERT OR REPLACE", - row_filter=keep_user_meta, - ) - result["source_meta_rows"] = result.pop("source_rows") - result["excluded_keys"] = sorted(_GENERATED_META_KEYS) - return result - - def _reconstruct_missing_sessions( destination: sqlite3.Connection, ) -> dict[str, Any]: @@ -1453,9 +1446,6 @@ def recover_session_database( topic_tables=any(inspection["tables"][table].get("available") for table in _TOPIC_TABLES), ) - copy_meta, copy_rows = ( - (_copy_state_meta_salvage, _copy_table_salvage) if allow_partial else (_copy_state_meta, _copy_table) - ) copy_report: dict[str, dict[str, Any]] = {} for table in (*_CANONICAL_TABLES, "state_meta", *_TOPIC_TABLES, *_AUXILIARY_TABLES): if table not in _CANONICAL_TABLES and not inspection["tables"][table].get("available"): @@ -1463,11 +1453,11 @@ def recover_session_database( continue if table in _AUXILIARY_TABLES: _ensure_auxiliary_destination_schema(destination_conn, table) - copy_function, args = (copy_meta, ()) if table == "state_meta" else (copy_rows, (table,)) - copy_report[table] = copy_function( + copy_report[table] = _copy_table( source_conn, destination_conn, - *args, + table, + salvage=allow_partial, chunk_size=chunk_size, progress_cb=progress_cb, source_rows=inspection["tables"][table].get("rows"), From 2a9ab0dc7fd765fe8922637b29d5e7c5e9f15d6d Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:01:40 -0700 Subject: [PATCH 2/7] refactor(hermes_cli): split session_recovery god functions into _RowidRangeSalvage and _verify_* phase helpers --- hermes_cli/session_recovery.py | 458 ++++++++++++++++++--------------- 1 file changed, 250 insertions(+), 208 deletions(-) diff --git a/hermes_cli/session_recovery.py b/hermes_cli/session_recovery.py index 2964503cdf..13129d8bb7 100644 --- a/hermes_cli/session_recovery.py +++ b/hermes_cli/session_recovery.py @@ -661,6 +661,112 @@ def _probe_populated_edge( span *= 2 +class _RowidRangeSalvage: + """Bisecting rowid-range copy for one table; counters live on the shared ``result`` dict.""" + + def __init__( + self, + source: sqlite3.Connection, + destination: sqlite3.Connection, + table: str, + columns: list[str], + *, + chunk_size: int, + progress_cb: Optional[ProgressCallback], + source_rows: Optional[int], + insert_prefix: str, + row_filter: Optional[Callable[[tuple[Any, ...], tuple[str, ...]], bool]], + result: dict[str, Any], + ) -> None: + self.source, self.destination, self.table = source, destination, table + self.chunk_size, self.progress_cb, self.source_rows = chunk_size, progress_cb, source_rows + self.row_filter, self.result = row_filter, result + self.column_names = tuple(columns) + quoted, placeholders = _quoted_columns(columns) + self.select_sql = f'SELECT rowid, {quoted} FROM "{table}" WHERE rowid BETWEEN ? AND ? ORDER BY rowid' + self.insert_sql = f'{insert_prefix} INTO "{table}" ({quoted}) VALUES ({placeholders})' + self.exact_sql = f'SELECT {quoted} FROM "{table}" WHERE rowid = ?' + self.stopped_at_query_limit = False + + def _keep(self, values: list[tuple[Any, ...]]) -> list[tuple[Any, ...]]: + if self.row_filter is None: + return values + return [row for row in values if self.row_filter(row, self.column_names)] + + def _skip(self, low: int, high: int, error: str) -> None: + _append_skipped_range(self.result["skipped_rowid_ranges"], low, high, error) + + def recover_exact_rowid(self, rowid: int) -> bool: + """Salvage one row by exact-key lookup (issue #80205). + + 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. + """ + result = self.result + result["range_queries"] += 1 + try: + row = self.source.execute(self.exact_sql, (rowid,)).fetchone() + except sqlite3.DatabaseError: + return False + if row is None: + return True # genuinely absent: nothing to skip + value = tuple(row) + if not self._keep([value]): + result["excluded_rows"] += 1 + return True + with _immediate_transaction(self.destination): + self.destination.execute(self.insert_sql, value) + result["copied_rows"] += 1 + result["exact_lookup_recovered"] += 1 + return True + + def copy_range(self, low: int, high: int) -> None: + """Copy ``[low, high]``; on a read error bisect the unread remainder, exact-lookup singletons.""" + result = self.result + if low > high: + return + if result["range_queries"] >= _MAX_SALVAGE_RANGE_QUERIES: + self.stopped_at_query_limit = True + self._skip(low, high, "salvage range query limit reached") + return + + result["range_queries"] += 1 + last_committed_rowid: Optional[int] = None + try: + cursor = self.source.execute(self.select_sql, (low, high)) + while True: + fetched = cursor.fetchmany(self.chunk_size) + if not fetched: + return + values = [tuple(row[1:]) for row in fetched] + included = self._keep(values) + if included: + with _immediate_transaction(self.destination): + self.destination.executemany(self.insert_sql, included) + result["copied_rows"] += len(included) + result["excluded_rows"] += len(values) - len(included) + last_committed_rowid = int(fetched[-1][0]) + if self.progress_cb is not None: + self.progress_cb({ + "table": self.table, + "copied_rows": result["copied_rows"], + "source_rows": self.source_rows, + "skipped_ranges": len(result["skipped_rowid_ranges"]), + }) + except sqlite3.DatabaseError as exc: + retry_low = last_committed_rowid + 1 if last_committed_rowid is not None else low + if retry_low > high: + return + if retry_low == high: + if not self.recover_exact_rowid(retry_low): + self._skip(retry_low, high, str(exc)) + return + midpoint = retry_low + (high - retry_low) // 2 + self.copy_range(retry_low, midpoint) + self.copy_range(midpoint + 1, high) + + def _copy_table_salvage( source: sqlite3.Connection, destination: sqlite3.Connection, @@ -703,102 +809,22 @@ def _copy_table_salvage( result["error"] += f": {details}" return result - quoted, placeholders = _quoted_columns(columns) - select_sql = ( - f'SELECT rowid, {quoted} FROM "{table}" ' - "WHERE rowid BETWEEN ? AND ? ORDER BY rowid" + salvage = _RowidRangeSalvage( + source, + destination, + table, + columns, + chunk_size=chunk_size, + progress_cb=progress_cb, + source_rows=source_rows, + insert_prefix=insert_prefix, + row_filter=row_filter, + result=result, ) - insert_sql = ( - f'{insert_prefix} INTO "{table}" ({quoted}) VALUES ({placeholders})' - ) - 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. - """ - 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 - with _immediate_transaction(destination): - destination.execute(insert_sql, value) - result["copied_rows"] += 1 - result["exact_lookup_recovered"] += 1 - return True - - def copy_range(low: int, high: int) -> None: - nonlocal stopped_at_query_limit - if low > high: - return - if result["range_queries"] >= _MAX_SALVAGE_RANGE_QUERIES: - stopped_at_query_limit = True - _append_skipped_range( - result["skipped_rowid_ranges"], low, high, "salvage range query limit reached" - ) - return - - result["range_queries"] += 1 - last_committed_rowid: Optional[int] = None - try: - cursor = source.execute(select_sql, (low, high)) - while True: - fetched = cursor.fetchmany(chunk_size) - if not fetched: - return - - values = [tuple(row[1:]) for row in fetched] - if row_filter is not None: - included = [ - row for row in values if row_filter(row, column_names) - ] - excluded_count = len(values) - len(included) - else: - included = values - excluded_count = 0 - - if included: - with _immediate_transaction(destination): - destination.executemany(insert_sql, included) - - result["copied_rows"] += len(included) - result["excluded_rows"] += excluded_count - last_committed_rowid = int(fetched[-1][0]) - if progress_cb is not None: - progress_cb({ - "table": table, - "copied_rows": result["copied_rows"], - "source_rows": source_rows, - "skipped_ranges": len(result["skipped_rowid_ranges"]), - }) - except sqlite3.DatabaseError as exc: - retry_low = last_committed_rowid + 1 if last_committed_rowid is not None else low - if retry_low > high: - return - if retry_low == high: - 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) - copy_range(midpoint + 1, high) - - copy_range(int(bounds["low"]), int(bounds["high"])) + salvage.copy_range(int(bounds["low"]), int(bounds["high"])) skipped_ranges = result["skipped_rowid_ranges"] result["skipped_rowid_span"] = sum(item["high"] - item["low"] + 1 for item in skipped_ranges) - result["query_limit_reached"] = stopped_at_query_limit + result["query_limit_reached"] = salvage.stopped_at_query_limit if skipped_ranges: result["status"] = "partial" if result["copied_rows"] else "failed" @@ -1052,19 +1078,68 @@ def _cleanup_partial_orphans( return result -def _verify_recovered_database( - output: Path, +def _verify_structure(conn: sqlite3.Connection, verification: dict[str, Any]) -> None: + """Integrity, foreign keys, journal mode, schema version and FTS meta of the recovered file.""" + errors = verification["errors"] + integrity_rows = [str(row[0]) for row in conn.execute("PRAGMA integrity_check").fetchall()] + verification["integrity_check"] = integrity_rows + if integrity_rows != ["ok"]: + errors.append("PRAGMA integrity_check did not return exactly 'ok'") + + foreign_key_rows = [list(row) for row in conn.execute("PRAGMA foreign_key_check").fetchall()] + verification["foreign_key_check"] = foreign_key_rows + if foreign_key_rows: + errors.append("foreign key violations remain") + + verification["journal_mode"] = _journal_mode(conn) + + schema_row = conn.execute("SELECT version FROM schema_version LIMIT 1").fetchone() + verification["schema_version"] = int(schema_row[0]) if schema_row else None + if verification["schema_version"] != SCHEMA_VERSION: + errors.append(f"schema version is {verification['schema_version']}, expected {SCHEMA_VERSION}") + + meta = { + str(row[0]): row[1] + for row in conn.execute("SELECT key, value FROM state_meta WHERE key LIKE 'fts_%'").fetchall() + } + verification["fts_meta"] = meta + if meta.get("fts_storage_version") != str(FTS_STORAGE_VERSION): + errors.append("fresh FTS storage version was not established") + pending_keys = sorted( + key + for key in _GENERATED_META_KEYS + if key.startswith("fts_") and key != "fts_storage_version" and key in meta + ) + verification["pending_fts_keys"] = pending_keys + if pending_keys: + errors.append("derived FTS transition markers remain in the recovered database") + + +def _verify_fts_indexes(conn: sqlite3.Connection, verification: dict[str, Any]) -> None: + fts_checks: dict[str, str] = {} + for table in ("messages_fts", "messages_fts_trigram", "messages_fts_cjk"): + if not _table_columns(conn, table): + continue + try: + conn.execute(f'INSERT INTO "{table}" ("{table}") VALUES (\'integrity-check\')') + conn.execute(f'SELECT 1 FROM "{table}" WHERE "{table}" MATCH \'""\' LIMIT 1').fetchone() + fts_checks[table] = "ok" + except sqlite3.DatabaseError as exc: + fts_checks[table] = str(exc) + verification["errors"].append(f"{table} integrity check failed: {exc}") + verification["fts_checks"] = fts_checks + + +def _verify_row_counts( + conn: sqlite3.Connection, + verification: dict[str, Any], *, expected_counts: dict[str, Optional[int]], copy_report: dict[str, dict[str, Any]], - allow_partial: bool = False, - orphan_cleanup: Optional[dict[str, Any]] = None, -) -> dict[str, Any]: - verification: dict[str, Any] = { - "errors": [], - "warnings": [], - "loss_detected": False, - } + allow_partial: bool, + orphan_cleanup: Optional[dict[str, Any]], +) -> None: + """Compare recovered counts and copy statuses against the source; classify shortfalls as loss.""" def flag(message: str, *, soft: bool) -> None: """Record data loss as a warning (``soft``) or as a verification error.""" @@ -1074,6 +1149,69 @@ def _verify_recovered_database( else: verification["errors"].append(message) + counts: dict[str, int] = {} + for table in _INVENTORY_TABLES: + if _table_columns(conn, table): + counts[table] = _count_rows(conn, table) + verification["table_counts"] = counts + + for table in ("sessions", "messages", *_AUXILIARY_TABLES): + expected = expected_counts.get(table) + if expected is not None and counts.get(table) != expected: + flag(f"{table} count is {counts.get(table)}, expected {expected}", soft=allow_partial) + + cleanup = orphan_cleanup or {} + rebuilt_sessions = int(cleanup.get("sessions_reconstructed") or 0) + retained_messages = int(cleanup.get("messages_retained") or 0) + removed_messages = int(cleanup.get("messages_removed") or 0) + # A wholly unreadable sessions b-tree is recoverable when every output parent was rebuilt from + # the surviving messages and none were dropped: data loss, but not structural failure. + sessions_fully_reconstructed = bool( + rebuilt_sessions > 0 + and counts.get("sessions") == rebuilt_sessions + and counts.get("messages") == retained_messages + and removed_messages == 0 + ) + + for table, table_report in copy_report.items(): + status = table_report.get("status") + if status not in {"failed", "partial"}: + continue + flag( + f"{table} copy status is {status}", + soft=allow_partial + and ( + status == "partial" + or table not in {"sessions", "messages"} + or (table == "sessions" and sessions_fully_reconstructed) + ), + ) + + if orphan_cleanup: + orphan_count = int(orphan_cleanup.get("total_removed_or_relinked") or 0) + if orphan_count: + flag(f"{orphan_count} orphaned reference(s) were removed or relinked", soft=True) + if rebuilt_sessions: + # Not a clean recovery: the conversation text survived but its session metadata did not. + flag( + f"{rebuilt_sessions} session(s) could not be salvaged and " + f"were reconstructed as placeholders to retain " + f"{retained_messages} message(s); their metadata (title, model, " + "timestamps, cost) is lost", + soft=True, + ) + + +def _verify_recovered_database( + output: Path, + *, + expected_counts: dict[str, Optional[int]], + copy_report: dict[str, dict[str, Any]], + allow_partial: bool = False, + orphan_cleanup: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + verification: dict[str, Any] = {"errors": [], "warnings": [], "loss_detected": False} + open_error = _db_opens_cleanly(output) verification["opens_cleanly"] = open_error is None if open_error is not None: @@ -1081,112 +1219,16 @@ def _verify_recovered_database( conn = sqlite3.connect(str(output), isolation_level=None) try: - integrity_rows = [str(row[0]) for row in conn.execute("PRAGMA integrity_check").fetchall()] - verification["integrity_check"] = integrity_rows - if integrity_rows != ["ok"]: - verification["errors"].append("PRAGMA integrity_check did not return exactly 'ok'") - - foreign_key_rows = [list(row) for row in conn.execute("PRAGMA foreign_key_check").fetchall()] - verification["foreign_key_check"] = foreign_key_rows - if foreign_key_rows: - verification["errors"].append("foreign key violations remain") - - verification["journal_mode"] = _journal_mode(conn) - - schema_row = conn.execute("SELECT version FROM schema_version LIMIT 1").fetchone() - verification["schema_version"] = int(schema_row[0]) if schema_row else None - if verification["schema_version"] != SCHEMA_VERSION: - verification["errors"].append( - f"schema version is {verification['schema_version']}, " - f"expected {SCHEMA_VERSION}" - ) - - meta = { - str(row[0]): row[1] - for row in conn.execute("SELECT key, value FROM state_meta WHERE key LIKE 'fts_%'").fetchall() - } - verification["fts_meta"] = meta - if meta.get("fts_storage_version") != str(FTS_STORAGE_VERSION): - verification["errors"].append("fresh FTS storage version was not established") - pending_keys = sorted( - key - for key in _GENERATED_META_KEYS - if key.startswith("fts_") and key != "fts_storage_version" and key in meta + _verify_structure(conn, verification) + _verify_row_counts( + conn, + verification, + expected_counts=expected_counts, + copy_report=copy_report, + allow_partial=allow_partial, + orphan_cleanup=orphan_cleanup, ) - verification["pending_fts_keys"] = pending_keys - if pending_keys: - verification["errors"].append("derived FTS transition markers remain in the recovered database") - - counts: dict[str, int] = {} - for table in _INVENTORY_TABLES: - if _table_columns(conn, table): - counts[table] = _count_rows(conn, table) - verification["table_counts"] = counts - - for table in ("sessions", "messages", *_AUXILIARY_TABLES): - expected = expected_counts.get(table) - if expected is not None and counts.get(table) != expected: - flag(f"{table} count is {counts.get(table)}, expected {expected}", soft=allow_partial) - - cleanup = orphan_cleanup or {} - rebuilt_sessions = int(cleanup.get("sessions_reconstructed") or 0) - retained_messages = int(cleanup.get("messages_retained") or 0) - removed_messages = int(cleanup.get("messages_removed") or 0) - # A wholly unreadable sessions b-tree is recoverable when every output - # parent was rebuilt from the surviving messages and none were dropped. - # This is still data loss, but it is not structural verification failure. - sessions_fully_reconstructed = bool( - rebuilt_sessions > 0 - and counts.get("sessions") == rebuilt_sessions - and counts.get("messages") == retained_messages - and removed_messages == 0 - ) - - for table, table_report in copy_report.items(): - status = table_report.get("status") - if status not in {"failed", "partial"}: - continue - flag( - f"{table} copy status is {status}", - soft=allow_partial - and ( - status == "partial" - or table not in {"sessions", "messages"} - or (table == "sessions" and sessions_fully_reconstructed) - ), - ) - - if orphan_cleanup: - orphan_count = int(orphan_cleanup.get("total_removed_or_relinked") or 0) - if orphan_count: - flag(f"{orphan_count} orphaned reference(s) were removed or relinked", soft=True) - if rebuilt_sessions: - # Not a clean recovery: the conversation text survived but its - # session metadata did not, so these rows are placeholders. - flag( - f"{rebuilt_sessions} session(s) could not be salvaged and " - f"were reconstructed as placeholders to retain " - f"{retained_messages} message(s); their metadata (title, model, " - "timestamps, cost) is lost", - soft=True, - ) - - fts_checks: dict[str, str] = {} - for table in ("messages_fts", "messages_fts_trigram", "messages_fts_cjk"): - if not _table_columns(conn, table): - continue - try: - conn.execute( - f'INSERT INTO "{table}" ("{table}") VALUES (\'integrity-check\')' - ) - conn.execute( - f'SELECT 1 FROM "{table}" WHERE "{table}" MATCH \'""\' LIMIT 1' - ).fetchone() - fts_checks[table] = "ok" - except sqlite3.DatabaseError as exc: - fts_checks[table] = str(exc) - verification["errors"].append(f"{table} integrity check failed: {exc}") - verification["fts_checks"] = fts_checks + _verify_fts_indexes(conn, verification) except sqlite3.DatabaseError as exc: verification["errors"].append(f"verification query failed: {exc}") finally: From 86d111d77ce0bc42cb3966a19700a1d6e7379cc1 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:12:14 -0700 Subject: [PATCH 3/7] refactor(hermes_cli): collapse defensive layers in sessions_cmd, join bracket groups in session recovery modules --- hermes_cli/session_lost_and_found.py | 92 ++++--------------- hermes_cli/session_recap.py | 8 +- hermes_cli/session_recovery.py | 131 ++++++--------------------- hermes_cli/sessions_cmd.py | 55 ++++------- 4 files changed, 62 insertions(+), 224 deletions(-) diff --git a/hermes_cli/session_lost_and_found.py b/hermes_cli/session_lost_and_found.py index 5503c9529e..422f6c7af7 100644 --- a/hermes_cli/session_lost_and_found.py +++ b/hermes_cli/session_lost_and_found.py @@ -82,11 +82,7 @@ def _cli_supports_recover(binary: str) -> bool: conn.commit() finally: conn.close() - probe = subprocess.run( - [binary, "-readonly", str(scratch), ".recover"], - capture_output=True, - timeout=30, - ) + probe = subprocess.run([binary, "-readonly", str(scratch), ".recover"], capture_output=True, timeout=30) if probe.returncode != 0: return False return b"sqlite_dbpage" not in probe.stderr @@ -129,9 +125,7 @@ def run_cli_lost_and_found_recover( except subprocess.TimeoutExpired: dump.kill() load.kill() - raise LostAndFoundError( - f"sqlite3 .recover timed out after {timeout:.0f}s" - ) + raise LostAndFoundError(f"sqlite3 .recover timed out after {timeout:.0f}s") attempt = { "command": command, "dump_returncode": dump.returncode, @@ -162,9 +156,7 @@ def _lost_and_found_db_usable(lf_path: Path) -> bool: try: conn = sqlite3.connect(str(lf_path)) try: - return conn.execute( - "SELECT 1 FROM sqlite_master WHERE type='table' LIMIT 1" - ).fetchone() is not None + return conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' LIMIT 1").fetchone() is not None finally: conn.close() except sqlite3.DatabaseError: @@ -218,22 +210,14 @@ def _looks_like_source(value: Any) -> bool: 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]: +def classify_lost_and_found_row(nfield: int, cells: tuple[Any, ...]) -> Optional[str]: """Classify one lost_and_found record by field count + sentinel values.""" 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 - ): + if (isinstance(cells[1], str) and cells[1] and isinstance(cells[2], str) and cells[2] in MESSAGE_ROLES): return "messages" return None @@ -281,26 +265,16 @@ def _insert_prefix_row( 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, - ) + 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]: +def _copy_direct_tables(lf_conn: sqlite3.Connection, dest: sqlite3.Connection) -> dict[str, int]: """Copy rows .recover managed to attribute to real canonical tables.""" # Lazy import: session_recovery imports this module inside a function, so # a module-level import here would be circular. - from hermes_cli.session_recovery import ( - _AUXILIARY_TABLE_SCHEMAS, - _AUXILIARY_TABLES, - _CANONICAL_TABLES, - ) + from hermes_cli.session_recovery import (_AUXILIARY_TABLE_SCHEMAS, _AUXILIARY_TABLES, _CANONICAL_TABLES) copied: dict[str, int] = {} for table in (*_CANONICAL_TABLES, *_AUXILIARY_TABLES): @@ -323,19 +297,13 @@ def _copy_direct_tables( 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, - ) + 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]: +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.""" report: dict[str, Any] = { @@ -356,11 +324,7 @@ def map_lost_and_found_rows( # genuinely NULL was already rejected by classify_lost_and_found_row, # so the substitutions only fill NOT NULL bookkeeping counters/flags. targets: dict[str, tuple[list[str], dict[int, Any]]] = {} - for kind_name, protected in ( - ("sessions", (0, 1)), - ("messages", (1, 2)), - ("session_model_usage", (0, 1)), - ): + for kind_name, protected in (("sessions", (0, 1)), ("messages", (1, 2)), ("session_model_usage", (0, 1))): defaults = _notnull_defaults(dest, kind_name) for index in protected: defaults.pop(index, None) @@ -447,11 +411,7 @@ def stub_missing_parent_sessions(dest: sqlite3.Connection) -> dict[str, Any]: clearly marked. """ - result: dict[str, Any] = { - "sessions_stubbed": 0, - "messages_retained": 0, - "usage_rows_retained": 0, - } + 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]] = {} @@ -470,42 +430,24 @@ def stub_missing_parent_sessions(dest: sqlite3.Connection) -> dict[str, Any]: "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} - ) + 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" - ) + 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 - ): + 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"], - ), + (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] - ) + 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( diff --git a/hermes_cli/session_recap.py b/hermes_cli/session_recap.py index 281202edb3..32d7553b7f 100644 --- a/hermes_cli/session_recap.py +++ b/hermes_cli/session_recap.py @@ -79,9 +79,7 @@ def _tool_call_name_and_args(tool_call: Any) -> Tuple[str, Mapping[str, Any]]: return name, raw_args if isinstance(raw_args, Mapping) else {} -def _iter_assistant_tool_calls( - messages: Sequence[Mapping[str, Any]], -) -> Iterable[Tuple[str, Mapping[str, Any]]]: +def _iter_assistant_tool_calls(messages: Sequence[Mapping[str, Any]]) -> Iterable[Tuple[str, Mapping[str, Any]]]: for msg in messages: if not isinstance(msg, Mapping): continue @@ -96,9 +94,7 @@ def _iter_assistant_tool_calls( yield name, args -def _count_visible_turns( - messages: Sequence[Mapping[str, Any]], -) -> Tuple[int, int, int]: +def _count_visible_turns(messages: Sequence[Mapping[str, Any]]) -> Tuple[int, int, int]: """Return ``(user_turn_count, assistant_turn_count, tool_message_count)``.""" roles = Counter(msg.get("role") for msg in messages if isinstance(msg, Mapping)) return roles["user"], roles["assistant"], roles["tool"] diff --git a/hermes_cli/session_recovery.py b/hermes_cli/session_recovery.py index 13129d8bb7..e67b1174ad 100644 --- a/hermes_cli/session_recovery.py +++ b/hermes_cli/session_recovery.py @@ -18,12 +18,7 @@ from contextlib import contextmanager from pathlib import Path from typing import Any, Callable, Iterator, Optional -from hermes_state import ( - FTS_STORAGE_VERSION, - SCHEMA_VERSION, - SessionDB, - _db_opens_cleanly, -) +from hermes_state import (FTS_STORAGE_VERSION, SCHEMA_VERSION, SessionDB, _db_opens_cleanly) ProgressCallback = Callable[[dict[str, Any]], None] @@ -38,10 +33,7 @@ _CANONICAL_TABLES = ( "async_delegations", ) -_TOPIC_TABLES = ( - "telegram_dm_topic_mode", - "telegram_dm_topic_bindings", -) +_TOPIC_TABLES = ("telegram_dm_topic_mode", "telegram_dm_topic_bindings") def _init_delivery_ledger_schema(conn: sqlite3.Connection) -> None: @@ -63,12 +55,7 @@ _AUXILIARY_TABLE_SCHEMAS: dict[str, Callable[[sqlite3.Connection], None]] = { _AUXILIARY_TABLES = tuple(_AUXILIARY_TABLE_SCHEMAS) -_INVENTORY_TABLES = ( - *_CANONICAL_TABLES, - "state_meta", - *_TOPIC_TABLES, - *_AUXILIARY_TABLES, -) +_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. @@ -134,9 +121,7 @@ def _validate_paths( for suffix in _SIDECAR_SUFFIXES: candidate = _sidecar_path(output, suffix) if os.path.lexists(candidate): - raise SessionRecoverySafetyError( - f"Refusing to overwrite existing recovery output: {candidate}" - ) + raise SessionRecoverySafetyError(f"Refusing to overwrite existing recovery output: {candidate}") work_root = ( work_dir.expanduser().resolve(strict=True) @@ -144,9 +129,7 @@ def _validate_paths( else (output.parent if output is not None else source.parent) ) if not work_root.is_dir(): - raise SessionRecoverySafetyError( - f"Recovery work directory is not a directory: {work_root}" - ) + raise SessionRecoverySafetyError(f"Recovery work directory is not a directory: {work_root}") return source, output, work_root @@ -180,11 +163,7 @@ def _same_filesystem(left: Path, right: Path) -> bool: return left.anchor.casefold() == right.anchor.casefold() -def _disk_space_preflight( - source: Path, - work_root: Path, - output_parent: Optional[Path], -) -> dict[str, Any]: +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( @@ -355,9 +334,7 @@ def _copy_rows( destination.executemany(insert_sql, rows) result["copied_rows"] += len(rows) if progress_cb is not None: - progress_cb( - {"table": table, "copied_rows": result["copied_rows"], "source_rows": expected_rows} - ) + progress_cb({"table": table, "copied_rows": result["copied_rows"], "source_rows": expected_rows}) except sqlite3.DatabaseError as exc: result["status"] = "partial" if result["copied_rows"] else "failed" result["error"] = str(exc) @@ -370,10 +347,7 @@ def _copy_rows( return result -def _table_inventory( - conn: sqlite3.Connection, - table: str, -) -> dict[str, Any]: +def _table_inventory(conn: sqlite3.Connection, table: str) -> dict[str, Any]: result: dict[str, Any] = {"available": False, "columns": [], "rows": None} try: columns = _table_columns(conn, table) @@ -419,10 +393,7 @@ def _snapshot_and_inspect( work_root: Path, ) -> tuple[tempfile.TemporaryDirectory[str], Path, dict[str, Any]]: before = _source_fingerprint(source) - temp_dir = tempfile.TemporaryDirectory( - prefix="hermes-session-recovery-", - dir=str(work_root), - ) + 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) @@ -454,11 +425,7 @@ def _snapshot_and_inspect( raise -def inspect_session_database( - source_path: Path, - *, - work_dir: Optional[Path] = None, -) -> dict[str, Any]: +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) @@ -476,10 +443,7 @@ def inspect_session_database( temp_dir.cleanup() -def _ensure_auxiliary_destination_schema( - destination: sqlite3.Connection, - table: str, -) -> None: +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 @@ -517,14 +481,11 @@ def _copy_table( source_rows: Optional[int], ) -> dict[str, Any]: """Copy one canonical table: straight chunked copy, or rowid-range salvage when ``salvage``.""" + copy_kwargs = dict(chunk_size=chunk_size, progress_cb=progress_cb, source_rows=source_rows) if table == "state_meta": - return _copy_state_meta( - source, destination, salvage=salvage, chunk_size=chunk_size, progress_cb=progress_cb, source_rows=source_rows - ) + return _copy_state_meta(source, destination, salvage=salvage, **copy_kwargs) if salvage: - return _copy_table_salvage( - source, destination, table, chunk_size=chunk_size, progress_cb=progress_cb, source_rows=source_rows - ) + return _copy_table_salvage(source, destination, table, **copy_kwargs) result: dict[str, Any] = {"source_rows": source_rows, "copied_rows": 0} columns = _compatible_columns(source, destination, table, result) if columns is None: @@ -544,12 +505,7 @@ def _copy_table( ) -def _append_skipped_range( - ranges: list[dict[str, Any]], - low: int, - high: int, - error: str, -) -> None: +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.""" if ranges and ranges[-1]["high"] + 1 == low and ranges[-1]["error"] == error: @@ -558,10 +514,7 @@ def _append_skipped_range( ranges.append({"low": low, "high": high, "error": error}) -def _salvage_rowid_bounds( - source: sqlite3.Connection, - table: str, -) -> dict[str, Any]: +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": []} @@ -606,13 +559,7 @@ def _salvage_rowid_bounds( return result -def _probe_populated_edge( - source: sqlite3.Connection, - table: str, - *, - edge: str, - anchor: int, -) -> dict[str, Any]: +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 @@ -634,9 +581,7 @@ def _probe_populated_edge( 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 - ): + 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 @@ -776,9 +721,7 @@ def _copy_table_salvage( progress_cb: Optional[ProgressCallback], source_rows: Optional[int], insert_prefix: str = "INSERT", - row_filter: Optional[ - Callable[[tuple[Any, ...], tuple[str, ...]], bool] - ] = None, + row_filter: Optional[Callable[[tuple[Any, ...], tuple[str, ...]], bool]] = None, ) -> dict[str, Any]: """Best-effort rowid-range copy that continues past damaged source pages.""" @@ -829,10 +772,7 @@ def _copy_table_salvage( if skipped_ranges: result["status"] = "partial" if result["copied_rows"] else "failed" result["error"] = f"{len(skipped_ranges)} rowid range(s) skipped" - elif ( - source_rows is not None - and result["copied_rows"] + result["excluded_rows"] != source_rows - ): + elif (source_rows is not None and result["copied_rows"] + result["excluded_rows"] != source_rows): result["status"] = "partial" result["error"] = ( f"copied {result['copied_rows']} and excluded " @@ -927,10 +867,7 @@ def _copy_state_meta( filtered_source_rows: Optional[int] = None try: filtered_source_rows = int( - source.execute( - f"SELECT COUNT(*) FROM state_meta WHERE key NOT IN ({placeholders})", - params, - ).fetchone()[0] + source.execute(f"SELECT COUNT(*) FROM state_meta WHERE key NOT IN ({placeholders})", params).fetchone()[0] ) except sqlite3.DatabaseError: pass # the copy loop below will return the concrete read error @@ -949,9 +886,7 @@ def _copy_state_meta( ) -def _reconstruct_missing_sessions( - destination: sqlite3.Connection, -) -> dict[str, Any]: +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 @@ -1008,18 +943,11 @@ def _reconcile(destination: sqlite3.Connection, table: str, where: str, mutation return count -_DEPENDENT_TABLES = ( - "messages", - "session_model_usage", - "compression_locks", - "telegram_dm_topic_bindings", -) +_DEPENDENT_TABLES = ("messages", "session_model_usage", "compression_locks", "telegram_dm_topic_bindings") _RELINK_COUNTERS = ("session_prompt_refs_cleared", "sessions_parent_cleared") -def _cleanup_partial_orphans( - destination: sqlite3.Connection, -) -> dict[str, Any]: +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 @@ -1257,10 +1185,7 @@ def _finalize_derived_metadata(destination: sqlite3.Connection) -> dict[str, Any fts_keys = tuple(key for key in _GENERATED_META_KEYS if key.startswith("fts_")) placeholders = ", ".join("?" for _ in fts_keys) with _immediate_transaction(destination): - destination.execute( - f"DELETE FROM state_meta WHERE key IN ({placeholders})", - fts_keys, - ) + destination.execute(f"DELETE FROM state_meta WHERE key IN ({placeholders})", fts_keys) destination.execute( "INSERT INTO state_meta(key, value) VALUES (?, ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", @@ -1442,11 +1367,7 @@ def recover_session_database( if chunk_size <= 0: raise SessionRecoverySafetyError("chunk_size must be greater than zero") - source, output, work_root = _validate_paths( - source_path, - output_path=output_path, - work_dir=work_dir, - ) + source, output, work_root = _validate_paths(source_path, output_path=output_path, work_dir=work_dir) assert output is not None disk_space = _disk_space_preflight(source, work_root, output.parent) diff --git a/hermes_cli/sessions_cmd.py b/hermes_cli/sessions_cmd.py index e130cc303d..c26e3dc5d0 100644 --- a/hermes_cli/sessions_cmd.py +++ b/hermes_cli/sessions_cmd.py @@ -10,6 +10,7 @@ working and imports stay one-way. Browse picker: :mod:`hermes_cli.sessions_cmd_b import json import os +import shutil import sys from functools import partial from pathlib import Path @@ -111,12 +112,9 @@ def _cmd_repair(args): try: from hermes_state import SessionDB - _repair_db = SessionDB() - try: + with SessionDB() as _repair_db: n = _repair_db._conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0] - print(f"✓ Repaired — {n} sessions recovered.") - finally: - _repair_db.close() + print(f"✓ Repaired — {n} sessions recovered.") except Exception: print("✓ Repaired.") return @@ -201,8 +199,7 @@ def _cmd_recover(args): class _RecoveryProgress: """`recover` progress printer: one live-updating `` : n/total`` line per table.""" - def __init__(self): - self.table = None + table = None def __call__(self, info): table = info.get("table") @@ -396,14 +393,11 @@ def _export_html(args, collect): sessions = collect() if sessions is None: return - if len(sessions) == 1: - content = generate_html_export(sessions[0]) - else: - content = generate_multi_session_html_export(sessions) + single = len(sessions) == 1 + content = generate_html_export(sessions[0]) if single else generate_multi_session_html_export(sessions) with open(args.output, "w", encoding="utf-8") as f: f.write(content) - suffix = "" if len(sessions) == 1 else "s" - print(f"Exported {len(sessions)} session{suffix} to {args.output} (HTML)") + print(f"Exported {len(sessions)} session{'' if single else 's'} to {args.output} (HTML)") def _export_jsonl(args, collect): @@ -577,9 +571,7 @@ def _export_markdown_single(db, args, export_one, output_dir, lineage_is_logical if db.delete_session(resolved_session_id, sessions_dir=_sessions_dir(), expected_delete_ids=delete_target_ids): delegate_count = len(delete_target_ids) - 1 delegate_suffix = ( - "" - if not delegate_count - else f" and {delegate_count} delegate session{'' if delegate_count == 1 else 's'}" + f" and {delegate_count} delegate session{'' if delegate_count == 1 else 's'}" if delegate_count else "" ) print(f"Deleted exported session '{resolved_session_id}'{delegate_suffix}.") else: @@ -594,8 +586,7 @@ def _cmd_delete(db, args): return _not_found(args.session_id) # The user named this id directly so the delete is honored, but a pin is a # "keep" flag and silently destroying it is surprising — say so. - _get_session = getattr(db, "get_session", None) - _meta = (_get_session(resolved_session_id) or {}) if callable(_get_session) else {} + _meta = db.get_session(resolved_session_id) or {} _pinned_note = " (this session is PINNED)" if _meta.get("pinned") else "" if not args.yes: if not _confirm_prompt( @@ -671,18 +662,12 @@ def _note_pinned_skipped(db, filters, action): """Pinned sessions are excluded by default from bulk prune/archive (pin = durable keep). `prune --include-pinned` opts in; archive has no such flag, so archive always spares pinned rows. Tell the user how many were spared.""" - _count_matches = getattr(db, "count_prune_matches", None) - if not callable(_count_matches): - return _base = {k: v for k, v in filters.items() if k != "include_pinned"} - try: - skipped = max( - int(_count_matches(**_base, include_pinned=True)) - - int(_count_matches(**_base, include_pinned=False)), - 0, - ) - except TypeError: - return # a db double without include_pinned support — skip the note + skipped = max( + int(db.count_prune_matches(**_base, include_pinned=True)) + - int(db.count_prune_matches(**_base, include_pinned=False)), + 0, + ) if not skipped: return suffix = "" if skipped == 1 else "s" @@ -778,12 +763,10 @@ def _cmd_prune_or_archive(db, args, action): return if action == "prune": - count = db.prune_sessions(sessions_dir=_sessions_dir(), **filters) - print(f"Pruned {count} session(s).") + print(f"Pruned {db.prune_sessions(sessions_dir=_sessions_dir(), **filters)} session(s).") else: - count = db.archive_sessions(**filters) print( - f"Archived {count} session(s). They're hidden from listings " + f"Archived {db.archive_sessions(**filters)} session(s). They're hidden from listings " "but fully recoverable (nothing was deleted)." ) @@ -996,8 +979,6 @@ def _cmd_optimize_storage(db, args): # VACUUM needs a full second copy — require headroom ≈ current file size. do_vacuum = not getattr(args, "no_vacuum", False) try: - import shutil - free_bytes = shutil.disk_usage(db_path.parent).free except Exception: free_bytes = None @@ -1146,11 +1127,9 @@ def cmd_sessions(args, sessions_parser=None): print(f"Error: Could not open session database: {e}") return 1 - try: + with db: handler = _DB_HANDLERS.get(action) if handler is None: sessions_parser.print_help() return return handler(db, args) - finally: - db.close() 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 4/7] 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) From 542819c4687b9c05a126762f62b09ae9ec39a728 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:39:50 -0700 Subject: [PATCH 5/7] refactor(hermes_cli): compact sessions_cmd docstrings, comments and printer bodies --- hermes_cli/sessions_cmd.py | 188 +++++++++++++------------------------ 1 file changed, 64 insertions(+), 124 deletions(-) diff --git a/hermes_cli/sessions_cmd.py b/hermes_cli/sessions_cmd.py index c26e3dc5d0..1460cc55c7 100644 --- a/hermes_cli/sessions_cmd.py +++ b/hermes_cli/sessions_cmd.py @@ -1,11 +1,9 @@ """``hermes sessions`` command. -``cmd_sessions`` routes ``args.sessions_action`` through ``_PRE_DB_HANDLERS`` -(repair / recover / import — must run without opening ``SessionDB()``, which is -exactly what a malformed schema prevents) and ``_DB_HANDLERS`` (everything else, -sharing one ``SessionDB`` that is always closed afterwards). ``get_hermes_home`` -is resolved through ``hermes_cli.main`` at call time so monkeypatches there keep -working and imports stay one-way. Browse picker: :mod:`hermes_cli.sessions_cmd_browse`. +``cmd_sessions`` routes ``args.sessions_action`` through ``_PRE_DB_HANDLERS`` (repair / recover / +import — must run without opening ``SessionDB()``, which a malformed schema prevents) and +``_DB_HANDLERS`` (everything else, sharing one ``SessionDB``). ``get_hermes_home`` is resolved through +``hermes_cli.main`` at call time so monkeypatches keep working. Picker: :mod:`hermes_cli.sessions_cmd_browse`. """ import json @@ -34,12 +32,9 @@ def _size_mb(path) -> float: def _size_delta_label(saved_mb: float) -> str: - """Size delta label. A negative delta means the file GREW (concurrent writes - during a long optimize can outweigh what the rebuild freed); "reclaimed - -163.0 MB" reads as data loss, so say "grew by" instead.""" - if saved_mb >= 0: - return f"reclaimed {saved_mb:.1f} MB" - return f"grew by {-saved_mb:.1f} MB" + """A negative delta means the file GREW (concurrent writes during a long optimize); "reclaimed + -163.0 MB" reads as data loss, so say "grew by".""" + return f"reclaimed {saved_mb:.1f} MB" if saved_mb >= 0 else f"grew by {-saved_mb:.1f} MB" def _confirm_prompt(prompt: str) -> bool: @@ -86,9 +81,7 @@ def _write_output(output, text, summary) -> None: print(summary) -# --------------------------------------------------------------------------- -# Handlers that must run BEFORE SessionDB() is opened -# --------------------------------------------------------------------------- +# -- handlers that must run BEFORE SessionDB() is opened ---------------------- def _cmd_repair(args): from hermes_state import DEFAULT_DB_PATH, _db_opens_cleanly, repair_state_db_schema @@ -122,8 +115,7 @@ def _cmd_repair(args): if report.get("backup_path"): print(f" A backup is preserved at: {report['backup_path']}") print(" Keep state.db and the backup; do not delete them.") - # Without this pointer the user is at a dead end after a failed in-place - # repair. Lead with --inspect-only so they confirm readability before writing. + # Without this pointer the user is at a dead end; lead with --inspect-only before writing. print("") print(" Next step — offline recovery (never modifies the source):") source_hint = report.get("backup_path") or db_path @@ -136,8 +128,7 @@ def _cmd_repair(args): def _cmd_recover(args): - """Offline recovery: never opens the supplied source directly (works on a - disposable copy) and never touches the active database.""" + """Offline recovery: works on a disposable copy of the source; never touches the active database.""" import sqlite3 from hermes_cli.session_recovery import ( @@ -232,10 +223,8 @@ def _print_recovery_verdict(report, output, allow_partial) -> int: ) else: print(f"✓ Partial recovery output verified at: {output}") - print( - f" Recovered {int(counts.get('sessions') or 0):,} sessions and " - f"{int(counts.get('messages') or 0):,} messages." - ) + sessions_n, messages_n = int(counts.get("sessions") or 0), int(counts.get("messages") or 0) + print(f" Recovered {sessions_n:,} sessions and {messages_n:,} messages.") print(" The active session database was not changed.") print( " This output is incomplete. Review every skipped range and orphan count in the " @@ -249,15 +238,12 @@ def _print_recovery_verdict(report, output, allow_partial) -> int: def _cmd_import(args): from hermes_cli.foreign_sessions import run_sessions_import - # Explicit path but nothing imported (bad path, unknown source, no turns) → - # non-zero so scripts can detect it. Picker cancel (no path) → normal exit 0. + # Explicit path but nothing imported -> non-zero for scripts. Picker cancel (no path) -> exit 0. if run_sessions_import(args) is None and getattr(args, "path", None): return 1 -# --------------------------------------------------------------------------- -# Handlers that receive an open SessionDB -# --------------------------------------------------------------------------- +# -- handlers that receive an open SessionDB ---------------------------------- def _default_exclude(args): """Hide third-party tool sessions by default, but honour explicit --source.""" @@ -283,8 +269,7 @@ def _cmd_list(db, args): print("No sessions found.") return - # The Workspace column only appears once at least one session carries a - # workspace key (or when filtering), so all-unbound listings read as before. + # Workspace column only when some session carries a key (or when filtering): unbound listings read as before. has_ws = bool(_ws_filter) or any(_ws_key(s) for s in sessions) has_titles = any(s.get("title") for s in sessions) @@ -339,8 +324,7 @@ def _cmd_export(db, args): return redact_session_data(data) def _collect_sessions(): - """Resolve --session-id / filters / bare export into a list of - redacted session dicts, or None after printing an error.""" + """--session-id / filters / bare export -> redacted session dicts, or None after printing an error.""" if args.session_id: resolved = db.resolve_session_id(args.session_id) data = _redact(db.export_session(resolved)) if resolved else None @@ -413,11 +397,10 @@ def _export_jsonl(args, collect): def _export_trace(db, args, filters): - """Claude Code JSONL trace export — local file or HF upload. Redaction is ON - by default (traces leave the machine with --upload); --no-redact opts out.""" + """Claude Code JSONL trace export — local file or HF upload. Redaction is ON by default (traces + leave the machine with --upload); --no-redact opts out.""" session_id = args.session_id - if not session_id and not filters: - # Match the shell's common intent: "the last thing I did". + if not session_id and not filters: # "the last thing I did" rows = db.list_sessions_rich(limit=1, order_by_last_active=True) session_id = rows[0].get("id") if rows else None if not session_id: @@ -501,7 +484,6 @@ def _export_markdown(db, args, filters, redact): if args.delete_after_verified and not args.session_id: print("--delete-after-verified is only supported with --session-id.") return - lineage_is_logical = getattr(args, "lineage", "single") == "logical" if args.session_id: @@ -584,8 +566,7 @@ def _cmd_delete(db, args): resolved_session_id = db.resolve_session_id(args.session_id) if not resolved_session_id: return _not_found(args.session_id) - # The user named this id directly so the delete is honored, but a pin is a - # "keep" flag and silently destroying it is surprising — say so. + # The delete is honored (explicit id), but a pin is a "keep" flag: say so instead of silently destroying it. _meta = db.get_session(resolved_session_id) or {} _pinned_note = " (this session is PINNED)" if _meta.get("pinned") else "" if not args.yes: @@ -603,16 +584,14 @@ def _cmd_delete(db, args): return _not_found(args.session_id) -#: Age floor for `prune --never-active`. Generous on purpose: the rows are harmless, -#: and a young never-active row may simply be a chat nobody has replied to yet. +#: Age floor for `prune --never-active`; generous: a young never-active row may be a chat nobody replied to yet. _NEVER_ACTIVE_DEFAULT_DAYS = 30.0 def _prune_never_active_keyed(db, args): - """`prune --never-active`: drop keyed gateway rows opened and never used - (mostly escaped test fixtures). Separate from the shared prune/archive - selector, which is pinned to `ended_at IS NOT NULL`: never-closed rows sit - outside it by construction and cannot be expressed as one more filter.""" + """`prune --never-active`: drop keyed gateway rows opened and never used (mostly escaped test + fixtures). Separate from the shared prune/archive selector, which is pinned to `ended_at IS NOT + NULL` — never-closed rows sit outside it by construction.""" from hermes_cli.session_filters import format_epoch, parse_duration_seconds older_than = getattr(args, "older_than", None) if older_than is None: @@ -659,9 +638,8 @@ def _prune_never_active_keyed(db, args): def _note_pinned_skipped(db, filters, action): - """Pinned sessions are excluded by default from bulk prune/archive (pin = - durable keep). `prune --include-pinned` opts in; archive has no such flag, - so archive always spares pinned rows. Tell the user how many were spared.""" + """Tell the user how many pinned rows bulk prune/archive spared (pin = durable keep; only + `prune --include-pinned` opts in, archive always spares them).""" _base = {k: v for k, v in filters.items() if k != "include_pinned"} skipped = max( int(db.count_prune_matches(**_base, include_pinned=True)) @@ -689,10 +667,9 @@ def _cmd_prune_or_archive(db, args, action): return _prune_never_active_keyed(db, args) from hermes_cli.session_filters import build_prune_filters, describe_filters, format_epoch - # Bare `prune` (no time window, no filters) keeps the historical "older - # than 90 days" default. ANY filter — including --source — suppresses the - # implicit cutoff (`prune --source cron` matches ALL cron sessions); the - # preview + confirmation below is the safety net. + # Bare `prune` keeps the historical "older than 90 days" default. ANY filter — including --source — + # suppresses the implicit cutoff (`prune --source cron` matches ALL cron sessions); the preview + + # confirmation below is the safety net. if action == "prune" and not _any_filter_args(args, _TIME_FILTER_ARGS + _NON_TIME_FILTER_ARGS): args.older_than = "90" @@ -709,8 +686,7 @@ def _cmd_prune_or_archive(db, args, action): ) return - # Prune skips archived rows unless --include-archived; archive only targets - # not-yet-archived rows (idempotent). + # Prune skips archived rows unless --include-archived; archive only targets not-yet-archived rows. filters["archived"] = None if action == "prune" and getattr(args, "include_archived", False) else False filters["include_pinned"] = getattr(args, "include_pinned", False) @@ -718,8 +694,8 @@ def _cmd_prune_or_archive(db, args, action): _note_pinned_skipped(db, filters, action) candidates = db.list_prune_candidates(**filters) - # Archive expands each row to its compression lineage (may include open - # continuations), so a direct-open count would misdescribe its effect. + # Archive expands each row to its compression lineage (may include open continuations), so a + # direct-open count would misdescribe its effect. skipped_open = db.count_open_prune_matches(**filters) if action == "prune" else 0 if skipped_open: suffix = "" if skipped_open == 1 else "s" @@ -734,8 +710,8 @@ def _cmd_prune_or_archive(db, args, action): print(f"No sessions match ({describe_filters(filters)}).") return - # Candidates are ordered oldest-activity-first; show the span so a long-lived - # but recently used conversation cannot look old merely by creation date. + # Candidates are oldest-activity-first; show the span so a long-lived but recently used + # conversation cannot look old merely by creation date. _span = ( f"oldest activity {format_epoch(candidates[0].get('last_active'))}, " f"newest activity {format_epoch(candidates[-1].get('last_active'))}" @@ -778,8 +754,7 @@ def _cmd_rename(db, args): if not resolved_session_id: return _not_found(args.session_id) title = " ".join(args.title) - # Empty titles render as "—" and newlines corrupt the `list` table; length - # is validated in set_session_title. + # Empty titles render as "—" and newlines corrupt the `list` table; length is validated in set_session_title. if not title.strip(): print("Error: title cannot be empty or whitespace-only.") return 1 @@ -797,8 +772,7 @@ def _cmd_rename(db, args): def _cmd_pin(db, args, pinning): - """Durable "keep" flag (exempt from sessions.auto_archive, always listed); - every surface — GUI, TUI, CLI, scripts — reads/writes the same store.""" + """Durable "keep" flag (exempt from sessions.auto_archive, always listed); every surface shares the store.""" failures = 0 for raw_id in args.session_ids: resolved = db.resolve_session_id(raw_id) @@ -814,22 +788,12 @@ def _cmd_pin(db, args, pinning): def _cmd_pinned(db, args): - # limit=1 keeps the recency page minimal; include_pinned back-fills ALL - # pinned rows the page missed, so old pins can't fall off a paging window. + # limit=1 keeps the recency page minimal; include_pinned back-fills ALL pinned rows the page missed. rows = db.list_sessions_rich(limit=1, include_pinned=True, exclude_sources=_default_exclude(args)) pinned_rows = [s for s in rows if s.get("pinned")] if getattr(args, "json", False): - payload = [ - { - "id": s["id"], - "title": s.get("title"), - "source": s.get("source"), - "last_active": s.get("last_active"), - "message_count": s.get("message_count"), - } - for s in pinned_rows - ] - print(json.dumps(payload, indent=2)) + keys = ("title", "source", "last_active", "message_count") + print(json.dumps([{"id": s["id"], **{k: s.get(k) for k in keys}} for s in pinned_rows], indent=2)) return if not pinned_rows: print("No pinned sessions. Pin one with: hermes sessions pin ") @@ -849,9 +813,8 @@ def _cmd_retitle_skills(db, args): apply_changes = bool(getattr(args, "apply", False)) def _is_titlelike(candidate: str) -> bool: - """Reject non-titles: an auxiliary model occasionally answers the prompt - instead of titling it ('$ df -h /'). The live path takes what it gets, - but this is a REPAIR — never replace a serviceable title with that.""" + """Reject non-titles: an auxiliary model occasionally answers the prompt instead of titling it + ('$ df -h /'). This is a REPAIR — never replace a serviceable title with that.""" return bool(candidate) and candidate[0].isalnum() candidates = db.list_skill_scaffolded_sessions(limit=limit) @@ -879,8 +842,7 @@ def _cmd_retitle_skills(db, args): continue try: db.set_session_title(session_id, new_title) - except ValueError: - # Unique-title collision: dedupe like the live auto-titler (base #2, #3, ...). + except ValueError: # unique-title collision: dedupe like the live auto-titler (base #2, #3, ...) deduped = db.get_next_title_in_lineage(new_title) try: db.set_session_title(session_id, deduped) @@ -905,16 +867,13 @@ def _cmd_browse(db, args): print("No sessions found.") return - # Keep the DB open: the picker uses it for status tags and 'd' delete. - try: + try: # keep the DB open: the picker uses it for status tags and 'd' delete selected_id = _session_browse_picker(sessions, session_db=db) finally: db.close() if not selected_id: print("Cancelled.") return - - # Launch hermes --resume by replacing the current process print(f"Resuming session: {selected_id}") from hermes_cli.relaunch import relaunch relaunch(["--resume", selected_id]) # won't return after execvp @@ -923,26 +882,20 @@ def _cmd_browse(db, args): # -- storage maintenance ----------------------------------------------------- def _print_size_change(db, before_mb, prefix=""): - """Report before/after size, preferring SQLite's page accounting over stat(): - in WAL mode a VACUUM's rewrite sits in the -wal file until a checkpoint, - which is refused while a live gateway holds a read-mark, so the main file - lags and stat() can even go negative. page_count * page_size is correct now.""" - after_mb = _size_mb(db.db_path) + """Report before/after size, preferring SQLite's page accounting over stat(): in WAL mode a VACUUM's + rewrite sits in the -wal file until a checkpoint (refused while a live gateway holds a read-mark), + so the main file lags and stat() can even go negative.""" logical_after = db.logical_size_bytes() - if logical_after is not None: - after_mb = logical_after / (1024 * 1024) - print( - f"{prefix}Database size: {before_mb:.1f} MB -> {after_mb:.1f} MB " - f"({_size_delta_label(before_mb - after_mb)})" - ) + after_mb = logical_after / (1024 * 1024) if logical_after is not None else _size_mb(db.db_path) + delta = _size_delta_label(before_mb - after_mb) + print(f"{prefix}Database size: {before_mb:.1f} MB -> {after_mb:.1f} MB ({delta})") def _cmd_optimize(db, args): before_mb = _size_mb(db.db_path) print("Optimizing session store (FTS merge + VACUUM)…") try: - # vacuum() merges FTS5 segments then VACUUMs; returns indexes merged. - n = db.vacuum() + n = db.vacuum() # merges FTS5 segments then VACUUMs; returns indexes merged except Exception as e: print(f"Error: optimization failed: {e}") return @@ -951,10 +904,7 @@ def _cmd_optimize(db, args): def _cmd_clean_markers(db, args): - if args.dry_run: - print("Dry run — scanning for stale tool-call marker rows (#78148)…") - else: - print("Scanning for stale tool-call marker rows (#78148)…") + print(f"{'Dry run — scanning' if args.dry_run else 'Scanning'} for stale tool-call marker rows (#78148)…") report = db.purge_stale_tool_call_markers(dry_run=args.dry_run, backup=not args.no_backup) if report["rows_affected"] == 0: print("✓ No affected rows found — nothing to clean.") @@ -975,8 +925,8 @@ def _cmd_optimize_storage(db, args): before_bytes = os.path.getsize(db_path) if db_path.exists() else 0 before_mb = before_bytes / (1024 * 1024) - # Disk preflight: the new index is built before the old is torn down, and - # VACUUM needs a full second copy — require headroom ≈ current file size. + # Disk preflight: the new index is built before the old is torn down, and VACUUM needs a full + # second copy — require headroom ≈ current file size. do_vacuum = not getattr(args, "no_vacuum", False) try: free_bytes = shutil.disk_usage(db_path.parent).free @@ -987,18 +937,15 @@ def _cmd_optimize_storage(db, args): print(f" Current database size: {before_mb:.1f} MB") if free_bytes is not None: print(f" Free disk: {free_bytes / (1024*1024):.0f} MB " - f"(need ~{need_bytes / (1024*1024):.0f} MB to complete" - f"{' incl. VACUUM' if do_vacuum else ''})") + f"(need ~{need_bytes / (1024*1024):.0f} MB to complete{' incl. VACUUM' if do_vacuum else ''})") if free_bytes < need_bytes: print() - print("⚠ Not enough free disk to complete safely. Free up " - "space, or run with --no-vacuum (rebuilds the index " - "but doesn't reclaim space until a later VACUUM).") + print("⚠ Not enough free disk to complete safely. Free up space, or run with --no-vacuum " + "(rebuilds the index but doesn't reclaim space until a later VACUUM).") return if before_mb > 500: - print(" This may take a while on a large database. It runs in " - "the foreground with progress below; safe to Ctrl-C and " - "re-run (it resumes).") + print(" This may take a while on a large database. It runs in the foreground with progress below; " + "safe to Ctrl-C and re-run (it resumes).") if not getattr(args, "yes", False): try: resp = input("Proceed? [y/N] ").strip().lower() @@ -1012,16 +959,12 @@ def _cmd_optimize_storage(db, args): def _progress(info): phase = info.get("phase") - pct = info.get("percent", 0) if phase == "backfill": - print(f"\r Rebuilding index: {pct:3d}% " - f"({info.get('indexed',0):,}/{info.get('total',0):,})", - end="", flush=True) + print(f"\r Rebuilding index: {info.get('percent', 0):3d}% " + f"({info.get('indexed',0):,}/{info.get('total',0):,})", end="", flush=True) elif phase != _last["phase"]: - label = {"teardown": "Reclaiming old index", - "vacuum": "Compacting database (VACUUM)", - "done": "Done"}.get(phase, phase) - print(f"\n {label}…", flush=True) + label = {"teardown": "Reclaiming old index", "vacuum": "Compacting database (VACUUM)", "done": "Done"} + print(f"\n {label.get(phase, phase)}…", flush=True) _last["phase"] = phase print("Optimizing search-index storage…") @@ -1046,8 +989,7 @@ def _cmd_repair_routing(db, args): for record in records: print(f"{record['orphan_id']} ({record['source']}, {record['message_count']} messages)") if record["adoptable"]: - print(f" → adopt into {record['session_key']} " - f"(from {record['donor_id']}, " + print(f" → adopt into {record['session_key']} (from {record['donor_id']}, " f"evidence: {record['evidence']})") else: print(f" ✗ not repairable — {record['reason']}") @@ -1087,9 +1029,7 @@ def _cmd_stats(db, args): print(f"Database size: {_size_mb(db.db_path):.1f} MB") -# --------------------------------------------------------------------------- -# Dispatch -# --------------------------------------------------------------------------- +# -- dispatch ----------------------------------------------------------------- _PRE_DB_HANDLERS = {"repair": _cmd_repair, "recover": _cmd_recover, "import": _cmd_import} From b7b543b25345cf91b34454aec99f9b153c993af3 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:51:48 -0700 Subject: [PATCH 6/7] refactor(hermes_cli): session_lost_and_found reuses session_recovery helpers; compact docs and mapper branches --- hermes_cli/session_lost_and_found.py | 188 ++++++++------------------- 1 file changed, 53 insertions(+), 135 deletions(-) diff --git a/hermes_cli/session_lost_and_found.py b/hermes_cli/session_lost_and_found.py index 422f6c7af7..03a9bfcc1d 100644 --- a/hermes_cli/session_lost_and_found.py +++ b/hermes_cli/session_lost_and_found.py @@ -1,8 +1,6 @@ -"""Last-resort page-level salvage for an unreadable session database schema. - -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:: -""" +"""Last-resort page-level salvage for an unreadable session database schema, via the sqlite3 shell's +``.recover`` (rows it cannot attribute to a schema land in ``lost_and_found`` tables: +``rootpgno, pgno, nfield, id, c0..cN``).""" from __future__ import annotations @@ -14,8 +12,12 @@ import tempfile 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. +from hermes_cli.session_recovery import ( + _AUXILIARY_TABLE_SCHEMAS, _AUXILIARY_TABLES, _CANONICAL_TABLES, _immediate_transaction, _quoted_columns, + _table_columns, +) + +# Hermes session ids are timestamps (20260812_135332_ab12cd): the strongest sentinel for schema-less rows. SESSION_ID_PATTERN = re.compile(r"^\d{8}_\d{6}_") MESSAGE_ROLES = frozenset({"user", "assistant", "tool", "system"}) @@ -27,9 +29,7 @@ KNOWN_SOURCES = frozenset({ "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. +# Historical sessions layouts. Columns are only ever appended, so an older record is a strict prefix. SESSIONS_LAYOUT_NFIELDS = frozenset({55, 54, 52}) SESSIONS_LEGACY_MINIMAL_NFIELD = 14 SESSION_MODEL_USAGE_NFIELD = 18 @@ -56,22 +56,14 @@ class LostAndFoundError(RuntimeError): def find_sqlite3_cli() -> Optional[str]: - """Return a ``.recover``-capable sqlite3 CLI path, or None. - - PATH presence is not enough: distro builds can ship a shell without the ``sqlite_dbpage`` - virtual table ``.recover`` needs, failing every recovery. Probe on a scratch DB once instead - of discovering it mid-recovery. - """ - + """A ``.recover``-capable sqlite3 CLI path, or None. PATH presence is not enough: distro builds can + lack the ``sqlite_dbpage`` virtual table ``.recover`` needs, so probe on a scratch DB once.""" binary = shutil.which("sqlite3") - if binary is None: - return None - return binary if _cli_supports_recover(binary) else None + return binary if binary is not None and _cli_supports_recover(binary) else None def _cli_supports_recover(binary: str) -> bool: """True when ``binary`` can run ``.recover`` (has sqlite_dbpage).""" - scratch_dir = tempfile.mkdtemp(prefix="hermes-recover-probe-") scratch = Path(scratch_dir) / "probe.db" try: @@ -83,9 +75,7 @@ def _cli_supports_recover(binary: str) -> bool: finally: conn.close() probe = subprocess.run([binary, "-readonly", str(scratch), ".recover"], capture_output=True, timeout=30) - if probe.returncode != 0: - return False - return b"sqlite_dbpage" not in probe.stderr + return probe.returncode == 0 and b"sqlite_dbpage" not in probe.stderr except (OSError, subprocess.SubprocessError, sqlite3.Error): return False finally: @@ -100,21 +90,15 @@ def run_cli_lost_and_found_recover( timeout: float = 3600.0, ) -> dict[str, Any]: """Run ``sqlite3 .recover`` streamed into a fresh scratch DB.""" - 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, + [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, + [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 @@ -127,17 +111,14 @@ def run_cli_lost_and_found_recover( load.kill() raise LostAndFoundError(f"sqlite3 .recover timed out after {timeout:.0f}s") attempt = { - "command": command, - "dump_returncode": dump.returncode, - "load_returncode": load.returncode, + "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 + attempt["usable"] = _lost_and_found_db_usable(lf_path) + if attempt["usable"]: return {"binary": sqlite3_bin, "attempts": attempts} - attempt["usable"] = False raise LostAndFoundError( "sqlite3 .recover did not produce a usable lost_and_found database: " @@ -163,18 +144,10 @@ def _lost_and_found_db_usable(lf_path: Path) -> bool: return False -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 return NULLs where the live schema says NOT NULL (torn cells, old rows). - Dropping a whole row over one damaged optional counter would defeat the lane, so such NULLs - get the schema default (or '' / 0 when none is declared). - """ - + """Column index -> substitute for NOT NULL columns. Salvage can return NULLs where the schema says + NOT NULL (torn cells, old rows); dropping a row over one damaged counter would defeat the lane, so + such NULLs get the schema default (or '' / 0 when none is declared).""" substitutes: dict[int, Any] = {} for index, row in enumerate(conn.execute(f'PRAGMA table_info("{table}")')): if not row[3]: # notnull flag @@ -214,28 +187,19 @@ def classify_lost_and_found_row(nfield: int, cells: tuple[Any, ...]) -> Optional """Classify one lost_and_found record by field count + sentinel values.""" 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): + # Rowid-alias tables store their INTEGER PRIMARY KEY as NULL; 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 - second = cells[1] if len(cells) > 1 else None - if nfield == SESSION_MODEL_USAGE_NFIELD: - # 18 fields, session id first, model string second. + if nfield == SESSION_MODEL_USAGE_NFIELD: # session id first, model string second return "session_model_usage" if isinstance(second, str) and second else None - - # Known sessions layouts, or an unknown historical one (>= 30 fields): a - # session-id first cell plus a recognizable source string is still strong - # enough for a prefix map. + # Known sessions layouts, or an unknown historical one (>= 30 fields): session id + source is enough. if ( - nfield in SESSIONS_LAYOUT_NFIELDS - or nfield == SESSIONS_LEGACY_MINIMAL_NFIELD - or nfield >= 30 + nfield in SESSIONS_LAYOUT_NFIELDS or nfield == SESSIONS_LEGACY_MINIMAL_NFIELD or nfield >= 30 ) and _looks_like_source(second): return "sessions" return None @@ -257,41 +221,29 @@ def _insert_prefix_row( ) -> bool: if notnull_substitutes: values = [ - notnull_substitutes[index] - if value is None and index in notnull_substitutes - else value + 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) + quoted, placeholders = _quoted_columns(dest_columns[: len(values)]) 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.""" - - # Lazy import: session_recovery imports this module inside a function, so - # a module-level import here would be circular. - from hermes_cli.session_recovery import (_AUXILIARY_TABLE_SCHEMAS, _AUXILIARY_TABLES, _CANONICAL_TABLES) - copied: dict[str, int] = {} for table in (*_CANONICAL_TABLES, *_AUXILIARY_TABLES): source_columns = _table_columns(lf_conn, table) if not source_columns: continue dest_columns = _table_columns(dest, table) - if not dest_columns and table in _AUXILIARY_TABLE_SCHEMAS: - # Lazily-created gateway table: base SessionDB never made it on - # the fresh destination, so create it before copying. + if not dest_columns and table in _AUXILIARY_TABLE_SCHEMAS: # lazily-created gateway table _AUXILIARY_TABLE_SCHEMAS[table](dest) 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) + quoted, placeholders = _quoted_columns(columns) rows = lf_conn.execute(f'SELECT {quoted} FROM "{table}"').fetchall() if not rows: copied[table] = 0 @@ -305,24 +257,15 @@ def _copy_direct_tables(lf_conn: sqlite3.Connection, dest: sqlite3.Connection) - 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.""" - 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": [], + "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: + with _immediate_transaction(dest): report["direct_table_rows"] = _copy_direct_tables(lf_conn, dest) - # Per-kind destination columns + NOT NULL substitutes. 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 the substitutions only fill NOT NULL bookkeeping counters/flags. + # Per-kind destination columns + NOT NULL substitutes. Identity fields are never fabricated: + # rows with a NULL session id / role / source were already rejected by classify_lost_and_found_row. targets: dict[str, tuple[list[str], dict[int, Any]]] = {} for kind_name, protected in (("sessions", (0, 1)), ("messages", (1, 2)), ("session_model_usage", (0, 1))): defaults = _notnull_defaults(dest, kind_name) @@ -340,8 +283,7 @@ def map_lost_and_found_rows(lf_conn: sqlite3.Connection, dest: sqlite3.Connectio 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"]: + if _table_columns(lf_conn, lf_table)[:3] != ["rootpgno", "pgno", "nfield"]: continue for row in lf_conn.execute(f'SELECT * FROM "{lf_table}"'): try: @@ -357,37 +299,28 @@ def map_lost_and_found_rows(lf_conn: sqlite3.Connection, dest: sqlite3.Connectio continue columns, defaults = targets[kind] try: - if kind == "messages": - # Rowid-alias PK is NULL in the record; use the lost_and_found rowid. - values = [lf_rowid, *cells[1 : min(nfield, len(columns))]] - inserted = _insert_prefix_row(dest, kind, columns, values, defaults) - elif kind == "session_model_usage": - values = list(cells[: len(columns)]) - inserted = _insert_prefix_row(dest, kind, columns, values, 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( + if kind == "sessions" and nfield == SESSIONS_LEGACY_MINIMAL_NFIELD: + # Pre-modern layout with unknown column order: salvage identity + timing only. + inserted = ( 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", + cells[1] if _looks_like_source(cells[1]) else "recovered", _heuristic_started_at(cells), - "[best-effort recovered] legacy session " - "row (layout unknown)", + "[best-effort recovered] legacy session row (layout unknown)", ), ).rowcount == 1 ) - if inserted: - report["legacy_minimal_sessions"] += 1 + report["legacy_minimal_sessions"] += int(inserted) else: - values = list(cells[: min(nfield, len(columns))]) + # messages: the rowid-alias PK is NULL in the record; use the lost_and_found rowid. + values = list(cells[:len(columns)]) + if kind == "messages": + values[0] = lf_rowid inserted = _insert_prefix_row(dest, kind, columns, values, defaults) except sqlite3.DatabaseError: report["unmapped_rows"] += 1 @@ -396,24 +329,14 @@ def map_lost_and_found_rows(lf_conn: sqlite3.Connection, dest: sqlite3.Connectio report["mapped"][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. - """ - + """Fabricate clearly-marked placeholder parents for salvaged child rows: children (messages, + model-usage rows) are NEVER deleted for FK cleanup — a stub parent beats losing the only copy.""" result: dict[str, Any] = {"sessions_stubbed": 0, "messages_retained": 0, "usage_rows_retained": 0} - dest.execute("BEGIN IMMEDIATE") - try: + with _immediate_transaction(dest): 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 " @@ -435,9 +358,9 @@ def stub_missing_parent_sessions(dest: sqlite3.Connection) -> dict[str, Any]: sequence = 1 for session_id, info in sorted(orphan_ids.items()): while True: - title = (f"[best-effort recovered {sequence}] session metadata " "was unreadable") + 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): + 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, " @@ -461,16 +384,11 @@ def stub_missing_parent_sessions(dest: sqlite3.Connection) -> dict[str, Any]: "(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): From a0e05e686a7fc340366454e1c2d0e3627b14124b Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:56:55 -0700 Subject: [PATCH 7/7] refactor(hermes_cli): compact sessions_cmd_browse and session_recap --- hermes_cli/session_recap.py | 99 ++++++++----------------------- hermes_cli/sessions_cmd_browse.py | 45 +++++--------- 2 files changed, 39 insertions(+), 105 deletions(-) diff --git a/hermes_cli/session_recap.py b/hermes_cli/session_recap.py index 32d7553b7f..399a41824e 100644 --- a/hermes_cli/session_recap.py +++ b/hermes_cli/session_recap.py @@ -1,8 +1,5 @@ -"""Session recap — summarize what's happened in the current session. - -Differences from Claude Code: - Pure local computation from the in-memory conversation history. No -LLM call, no auxiliary model, no prompt-cache invalidation. A recap should be instant and free. -""" +"""Session recap — summarize the current session from in-memory history. Pure local computation: no +LLM call, no auxiliary model, no prompt-cache invalidation. A recap should be instant and free.""" from __future__ import annotations import json @@ -12,35 +9,20 @@ from typing import Any, Iterable, List, Mapping, Optional, Sequence, Tuple from tools.ansi_strip import sanitize_display_text -# How many recent user/assistant turns we consider "recent activity". -_RECENT_TURN_WINDOW = 20 - -# How many characters of the latest user prompt to show. +_RECENT_TURN_WINDOW = 20 # user/assistant turns considered "recent activity" _PROMPT_PREVIEW_CHARS = 140 - -# How many characters of the latest assistant text to show. _ASSISTANT_PREVIEW_CHARS = 200 - -# How many recently-touched files to list. _MAX_FILES_LISTED = 5 -# Tool names that identify a file-editing action and the argument key that -# holds the path. +# File-touching tool name -> argument key holding the path. _FILE_EDIT_TOOLS: Mapping[str, str] = { - "write_file": "path", - "patch": "path", - "read_file": "path", - "skill_manage": "file_path", - "skill_view": "file_path", + "write_file": "path", "patch": "path", "read_file": "path", + "skill_manage": "file_path", "skill_view": "file_path", } def _coerce_text(value: Any) -> str: - """Flatten assistant/user ``content`` into a plain string. - - Content may be a string or a list of blocks (multimodal/reasoning models); text-like blocks - are concatenated and the rest ignored. - """ + """Flatten ``content`` (string, or list of blocks whose text-like parts are joined) into a string.""" if value is None: return "" if isinstance(value, str): @@ -59,14 +41,8 @@ def _coerce_text(value: Any) -> str: def _tool_call_name_and_args(tool_call: Any) -> Tuple[str, Mapping[str, Any]]: - """Extract ``(name, arguments_dict)`` from a tool_call entry. - - ``arguments`` may be a JSON string or a dict depending on provider. Return an empty dict if it - cannot be parsed. - """ - if not isinstance(tool_call, Mapping): - return "", {} - fn = tool_call.get("function") or {} + """``(name, arguments_dict)`` from a tool_call; ``arguments`` may be a JSON string or dict ({} if unparsable).""" + fn = tool_call.get("function") if isinstance(tool_call, Mapping) else None if not isinstance(fn, Mapping): return "", {} name = str(fn.get("name") or "") @@ -81,9 +57,7 @@ def _tool_call_name_and_args(tool_call: Any) -> Tuple[str, Mapping[str, Any]]: def _iter_assistant_tool_calls(messages: Sequence[Mapping[str, Any]]) -> Iterable[Tuple[str, Mapping[str, Any]]]: for msg in messages: - if not isinstance(msg, Mapping): - continue - if msg.get("role") != "assistant": + if not isinstance(msg, Mapping) or msg.get("role") != "assistant": continue tool_calls = msg.get("tool_calls") or [] if not isinstance(tool_calls, list): @@ -113,21 +87,15 @@ def _latest_text(messages: Sequence[Mapping[str, Any]], role: str) -> Optional[s def _recent_window( messages: Sequence[Mapping[str, Any]], window: int = _RECENT_TURN_WINDOW ) -> List[Mapping[str, Any]]: - """Return the tail slice of ``messages`` covering at most ``window`` user+assistant turns (tool - messages ride along inside the window). - """ + """Tail slice covering at most ``window`` user+assistant turns (tool messages ride along).""" count = 0 - cut = 0 for i in range(len(messages) - 1, -1, -1): msg = messages[i] if isinstance(msg, Mapping) and msg.get("role") in {"user", "assistant"}: count += 1 if count >= window: - cut = i - break - else: - return list(messages) - return list(messages[cut:]) + return list(messages[i:]) + return list(messages) def _shortened_path(path: str) -> str: @@ -152,16 +120,11 @@ def _shortened_path(path: str) -> str: def _summarise_tool_activity( tool_calls: Sequence[Tuple[str, Mapping[str, Any]]], ) -> Tuple[List[Tuple[str, int]], List[str]]: - """Return ``(tool_counts_sorted, recently_edited_files)``. - - Counts are descending and kept in full so callers truncate for display; files are distinct - paths, most recent first, from file-editing tools. - """ + """``(tool_counts_sorted_desc, recently_edited_files)`` — files are distinct paths, newest first.""" counter: Counter[str] = Counter() files_seen: List[str] = [] files_set: set[str] = set() - # Walk in reverse so files_seen comes out newest→oldest (Counter ignores order). - for name, args in reversed(list(tool_calls)): + for name, args in reversed(list(tool_calls)): # reversed so files_seen comes out newest -> oldest counter[name] += 1 arg_key = _FILE_EDIT_TOOLS.get(name) if arg_key: @@ -181,14 +144,10 @@ def _join_capped(items: List[str], limit: int) -> str: def _truncate(text: str, limit: int) -> str: - # Stored history is untrusted for display — remove escape sequences and - # control chars so a recap line can't clear the screen / retitle the - # window when echoed to a terminal (openai/codex#31494 bug class). - text = sanitize_display_text(text) - text = " ".join(text.split()) # collapse newlines for a compact one-liner - if len(text) <= limit: - return text - return text[: limit - 1].rstrip() + "…" + # Stored history is untrusted for display: strip escapes/control chars so a recap line can't clear + # the screen or retitle the window when echoed to a terminal. + text = " ".join(sanitize_display_text(text).split()) # collapse newlines for a compact one-liner + return text if len(text) <= limit else text[: limit - 1].rstrip() + "…" def build_recap( @@ -198,20 +157,13 @@ def build_recap( session_id: Optional[str] = None, platform: Optional[str] = None, ) -> str: - """Build a multi-line recap of recent activity from chat-completion-style ``messages``. - - ``platform`` is accepted for forward compat but does not change behavior. Output is plain - text that renders well both in an 80-col terminal and in a gateway message bubble. - """ - _ = platform # reserved for future use - lines: List[str] = [] - - header_bits: List[str] = ["Session recap"] + """Multi-line plain-text recap of recent activity (80-col terminal / gateway bubble friendly). + ``platform`` is accepted for forward compat and does not change behavior.""" + lines: List[str] = ["Session recap"] if session_title: - header_bits.append(f"— {session_title}") + lines[0] += f" — {session_title}" elif session_id: - header_bits.append(f"— {session_id[:8]}") - lines.append(" ".join(header_bits)) + lines[0] += f" — {session_id[:8]}" if not messages: lines.append(" (nothing to recap — no messages yet)") @@ -245,8 +197,7 @@ def build_recap( if latest_reply: lines.append(f" Last reply: {_truncate(latest_reply, _ASSISTANT_PREVIEW_CHARS)}") - if len(lines) == 2: - # Only the header + scope line — nothing substantive to show. + if len(lines) == 2: # only header + scope line: nothing substantive to show lines.append(" (no assistant activity yet in this window)") return "\n".join(lines) diff --git a/hermes_cli/sessions_cmd_browse.py b/hermes_cli/sessions_cmd_browse.py index fb014f6097..c34c31715f 100644 --- a/hermes_cli/sessions_cmd_browse.py +++ b/hermes_cli/sessions_cmd_browse.py @@ -1,8 +1,5 @@ -"""Interactive picker for ``hermes sessions browse`` (extracted from sessions_cmd). - -Curses UI with live search filtering and ``d`` delete-with-confirmation; a -numbered-list fallback when curses is unavailable (Windows, etc.). -""" +"""Interactive picker for ``hermes sessions browse``: curses UI with live search filtering and ``d`` +delete-with-confirmation; numbered-list fallback when curses is unavailable (Windows, etc.).""" from typing import Optional @@ -15,8 +12,7 @@ def _session_status_tag(status: Optional[str]) -> str: def _annotate_session_statuses(sessions: list, session_db) -> None: - """Attach a ``_status`` key to each row via one indexed lookup per session - (never a transcript scan). On failure rows stay untagged and render '-'.""" + """Attach ``_status`` per row via one indexed lookup (never a transcript scan); on failure rows render '-'.""" if session_db is None or not sessions: return try: @@ -29,7 +25,7 @@ def _annotate_session_statuses(sessions: list, session_db) -> None: def _label(s: dict) -> str: """Title, else preview, else id.""" - return ((s.get("title") or "").strip() or (s.get("preview") or "").strip() or s["id"]) + return (s.get("title") or "").strip() or (s.get("preview") or "").strip() or s["id"] def _msgs_str(s: dict) -> str: @@ -48,18 +44,14 @@ def _match(s: dict, query: str) -> bool: ) -# Layout: [arrow 3] [title/preview flexible] [status 5] [msgs 5] -# [active 12] [src 6] [id 18] +# Layout: [arrow 3] [title/preview flexible] [status 5] [msgs 5] [active 12] [src 6] [id 18] _FIXED_COLS = 3 + 5 + 2 + 5 + 2 + 12 + 6 + 18 + 6 def _format_row(s: dict, max_x: int) -> str: - """Format a session row for display.""" name_width = max(20, max_x - _FIXED_COLS) - title = (s.get("title") or "").strip() - preview = (s.get("preview") or "").strip() sid = s["id"][:18] - name = (title or preview)[:name_width] or sid + name = ((s.get("title") or "").strip() or (s.get("preview") or "").strip())[:name_width] or sid return ( f"{name:<{name_width}} {_session_status_tag(s.get('_status')):<5} " f"{_msgs_str(s):>5} {_relative_time(s.get('last_active')):<10} " @@ -71,12 +63,9 @@ class _CursesBrowser: """State + render loop for the curses picker. ``run`` is the wrapper target.""" def __init__(self, curses, sessions, delete_fn): - self.curses = curses - self.sessions = sessions - self.delete_fn = delete_fn # None => no delete support + self.curses, self.sessions, self.delete_fn = curses, sessions, delete_fn # delete_fn None => no delete self.result = None - self.cursor = 0 - self.scroll = 0 + self.cursor = self.scroll = 0 self.search = "" self.confirm_delete = None # session dict pending y/n confirmation self.flash = "" # one-frame notice (e.g. "Deleted.") @@ -97,10 +86,9 @@ class _CursesBrowser: pass def _refilter(self, reset_cursor=True): - self.filtered = ([s for s in self.sessions if _match(s, self.search)] if self.search else list(self.sessions)) + self.filtered = [s for s in self.sessions if _match(s, self.search)] if self.search else list(self.sessions) if reset_cursor: - self.cursor = 0 - self.scroll = 0 + self.cursor = self.scroll = 0 def _draw(self, stdscr, max_y, max_x): c = self.curses @@ -163,8 +151,7 @@ class _CursesBrowser: def _handle_key(self, key) -> bool: """Apply one keypress; return True when the picker should exit.""" c = self.curses - if self.confirm_delete is not None: - # y/n confirmation mode — only an explicit 'y' deletes. + if self.confirm_delete is not None: # y/n confirmation mode — only an explicit 'y' deletes target, self.confirm_delete = self.confirm_delete, None if key in {ord("y"), ord("Y")}: if self.delete_fn(target["id"]): @@ -261,14 +248,10 @@ def _fallback_picker(sessions: list) -> Optional[str]: def _session_browse_picker(sessions: list, session_db=None) -> Optional[str]: - """Interactive curses-based session browser with live search filtering. + """Curses session browser with live search; returns the selected session ID, or None if cancelled. - Shows lifecycle status (done / intr / err / empty) and message count per - session when *session_db* is provided. With a live *session_db*, pressing - ``d`` on a row (while the search filter is empty) prompts y/n and deletes - the session via ``SessionDB.delete_session``. - - Returns the selected session ID, or None if cancelled. + With *session_db*: shows lifecycle status / message count per row, and ``d`` (while the filter is + empty) prompts y/n and deletes via ``SessionDB.delete_session``. """ if not sessions: print("No sessions found.")