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

refactor(hermes_cli): collapse defensive layers in sessions_cmd, join bracket groups in session recovery modules

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:12:14 -07:00
parent 2a9ab0dc7f
commit 86d111d77c
4 changed files with 62 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
hermes_cli/session_lost_and_found.py
View File

@@ -82,11 +82,7 @@ def _cli_supports_recover(binary: str) -> bool:
conn.commit()
finally:
conn.close()
probe = subprocess.run(
[binary, "-readonly", str(scratch), ".recover"],
capture_output=True,
timeout=30,
)
probe = subprocess.run([binary, "-readonly", str(scratch), ".recover"], capture_output=True, timeout=30)
if probe.returncode != 0:
return False
return b"sqlite_dbpage" not in probe.stderr
@@ -129,9 +125,7 @@ def run_cli_lost_and_found_recover(
except subprocess.TimeoutExpired:
dump.kill()
load.kill()
raise LostAndFoundError(
f"sqlite3 .recover timed out after {timeout:.0f}s"
)
raise LostAndFoundError(f"sqlite3 .recover timed out after {timeout:.0f}s")
attempt = {
"command": command,
"dump_returncode": dump.returncode,
@@ -162,9 +156,7 @@ def _lost_and_found_db_usable(lf_path: Path) -> bool:
try:
conn = sqlite3.connect(str(lf_path))
try:
return conn.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' LIMIT 1"
).fetchone() is not None
return conn.execute("SELECT 1 FROM sqlite_master WHERE type='table' LIMIT 1").fetchone() is not None
finally:
conn.close()
except sqlite3.DatabaseError:
@@ -218,22 +210,14 @@ def _looks_like_source(value: Any) -> bool:
return value in KNOWN_SOURCES or bool(re.fullmatch(r"[a-z][a-z0-9_-]{0,31}", value))
def classify_lost_and_found_row(
nfield: int,
cells: tuple[Any, ...],
) -> Optional[str]:
def classify_lost_and_found_row(nfield: int, cells: tuple[Any, ...]) -> Optional[str]:
"""Classify one lost_and_found record by field count + sentinel values."""
if len(cells) >= 3 and cells[0] is None:
# Rowid-alias tables store their INTEGER PRIMARY KEY as NULL in the
# record; messages is the only canonical table shaped like that with
# a session id second and a role third.
if (
isinstance(cells[1], str)
and cells[1]
and isinstance(cells[2], str)
and cells[2] in MESSAGE_ROLES
):
if (isinstance(cells[1], str) and cells[1] and isinstance(cells[2], str) and cells[2] in MESSAGE_ROLES):
return "messages"
return None
@@ -281,26 +265,16 @@ def _insert_prefix_row(
columns = dest_columns[: len(values)]
quoted = ", ".join(f'"{column}"' for column in columns)
placeholders = ", ".join("?" for _ in columns)
cursor = dest.execute(
f'INSERT OR IGNORE INTO "{table}" ({quoted}) VALUES ({placeholders})',
values,
)
cursor = dest.execute(f'INSERT OR IGNORE INTO "{table}" ({quoted}) VALUES ({placeholders})', values)
return cursor.rowcount == 1
def _copy_direct_tables(
lf_conn: sqlite3.Connection,
dest: sqlite3.Connection,
) -> dict[str, int]:
def _copy_direct_tables(lf_conn: sqlite3.Connection, dest: sqlite3.Connection) -> dict[str, int]:
"""Copy rows .recover managed to attribute to real canonical tables."""
# Lazy import: session_recovery imports this module inside a function, so
# a module-level import here would be circular.
from hermes_cli.session_recovery import (
_AUXILIARY_TABLE_SCHEMAS,
_AUXILIARY_TABLES,
_CANONICAL_TABLES,
)
from hermes_cli.session_recovery import (_AUXILIARY_TABLE_SCHEMAS, _AUXILIARY_TABLES, _CANONICAL_TABLES)
copied: dict[str, int] = {}
for table in (*_CANONICAL_TABLES, *_AUXILIARY_TABLES):
@@ -323,19 +297,13 @@ def _copy_direct_tables(
copied[table] = 0
continue
before = int(dest.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0])
dest.executemany(
f'INSERT OR IGNORE INTO "{table}" ({quoted}) VALUES ({placeholders})',
rows,
)
dest.executemany(f'INSERT OR IGNORE INTO "{table}" ({quoted}) VALUES ({placeholders})', rows)
after = int(dest.execute(f'SELECT COUNT(*) FROM "{table}"').fetchone()[0])
copied[table] = after - before
return copied
def map_lost_and_found_rows(
lf_conn: sqlite3.Connection,
dest: sqlite3.Connection,
) -> dict[str, Any]:
def map_lost_and_found_rows(lf_conn: sqlite3.Connection, dest: sqlite3.Connection) -> dict[str, Any]:
"""Best-effort mapping of a .recover output DB into a fresh SessionDB."""
report: dict[str, Any] = {
@@ -356,11 +324,7 @@ def map_lost_and_found_rows(
# genuinely NULL was already rejected by classify_lost_and_found_row,
# so the substitutions only fill NOT NULL bookkeeping counters/flags.
targets: dict[str, tuple[list[str], dict[int, Any]]] = {}
for kind_name, protected in (
("sessions", (0, 1)),
("messages", (1, 2)),
("session_model_usage", (0, 1)),
):
for kind_name, protected in (("sessions", (0, 1)), ("messages", (1, 2)), ("session_model_usage", (0, 1))):
defaults = _notnull_defaults(dest, kind_name)
for index in protected:
defaults.pop(index, None)
@@ -447,11 +411,7 @@ def stub_missing_parent_sessions(dest: sqlite3.Connection) -> dict[str, Any]:
clearly marked.
"""
result: dict[str, Any] = {
"sessions_stubbed": 0,
"messages_retained": 0,
"usage_rows_retained": 0,
}
result: dict[str, Any] = {"sessions_stubbed": 0, "messages_retained": 0, "usage_rows_retained": 0}
dest.execute("BEGIN IMMEDIATE")
try:
orphan_ids: dict[str, dict[str, Any]] = {}
@@ -470,42 +430,24 @@ def stub_missing_parent_sessions(dest: sqlite3.Connection) -> dict[str, Any]:
"WHERE u.session_id IS NOT NULL AND NOT EXISTS "
"(SELECT 1 FROM sessions WHERE sessions.id = u.session_id)"
):
orphan_ids.setdefault(
str(session_id), {"started_at": 0.0, "message_count": 0}
)
orphan_ids.setdefault(str(session_id), {"started_at": 0.0, "message_count": 0})
sequence = 1
for session_id, info in sorted(orphan_ids.items()):
while True:
title = (
f"[best-effort recovered {sequence}] session metadata "
"was unreadable"
)
title = (f"[best-effort recovered {sequence}] session metadata " "was unreadable")
sequence += 1
if (
dest.execute(
"SELECT 1 FROM sessions WHERE title = ? LIMIT 1",
(title,),
).fetchone()
is None
):
if (dest.execute("SELECT 1 FROM sessions WHERE title = ? LIMIT 1", (title,),).fetchone() is None):
break
dest.execute(
"INSERT INTO sessions (id, source, started_at, title, "
"message_count) VALUES (?, 'recovered', ?, ?, ?)",
(
session_id,
info["started_at"],
title,
info["message_count"],
),
(session_id, info["started_at"], title, info["message_count"]),
)
result["sessions_stubbed"] += 1
result["messages_retained"] += info["message_count"]
result["usage_rows_retained"] = int(
dest.execute("SELECT COUNT(*) FROM session_model_usage").fetchone()[0]
)
result["usage_rows_retained"] = int(dest.execute("SELECT COUNT(*) FROM session_model_usage").fetchone()[0])
# Repair dangling intra-sessions references without deleting rows.
dest.execute(

8
hermes_cli/session_recap.py
View File

@@ -79,9 +79,7 @@ def _tool_call_name_and_args(tool_call: Any) -> Tuple[str, Mapping[str, Any]]:
return name, raw_args if isinstance(raw_args, Mapping) else {}
def _iter_assistant_tool_calls(
messages: Sequence[Mapping[str, Any]],
) -> Iterable[Tuple[str, Mapping[str, Any]]]:
def _iter_assistant_tool_calls(messages: Sequence[Mapping[str, Any]]) -> Iterable[Tuple[str, Mapping[str, Any]]]:
for msg in messages:
if not isinstance(msg, Mapping):
continue
@@ -96,9 +94,7 @@ def _iter_assistant_tool_calls(
yield name, args
def _count_visible_turns(
messages: Sequence[Mapping[str, Any]],
) -> Tuple[int, int, int]:
def _count_visible_turns(messages: Sequence[Mapping[str, Any]]) -> Tuple[int, int, int]:
"""Return ``(user_turn_count, assistant_turn_count, tool_message_count)``."""
roles = Counter(msg.get("role") for msg in messages if isinstance(msg, Mapping))
return roles["user"], roles["assistant"], roles["tool"]

131
hermes_cli/session_recovery.py
View File

@@ -18,12 +18,7 @@ from contextlib import contextmanager
from pathlib import Path
from typing import Any, Callable, Iterator, Optional
from hermes_state import (
FTS_STORAGE_VERSION,
SCHEMA_VERSION,
SessionDB,
_db_opens_cleanly,
)
from hermes_state import (FTS_STORAGE_VERSION, SCHEMA_VERSION, SessionDB, _db_opens_cleanly)
ProgressCallback = Callable[[dict[str, Any]], None]
@@ -38,10 +33,7 @@ _CANONICAL_TABLES = (
"async_delegations",
)
_TOPIC_TABLES = (
"telegram_dm_topic_mode",
"telegram_dm_topic_bindings",
)
_TOPIC_TABLES = ("telegram_dm_topic_mode", "telegram_dm_topic_bindings")
def _init_delivery_ledger_schema(conn: sqlite3.Connection) -> None:
@@ -63,12 +55,7 @@ _AUXILIARY_TABLE_SCHEMAS: dict[str, Callable[[sqlite3.Connection], None]] = {
_AUXILIARY_TABLES = tuple(_AUXILIARY_TABLE_SCHEMAS)
_INVENTORY_TABLES = (
*_CANONICAL_TABLES,
"state_meta",
*_TOPIC_TABLES,
*_AUXILIARY_TABLES,
)
_INVENTORY_TABLES = (*_CANONICAL_TABLES, "state_meta", *_TOPIC_TABLES, *_AUXILIARY_TABLES)
# These values describe derived indexes or the schema that owns an optional
# table. A fresh destination must generate them from its own current schema.
@@ -134,9 +121,7 @@ def _validate_paths(
for suffix in _SIDECAR_SUFFIXES:
candidate = _sidecar_path(output, suffix)
if os.path.lexists(candidate):
raise SessionRecoverySafetyError(
f"Refusing to overwrite existing recovery output: {candidate}"
)
raise SessionRecoverySafetyError(f"Refusing to overwrite existing recovery output: {candidate}")
work_root = (
work_dir.expanduser().resolve(strict=True)
@@ -144,9 +129,7 @@ def _validate_paths(
else (output.parent if output is not None else source.parent)
)
if not work_root.is_dir():
raise SessionRecoverySafetyError(
f"Recovery work directory is not a directory: {work_root}"
)
raise SessionRecoverySafetyError(f"Recovery work directory is not a directory: {work_root}")
return source, output, work_root
@@ -180,11 +163,7 @@ def _same_filesystem(left: Path, right: Path) -> bool:
return left.anchor.casefold() == right.anchor.casefold()
def _disk_space_preflight(
source: Path,
work_root: Path,
output_parent: Optional[Path],
) -> dict[str, Any]:
def _disk_space_preflight(source: Path, work_root: Path, output_parent: Optional[Path]) -> dict[str, Any]:
"""Require space for the disposable bundle, output, and safety headroom."""
bundle_bytes = sum(
@@ -355,9 +334,7 @@ def _copy_rows(
destination.executemany(insert_sql, rows)
result["copied_rows"] += len(rows)
if progress_cb is not None:
progress_cb(
{"table": table, "copied_rows": result["copied_rows"], "source_rows": expected_rows}
)
progress_cb({"table": table, "copied_rows": result["copied_rows"], "source_rows": expected_rows})
except sqlite3.DatabaseError as exc:
result["status"] = "partial" if result["copied_rows"] else "failed"
result["error"] = str(exc)
@@ -370,10 +347,7 @@ def _copy_rows(
return result
def _table_inventory(
conn: sqlite3.Connection,
table: str,
) -> dict[str, Any]:
def _table_inventory(conn: sqlite3.Connection, table: str) -> dict[str, Any]:
result: dict[str, Any] = {"available": False, "columns": [], "rows": None}
try:
columns = _table_columns(conn, table)
@@ -419,10 +393,7 @@ def _snapshot_and_inspect(
work_root: Path,
) -> tuple[tempfile.TemporaryDirectory[str], Path, dict[str, Any]]:
before = _source_fingerprint(source)
temp_dir = tempfile.TemporaryDirectory(
prefix="hermes-session-recovery-",
dir=str(work_root),
)
temp_dir = tempfile.TemporaryDirectory(prefix="hermes-session-recovery-", dir=str(work_root))
snapshot_dir = Path(temp_dir.name)
try:
snapshot_source, copied = _copy_source_bundle(source, snapshot_dir)
@@ -454,11 +425,7 @@ def _snapshot_and_inspect(
raise
def inspect_session_database(
source_path: Path,
*,
work_dir: Optional[Path] = None,
) -> dict[str, Any]:
def inspect_session_database(source_path: Path, *, work_dir: Optional[Path] = None) -> dict[str, Any]:
"""Inspect canonical table readability without opening the source itself."""
source, _, work_root = _validate_paths(source_path, work_dir=work_dir)
@@ -476,10 +443,7 @@ def inspect_session_database(
temp_dir.cleanup()
def _ensure_auxiliary_destination_schema(
destination: sqlite3.Connection,
table: str,
) -> None:
def _ensure_auxiliary_destination_schema(destination: sqlite3.Connection, table: str) -> None:
"""Create a lazy auxiliary table on the recovered destination.
Recovery initializes the destination via base ``SessionDB``, which does not create gateway-owned
@@ -517,14 +481,11 @@ def _copy_table(
source_rows: Optional[int],
) -> dict[str, Any]:
"""Copy one canonical table: straight chunked copy, or rowid-range salvage when ``salvage``."""
copy_kwargs = dict(chunk_size=chunk_size, progress_cb=progress_cb, source_rows=source_rows)
if table == "state_meta":
return _copy_state_meta(
source, destination, salvage=salvage, chunk_size=chunk_size, progress_cb=progress_cb, source_rows=source_rows
)
return _copy_state_meta(source, destination, salvage=salvage, **copy_kwargs)
if salvage:
return _copy_table_salvage(
source, destination, table, chunk_size=chunk_size, progress_cb=progress_cb, source_rows=source_rows
)
return _copy_table_salvage(source, destination, table, **copy_kwargs)
result: dict[str, Any] = {"source_rows": source_rows, "copied_rows": 0}
columns = _compatible_columns(source, destination, table, result)
if columns is None:
@@ -544,12 +505,7 @@ def _copy_table(
)
def _append_skipped_range(
ranges: list[dict[str, Any]],
low: int,
high: int,
error: str,
) -> None:
def _append_skipped_range(ranges: list[dict[str, Any]], low: int, high: int, error: str) -> None:
"""Record skipped rowid ranges without producing one entry per row."""
if ranges and ranges[-1]["high"] + 1 == low and ranges[-1]["error"] == error:
@@ -558,10 +514,7 @@ def _append_skipped_range(
ranges.append({"low": low, "high": high, "error": error})
def _salvage_rowid_bounds(
source: sqlite3.Connection,
table: str,
) -> dict[str, Any]:
def _salvage_rowid_bounds(source: sqlite3.Connection, table: str) -> dict[str, Any]:
"""Find the readable rowid edges without scanning the complete table."""
result: dict[str, Any] = {"errors": [], "fallback_edges": []}
@@ -606,13 +559,7 @@ def _salvage_rowid_bounds(
return result
def _probe_populated_edge(
source: sqlite3.Connection,
table: str,
*,
edge: str,
anchor: int,
) -> dict[str, Any]:
def _probe_populated_edge(source: sqlite3.Connection, table: str, *, edge: str, anchor: int) -> dict[str, Any]:
"""Find a finite bound for a damaged rowid edge (issue #80205).
Substituting the whole rowid domain for a failed edge probe made range bisection burn the entire
@@ -634,9 +581,7 @@ def _probe_populated_edge(
span = 1
while True:
candidate = position + span if ascending else position - span
if (ascending and candidate >= domain_limit) or (
not ascending and candidate <= domain_limit
):
if (ascending and candidate >= domain_limit) or (not ascending and candidate <= domain_limit):
# No clean empty-tail answer before the domain edge; keep the
# domain fallback rather than inventing a bound.
result["bound"] = domain_limit
@@ -776,9 +721,7 @@ def _copy_table_salvage(
progress_cb: Optional[ProgressCallback],
source_rows: Optional[int],
insert_prefix: str = "INSERT",
row_filter: Optional[
Callable[[tuple[Any, ...], tuple[str, ...]], bool]
] = None,
row_filter: Optional[Callable[[tuple[Any, ...], tuple[str, ...]], bool]] = None,
) -> dict[str, Any]:
"""Best-effort rowid-range copy that continues past damaged source pages."""
@@ -829,10 +772,7 @@ def _copy_table_salvage(
if skipped_ranges:
result["status"] = "partial" if result["copied_rows"] else "failed"
result["error"] = f"{len(skipped_ranges)} rowid range(s) skipped"
elif (
source_rows is not None
and result["copied_rows"] + result["excluded_rows"] != source_rows
):
elif (source_rows is not None and result["copied_rows"] + result["excluded_rows"] != source_rows):
result["status"] = "partial"
result["error"] = (
f"copied {result['copied_rows']} and excluded "
@@ -927,10 +867,7 @@ def _copy_state_meta(
filtered_source_rows: Optional[int] = None
try:
filtered_source_rows = int(
source.execute(
f"SELECT COUNT(*) FROM state_meta WHERE key NOT IN ({placeholders})",
params,
).fetchone()[0]
source.execute(f"SELECT COUNT(*) FROM state_meta WHERE key NOT IN ({placeholders})", params).fetchone()[0]
)
except sqlite3.DatabaseError:
pass # the copy loop below will return the concrete read error
@@ -949,9 +886,7 @@ def _copy_state_meta(
)
def _reconstruct_missing_sessions(
destination: sqlite3.Connection,
) -> dict[str, Any]:
def _reconstruct_missing_sessions(destination: sqlite3.Connection) -> dict[str, Any]:
"""Recreate placeholder session rows for salvaged orphaned messages.
When the ``sessions`` b-tree is damaged worse than ``messages``, salvage can recover the
@@ -1008,18 +943,11 @@ def _reconcile(destination: sqlite3.Connection, table: str, where: str, mutation
return count
_DEPENDENT_TABLES = (
"messages",
"session_model_usage",
"compression_locks",
"telegram_dm_topic_bindings",
)
_DEPENDENT_TABLES = ("messages", "session_model_usage", "compression_locks", "telegram_dm_topic_bindings")
_RELINK_COUNTERS = ("session_prompt_refs_cleared", "sessions_parent_cleared")
def _cleanup_partial_orphans(
destination: sqlite3.Connection,
) -> dict[str, Any]:
def _cleanup_partial_orphans(destination: sqlite3.Connection) -> dict[str, Any]:
"""Reconcile references to sessions that could not be salvaged.
Messages are never discarded for lack of a session row: their owning session is reconstructed as
@@ -1257,10 +1185,7 @@ def _finalize_derived_metadata(destination: sqlite3.Connection) -> dict[str, Any
fts_keys = tuple(key for key in _GENERATED_META_KEYS if key.startswith("fts_"))
placeholders = ", ".join("?" for _ in fts_keys)
with _immediate_transaction(destination):
destination.execute(
f"DELETE FROM state_meta WHERE key IN ({placeholders})",
fts_keys,
)
destination.execute(f"DELETE FROM state_meta WHERE key IN ({placeholders})", fts_keys)
destination.execute(
"INSERT INTO state_meta(key, value) VALUES (?, ?) "
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
@@ -1442,11 +1367,7 @@ def recover_session_database(
if chunk_size <= 0:
raise SessionRecoverySafetyError("chunk_size must be greater than zero")
source, output, work_root = _validate_paths(
source_path,
output_path=output_path,
work_dir=work_dir,
)
source, output, work_root = _validate_paths(source_path, output_path=output_path, work_dir=work_dir)
assert output is not None
disk_space = _disk_space_preflight(source, work_root, output.parent)

55
hermes_cli/sessions_cmd.py
View File

@@ -10,6 +10,7 @@ working and imports stay one-way. Browse picker: :mod:`hermes_cli.sessions_cmd_b
import json
import os
import shutil
import sys
from functools import partial
from pathlib import Path
@@ -111,12 +112,9 @@ def _cmd_repair(args):
try:
from hermes_state import SessionDB
_repair_db = SessionDB()
try:
with SessionDB() as _repair_db:
n = _repair_db._conn.execute("SELECT COUNT(*) FROM sessions").fetchone()[0]
print(f"✓ Repaired — {n} sessions recovered.")
finally:
_repair_db.close()
print(f"✓ Repaired — {n} sessions recovered.")
except Exception:
print("✓ Repaired.")
return
@@ -201,8 +199,7 @@ def _cmd_recover(args):
class _RecoveryProgress:
"""`recover` progress printer: one live-updating `` <table>: n/total`` line per table."""
def __init__(self):
self.table = None
table = None
def __call__(self, info):
table = info.get("table")
@@ -396,14 +393,11 @@ def _export_html(args, collect):
sessions = collect()
if sessions is None:
return
if len(sessions) == 1:
content = generate_html_export(sessions[0])
else:
content = generate_multi_session_html_export(sessions)
single = len(sessions) == 1
content = generate_html_export(sessions[0]) if single else generate_multi_session_html_export(sessions)
with open(args.output, "w", encoding="utf-8") as f:
f.write(content)
suffix = "" if len(sessions) == 1 else "s"
print(f"Exported {len(sessions)} session{suffix} to {args.output} (HTML)")
print(f"Exported {len(sessions)} session{'' if single else 's'} to {args.output} (HTML)")
def _export_jsonl(args, collect):
@@ -577,9 +571,7 @@ def _export_markdown_single(db, args, export_one, output_dir, lineage_is_logical
if db.delete_session(resolved_session_id, sessions_dir=_sessions_dir(), expected_delete_ids=delete_target_ids):
delegate_count = len(delete_target_ids) - 1
delegate_suffix = (
""
if not delegate_count
else f" and {delegate_count} delegate session{'' if delegate_count == 1 else 's'}"
f" and {delegate_count} delegate session{'' if delegate_count == 1 else 's'}" if delegate_count else ""
)
print(f"Deleted exported session '{resolved_session_id}'{delegate_suffix}.")
else:
@@ -594,8 +586,7 @@ def _cmd_delete(db, args):
return _not_found(args.session_id)
# The user named this id directly so the delete is honored, but a pin is a
# "keep" flag and silently destroying it is surprising — say so.
_get_session = getattr(db, "get_session", None)
_meta = (_get_session(resolved_session_id) or {}) if callable(_get_session) else {}
_meta = db.get_session(resolved_session_id) or {}
_pinned_note = " (this session is PINNED)" if _meta.get("pinned") else ""
if not args.yes:
if not _confirm_prompt(
@@ -671,18 +662,12 @@ def _note_pinned_skipped(db, filters, action):
"""Pinned sessions are excluded by default from bulk prune/archive (pin =
durable keep). `prune --include-pinned` opts in; archive has no such flag,
so archive always spares pinned rows. Tell the user how many were spared."""
_count_matches = getattr(db, "count_prune_matches", None)
if not callable(_count_matches):
return
_base = {k: v for k, v in filters.items() if k != "include_pinned"}
try:
skipped = max(
int(_count_matches(**_base, include_pinned=True))
- int(_count_matches(**_base, include_pinned=False)),
0,
)
except TypeError:
return # a db double without include_pinned support — skip the note
skipped = max(
int(db.count_prune_matches(**_base, include_pinned=True))
- int(db.count_prune_matches(**_base, include_pinned=False)),
0,
)
if not skipped:
return
suffix = "" if skipped == 1 else "s"
@@ -778,12 +763,10 @@ def _cmd_prune_or_archive(db, args, action):
return
if action == "prune":
count = db.prune_sessions(sessions_dir=_sessions_dir(), **filters)
print(f"Pruned {count} session(s).")
print(f"Pruned {db.prune_sessions(sessions_dir=_sessions_dir(), **filters)} session(s).")
else:
count = db.archive_sessions(**filters)
print(
f"Archived {count} session(s). They're hidden from listings "
f"Archived {db.archive_sessions(**filters)} session(s). They're hidden from listings "
"but fully recoverable (nothing was deleted)."
)
@@ -996,8 +979,6 @@ def _cmd_optimize_storage(db, args):
# VACUUM needs a full second copy — require headroom ≈ current file size.
do_vacuum = not getattr(args, "no_vacuum", False)
try:
import shutil
free_bytes = shutil.disk_usage(db_path.parent).free
except Exception:
free_bytes = None
@@ -1146,11 +1127,9 @@ def cmd_sessions(args, sessions_parser=None):
print(f"Error: Could not open session database: {e}")
return 1
try:
with db:
handler = _DB_HANDLERS.get(action)
if handler is None:
sessions_parser.print_help()
return
return handler(db, args)
finally:
db.close()