ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 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 21:59:06 -07:00
parent d41d56b748 a0e05e686a
commit d2bc08bbfc
5 changed files with 590 additions and 1080 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/session_lost_and_found.py
+66 -206
View File
@@ -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:
@@ -82,14 +74,8 @@ 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,
)
if probe.returncode != 0:
return False
return b"sqlite_dbpage" not in probe.stderr
probe = subprocess.run([binary, "-readonly", str(scratch), ".recover"], capture_output=True, timeout=30)
return probe.returncode == 0 and b"sqlite_dbpage" not in probe.stderr
except (OSError, subprocess.SubprocessError, sqlite3.Error):
return False
finally:
@@ -104,21 +90,15 @@ def run_cli_lost_and_found_recover(
timeout: float = 3600.0,
) -> dict[str, Any]:
"""Run ``sqlite3 <source> .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
@@ -129,21 +109,16 @@ 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,
"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: "
@@ -162,27 +137,17 @@ 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:
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
@@ -218,40 +183,23 @@ 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
):
# 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
@@ -273,94 +221,53 @@ 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)
cursor = dest.execute(
f'INSERT OR IGNORE INTO "{table}" ({quoted}) VALUES ({placeholders})',
values,
)
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]:
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
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] = {
"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)),
):
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)
@@ -376,8 +283,7 @@ def map_lost_and_found_rows(
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:
@@ -393,37 +299,28 @@ def map_lost_and_found_rows(
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
@@ -432,28 +329,14 @@ def map_lost_and_found_rows(
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.
"""
result: dict[str, Any] = {
"sessions_stubbed": 0,
"messages_retained": 0,
"usage_rows_retained": 0,
}
dest.execute("BEGIN IMMEDIATE")
try:
"""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}
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 "
@@ -470,42 +353,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(
@@ -519,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):
hermes_cli/session_recap.py
+27 -80
View File
@@ -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 "")
@@ -79,13 +55,9 @@ 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
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):
@@ -96,9 +68,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"]
@@ -117,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:
@@ -156,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:
@@ -185,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(
@@ -202,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)")
@@ -249,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)
hermes_cli/session_recovery.py
+402 -601
View File
File diff suppressed because it is too large Load Diff
hermes_cli/sessions_cmd.py
+81 -162
View File
@@ -1,15 +1,14 @@
"""``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
import os
import shutil
import sys
from functools import partial
from pathlib import Path
@@ -33,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:
@@ -85,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
@@ -111,12 +105,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
@@ -124,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
@@ -138,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 (
@@ -201,8 +190,7 @@ def _cmd_recover(args):
class _RecoveryProgress:
"""`recover` progress printer: one live-updating `` <table>: n/total`` line per table."""
def __init__(self):
self.table = None
table = None
def __call__(self, info):
table = info.get("table")
@@ -235,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 "
@@ -252,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."""
@@ -286,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)
@@ -342,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
@@ -396,14 +377,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):
@@ -419,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:
@@ -507,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:
@@ -577,9 +553,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:
@@ -592,10 +566,8 @@ 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.
_get_session = getattr(db, "get_session", None)
_meta = (_get_session(resolved_session_id) or {}) if callable(_get_session) else {}
# 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:
if not _confirm_prompt(
@@ -612,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:
@@ -668,21 +638,14 @@ 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."""
_count_matches = getattr(db, "count_prune_matches", None)
if not callable(_count_matches):
return
"""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"}
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"
@@ -704,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"
@@ -724,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)
@@ -733,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"
@@ -749,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,12 +739,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)."
)
@@ -795,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
@@ -814,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)
@@ -831,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 <session_id>")
@@ -866,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)
@@ -896,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)
@@ -922,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 <id> 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
@@ -940,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
@@ -968,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.")
@@ -992,12 +925,10 @@ 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:
import shutil
free_bytes = shutil.disk_usage(db_path.parent).free
except Exception:
free_bytes = None
@@ -1006,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()
@@ -1031,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…")
@@ -1065,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']}")
@@ -1106,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}
@@ -1146,11 +1067,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()
hermes_cli/sessions_cmd_browse.py
+14 -31
View File
@@ -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.")