ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-23-g2' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:42:13 -07:00
parent d86ffe80d4 0bcade22e0
commit d5c07a8037
4 changed files with 195 additions and 408 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/session_lost_and_found.py
+23 -47
View File
@@ -13,8 +13,8 @@ from pathlib import Path
from typing import Any, Optional
from hermes_cli.session_recovery import (
_AUXILIARY_TABLE_SCHEMAS, _AUXILIARY_TABLES, _CANONICAL_TABLES, _immediate_transaction, _quoted_columns,
_table_columns,
_AUXILIARY_TABLE_SCHEMAS, _AUXILIARY_TABLES, _CANONICAL_TABLES, _immediate_transaction, _placeholder_titles,
_quoted_columns, _table_columns,
)
# Hermes session ids are timestamps (20260812_135332_ab12cd): the strongest sentinel for schema-less rows.
@@ -39,15 +39,12 @@ _EPOCH_LOW = 1_000_000_000.0 # 2001
_EPOCH_HIGH = 4_000_000_000.0 # 2096
SQLITE3_CLI_GUIDANCE = (
"A last-resort page-level salvage is available when a `.recover`-capable "
"`sqlite3` command-line shell is installed: its `.recover` command can "
"rebuild rows into lost_and_found tables even when the table schemas are "
"unreadable (this is a CLI-only feature, not part of Python's sqlite3 "
"module, and some distro builds lack it — the shell must include the "
"sqlite_dbpage extension, as the official builds from sqlite.org do). "
"Install such a sqlite3 CLI (e.g. `brew install sqlite` or the "
"precompiled sqlite-tools from sqlite.org) so it is on PATH, then re-run "
"with --allow-partial."
"A last-resort page-level salvage is available when a `.recover`-capable `sqlite3` command-line shell is "
"installed: its `.recover` command can rebuild rows into lost_and_found tables even when the table schemas are "
"unreadable (this is a CLI-only feature, not part of Python's sqlite3 module, and some distro builds lack it — "
"the shell must include the sqlite_dbpage extension, as the official builds from sqlite.org do). Install such a "
"sqlite3 CLI (e.g. `brew install sqlite` or the precompiled sqlite-tools from sqlite.org) so it is on PATH, then "
"re-run with --allow-partial."
)
@@ -83,11 +80,7 @@ def _cli_supports_recover(binary: str) -> bool:
def run_cli_lost_and_found_recover(
source: Path,
lf_path: Path,
sqlite3_bin: str,
*,
timeout: float = 3600.0,
source: Path, lf_path: Path, sqlite3_bin: str, *, timeout: float = 3600.0,
) -> dict[str, Any]:
"""Run ``sqlite3 <source> .recover`` streamed into a fresh scratch DB."""
attempts: list[dict[str, Any]] = []
@@ -213,10 +206,7 @@ def _heuristic_started_at(cells: tuple[Any, ...]) -> float:
def _insert_prefix_row(
dest: sqlite3.Connection,
table: str,
dest_columns: list[str],
values: list[Any],
dest: sqlite3.Connection, table: str, dest_columns: list[str], values: list[Any],
notnull_substitutes: Optional[dict[int, Any]] = None,
) -> bool:
if notnull_substitutes:
@@ -276,8 +266,7 @@ def map_lost_and_found_rows(lf_conn: sqlite3.Connection, dest: sqlite3.Connectio
lf_tables = [
str(row[0])
for row in lf_conn.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name LIKE 'lost_and_found%'"
"SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'lost_and_found%'"
)
]
report["lost_and_found_tables"] = lf_tables
@@ -303,9 +292,7 @@ def map_lost_and_found_rows(lf_conn: sqlite3.Connection, dest: sqlite3.Connectio
# 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 (?, ?, ?, ?)",
"INSERT OR IGNORE INTO sessions (id, source, started_at, title) VALUES (?, ?, ?, ?)",
(
cells[0],
cells[1] if _looks_like_source(cells[1]) else "recovered",
@@ -339,32 +326,24 @@ def stub_missing_parent_sessions(dest: sqlite3.Connection) -> dict[str, Any]:
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 "
"WHERE m.session_id IS NOT NULL AND NOT EXISTS "
"(SELECT 1 FROM sessions WHERE sessions.id = m.session_id) "
"GROUP BY m.session_id"
"SELECT m.session_id, MIN(m.timestamp), COUNT(*) FROM messages AS m WHERE m.session_id IS NOT NULL AND "
"NOT EXISTS (SELECT 1 FROM sessions WHERE sessions.id = m.session_id) GROUP BY m.session_id"
):
orphan_ids[str(session_id)] = {
"started_at": float(first_ts) if first_ts is not None else 0.0,
"message_count": int(count),
}
for (session_id,) in dest.execute(
"SELECT DISTINCT u.session_id FROM session_model_usage AS u "
"WHERE u.session_id IS NOT NULL AND NOT EXISTS "
"(SELECT 1 FROM sessions WHERE sessions.id = u.session_id)"
"SELECT DISTINCT u.session_id FROM session_model_usage AS u WHERE u.session_id IS NOT NULL AND NOT "
"EXISTS (SELECT 1 FROM sessions WHERE sessions.id = u.session_id)"
):
orphan_ids.setdefault(str(session_id), {"started_at": 0.0, "message_count": 0})
sequence = 1
titles = _placeholder_titles(dest, "best-effort recovered")
for session_id, info in sorted(orphan_ids.items()):
while True:
title = f"[best-effort recovered {sequence}] session metadata was unreadable"
sequence += 1
if dest.execute("SELECT 1 FROM sessions WHERE title = ? LIMIT 1", (title,)).fetchone() is None:
break
title = next(titles)
dest.execute(
"INSERT INTO sessions (id, source, started_at, title, "
"message_count) VALUES (?, 'recovered', ?, ?, ?)",
"INSERT INTO sessions (id, source, started_at, title, message_count) VALUES (?, 'recovered', ?, ?, ?)",
(session_id, info["started_at"], title, info["message_count"]),
)
result["sessions_stubbed"] += 1
@@ -374,15 +353,12 @@ def stub_missing_parent_sessions(dest: sqlite3.Connection) -> dict[str, Any]:
# Repair dangling intra-sessions references without deleting rows.
dest.execute(
"UPDATE sessions SET parent_session_id = NULL "
"WHERE parent_session_id IS NOT NULL AND NOT EXISTS "
"(SELECT 1 FROM sessions AS p WHERE p.id = sessions.parent_session_id)"
"UPDATE sessions SET parent_session_id = NULL WHERE parent_session_id IS NOT NULL AND NOT EXISTS (SELECT "
"1 FROM sessions AS p WHERE p.id = sessions.parent_session_id)"
)
dest.execute(
"UPDATE sessions SET system_prompt_hash = NULL "
"WHERE system_prompt_hash IS NOT NULL AND NOT EXISTS "
"(SELECT 1 FROM system_prompts "
"WHERE system_prompts.hash = sessions.system_prompt_hash)"
"UPDATE sessions SET system_prompt_hash = NULL WHERE system_prompt_hash IS NOT NULL AND NOT EXISTS "
"(SELECT 1 FROM system_prompts WHERE system_prompts.hash = sessions.system_prompt_hash)"
)
return result
hermes_cli/session_recap.py
+1 -4
View File
@@ -151,10 +151,7 @@ def _truncate(text: str, limit: int) -> str:
def build_recap(
messages: Sequence[Mapping[str, Any]],
*,
session_title: Optional[str] = None,
session_id: Optional[str] = None,
messages: Sequence[Mapping[str, Any]], *, session_title: Optional[str] = None, session_id: Optional[str] = None,
platform: Optional[str] = None,
) -> str:
"""Multi-line plain-text recap of recent activity (80-col terminal / gateway bubble friendly).
hermes_cli/session_recovery.py
+100 -255
View File
@@ -81,9 +81,7 @@ def _resolved_output_path(path: Path) -> Path:
def _validate_paths(
source_path: Path,
output_path: Optional[Path] = None,
work_dir: Optional[Path] = None,
source_path: Path, output_path: Optional[Path] = None, work_dir: Optional[Path] = None,
) -> tuple[Path, Optional[Path], Path]:
source = source_path.expanduser().resolve(strict=True)
if not source.is_file():
@@ -95,8 +93,7 @@ def _validate_paths(
protected = {_sidecar_path(source, suffix).resolve(strict=False) for suffix in _SIDECAR_SUFFIXES}
if output.resolve(strict=False) in protected:
raise SessionRecoverySafetyError(
"The recovery output must not be the source database or one of "
"its journal sidecars."
"The recovery output must not be the source database or one of its journal sidecars."
)
for suffix in _SIDECAR_SUFFIXES:
candidate = _sidecar_path(output, suffix)
@@ -161,17 +158,13 @@ def _disk_space_preflight(source: Path, work_root: Path, output_parent: Optional
if output_parent is None or _same_filesystem(work_root, output_parent):
required = bundle_bytes + output_allowance + headroom
report["shared_filesystem"] = True
report["work_dir_required_bytes"] = required
report.update(shared_filesystem=True, work_dir_required_bytes=required)
if work_free < required:
raise SessionRecoverySafetyError(
"Not enough free disk space for a safe recovery copy: "
f"{_format_bytes(work_free)} available at {work_root}, "
f"{_format_bytes(required)} required "
f"({_format_bytes(bundle_bytes)} source bundle + "
f"{_format_bytes(output_allowance)} output allowance + "
f"{_format_bytes(headroom)} headroom). Use --work-dir or "
"--output on a filesystem with more free space."
f"Not enough free disk space for a safe recovery copy: {_format_bytes(work_free)} available at "
f"{work_root}, {_format_bytes(required)} required ({_format_bytes(bundle_bytes)} source bundle + "
f"{_format_bytes(output_allowance)} output allowance + {_format_bytes(headroom)} headroom). "
"Use --work-dir or --output on a filesystem with more free space."
)
return report
@@ -248,10 +241,7 @@ def _immediate_transaction(conn: sqlite3.Connection) -> Iterator[None]:
def _compatible_columns(
source: sqlite3.Connection,
destination: sqlite3.Connection,
table: str,
result: dict[str, Any],
source: sqlite3.Connection, destination: sqlite3.Connection, table: str, result: dict[str, Any],
) -> Optional[list[str]]:
"""Columns shared by source and destination; sets a terminal status and returns None otherwise."""
source_columns = _table_columns(source, table)
@@ -273,17 +263,9 @@ def _quoted_columns(columns: list[str]) -> tuple[str, str]:
def _copy_rows(
source: sqlite3.Connection,
destination: sqlite3.Connection,
select_sql: str,
params: tuple[Any, ...],
insert_sql: str,
*,
table: str,
chunk_size: int,
progress_cb: Optional[ProgressCallback],
expected_rows: Optional[int],
result: dict[str, Any],
source: sqlite3.Connection, destination: sqlite3.Connection, select_sql: str, params: tuple[Any, ...],
insert_sql: str, *, table: str, chunk_size: int, progress_cb: Optional[ProgressCallback],
expected_rows: Optional[int], result: dict[str, Any],
) -> dict[str, Any]:
"""Chunked straight copy; fills ``status``/``error`` on ``result``."""
try:
@@ -351,8 +333,7 @@ def _inspect_connection(conn: sqlite3.Connection) -> dict[str, Any]:
def _snapshot_and_inspect(
source: Path,
work_root: Path,
source: Path, 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))
@@ -360,16 +341,12 @@ def _snapshot_and_inspect(
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. "
"This includes the interactive `hermes` CLI session this "
"command may have been launched from: a running parent CLI "
"writes session bookkeeping (compression ticks, context "
"tracking) to state.db in the background and counts as a "
"Hermes process even after the gateway is stopped. Run the "
"recovery from a fresh shell with no `hermes` session open, "
"or point --source at an immutable snapshot copy of the "
"database."
"The source database bundle changed while it was being copied. Stop every Hermes process using this "
"profile and retry. This includes the interactive `hermes` CLI session this command may have been "
"launched from: a running parent CLI writes session bookkeeping (compression ticks, context "
"tracking) to state.db in the background and counts as a Hermes process even after the gateway is "
"stopped. Run the recovery from a fresh shell with no `hermes` session open, or point --source at an "
"immutable snapshot copy of the database."
)
conn = _connect(snapshot_source)
@@ -377,8 +354,7 @@ def _snapshot_and_inspect(
inspection = _inspect_connection(conn)
finally:
conn.close()
inspection["source_bundle"] = copied
inspection["source_fingerprint"] = before
inspection.update(source_bundle=copied, source_fingerprint=before)
return temp_dir, snapshot_source, inspection
except BaseException:
temp_dir.cleanup()
@@ -425,14 +401,8 @@ def _fresh_destination(output: Path, *, topic_tables: bool = False) -> sqlite3.C
def _copy_table(
source: sqlite3.Connection,
destination: sqlite3.Connection,
table: str,
*,
salvage: bool,
chunk_size: int,
progress_cb: Optional[ProgressCallback],
source_rows: Optional[int],
source: sqlite3.Connection, destination: sqlite3.Connection, table: str, *, salvage: bool, chunk_size: int,
progress_cb: Optional[ProgressCallback], 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)
@@ -446,16 +416,9 @@ def _copy_table(
return result
quoted, placeholders = _quoted_columns(columns)
return _copy_rows(
source,
destination,
f'SELECT {quoted} FROM "{table}"',
(),
f'INSERT INTO "{table}" ({quoted}) VALUES ({placeholders})',
table=table,
chunk_size=chunk_size,
progress_cb=progress_cb,
expected_rows=source_rows,
result=result,
source, destination, f'SELECT {quoted} FROM "{table}"', (),
f'INSERT INTO "{table}" ({quoted}) VALUES ({placeholders})', table=table, chunk_size=chunk_size,
progress_cb=progress_cb, expected_rows=source_rows, result=result,
)
@@ -544,18 +507,9 @@ 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],
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
@@ -628,9 +582,7 @@ class _RowidRangeSalvage:
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,
"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:
@@ -647,14 +599,8 @@ class _RowidRangeSalvage:
def _copy_table_salvage(
source: sqlite3.Connection,
destination: sqlite3.Connection,
table: str,
*,
chunk_size: int,
progress_cb: Optional[ProgressCallback],
source_rows: Optional[int],
insert_prefix: str = "INSERT",
source: sqlite3.Connection, destination: sqlite3.Connection, table: str, *, chunk_size: int,
progress_cb: Optional[ProgressCallback], source_rows: Optional[int], insert_prefix: str = "INSERT",
row_filter: Optional[Callable[[tuple[Any, ...], tuple[str, ...]], bool]] = None,
) -> dict[str, Any]:
"""Best-effort rowid-range copy that continues past damaged source pages."""
@@ -672,24 +618,14 @@ def _copy_table_salvage(
result["status"] = "complete"
return result
if bounds.get("low") is None or bounds.get("high") is None:
result["status"] = "failed"
details = "; ".join(bounds.get("errors") or [])
result["error"] = "could not determine a rowid range for salvage"
if details:
result["error"] += f": {details}"
result["status"] = "failed"
result["error"] = "could not determine a rowid range for salvage" + (f": {details}" if details else "")
return result
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,
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,
)
salvage.copy_range(int(bounds["low"]), int(bounds["high"]))
skipped_ranges = result["skipped_rowid_ranges"]
@@ -699,12 +635,10 @@ 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 "
f"{result['excluded_rows']} of {source_rows} source rows"
)
copied, excluded = result["copied_rows"], result["excluded_rows"]
result["error"] = f"copied {copied} and excluded {excluded} of {source_rows} source rows"
else:
result["status"] = "complete"
return result
@@ -712,20 +646,13 @@ def _copy_table_salvage(
def _state_meta_result(source_rows: Optional[int], **extra: Any) -> dict[str, Any]:
return {
"source_meta_rows": source_rows,
"copied_rows": 0,
"columns": ["key", "value"],
"excluded_keys": sorted(_GENERATED_META_KEYS),
**extra,
"source_meta_rows": source_rows, "copied_rows": 0, "columns": ["key", "value"],
"excluded_keys": sorted(_GENERATED_META_KEYS), **extra,
}
def _state_meta_precheck(
source: sqlite3.Connection,
destination: sqlite3.Connection,
source_rows: Optional[int],
*,
salvage: bool,
source: sqlite3.Connection, destination: sqlite3.Connection, source_rows: Optional[int], *, salvage: bool,
) -> Optional[dict[str, Any]]:
"""Terminal ``state_meta`` result when the key/value schema is unusable, else ``None``.
@@ -737,31 +664,19 @@ def _state_meta_precheck(
source_columns = _table_columns(source, "state_meta")
if not {"key", "value"}.issubset(source_columns):
if salvage and source_columns: # present but unusable: real data loss
return _state_meta_result(
source_rows,
**extra,
status="failed",
error=(
"source state_meta exists but is missing the key/value "
f"columns (found: {', '.join(source_columns) or 'none'})"
),
)
found = ", ".join(source_columns) or "none"
error = f"source state_meta exists but is missing the key/value columns (found: {found})"
return _state_meta_result(source_rows, **extra, status="failed", error=error)
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"
)
error = "destination state_meta schema is incomplete"
return _state_meta_result(source_rows, **extra, status="failed", error=error)
return None
def _copy_state_meta(
source: sqlite3.Connection,
destination: sqlite3.Connection,
*,
salvage: bool,
chunk_size: int,
progress_cb: Optional[ProgressCallback],
source_rows: Optional[int],
source: sqlite3.Connection, destination: sqlite3.Connection, *, salvage: bool, chunk_size: int,
progress_cb: Optional[ProgressCallback], source_rows: Optional[int],
) -> dict[str, Any]:
"""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)
@@ -773,14 +688,8 @@ def _copy_state_meta(
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,
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)
@@ -797,19 +706,22 @@ def _copy_state_meta(
pass # the copy loop below will return the concrete read error
return _copy_rows(
source,
destination,
f"SELECT key, value FROM state_meta WHERE key NOT IN ({placeholders})",
params,
"INSERT OR REPLACE INTO state_meta(key, value) VALUES (?, ?)",
table="state_meta",
chunk_size=chunk_size,
progress_cb=progress_cb,
expected_rows=filtered_source_rows,
result=_state_meta_result(source_rows),
source, destination, f"SELECT key, value FROM state_meta WHERE key NOT IN ({placeholders})", params,
"INSERT OR REPLACE INTO state_meta(key, value) VALUES (?, ?)", table="state_meta", chunk_size=chunk_size,
progress_cb=progress_cb, expected_rows=filtered_source_rows, result=_state_meta_result(source_rows),
)
def _placeholder_titles(destination: sqlite3.Connection, prefix: str) -> Iterator[str]:
"""Yield ``[<prefix> N] session metadata was unreadable`` titles not yet present in ``sessions``."""
sequence = 1
while True:
title = f"[{prefix} {sequence}] session metadata was unreadable"
sequence += 1
if destination.execute("SELECT 1 FROM sessions WHERE title = ? LIMIT 1", (title,)).fetchone() is None:
yield title
def _reconstruct_missing_sessions(destination: sqlite3.Connection) -> dict[str, Any]:
"""Recreate placeholder session rows for salvaged orphaned messages.
@@ -822,29 +734,18 @@ def _reconstruct_missing_sessions(destination: sqlite3.Connection) -> dict[str,
return result
orphaned = destination.execute(
"SELECT m.session_id, MIN(m.timestamp), COUNT(*) "
"FROM messages AS m "
"WHERE m.session_id IS NOT NULL AND NOT EXISTS ("
"SELECT 1 FROM sessions WHERE sessions.id = m.session_id) "
"GROUP BY m.session_id"
"SELECT m.session_id, MIN(m.timestamp), COUNT(*) FROM messages AS m WHERE m.session_id IS NOT NULL AND NOT "
"EXISTS (SELECT 1 FROM sessions WHERE sessions.id = m.session_id) GROUP BY m.session_id"
).fetchall()
if not orphaned:
return result
title_sequence = 1
titles = _placeholder_titles(destination, "recovered")
for session_id, first_timestamp, message_count in orphaned:
started_at = float(first_timestamp) if first_timestamp is not None else 0.0
while True:
title = f"[recovered {title_sequence}] session metadata was unreadable"
title_sequence += 1
taken = destination.execute("SELECT 1 FROM sessions WHERE title = ? LIMIT 1", (title,)).fetchone()
if taken is None:
break
title = next(titles)
cursor = destination.execute(
"INSERT INTO sessions "
"(id, source, started_at, title, message_count) "
"VALUES (?, 'recovered', ?, ?, ?)",
"INSERT INTO sessions (id, source, started_at, title, message_count) VALUES (?, 'recovered', ?, ?, ?)",
(session_id, started_at, title, int(message_count)),
)
if cursor.rowcount != 1:
@@ -934,17 +835,13 @@ def _verify_structure(conn: sqlite3.Connection, verification: dict[str, Any]) ->
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()
}
meta = dict(conn.execute("SELECT key, value FROM state_meta WHERE key LIKE 'fts_%'").fetchall())
meta = {str(key): value for key, value in meta.items()}
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
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:
@@ -967,13 +864,8 @@ def _verify_fts_indexes(conn: sqlite3.Connection, verification: dict[str, Any])
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,
orphan_cleanup: Optional[dict[str, Any]],
conn: sqlite3.Connection, verification: dict[str, Any], *, expected_counts: dict[str, Optional[int]],
copy_report: dict[str, dict[str, Any]], allow_partial: bool, orphan_cleanup: Optional[dict[str, Any]],
) -> None:
"""Compare recovered counts and copy statuses against the source; classify shortfalls as loss."""
@@ -985,10 +877,7 @@ def _verify_row_counts(
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)
counts = {table: _count_rows(conn, table) for table in _INVENTORY_TABLES if _table_columns(conn, table)}
verification["table_counts"] = counts
for table in ("sessions", "messages", *_AUXILIARY_TABLES):
@@ -1003,25 +892,18 @@ def _verify_row_counts(
# 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
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)
),
tolerable = (
status == "partial" or table not in {"sessions", "messages"}
or (table == "sessions" and sessions_fully_reconstructed)
)
flag(f"{table} copy status is {status}", soft=allow_partial and tolerable)
if orphan_cleanup:
orphan_count = int(orphan_cleanup.get("total_removed_or_relinked") or 0)
@@ -1030,21 +912,15 @@ def _verify_row_counts(
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",
f"{rebuilt_sessions} session(s) could not be salvaged and were reconstructed as placeholders to "
f"retain {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,
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}
@@ -1057,11 +933,7 @@ def _verify_recovered_database(
try:
_verify_structure(conn, verification)
_verify_row_counts(
conn,
verification,
expected_counts=expected_counts,
copy_report=copy_report,
allow_partial=allow_partial,
conn, verification, expected_counts=expected_counts, copy_report=copy_report, allow_partial=allow_partial,
orphan_cleanup=orphan_cleanup,
)
_verify_fts_indexes(conn, verification)
@@ -1080,8 +952,7 @@ def _finalize_derived_metadata(destination: sqlite3.Connection) -> dict[str, Any
fts_tables = {
str(row[0])
for row in destination.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name IN ('messages_fts', 'messages_fts_trigram')"
"SELECT name FROM sqlite_master WHERE type='table' AND name IN ('messages_fts', 'messages_fts_trigram')"
).fetchall()
}
result: dict[str, Any] = {"fts_tables": sorted(fts_tables), "finalized": False}
@@ -1094,8 +965,7 @@ def _finalize_derived_metadata(destination: sqlite3.Connection) -> dict[str, Any
with _immediate_transaction(destination):
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",
"INSERT INTO state_meta(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
("fts_storage_version", str(FTS_STORAGE_VERSION)),
)
result["finalized"] = True
@@ -1103,14 +973,8 @@ def _finalize_derived_metadata(destination: sqlite3.Connection) -> dict[str, Any
def _recover_via_lost_and_found(
*,
source: Path,
snapshot_source: Path,
snapshot_dir: Path,
output: Path,
inspection: dict[str, Any],
disk_space: dict[str, Any],
missing_required: list[str],
*, source: Path, snapshot_source: Path, snapshot_dir: Path, output: Path, inspection: dict[str, Any],
disk_space: dict[str, Any], missing_required: list[str],
) -> dict[str, Any]:
"""Best-effort page-level salvage when table schemas are unreadable: the sqlite3 CLI's ``.recover``
(shell-only, not in Python's ``sqlite3``) rebuilds rows into a scratch lost_and_found database which
@@ -1120,22 +984,19 @@ def _recover_via_lost_and_found(
run_cli_lost_and_found_recover, stub_missing_parent_sessions,
)
missing = ", ".join(missing_required)
sqlite3_bin = find_sqlite3_cli()
if sqlite3_bin is None:
raise SessionRecoverySourceError(
"Partial recovery still requires readable table schemas for: "
+ ", ".join(missing_required)
+ ". " + SQLITE3_CLI_GUIDANCE
f"Partial recovery still requires readable table schemas for: {missing}. {SQLITE3_CLI_GUIDANCE}"
)
lf_path = snapshot_dir / "lost_and_found.db"
try:
cli_report = run_cli_lost_and_found_recover(snapshot_source, lf_path, sqlite3_bin)
except (LostAndFoundError, OSError) as exc:
raise SessionRecoverySourceError(
"Partial recovery could not read the table schemas for: "
+ ", ".join(missing_required)
+ f", and page-level .recover salvage failed: {exc}"
f"Partial recovery could not read the table schemas for: {missing}, "
f"and page-level .recover salvage failed: {exc}"
) from exc
lf_conn = sqlite3.connect(str(lf_path), isolation_level=None)
@@ -1164,17 +1025,13 @@ def _recover_via_lost_and_found(
"messages_removed": 0, "total_removed_or_relinked": 0,
}
verification = _verify_recovered_database(
output,
expected_counts={"sessions": None, "messages": None},
copy_report=copy_report,
allow_partial=True,
output, expected_counts={"sessions": None, "messages": None}, copy_report=copy_report, allow_partial=True,
orphan_cleanup=orphan_cleanup,
)
verification["loss_detected"] = True
verification["warnings"].append(
"BEST-EFFORT page-level salvage: the source table schemas were "
"unreadable, so rows were rebuilt from raw pages and mapped "
"heuristically. Review every count before trusting this output."
"BEST-EFFORT page-level salvage: the source table schemas were unreadable, so rows were rebuilt from raw "
"pages and mapped heuristically. Review every count before trusting this output."
)
verification["complete"] = False
@@ -1187,14 +1044,8 @@ def _recover_via_lost_and_found(
def _recovery_report(
source: Path,
output: Path,
inspection: dict[str, Any],
disk_space: dict[str, Any],
verification: dict[str, Any],
*,
on_source_change: str,
**fields: Any,
source: Path, output: Path, inspection: dict[str, Any], disk_space: dict[str, Any], verification: dict[str, Any],
*, 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 also clears ``verification[on_source_change]``.
@@ -1225,13 +1076,8 @@ def _recovery_report(
def recover_session_database(
source_path: Path,
output_path: Path,
*,
work_dir: Optional[Path] = None,
chunk_size: int = 1_000,
progress_cb: Optional[ProgressCallback] = None,
allow_partial: bool = False,
source_path: Path, output_path: Path, *, work_dir: Optional[Path] = None, chunk_size: int = 1_000,
progress_cb: Optional[ProgressCallback] = None, allow_partial: bool = False,
) -> dict[str, Any]:
"""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."""
@@ -1266,8 +1112,7 @@ def recover_session_database(
destination_conn: Optional[sqlite3.Connection] = None
try:
destination_conn = _fresh_destination(
output,
topic_tables=any(inspection["tables"][table].get("available") for table in _TOPIC_TABLES),
output, topic_tables=any(inspection["tables"][table].get("available") for table in _TOPIC_TABLES),
)
copy_report: dict[str, dict[str, Any]] = {}
hermes_cli/sessions_cmd.py
+71 -102
View File
@@ -71,6 +71,11 @@ def _any_filter_args(args, names) -> bool:
return any(getattr(args, a, None) is not None for a in names)
def _export_dir(output) -> Path:
"""``--output`` dir for multi-file exports; ``~/.hermes/session-exports`` when empty or ``-``."""
return Path(output).expanduser() if output and output != "-" else get_hermes_home() / "session-exports"
def _write_output(output, text, summary) -> None:
"""Write to stdout when *output* is empty or ``-``; else to the file + print *summary*."""
if not output or output == "-":
@@ -116,15 +121,16 @@ def _cmd_repair(args):
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; 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
print(f" hermes sessions recover --source {source_hint} \\")
print(" --inspect-only")
print(" If that reports the data is recoverable, rebuild it into")
print(" a NEW database (the active one is left untouched):")
print(f" hermes sessions recover --source {source_hint} \\")
print(" --output recovered-state.db")
print(
"\n Next step — offline recovery (never modifies the source):\n"
f" hermes sessions recover --source {source_hint} \\\n"
" --inspect-only\n"
" If that reports the data is recoverable, rebuild it into\n"
" a NEW database (the active one is left untouched):\n"
f" hermes sessions recover --source {source_hint} \\\n"
" --output recovered-state.db"
)
def _cmd_recover(args):
@@ -134,8 +140,7 @@ def _cmd_recover(args):
from hermes_cli.session_recovery import (
SessionRecoveryError, inspect_session_database, recover_session_database, write_recovery_report,
)
source = args.source
output = getattr(args, "output", None)
source, output = args.source, getattr(args, "output", None)
inspect_only = bool(getattr(args, "inspect_only", False))
allow_partial = bool(getattr(args, "allow_partial", False))
report_path = getattr(args, "report", None)
@@ -209,30 +214,34 @@ class _RecoveryProgress:
def _print_recovery_verdict(report, output, allow_partial) -> int:
if report.get("complete"):
print(f"✓ Recovered database verified at: {output}")
print(" The active session database was not changed.")
print(" Review the JSON report before installing this database.")
print(
f"✓ Recovered database verified at: {output}\n"
" The active session database was not changed.\n"
" Review the JSON report before installing this database."
)
return 0
if allow_partial and report.get("verified"):
counts = report.get("verification", {}).get("table_counts", {})
if report.get("best_effort"):
print(f"✓ BEST-EFFORT page-level salvage verified at: {output}")
print(
f"✓ BEST-EFFORT page-level salvage verified at: {output}\n"
" The source table schemas were unreadable; rows were rebuilt from raw pages "
"via sqlite3 .recover and mapped heuristically."
)
else:
print(f"✓ Partial recovery output verified at: {output}")
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(
f" Recovered {sessions_n:,} sessions and {messages_n:,} messages.\n"
" The active session database was not changed.\n"
" This output is incomplete. Review every skipped range and orphan count in the "
"JSON report before installing it."
)
return 0
print("✗ Recovery output did not pass every verification check.")
print(" Do not install it. Review the JSON report for partial data or errors.")
print(
"✗ Recovery output did not pass every verification check.\n"
" Do not install it. Review the JSON report for partial data or errors."
)
return 1
@@ -277,14 +286,9 @@ def _cmd_list(db, args):
key = _ws_key(s)
return ((os.path.basename(key.rstrip("/\\")) or key) if key else "—")[:16]
def _title(s, n):
return (s.get("title") or "—")[:n]
def _preview(s, n):
return s.get("preview", "")[:n]
def _ago(s):
return _relative_time(s.get("last_active"))
_title = lambda s, n: (s.get("title") or "—")[:n] # noqa: E731
_preview = lambda s, n: s.get("preview", "")[:n] # noqa: E731
_ago = lambda s: _relative_time(s.get("last_active")) # noqa: E731
layouts = { # (has_ws, has_titles): header, rule width, row formatter
(True, True): (f"{'Title':<28} {'Workspace':<18} {'Last Active':<13} {'ID'}", 110,
@@ -345,12 +349,10 @@ def _cmd_export(db, args):
if getattr(args, "only", None):
return _export_only(args, _collect_sessions)
if args.format == "html":
return _export_html(args, _collect_sessions)
if args.format == "trace":
return _export_trace(db, args, filters)
if args.format == "jsonl":
return _export_jsonl(args, _collect_sessions)
if args.format in ("html", "jsonl"):
return (_export_html if args.format == "html" else _export_jsonl)(args, _collect_sessions)
return _export_markdown(db, args, filters, _redact)
@@ -379,9 +381,8 @@ def _export_html(args, collect):
return
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)
print(f"Exported {len(sessions)} session{'' if single else 's'} to {args.output} (HTML)")
noun = "session" if single else "sessions"
_write_output(args.output, content, f"Exported {len(sessions)} {noun} to {args.output} (HTML)")
def _export_jsonl(args, collect):
@@ -438,7 +439,6 @@ def _export_trace(db, args, filters):
return None
return build_trace_jsonl(messages, session_id=sid, model=meta.get("model") or "", cwd="", redact=redact_trace)
to_stdout = not args.output or args.output == "-"
try:
if len(ids) == 1:
jsonl = _render_trace(ids[0])
@@ -447,7 +447,7 @@ def _export_trace(db, args, filters):
return
_write_output(args.output, jsonl, f"Exported 1 session trace to {args.output}")
else:
out_dir = (get_hermes_home() / "session-exports" if to_stdout else Path(args.output).expanduser())
out_dir = _export_dir(args.output)
out_dir.mkdir(parents=True, exist_ok=True)
exported = 0
for sid in ids:
@@ -467,7 +467,7 @@ def _export_markdown(db, args, filters, redact):
if args.output == "-":
print("Markdown/QMD export writes files; stdout (-) is only supported with --format jsonl.")
return
output_dir = Path(args.output).expanduser() if args.output else get_hermes_home() / "session-exports"
output_dir = _export_dir(args.output)
def _export_one(session_id: str, *, include_lineage: bool = False):
data = (db.export_session_lineage(session_id) if include_lineage else db.export_session(session_id))
@@ -526,8 +526,7 @@ def _export_markdown_single(db, args, export_one, output_dir, lineage_is_logical
for target_id in delete_target_ids:
try:
data, exported_path = export_one(
target_id,
include_lineage=(target_id == resolved_session_id and lineage_is_logical),
target_id, include_lineage=(target_id == resolved_session_id and lineage_is_logical),
)
except FileExistsError as e:
print(f"Export already exists: {e}. Pass --force to overwrite.")
@@ -539,10 +538,9 @@ def _export_markdown_single(db, args, export_one, output_dir, lineage_is_logical
message_count = sum(len(data.get("messages") or []) for data, _path in exported_items)
suffix = "" if message_count == 1 else "s"
if len(exported_items) == 1:
print(f"Exported 1 session ({message_count} message{suffix}) to {exported_items[0][1]}")
else:
print(f"Exported {len(exported_items)} sessions ({message_count} message{suffix}) to {output_dir}")
n = len(exported_items)
where = exported_items[0][1] if n == 1 else output_dir
print(f"Exported {n} session{'' if n == 1 else 's'} ({message_count} message{suffix}) to {where}")
if not args.delete_after_verified:
return
for data, exported_path in exported_items:
@@ -570,18 +568,14 @@ def _cmd_delete(db, args):
_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(
f"Delete session '{resolved_session_id}'{_pinned_note} "
"and all its messages? [y/N] "
):
if not _confirm_prompt(f"Delete session '{resolved_session_id}'{_pinned_note} and all its messages? [y/N] "):
print("Cancelled.")
return
elif _pinned_note:
print(f"Warning: deleting a pinned session '{resolved_session_id}'.")
if db.delete_session(resolved_session_id, sessions_dir=_sessions_dir()):
print(f"Deleted session '{resolved_session_id}'.")
else:
if not db.delete_session(resolved_session_id, sessions_dir=_sessions_dir()):
return _not_found(args.session_id)
print(f"Deleted session '{resolved_session_id}'.")
#: Age floor for `prune --never-active`; generous: a young never-active row may be a chat nobody replied to yet.
@@ -594,9 +588,8 @@ def _prune_never_active_keyed(db, args):
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:
days = _NEVER_ACTIVE_DEFAULT_DAYS
else:
days = _NEVER_ACTIVE_DEFAULT_DAYS
if older_than is not None:
seconds = parse_duration_seconds(str(older_than))
if seconds is None:
print(
@@ -612,15 +605,11 @@ def _prune_never_active_keyed(db, args):
return
shown = candidates if args.dry_run else candidates[:15]
print(
f"{len(candidates)} never-active keyed session(s) older than "
f"{days:g} day(s) — no messages, tokens, tool calls or title:"
)
print(f"{len(candidates)} never-active keyed session(s) older than {days:g} day(s) "
"— no messages, tokens, tool calls or title:")
for s in shown:
print(
f" {s['id']} {format_epoch(s.get('started_at')):<17} "
f"{(s.get('source') or '-'):<10} {s.get('session_key') or '-'}"
)
print(f" {s['id']} {format_epoch(s.get('started_at')):<17} {(s.get('source') or '-'):<10} "
f"{s.get('session_key') or '-'}")
if len(candidates) > len(shown):
print(f" … {len(candidates) - len(shown)} more")
@@ -650,16 +639,12 @@ def _note_pinned_skipped(db, filters, action):
return
suffix = "" if skipped == 1 else "s"
if action == "prune":
verb, optin = "deleted", (
"Pass --include-pinned to delete them anyway, or unpin "
"first with `hermes sessions unpin <id>`."
)
verb = "deleted"
optin = "Pass --include-pinned to delete them anyway, or unpin first with `hermes sessions unpin <id>`."
else:
verb, optin = "archived", "Unpin first with `hermes sessions unpin <id>` to include them."
print(
f"Note: {skipped} pinned session{suffix} also match these filters but "
f"will NOT be {verb} (pin is a keep flag). {optin}"
)
print(f"Note: {skipped} pinned session{suffix} also match these filters but will NOT be {verb} "
f"(pin is a keep flag). {optin}")
def _cmd_prune_or_archive(db, args, action):
@@ -680,10 +665,8 @@ def _cmd_prune_or_archive(db, args, action):
return 1
if action == "archive" and not any(v for k, v in filters.items() if k != "older_than_days"):
print(
"Refusing to archive every ended session: pass at least one "
"filter (e.g. --newer-than 5h, --source cli, --title codex)."
)
print("Refusing to archive every ended session: pass at least one "
"filter (e.g. --newer-than 5h, --source cli, --title codex).")
return
# Prune skips archived rows unless --include-archived; archive only targets not-yet-archived rows.
@@ -698,11 +681,9 @@ def _cmd_prune_or_archive(db, args, action):
# 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"
print(
f"Note: {skipped_open} open session{suffix} also match these "
"filters but will be skipped because prune only deletes ended "
"sessions. Use `hermes sessions delete <id>` "
f"Note: {skipped_open} open session{'' if skipped_open == 1 else 's'} also match these filters but "
"will be skipped because prune only deletes ended sessions. Use `hermes sessions delete <id>` "
"to remove one explicitly."
)
verb = "Delete" if action == "prune" else "Archive"
@@ -724,8 +705,7 @@ def _cmd_prune_or_archive(db, args, action):
title = (s.get("title") or "")[:36]
model = (s.get("model") or "-").split("/")[-1][:24]
print(
f" {s['id']} {format_epoch(s.get('last_active')):<17} "
f"{s['source']:<10} {model:<24} "
f" {s['id']} {format_epoch(s.get('last_active')):<17} {s['source']:<10} {model:<24} "
f"{s['message_count']:>4} msgs {title}"
)
if len(candidates) > len(shown):
@@ -741,10 +721,8 @@ def _cmd_prune_or_archive(db, args, action):
if action == "prune":
print(f"Pruned {db.prune_sessions(sessions_dir=_sessions_dir(), **filters)} session(s).")
else:
print(
f"Archived {db.archive_sessions(**filters)} session(s). They're hidden from listings "
"but fully recoverable (nothing was deleted)."
)
print(f"Archived {db.archive_sessions(**filters)} session(s). They're hidden from listings "
"but fully recoverable (nothing was deleted).")
# -- titles / pins -----------------------------------------------------------
@@ -762,13 +740,12 @@ def _cmd_rename(db, args):
print("Error: title cannot contain newlines.")
return 1
try:
if db.set_session_title(resolved_session_id, title):
print(f"Session '{resolved_session_id}' renamed to: {title}")
else:
if not db.set_session_title(resolved_session_id, title):
return _not_found(args.session_id)
except ValueError as e:
print(f"Error: {e}")
return 1
print(f"Session '{resolved_session_id}' renamed to: {title}")
def _cmd_pin(db, args, pinning):
@@ -777,10 +754,8 @@ def _cmd_pin(db, args, pinning):
for raw_id in args.session_ids:
resolved = db.resolve_session_id(raw_id)
if resolved and db.set_session_pinned(resolved, pinning):
verb = "Pinned" if pinning else "Unpinned"
title = db.get_session_title(resolved)
suffix = f" ({title})" if title else ""
print(f"{verb} session '{resolved}'.{suffix}")
print(f"{'Pinned' if pinning else 'Unpinned'} session '{resolved}'.{f' ({title})' if title else ''}")
else:
failures += _not_found(raw_id)
if failures:
@@ -798,12 +773,10 @@ def _cmd_pinned(db, args):
if not pinned_rows:
print("No pinned sessions. Pin one with: hermes sessions pin <session_id>")
return
print(f"{'Title':<32} {'Last Active':<13} {'Src':<9} {'ID'}")
print("─" * 100)
print(f"{'Title':<32} {'Last Active':<13} {'Src':<9} {'ID'}\n" + "─" * 100)
for s in pinned_rows:
title = (s.get("title") or s.get("preview", "") or "—")[:30]
last_active = _relative_time(s.get("last_active"))
print(f"{title:<32} {last_active:<13} {(s.get('source') or '-'):<9} {s['id']}")
print(f"{title:<32} {_relative_time(s.get('last_active')):<13} {(s.get('source') or '-'):<9} {s['id']}")
def _cmd_retitle_skills(db, args):
@@ -822,10 +795,8 @@ def _cmd_retitle_skills(db, args):
print("No sessions were titled from a /skill invocation.")
return
print(
f"{len(candidates)} session(s) opened with a /skill"
f"{'' if apply_changes else ' (dry run — pass --apply to write)'}:"
)
mode = "" if apply_changes else " (dry run — pass --apply to write)"
print(f"{len(candidates)} session(s) opened with a /skill{mode}:")
changed = 0
for row in candidates:
session_id = row["id"]
@@ -921,7 +892,6 @@ def _cmd_optimize_storage(db, args):
if not db.fts_optimize_available():
print("Search index is already on the compact layout — nothing to do.")
return
before_bytes = os.path.getsize(db_path) if db_path.exists() else 0
before_mb = before_bytes / (1024 * 1024)
@@ -1001,8 +971,8 @@ def _cmd_repair_routing(db, args):
print(f"\n{len(records)} orphaned session(s) found, none unambiguously repairable. Nothing to do.")
return
if not getattr(args, "apply", False):
print(f"\n{len(adoptable)} of {len(records)} orphaned session(s) "
"can be repaired. Re-run with --apply to perform them.")
print(f"\n{len(adoptable)} of {len(records)} orphaned session(s) can be repaired. "
"Re-run with --apply to perform them.")
return
print("\nStop the gateway before applying — a running gateway still holds the old routing mapping in memory.")
if not _confirm_prompt(f"Adopt {len(adoptable)} orphaned session(s)? [y/N] "):
@@ -1019,9 +989,8 @@ def _cmd_repair_routing(db, args):
def _cmd_stats(db, args):
print(f"Total sessions: {db.session_count()}")
print(f"Total messages: {db.message_count()}")
for src in ["cli", "telegram", "discord", "whatsapp", "slack"]:
print(f"Total sessions: {db.session_count()}\nTotal messages: {db.message_count()}")
for src in ("cli", "telegram", "discord", "whatsapp", "slack"):
c = db.session_count(source=src)
if c > 0:
print(f" {src}: {c} sessions")