diff --git a/hermes_cli/main.py b/hermes_cli/main.py index b0e47eee609..1b9147abf90 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -428,6 +428,9 @@ from pathlib import Path from typing import Optional +import functools as _functools + +from hermes_cli.sessions_cmd import cmd_sessions # noqa: F401 from hermes_cli.subcommands._shared import add_accept_hooks_flag as _add_accept_hooks_flag from hermes_cli.subcommands.cron import build_cron_parser from hermes_cli.subcommands.gateway import build_gateway_parser @@ -12365,1117 +12368,14 @@ def main(): "--limit", type=int, default=500, help="Max sessions to load (default: 500)" ) - def _confirm_prompt(prompt: str) -> bool: - """Prompt for y/N confirmation, safe against non-TTY environments.""" - try: - return input(prompt).strip().lower() in {"y", "yes"} - except (EOFError, KeyboardInterrupt): - return False - def cmd_sessions(args): - import json as _json - - action = args.sessions_action - - # 'repair' and 'recover' must run BEFORE opening SessionDB(): a - # malformed schema is exactly the case where SessionDB() can't open. - # Recovery additionally promises never to open the supplied source - # directly, so it operates through its own disposable source copy. - if action == "repair": - from hermes_state import ( - DEFAULT_DB_PATH, - _db_opens_cleanly, - repair_state_db_schema, - ) - - db_path = DEFAULT_DB_PATH - if not db_path.exists(): - print(f"No session database at {db_path} (nothing to repair).") - return - reason = _db_opens_cleanly(db_path) - if reason is None: - print(f"✓ {db_path} opens cleanly — no repair needed.") - return - print(f"✗ {db_path} does not open cleanly: {reason}") - if getattr(args, "check_only", False): - return - print("Repairing (a backup copy is made first)…") - report = repair_state_db_schema( - db_path, backup=not getattr(args, "no_backup", False) - ) - if report.get("repaired"): - if report.get("backup_path"): - print(f" backup: {report['backup_path']}") - print(f" strategy: {report.get('strategy')}") - try: - from hermes_state import SessionDB - - n = SessionDB()._conn.execute( - "SELECT COUNT(*) FROM sessions" - ).fetchone()[0] - print(f"✓ Repaired — {n} sessions recovered.") - except Exception: - print("✓ Repaired.") - else: - print(f"✗ Repair failed: {report.get('error')}") - 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: in-place - # repair has failed and nothing tells them the non-destructive - # offline recovery path exists. Lead with --inspect-only so - # they confirm the data is readable before writing anything. - 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") - return - - if action == "recover": - import sqlite3 as _sqlite3 - - from hermes_cli.session_recovery import ( - SessionRecoveryError, - inspect_session_database, - recover_session_database, - write_recovery_report, - ) - - source = args.source - output = 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) - if inspect_only and output is not None: - print("Error: --output cannot be used with --inspect-only.") - return 2 - if inspect_only and allow_partial: - print("Error: --allow-partial cannot be used with --inspect-only.") - return 2 - if not inspect_only and output is None: - print("Error: --output is required unless --inspect-only is used.") - return 2 - if not inspect_only and report_path is None: - report_path = output.with_name(output.name + ".recovery.json") - if ( - report_path is not None - and os.path.lexists(report_path.expanduser()) - ): - print(f"Error: refusing to overwrite existing report: {report_path}") - return 2 - - try: - if inspect_only: - report = inspect_session_database( - source, - work_dir=getattr(args, "work_dir", None), - ) - else: - last_progress = {"table": None} - - def _recovery_progress(info): - table = info.get("table") - copied = int(info.get("copied_rows") or 0) - total = info.get("source_rows") - if table != last_progress["table"]: - if last_progress["table"] is not None: - print() - print(f" {table}: ", end="", flush=True) - last_progress["table"] = table - suffix = f"/{int(total):,}" if total is not None else "" - print(f"\r {table}: {copied:,}{suffix}", end="", flush=True) - - print("Recovering canonical session data into a new database…") - report = recover_session_database( - source, - output, - work_dir=getattr(args, "work_dir", None), - chunk_size=getattr(args, "chunk_size", 1000), - progress_cb=_recovery_progress, - allow_partial=allow_partial, - ) - if last_progress["table"] is not None: - print() - except (SessionRecoveryError, OSError, _sqlite3.DatabaseError) as exc: - print(f"Error: session recovery failed: {exc}") - print("The supplied source database was not replaced or deleted.") - return 1 - - if report_path is not None: - try: - written_report = write_recovery_report(report_path, report) - except (FileExistsError, OSError) as exc: - print(f"Error: could not write recovery report: {exc}") - return 1 - print(f"Recovery report: {written_report}") - else: - print(_json.dumps(report, indent=2, sort_keys=True)) - - if inspect_only: - return 0 if report.get("recoverable") else 1 - 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.") - return 0 - if allow_partial and report.get("verified"): - counts = report.get("verification", {}).get("table_counts", {}) - print(f"✓ Partial recovery output verified at: {output}") - print( - " Recovered " - f"{int(counts.get('sessions') or 0):,} sessions and " - f"{int(counts.get('messages') or 0):,} messages." - ) - print(" The active session database was not changed.") - print( - " 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.") - return 1 - - try: - from hermes_state import SessionDB - - db = SessionDB() - except Exception as e: - print(f"Error: Could not open session database: {e}") - return - - # Hide third-party tool sessions by default, but honour explicit --source - _source = getattr(args, "source", None) - _exclude = None if _source else ["tool"] - - if action == "list": - from hermes_state import workspace_key as _ws_key - - sessions = db.list_sessions_rich( - source=args.source, exclude_sources=_exclude, limit=args.limit - ) - - # Workspace filter: match a session by its workspace key (git repo - # root, else cwd) — path substring or exact basename. - _ws_filter = (getattr(args, "workspace", None) or "").strip() - if _ws_filter: - _needle = _ws_filter.lower() - - def _in_workspace(s): - key = (_ws_key(s) or "").lower() - return bool(key) and ( - _needle in key or _needle == os.path.basename(key.rstrip("/\\")) - ) - - sessions = [s for s in sessions if _in_workspace(s)] - - if not sessions: - print("No sessions found.") - return - - # Short workspace label: the repo/dir basename, "—" when unbound. The - # Workspace column only appears once at least one session carries one - # (or when filtering), so all-unbound listings read as before. - def _ws_label(s): - key = _ws_key(s) - return (os.path.basename(key.rstrip("/\\")) or key) if key else "—" - - has_ws = bool(_ws_filter) or any(_ws_key(s) for s in sessions) - has_titles = any(s.get("title") for s in sessions) - - if has_ws: - if has_titles: - print(f"{'Title':<28} {'Workspace':<18} {'Last Active':<13} {'ID'}") - print("─" * 110) - else: - print(f"{'Preview':<38} {'Workspace':<18} {'Last Active':<13} {'Src':<6} {'ID'}") - print("─" * 100) - for s in sessions: - last_active = _relative_time(s.get("last_active")) - ws = _ws_label(s)[:16] - if has_titles: - title = (s.get("title") or "—")[:26] - print(f"{title:<28} {ws:<18} {last_active:<13} {s['id']}") - else: - preview = s.get("preview", "")[:36] - print(f"{preview:<38} {ws:<18} {last_active:<13} {s['source']:<6} {s['id']}") - return - - if has_titles: - print(f"{'Title':<32} {'Preview':<40} {'Last Active':<13} {'ID'}") - print("─" * 110) - else: - print(f"{'Preview':<50} {'Last Active':<13} {'Src':<6} {'ID'}") - print("─" * 95) - for s in sessions: - last_active = _relative_time(s.get("last_active")) - preview = ( - s.get("preview", "")[:38] - if has_titles - else s.get("preview", "")[:48] - ) - if has_titles: - title = (s.get("title") or "—")[:30] - sid = s["id"] - print(f"{title:<32} {preview:<40} {last_active:<13} {sid}") - else: - sid = s["id"] - print(f"{preview:<50} {last_active:<13} {s['source']:<6} {sid}") - - elif action == "export": - from hermes_cli.session_filters import ( - build_prune_filters, - describe_filters, - ) - - _filter_arg_names = ( - "older_than", "newer_than", "before", "after", - "source", "title", "end_reason", "cwd", - "min_messages", "max_messages", "model", "provider", - "user", "chat_id", "chat_type", "branch", - "min_tokens", "max_tokens", "min_cost", "max_cost", - "min_tool_calls", "max_tool_calls", - ) - _any_filters = any( - getattr(args, a, None) is not None for a in _filter_arg_names - ) - filters = None - if _any_filters: - try: - filters = build_prune_filters(args) - except ValueError as e: - print(f"Error: {e}") - return - # Unlike prune/archive, export includes archived sessions. - filters["archived"] = None - - def _redact(data): - if not args.redact or data is None: - return data - from hermes_cli.session_export_md import redact_session_data - - 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.""" - if args.session_id: - resolved = db.resolve_session_id(args.session_id) - data = _redact(db.export_session(resolved)) if resolved else None - if not data: - print(f"Session '{args.session_id}' not found.") - return None - return [data] - if filters: - candidates = db.list_prune_candidates(**filters) - if args.dry_run: - print( - f"Would export {len(candidates)} session(s) " - f"({describe_filters(filters)})." - ) - for row in candidates[:100]: - print(f" {row.get('id')} {row.get('source', '')}") - if len(candidates) > 100: - print(f" ... {len(candidates) - 100} more") - return None - return [ - s - for s in ( - _redact(db.export_session(row["id"])) for row in candidates - ) - if s - ] - if args.dry_run: - print("--dry-run requires at least one filter.") - return None - return [_redact(s) for s in db.export_all(source=None)] - - # Prompt-only export (--only user-prompts): one prompt record per - # line (jsonl) or headed sections (md). Delegates rendering to - # hermes_cli.session_export. - if getattr(args, "only", None): - if args.format not in ("jsonl", "md"): - print("--only user-prompts supports --format jsonl or md.") - return - from hermes_cli.session_export import ( - export_record_count, - render_sessions_export, - ) - - sessions = _collect_sessions() - if sessions is None: - db.close() - return - rendered = render_sessions_export( - sessions, - fmt="markdown" if args.format == "md" else "jsonl", - only=args.only, - ) - if not args.output or args.output == "-": - sys.stdout.write(rendered) - db.close() - return - with open(args.output, "w", encoding="utf-8") as f: - f.write(rendered) - count, noun = export_record_count(sessions, only=args.only) - suffix = "" if count == 1 else "s" - print(f"Exported {count} {noun}{suffix} to {args.output}") - db.close() - return - - # Standalone HTML export: one self-contained file (single session - # or multi-session with sidebar navigation). - if args.format == "html": - if not args.output or args.output == "-": - print("HTML export requires an output file path.") - return - from hermes_cli.session_export_html import ( - generate_html_export, - generate_multi_session_html_export, - ) - - sessions = _collect_sessions() - if sessions is None: - db.close() - return - if len(sessions) == 1: - content = generate_html_export(sessions[0]) - else: - content = 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)") - db.close() - return - - # Claude Code JSONL trace export — local file or HF upload. - # Redaction is ON by default for traces (they leave the machine - # when --upload is used); --no-redact opts out after review. - if args.format == "trace": - if getattr(args, "only", None): - print("--only user-prompts supports --format jsonl or md.") - db.close() - return - session_id = args.session_id - if not session_id and not filters: - # Match the shell's common intent: "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: - print("No session found to export. Pass --session-id.") - db.close() - return - if session_id and not db.resolve_session_id(session_id): - print(f"Session '{session_id}' not found.") - db.close() - return - - from agent.trace_upload import ( - TraceRedactionError, - build_trace_jsonl, - upload_session_trace, - ) - - redact_trace = not getattr(args, "no_redact", False) - - if getattr(args, "upload", False): - if not session_id: - print("--upload exports one session: pass --session-id (or drop filters to use the most recent).") - db.close() - return - resolved = db.resolve_session_id(session_id) - db.close() - status = upload_session_trace( - resolved, - cwd="", - redact=redact_trace, - private=not getattr(args, "public", False), - ) - print(status) - return - - # Local trace file(s) - def _trace_ids(): - if session_id: - return [db.resolve_session_id(session_id)] - candidates = db.list_prune_candidates(**filters) - if args.dry_run: - print( - f"Would export {len(candidates)} session(s) " - f"({describe_filters(filters)})." - ) - for row in candidates[:100]: - print(f" {row.get('id')} {row.get('source', '')}") - if len(candidates) > 100: - print(f" ... {len(candidates) - 100} more") - return None - return [row["id"] for row in candidates] - - ids = _trace_ids() - if ids is None: - db.close() - return - - def _render_trace(sid): - meta = db.get_session(sid) or {} - messages = db.get_messages_as_conversation(sid) - if not messages: - return None - return build_trace_jsonl( - messages, - session_id=sid, - model=meta.get("model") or "", - cwd="", - redact=redact_trace, - ) - - try: - if len(ids) == 1: - jsonl = _render_trace(ids[0]) - if not jsonl: - print(f"No transcript to export for session '{ids[0]}'.") - db.close() - return - if not args.output or args.output == "-": - sys.stdout.write(jsonl) - else: - with open(args.output, "w", encoding="utf-8") as f: - f.write(jsonl) - print(f"Exported 1 session trace to {args.output}") - else: - out_dir = ( - Path(args.output).expanduser() - if args.output and args.output != "-" - else get_hermes_home() / "session-exports" - ) - out_dir.mkdir(parents=True, exist_ok=True) - exported = 0 - for sid in ids: - jsonl = _render_trace(sid) - if not jsonl: - continue - (out_dir / f"{sid}.trace.jsonl").write_text( - jsonl, encoding="utf-8" - ) - exported += 1 - print(f"Exported {exported} session trace(s) to {out_dir}") - except TraceRedactionError: - print("Redaction failed; refusing to export unredacted trace content.") - db.close() - return - - if args.format == "jsonl": - if not args.output: - print("JSONL export requires an output path (use - for stdout).") - return - if args.session_id: - resolved_session_id = db.resolve_session_id(args.session_id) - if not resolved_session_id: - print(f"Session '{args.session_id}' not found.") - return - data = _redact(db.export_session(resolved_session_id)) - if not data: - print(f"Session '{args.session_id}' not found.") - return - line = _json.dumps(data, ensure_ascii=False) + "\n" - if args.output == "-": - - sys.stdout.write(line) - else: - with open(args.output, "w", encoding="utf-8") as f: - f.write(line) - print(f"Exported 1 session to {args.output}") - else: - if filters: - candidates = db.list_prune_candidates(**filters) - if args.dry_run: - print( - f"Would export {len(candidates)} session(s) " - f"({describe_filters(filters)})." - ) - for row in candidates[:100]: - print(f" {row.get('id')} {row.get('source', '')}") - if len(candidates) > 100: - print(f" ... {len(candidates) - 100} more") - return - sessions = [ - s - for s in ( - db.export_session(row["id"]) for row in candidates - ) - if s - ] - else: - if args.dry_run: - print("--dry-run requires at least one filter.") - return - sessions = db.export_all(source=None) - if args.output == "-": - - for s in sessions: - sys.stdout.write( - _json.dumps(_redact(s), ensure_ascii=False) + "\n" - ) - else: - with open(args.output, "w", encoding="utf-8") as f: - for s in sessions: - f.write( - _json.dumps(_redact(s), ensure_ascii=False) + "\n" - ) - print(f"Exported {len(sessions)} sessions to {args.output}") - return - - # Markdown / QMD export - from hermes_cli.session_export_md import ( - append_manifest_entry, - verify_export_file, - write_session_markdown, - ) - - if args.output == "-": - print("Markdown/QMD export writes files; stdout (-) is only supported with --format jsonl.") - db.close() - return - output_dir = Path(args.output).expanduser() if args.output else get_hermes_home() / "session-exports" - - 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) - ) - if not data: - return None, None - data = _redact(data) - path = write_session_markdown( - data, - output_dir, - fmt=args.format, - force=args.force, - ) - append_manifest_entry(output_dir, data, path, fmt=args.format) - return data, path - - if args.delete_after_verified and not args.yes: - print("--delete-after-verified requires --yes.") - db.close() - return - if args.delete_after_verified and not args.session_id: - print("--delete-after-verified is only supported with --session-id.") - db.close() - return - - lineage_is_logical = getattr(args, "lineage", "single") == "logical" - - if args.session_id: - resolved_session_id = db.resolve_session_id(args.session_id) - if not resolved_session_id: - print(f"Session '{args.session_id}' not found.") - db.close() - return - delete_target_ids = [resolved_session_id] - if args.delete_after_verified: - delete_target_ids = db.get_session_delete_targets( - resolved_session_id - ) - - exported_items = [] - 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 - ), - ) - except FileExistsError as e: - print( - f"Export already exists: {e}. " - "Pass --force to overwrite." - ) - db.close() - return - if not data or not exported_path: - print( - f"Session '{target_id}' disappeared during export; " - "nothing was deleted." - ) - db.close() - return - exported_items.append((data, exported_path)) - - 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}) " - f"to {exported_items[0][1]}" - ) - else: - print( - f"Exported {len(exported_items)} sessions " - f"({message_count} message{suffix}) to {output_dir}" - ) - if args.delete_after_verified: - for data, exported_path in exported_items: - ok, reason = verify_export_file(exported_path, data) - if not ok: - print( - "Export verification failed; not deleting " - f"session '{data.get('id')}': {reason}" - ) - db.close() - return - sessions_dir = get_hermes_home() / "sessions" - 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" - f"{'' if delegate_count == 1 else 's'}" - ) - print( - f"Deleted exported session '{resolved_session_id}'" - f"{delegate_suffix}." - ) - else: - print( - f"Exported, but session '{resolved_session_id}' was " - "not deleted because its delegate set changed." - ) - db.close() - return - - if not filters: - print( - "Refusing bulk export without a filter. Pass --session-id or " - "at least one filter (e.g. --older-than 90, --source telegram)." - ) - db.close() - return - candidates = db.list_prune_candidates(**filters) - if args.dry_run: - print( - f"Would export {len(candidates)} session(s) " - f"({describe_filters(filters)})." - ) - for row in candidates[:100]: - print(f" {row.get('id')} {row.get('source', '')}") - if len(candidates) > 100: - print(f" ... {len(candidates) - 100} more") - db.close() - return - exported = 0 - for row in candidates: - try: - data, exported_path = _export_one( - row["id"], - include_lineage=lineage_is_logical, - ) - except FileExistsError as e: - print(f"Skipping existing export: {e}. Pass --force to overwrite.") - continue - if data and exported_path: - exported += 1 - print(f"Exported {exported} session(s) to {output_dir}") - - elif action == "delete": - resolved_session_id = db.resolve_session_id(args.session_id) - if not resolved_session_id: - print(f"Session '{args.session_id}' not found.") - return - if not args.yes: - if not _confirm_prompt( - f"Delete session '{resolved_session_id}' and all its messages? [y/N] " - ): - print("Cancelled.") - return - sessions_dir = get_hermes_home() / "sessions" - if db.delete_session(resolved_session_id, sessions_dir=sessions_dir): - print(f"Deleted session '{resolved_session_id}'.") - else: - print(f"Session '{args.session_id}' not found.") - - elif action in ("prune", "archive"): - from hermes_cli.session_filters import ( - build_prune_filters, - describe_filters, - format_epoch, - ) - - # Preserve the historical default ONLY for a truly bare - # `hermes sessions prune`: no time window and no filters at all - # means "older than 90 days". ANY filter — including --source — - # suppresses the implicit cutoff, so `prune --source cron` - # matches ALL cron sessions regardless of age. The preview + - # confirmation below (count, oldest/newest) is the safety net. - _non_time_filters = any( - getattr(args, a, None) is not None - for a in ( - "source", "title", "end_reason", "cwd", - "min_messages", "max_messages", "model", "provider", - "user", "chat_id", "chat_type", "branch", - "min_tokens", "max_tokens", "min_cost", "max_cost", - "min_tool_calls", "max_tool_calls", - ) - ) - if ( - action == "prune" - and args.older_than is None - and args.newer_than is None - and args.before is None - and args.after is None - and not _non_time_filters - ): - args.older_than = "90" - - try: - filters = build_prune_filters(args) - except ValueError as e: - print(f"Error: {e}") - return - - 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)." - ) - return - - # Prune skips archived sessions unless --include-archived; - # archive only targets not-yet-archived rows (idempotent). - if action == "prune": - filters["archived"] = ( - None if getattr(args, "include_archived", False) else False - ) - else: - filters["archived"] = False - - candidates = db.list_prune_candidates(**filters) - verb = "Delete" if action == "prune" else "Archive" - if not candidates: - print(f"No sessions match ({describe_filters(filters)}).") - return - - # Candidates are ordered by activity oldest-first. Surface that - # span so a long-lived but recently used conversation cannot look - # old merely because of its creation date. - _oldest = candidates[0].get("last_active") - _newest = candidates[-1].get("last_active") - _span = ( - f"oldest activity {format_epoch(_oldest)}, " - f"newest activity {format_epoch(_newest)}" - ) - - if args.dry_run or not args.yes: - shown = candidates if args.dry_run else candidates[:15] - print( - f"{len(candidates)} session(s) match " - f"({describe_filters(filters)}; {_span}):" - ) - for s in shown: - 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['message_count']:>4} msgs {title}" - ) - if len(candidates) > len(shown): - print(f" … and {len(candidates) - len(shown)} more") - if args.dry_run: - print(f"Dry run — nothing {'deleted' if action == 'prune' else 'archived'}.") - return - - if not args.yes: - if not _confirm_prompt( - f"{verb} these {len(candidates)} session(s) ({_span})? [y/N] " - ): - print("Cancelled.") - return - - if action == "prune": - sessions_dir = get_hermes_home() / "sessions" - count = db.prune_sessions(sessions_dir=sessions_dir, **filters) - print(f"Pruned {count} session(s).") - else: - count = db.archive_sessions(**filters) - print( - f"Archived {count} session(s). They're hidden from listings " - "but fully recoverable (nothing was deleted)." - ) - - elif action == "rename": - resolved_session_id = db.resolve_session_id(args.session_id) - if not resolved_session_id: - print(f"Session '{args.session_id}' not found.") - return - title = " ".join(args.title) - try: - if db.set_session_title(resolved_session_id, title): - print(f"Session '{resolved_session_id}' renamed to: {title}") - else: - print(f"Session '{args.session_id}' not found.") - except ValueError as e: - print(f"Error: {e}") - - elif action == "retitle-skills": - from agent.skill_commands import describe_skill_invocation - from agent.title_generator import generate_title - - limit = max(1, int(getattr(args, "limit", 200) or 200)) - apply_changes = bool(getattr(args, "apply", False)) - - def _is_titlelike(candidate: str) -> bool: - """Reject a candidate that isn't a title at all. - - An auxiliary model occasionally answers the prompt instead of - titling it and echoes the assistant's output ('$ df -h /'). The - live path has no alternative and takes what it gets, but this is - a REPAIR — replacing a serviceable title with command output - would make things worse, so keep the old one. - """ - return bool(candidate) and candidate[0].isalnum() - - candidates = db.list_skill_scaffolded_sessions(limit=limit) - if not candidates: - 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)'}:" - ) - changed = 0 - for row in candidates: - session_id = row["id"] - typed = describe_skill_invocation(row["content"]) or "" - first_reply = db.get_first_assistant_text(session_id) or "" - new_title = generate_title(typed, first_reply) - if not new_title or new_title == row["title"]: - continue - if not _is_titlelike(new_title): - print(f" {session_id}\n kept {row['title']!r} — got {new_title!r}") - continue - print(f" {session_id}\n {row['title']!r}\n → {new_title!r}") - changed += 1 - if not apply_changes: - continue - try: - db.set_session_title(session_id, new_title) - except ValueError: - # Unique-title collision. Dedupe the same way the live - # auto-titler does (base #2, base #3, ...) rather than - # leaving the leaked title in place. - deduped = db.get_next_title_in_lineage(new_title) - try: - db.set_session_title(session_id, deduped) - print(f" (renamed to {deduped!r} — title was taken)") - except ValueError as e: - print(f" skipped: {e}") - changed -= 1 - - if not changed: - print(" every title already reflects the user's request.") - elif apply_changes: - print(f"✓ Re-titled {changed} session(s).") - - elif action == "browse": - limit = getattr(args, "limit", 500) or 500 - source = getattr(args, "source", None) - _browse_exclude = None if source else ["tool"] - sessions = db.list_sessions_rich( - source=source, exclude_sources=_browse_exclude, limit=limit - ) - db.close() - if not sessions: - print("No sessions found.") - return - - selected_id = _session_browse_picker(sessions) - if not selected_id: - print("Cancelled.") - return - - # Launch hermes --resume by replacing the current process - print(f"Resuming session: {selected_id}") - from hermes_cli.relaunch import relaunch - - relaunch(["--resume", selected_id]) - return # won't reach here after execvp - - elif action == "optimize": - db_path = db.db_path - before_mb = ( - os.path.getsize(db_path) / (1024 * 1024) - if db_path.exists() - else 0.0 - ) - print("Optimizing session store (FTS merge + VACUUM)…") - try: - # vacuum() merges FTS5 segments (optimize_fts) then VACUUMs, - # and returns the number of indexes it merged. - n = db.vacuum() - except Exception as e: - print(f"Error: optimization failed: {e}") - db.close() - return - after_mb = ( - os.path.getsize(db_path) / (1024 * 1024) - if db_path.exists() - else 0.0 - ) - # Same WAL caveat as optimize-storage: after a VACUUM the main file - # on disk lags until the WAL is checkpointed back (refused while a - # live gateway holds a read-mark), so stat() understates the win and - # can go negative. SQLite's page accounting is correct immediately. - logical_after = db.logical_size_bytes() - if logical_after is not None: - after_mb = logical_after / (1024 * 1024) - saved = before_mb - after_mb - print(f"Optimized {n} FTS index(es).") - print( - f"Database size: {before_mb:.1f} MB -> {after_mb:.1f} MB " - f"({_size_delta_label(saved)})" - ) - - elif action == "optimize-storage": - db_path = db.db_path - if not db.fts_optimize_available(): - print("Search index is already on the compact layout — nothing to do.") - db.close() - return - - before_bytes = os.path.getsize(db_path) if db_path.exists() else 0 - before_mb = before_bytes / (1024 * 1024) - - # Disk preflight: the rebuild adds the new index before the old is - # torn down, and the final VACUUM needs a full second copy of the - # file. Require headroom ≈ current file size to finish cleanly. - do_vacuum = not getattr(args, "no_vacuum", False) - try: - import shutil as _shutil - free_bytes = _shutil.disk_usage(db_path.parent).free - except Exception: - free_bytes = None - need_bytes = before_bytes if do_vacuum else int(before_bytes * 0.3) - print(f"Search-index optimization for {db_path}") - 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 ''})") - 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).") - db.close() - 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).") - if not getattr(args, "yes", False): - try: - resp = input("Proceed? [y/N] ").strip().lower() - except EOFError: - resp = "" - if resp not in ("y", "yes"): - print("Cancelled.") - db.close() - return - - _last = {"phase": None} - - 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) - elif phase != _last["phase"]: - label = {"teardown": "Reclaiming old index", - "vacuum": "Compacting database (VACUUM)", - "done": "Done"}.get(phase, phase) - print(f"\n {label}…", flush=True) - _last["phase"] = phase - - print("Optimizing search-index storage…") - try: - result = db.optimize_fts_storage( - progress_cb=_progress, vacuum=do_vacuum - ) - except Exception as e: - print(f"\nError: optimization failed: {e}") - print("No data was lost. Re-run to resume.") - db.close() - return - if not result.get("ok"): - print(f"\nCould not optimize: {result.get('reason', 'unknown')}") - db.close() - return - after_mb = ( - os.path.getsize(db_path) / (1024 * 1024) if db_path.exists() else 0.0 - ) - # Prefer SQLite's own page accounting over stat(). In WAL mode a - # VACUUM's rewrite sits in the -wal file until a checkpoint folds it - # back, and that checkpoint is refused while another connection (a - # live gateway) holds a read-mark — so the main file on disk still - # reads at its pre-VACUUM size and keeps growing. stat()ing it here - # reported "reclaimed -3820.1 MB" on a DB that had actually shrunk - # 60%. page_count * page_size is correct immediately. - logical_after = db.logical_size_bytes() - if logical_after is not None: - after_mb = logical_after / (1024 * 1024) - saved = before_mb - after_mb - print(f"\n✓ Search index optimized.") - print( - f" Database size: {before_mb:.1f} MB -> {after_mb:.1f} MB " - f"({_size_delta_label(saved)})" - ) - if result.get("vacuumed") is False: - print(" (VACUUM was skipped or failed — run " - "`hermes sessions optimize` later to reclaim freed space.)") - - elif action == "stats": - total = db.session_count() - msgs = db.message_count() - print(f"Total sessions: {total}") - print(f"Total messages: {msgs}") - for src in ["cli", "telegram", "discord", "whatsapp", "slack"]: - c = db.session_count(source=src) - if c > 0: - print(f" {src}: {c} sessions") - db_path = db.db_path - if db_path.exists(): - size_mb = os.path.getsize(db_path) / (1024 * 1024) - print(f"Database size: {size_mb:.1f} MB") - - else: - sessions_parser.print_help() - - db.close() - - sessions_parser.set_defaults(func=cmd_sessions) + # cmd_sessions lives in hermes_cli/sessions_cmd.py (main.py decomposition). + # sessions_parser is threaded in via functools.partial because the + # fallthrough branch calls sessions_parser.print_help() (formerly a + # closure capture of this main()-local). + sessions_parser.set_defaults( + func=_functools.partial(cmd_sessions, sessions_parser=sessions_parser) + ) # ========================================================================= # insights command (parser built in hermes_cli/subcommands/insights.py) diff --git a/hermes_cli/sessions_cmd.py b/hermes_cli/sessions_cmd.py new file mode 100644 index 00000000000..0338585a688 --- /dev/null +++ b/hermes_cli/sessions_cmd.py @@ -0,0 +1,1159 @@ +"""``hermes sessions`` command — extracted from ``hermes_cli/main.py``. + +Mechanical move (main.py decomposition): ``cmd_sessions`` was a ``def`` nested +inside ``main()``'s body; its dispatch on ``args.sessions_action`` is lifted +byte-identical. A symtable/AST closure check found exactly two free variables: + +* ``_confirm_prompt`` — a sibling nested def with zero captures of its own; + moved here to module level, byte-identical. +* ``sessions_parser`` — a ``main()``-local (the argparse subparser, used only + for ``sessions_parser.print_help()`` in the fallthrough branch). It is + threaded as a keyword parameter via ``functools.partial`` at the + ``set_defaults(func=...)`` wiring site in ``main()``. + +Helpers that stay in ``hermes_cli.main`` (``get_hermes_home``, +``_relative_time``, ``_session_browse_picker``, ``_size_delta_label``) are +delegated through call-time wrappers below so existing test monkeypatches on +``hermes_cli.main.`` keep reaching this code path, and so imports stay +one-way (main.py imports this module; the reverse happens only lazily at call +time — no import cycle). +""" + +import os +import sys +from pathlib import Path + + +def _m(): + """Lazy ``hermes_cli.main`` reference (call-time, keeps patches working).""" + from hermes_cli import main + + return main + + +def get_hermes_home(): + return _m().get_hermes_home() + + +def _relative_time(ts): + return _m()._relative_time(ts) + + +def _session_browse_picker(sessions): + return _m()._session_browse_picker(sessions) + + +def _size_delta_label(saved_mb): + return _m()._size_delta_label(saved_mb) + + +def _confirm_prompt(prompt: str) -> bool: + """Prompt for y/N confirmation, safe against non-TTY environments.""" + try: + return input(prompt).strip().lower() in {"y", "yes"} + except (EOFError, KeyboardInterrupt): + return False + + +def cmd_sessions(args, sessions_parser=None): + import json as _json + + action = args.sessions_action + + # 'repair' and 'recover' must run BEFORE opening SessionDB(): a + # malformed schema is exactly the case where SessionDB() can't open. + # Recovery additionally promises never to open the supplied source + # directly, so it operates through its own disposable source copy. + if action == "repair": + from hermes_state import ( + DEFAULT_DB_PATH, + _db_opens_cleanly, + repair_state_db_schema, + ) + + db_path = DEFAULT_DB_PATH + if not db_path.exists(): + print(f"No session database at {db_path} (nothing to repair).") + return + reason = _db_opens_cleanly(db_path) + if reason is None: + print(f"✓ {db_path} opens cleanly — no repair needed.") + return + print(f"✗ {db_path} does not open cleanly: {reason}") + if getattr(args, "check_only", False): + return + print("Repairing (a backup copy is made first)…") + report = repair_state_db_schema( + db_path, backup=not getattr(args, "no_backup", False) + ) + if report.get("repaired"): + if report.get("backup_path"): + print(f" backup: {report['backup_path']}") + print(f" strategy: {report.get('strategy')}") + try: + from hermes_state import SessionDB + + n = SessionDB()._conn.execute( + "SELECT COUNT(*) FROM sessions" + ).fetchone()[0] + print(f"✓ Repaired — {n} sessions recovered.") + except Exception: + print("✓ Repaired.") + else: + print(f"✗ Repair failed: {report.get('error')}") + 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: in-place + # repair has failed and nothing tells them the non-destructive + # offline recovery path exists. Lead with --inspect-only so + # they confirm the data is readable before writing anything. + 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") + return + + if action == "recover": + import sqlite3 as _sqlite3 + + from hermes_cli.session_recovery import ( + SessionRecoveryError, + inspect_session_database, + recover_session_database, + write_recovery_report, + ) + + source = args.source + output = 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) + if inspect_only and output is not None: + print("Error: --output cannot be used with --inspect-only.") + return 2 + if inspect_only and allow_partial: + print("Error: --allow-partial cannot be used with --inspect-only.") + return 2 + if not inspect_only and output is None: + print("Error: --output is required unless --inspect-only is used.") + return 2 + if not inspect_only and report_path is None: + report_path = output.with_name(output.name + ".recovery.json") + if ( + report_path is not None + and os.path.lexists(report_path.expanduser()) + ): + print(f"Error: refusing to overwrite existing report: {report_path}") + return 2 + + try: + if inspect_only: + report = inspect_session_database( + source, + work_dir=getattr(args, "work_dir", None), + ) + else: + last_progress = {"table": None} + + def _recovery_progress(info): + table = info.get("table") + copied = int(info.get("copied_rows") or 0) + total = info.get("source_rows") + if table != last_progress["table"]: + if last_progress["table"] is not None: + print() + print(f" {table}: ", end="", flush=True) + last_progress["table"] = table + suffix = f"/{int(total):,}" if total is not None else "" + print(f"\r {table}: {copied:,}{suffix}", end="", flush=True) + + print("Recovering canonical session data into a new database…") + report = recover_session_database( + source, + output, + work_dir=getattr(args, "work_dir", None), + chunk_size=getattr(args, "chunk_size", 1000), + progress_cb=_recovery_progress, + allow_partial=allow_partial, + ) + if last_progress["table"] is not None: + print() + except (SessionRecoveryError, OSError, _sqlite3.DatabaseError) as exc: + print(f"Error: session recovery failed: {exc}") + print("The supplied source database was not replaced or deleted.") + return 1 + + if report_path is not None: + try: + written_report = write_recovery_report(report_path, report) + except (FileExistsError, OSError) as exc: + print(f"Error: could not write recovery report: {exc}") + return 1 + print(f"Recovery report: {written_report}") + else: + print(_json.dumps(report, indent=2, sort_keys=True)) + + if inspect_only: + return 0 if report.get("recoverable") else 1 + 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.") + return 0 + if allow_partial and report.get("verified"): + counts = report.get("verification", {}).get("table_counts", {}) + print(f"✓ Partial recovery output verified at: {output}") + print( + " Recovered " + f"{int(counts.get('sessions') or 0):,} sessions and " + f"{int(counts.get('messages') or 0):,} messages." + ) + print(" The active session database was not changed.") + print( + " 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.") + return 1 + + try: + from hermes_state import SessionDB + + db = SessionDB() + except Exception as e: + print(f"Error: Could not open session database: {e}") + return + + # Hide third-party tool sessions by default, but honour explicit --source + _source = getattr(args, "source", None) + _exclude = None if _source else ["tool"] + + if action == "list": + from hermes_state import workspace_key as _ws_key + + sessions = db.list_sessions_rich( + source=args.source, exclude_sources=_exclude, limit=args.limit + ) + + # Workspace filter: match a session by its workspace key (git repo + # root, else cwd) — path substring or exact basename. + _ws_filter = (getattr(args, "workspace", None) or "").strip() + if _ws_filter: + _needle = _ws_filter.lower() + + def _in_workspace(s): + key = (_ws_key(s) or "").lower() + return bool(key) and ( + _needle in key or _needle == os.path.basename(key.rstrip("/\\")) + ) + + sessions = [s for s in sessions if _in_workspace(s)] + + if not sessions: + print("No sessions found.") + return + + # Short workspace label: the repo/dir basename, "—" when unbound. The + # Workspace column only appears once at least one session carries one + # (or when filtering), so all-unbound listings read as before. + def _ws_label(s): + key = _ws_key(s) + return (os.path.basename(key.rstrip("/\\")) or key) if key else "—" + + has_ws = bool(_ws_filter) or any(_ws_key(s) for s in sessions) + has_titles = any(s.get("title") for s in sessions) + + if has_ws: + if has_titles: + print(f"{'Title':<28} {'Workspace':<18} {'Last Active':<13} {'ID'}") + print("─" * 110) + else: + print(f"{'Preview':<38} {'Workspace':<18} {'Last Active':<13} {'Src':<6} {'ID'}") + print("─" * 100) + for s in sessions: + last_active = _relative_time(s.get("last_active")) + ws = _ws_label(s)[:16] + if has_titles: + title = (s.get("title") or "—")[:26] + print(f"{title:<28} {ws:<18} {last_active:<13} {s['id']}") + else: + preview = s.get("preview", "")[:36] + print(f"{preview:<38} {ws:<18} {last_active:<13} {s['source']:<6} {s['id']}") + return + + if has_titles: + print(f"{'Title':<32} {'Preview':<40} {'Last Active':<13} {'ID'}") + print("─" * 110) + else: + print(f"{'Preview':<50} {'Last Active':<13} {'Src':<6} {'ID'}") + print("─" * 95) + for s in sessions: + last_active = _relative_time(s.get("last_active")) + preview = ( + s.get("preview", "")[:38] + if has_titles + else s.get("preview", "")[:48] + ) + if has_titles: + title = (s.get("title") or "—")[:30] + sid = s["id"] + print(f"{title:<32} {preview:<40} {last_active:<13} {sid}") + else: + sid = s["id"] + print(f"{preview:<50} {last_active:<13} {s['source']:<6} {sid}") + + elif action == "export": + from hermes_cli.session_filters import ( + build_prune_filters, + describe_filters, + ) + + _filter_arg_names = ( + "older_than", "newer_than", "before", "after", + "source", "title", "end_reason", "cwd", + "min_messages", "max_messages", "model", "provider", + "user", "chat_id", "chat_type", "branch", + "min_tokens", "max_tokens", "min_cost", "max_cost", + "min_tool_calls", "max_tool_calls", + ) + _any_filters = any( + getattr(args, a, None) is not None for a in _filter_arg_names + ) + filters = None + if _any_filters: + try: + filters = build_prune_filters(args) + except ValueError as e: + print(f"Error: {e}") + return + # Unlike prune/archive, export includes archived sessions. + filters["archived"] = None + + def _redact(data): + if not args.redact or data is None: + return data + from hermes_cli.session_export_md import redact_session_data + + 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.""" + if args.session_id: + resolved = db.resolve_session_id(args.session_id) + data = _redact(db.export_session(resolved)) if resolved else None + if not data: + print(f"Session '{args.session_id}' not found.") + return None + return [data] + if filters: + candidates = db.list_prune_candidates(**filters) + if args.dry_run: + print( + f"Would export {len(candidates)} session(s) " + f"({describe_filters(filters)})." + ) + for row in candidates[:100]: + print(f" {row.get('id')} {row.get('source', '')}") + if len(candidates) > 100: + print(f" ... {len(candidates) - 100} more") + return None + return [ + s + for s in ( + _redact(db.export_session(row["id"])) for row in candidates + ) + if s + ] + if args.dry_run: + print("--dry-run requires at least one filter.") + return None + return [_redact(s) for s in db.export_all(source=None)] + + # Prompt-only export (--only user-prompts): one prompt record per + # line (jsonl) or headed sections (md). Delegates rendering to + # hermes_cli.session_export. + if getattr(args, "only", None): + if args.format not in ("jsonl", "md"): + print("--only user-prompts supports --format jsonl or md.") + return + from hermes_cli.session_export import ( + export_record_count, + render_sessions_export, + ) + + sessions = _collect_sessions() + if sessions is None: + db.close() + return + rendered = render_sessions_export( + sessions, + fmt="markdown" if args.format == "md" else "jsonl", + only=args.only, + ) + if not args.output or args.output == "-": + sys.stdout.write(rendered) + db.close() + return + with open(args.output, "w", encoding="utf-8") as f: + f.write(rendered) + count, noun = export_record_count(sessions, only=args.only) + suffix = "" if count == 1 else "s" + print(f"Exported {count} {noun}{suffix} to {args.output}") + db.close() + return + + # Standalone HTML export: one self-contained file (single session + # or multi-session with sidebar navigation). + if args.format == "html": + if not args.output or args.output == "-": + print("HTML export requires an output file path.") + return + from hermes_cli.session_export_html import ( + generate_html_export, + generate_multi_session_html_export, + ) + + sessions = _collect_sessions() + if sessions is None: + db.close() + return + if len(sessions) == 1: + content = generate_html_export(sessions[0]) + else: + content = 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)") + db.close() + return + + # Claude Code JSONL trace export — local file or HF upload. + # Redaction is ON by default for traces (they leave the machine + # when --upload is used); --no-redact opts out after review. + if args.format == "trace": + if getattr(args, "only", None): + print("--only user-prompts supports --format jsonl or md.") + db.close() + return + session_id = args.session_id + if not session_id and not filters: + # Match the shell's common intent: "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: + print("No session found to export. Pass --session-id.") + db.close() + return + if session_id and not db.resolve_session_id(session_id): + print(f"Session '{session_id}' not found.") + db.close() + return + + from agent.trace_upload import ( + TraceRedactionError, + build_trace_jsonl, + upload_session_trace, + ) + + redact_trace = not getattr(args, "no_redact", False) + + if getattr(args, "upload", False): + if not session_id: + print("--upload exports one session: pass --session-id (or drop filters to use the most recent).") + db.close() + return + resolved = db.resolve_session_id(session_id) + db.close() + status = upload_session_trace( + resolved, + cwd="", + redact=redact_trace, + private=not getattr(args, "public", False), + ) + print(status) + return + + # Local trace file(s) + def _trace_ids(): + if session_id: + return [db.resolve_session_id(session_id)] + candidates = db.list_prune_candidates(**filters) + if args.dry_run: + print( + f"Would export {len(candidates)} session(s) " + f"({describe_filters(filters)})." + ) + for row in candidates[:100]: + print(f" {row.get('id')} {row.get('source', '')}") + if len(candidates) > 100: + print(f" ... {len(candidates) - 100} more") + return None + return [row["id"] for row in candidates] + + ids = _trace_ids() + if ids is None: + db.close() + return + + def _render_trace(sid): + meta = db.get_session(sid) or {} + messages = db.get_messages_as_conversation(sid) + if not messages: + return None + return build_trace_jsonl( + messages, + session_id=sid, + model=meta.get("model") or "", + cwd="", + redact=redact_trace, + ) + + try: + if len(ids) == 1: + jsonl = _render_trace(ids[0]) + if not jsonl: + print(f"No transcript to export for session '{ids[0]}'.") + db.close() + return + if not args.output or args.output == "-": + sys.stdout.write(jsonl) + else: + with open(args.output, "w", encoding="utf-8") as f: + f.write(jsonl) + print(f"Exported 1 session trace to {args.output}") + else: + out_dir = ( + Path(args.output).expanduser() + if args.output and args.output != "-" + else get_hermes_home() / "session-exports" + ) + out_dir.mkdir(parents=True, exist_ok=True) + exported = 0 + for sid in ids: + jsonl = _render_trace(sid) + if not jsonl: + continue + (out_dir / f"{sid}.trace.jsonl").write_text( + jsonl, encoding="utf-8" + ) + exported += 1 + print(f"Exported {exported} session trace(s) to {out_dir}") + except TraceRedactionError: + print("Redaction failed; refusing to export unredacted trace content.") + db.close() + return + + if args.format == "jsonl": + if not args.output: + print("JSONL export requires an output path (use - for stdout).") + return + if args.session_id: + resolved_session_id = db.resolve_session_id(args.session_id) + if not resolved_session_id: + print(f"Session '{args.session_id}' not found.") + return + data = _redact(db.export_session(resolved_session_id)) + if not data: + print(f"Session '{args.session_id}' not found.") + return + line = _json.dumps(data, ensure_ascii=False) + "\n" + if args.output == "-": + + sys.stdout.write(line) + else: + with open(args.output, "w", encoding="utf-8") as f: + f.write(line) + print(f"Exported 1 session to {args.output}") + else: + if filters: + candidates = db.list_prune_candidates(**filters) + if args.dry_run: + print( + f"Would export {len(candidates)} session(s) " + f"({describe_filters(filters)})." + ) + for row in candidates[:100]: + print(f" {row.get('id')} {row.get('source', '')}") + if len(candidates) > 100: + print(f" ... {len(candidates) - 100} more") + return + sessions = [ + s + for s in ( + db.export_session(row["id"]) for row in candidates + ) + if s + ] + else: + if args.dry_run: + print("--dry-run requires at least one filter.") + return + sessions = db.export_all(source=None) + if args.output == "-": + + for s in sessions: + sys.stdout.write( + _json.dumps(_redact(s), ensure_ascii=False) + "\n" + ) + else: + with open(args.output, "w", encoding="utf-8") as f: + for s in sessions: + f.write( + _json.dumps(_redact(s), ensure_ascii=False) + "\n" + ) + print(f"Exported {len(sessions)} sessions to {args.output}") + return + + # Markdown / QMD export + from hermes_cli.session_export_md import ( + append_manifest_entry, + verify_export_file, + write_session_markdown, + ) + + if args.output == "-": + print("Markdown/QMD export writes files; stdout (-) is only supported with --format jsonl.") + db.close() + return + output_dir = Path(args.output).expanduser() if args.output else get_hermes_home() / "session-exports" + + 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) + ) + if not data: + return None, None + data = _redact(data) + path = write_session_markdown( + data, + output_dir, + fmt=args.format, + force=args.force, + ) + append_manifest_entry(output_dir, data, path, fmt=args.format) + return data, path + + if args.delete_after_verified and not args.yes: + print("--delete-after-verified requires --yes.") + db.close() + return + if args.delete_after_verified and not args.session_id: + print("--delete-after-verified is only supported with --session-id.") + db.close() + return + + lineage_is_logical = getattr(args, "lineage", "single") == "logical" + + if args.session_id: + resolved_session_id = db.resolve_session_id(args.session_id) + if not resolved_session_id: + print(f"Session '{args.session_id}' not found.") + db.close() + return + delete_target_ids = [resolved_session_id] + if args.delete_after_verified: + delete_target_ids = db.get_session_delete_targets( + resolved_session_id + ) + + exported_items = [] + 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 + ), + ) + except FileExistsError as e: + print( + f"Export already exists: {e}. " + "Pass --force to overwrite." + ) + db.close() + return + if not data or not exported_path: + print( + f"Session '{target_id}' disappeared during export; " + "nothing was deleted." + ) + db.close() + return + exported_items.append((data, exported_path)) + + 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}) " + f"to {exported_items[0][1]}" + ) + else: + print( + f"Exported {len(exported_items)} sessions " + f"({message_count} message{suffix}) to {output_dir}" + ) + if args.delete_after_verified: + for data, exported_path in exported_items: + ok, reason = verify_export_file(exported_path, data) + if not ok: + print( + "Export verification failed; not deleting " + f"session '{data.get('id')}': {reason}" + ) + db.close() + return + sessions_dir = get_hermes_home() / "sessions" + 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" + f"{'' if delegate_count == 1 else 's'}" + ) + print( + f"Deleted exported session '{resolved_session_id}'" + f"{delegate_suffix}." + ) + else: + print( + f"Exported, but session '{resolved_session_id}' was " + "not deleted because its delegate set changed." + ) + db.close() + return + + if not filters: + print( + "Refusing bulk export without a filter. Pass --session-id or " + "at least one filter (e.g. --older-than 90, --source telegram)." + ) + db.close() + return + candidates = db.list_prune_candidates(**filters) + if args.dry_run: + print( + f"Would export {len(candidates)} session(s) " + f"({describe_filters(filters)})." + ) + for row in candidates[:100]: + print(f" {row.get('id')} {row.get('source', '')}") + if len(candidates) > 100: + print(f" ... {len(candidates) - 100} more") + db.close() + return + exported = 0 + for row in candidates: + try: + data, exported_path = _export_one( + row["id"], + include_lineage=lineage_is_logical, + ) + except FileExistsError as e: + print(f"Skipping existing export: {e}. Pass --force to overwrite.") + continue + if data and exported_path: + exported += 1 + print(f"Exported {exported} session(s) to {output_dir}") + + elif action == "delete": + resolved_session_id = db.resolve_session_id(args.session_id) + if not resolved_session_id: + print(f"Session '{args.session_id}' not found.") + return + if not args.yes: + if not _confirm_prompt( + f"Delete session '{resolved_session_id}' and all its messages? [y/N] " + ): + print("Cancelled.") + return + sessions_dir = get_hermes_home() / "sessions" + if db.delete_session(resolved_session_id, sessions_dir=sessions_dir): + print(f"Deleted session '{resolved_session_id}'.") + else: + print(f"Session '{args.session_id}' not found.") + + elif action in ("prune", "archive"): + from hermes_cli.session_filters import ( + build_prune_filters, + describe_filters, + format_epoch, + ) + + # Preserve the historical default ONLY for a truly bare + # `hermes sessions prune`: no time window and no filters at all + # means "older than 90 days". ANY filter — including --source — + # suppresses the implicit cutoff, so `prune --source cron` + # matches ALL cron sessions regardless of age. The preview + + # confirmation below (count, oldest/newest) is the safety net. + _non_time_filters = any( + getattr(args, a, None) is not None + for a in ( + "source", "title", "end_reason", "cwd", + "min_messages", "max_messages", "model", "provider", + "user", "chat_id", "chat_type", "branch", + "min_tokens", "max_tokens", "min_cost", "max_cost", + "min_tool_calls", "max_tool_calls", + ) + ) + if ( + action == "prune" + and args.older_than is None + and args.newer_than is None + and args.before is None + and args.after is None + and not _non_time_filters + ): + args.older_than = "90" + + try: + filters = build_prune_filters(args) + except ValueError as e: + print(f"Error: {e}") + return + + 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)." + ) + return + + # Prune skips archived sessions unless --include-archived; + # archive only targets not-yet-archived rows (idempotent). + if action == "prune": + filters["archived"] = ( + None if getattr(args, "include_archived", False) else False + ) + else: + filters["archived"] = False + + candidates = db.list_prune_candidates(**filters) + verb = "Delete" if action == "prune" else "Archive" + if not candidates: + print(f"No sessions match ({describe_filters(filters)}).") + return + + # Candidates are ordered by activity oldest-first. Surface that + # span so a long-lived but recently used conversation cannot look + # old merely because of its creation date. + _oldest = candidates[0].get("last_active") + _newest = candidates[-1].get("last_active") + _span = ( + f"oldest activity {format_epoch(_oldest)}, " + f"newest activity {format_epoch(_newest)}" + ) + + if args.dry_run or not args.yes: + shown = candidates if args.dry_run else candidates[:15] + print( + f"{len(candidates)} session(s) match " + f"({describe_filters(filters)}; {_span}):" + ) + for s in shown: + 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['message_count']:>4} msgs {title}" + ) + if len(candidates) > len(shown): + print(f" … and {len(candidates) - len(shown)} more") + if args.dry_run: + print(f"Dry run — nothing {'deleted' if action == 'prune' else 'archived'}.") + return + + if not args.yes: + if not _confirm_prompt( + f"{verb} these {len(candidates)} session(s) ({_span})? [y/N] " + ): + print("Cancelled.") + return + + if action == "prune": + sessions_dir = get_hermes_home() / "sessions" + count = db.prune_sessions(sessions_dir=sessions_dir, **filters) + print(f"Pruned {count} session(s).") + else: + count = db.archive_sessions(**filters) + print( + f"Archived {count} session(s). They're hidden from listings " + "but fully recoverable (nothing was deleted)." + ) + + elif action == "rename": + resolved_session_id = db.resolve_session_id(args.session_id) + if not resolved_session_id: + print(f"Session '{args.session_id}' not found.") + return + title = " ".join(args.title) + try: + if db.set_session_title(resolved_session_id, title): + print(f"Session '{resolved_session_id}' renamed to: {title}") + else: + print(f"Session '{args.session_id}' not found.") + except ValueError as e: + print(f"Error: {e}") + + elif action == "retitle-skills": + from agent.skill_commands import describe_skill_invocation + from agent.title_generator import generate_title + + limit = max(1, int(getattr(args, "limit", 200) or 200)) + apply_changes = bool(getattr(args, "apply", False)) + + def _is_titlelike(candidate: str) -> bool: + """Reject a candidate that isn't a title at all. + + An auxiliary model occasionally answers the prompt instead of + titling it and echoes the assistant's output ('$ df -h /'). The + live path has no alternative and takes what it gets, but this is + a REPAIR — replacing a serviceable title with command output + would make things worse, so keep the old one. + """ + return bool(candidate) and candidate[0].isalnum() + + candidates = db.list_skill_scaffolded_sessions(limit=limit) + if not candidates: + 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)'}:" + ) + changed = 0 + for row in candidates: + session_id = row["id"] + typed = describe_skill_invocation(row["content"]) or "" + first_reply = db.get_first_assistant_text(session_id) or "" + new_title = generate_title(typed, first_reply) + if not new_title or new_title == row["title"]: + continue + if not _is_titlelike(new_title): + print(f" {session_id}\n kept {row['title']!r} — got {new_title!r}") + continue + print(f" {session_id}\n {row['title']!r}\n → {new_title!r}") + changed += 1 + if not apply_changes: + continue + try: + db.set_session_title(session_id, new_title) + except ValueError: + # Unique-title collision. Dedupe the same way the live + # auto-titler does (base #2, base #3, ...) rather than + # leaving the leaked title in place. + deduped = db.get_next_title_in_lineage(new_title) + try: + db.set_session_title(session_id, deduped) + print(f" (renamed to {deduped!r} — title was taken)") + except ValueError as e: + print(f" skipped: {e}") + changed -= 1 + + if not changed: + print(" every title already reflects the user's request.") + elif apply_changes: + print(f"✓ Re-titled {changed} session(s).") + + elif action == "browse": + limit = getattr(args, "limit", 500) or 500 + source = getattr(args, "source", None) + _browse_exclude = None if source else ["tool"] + sessions = db.list_sessions_rich( + source=source, exclude_sources=_browse_exclude, limit=limit + ) + db.close() + if not sessions: + print("No sessions found.") + return + + selected_id = _session_browse_picker(sessions) + if not selected_id: + print("Cancelled.") + return + + # Launch hermes --resume by replacing the current process + print(f"Resuming session: {selected_id}") + from hermes_cli.relaunch import relaunch + + relaunch(["--resume", selected_id]) + return # won't reach here after execvp + + elif action == "optimize": + db_path = db.db_path + before_mb = ( + os.path.getsize(db_path) / (1024 * 1024) + if db_path.exists() + else 0.0 + ) + print("Optimizing session store (FTS merge + VACUUM)…") + try: + # vacuum() merges FTS5 segments (optimize_fts) then VACUUMs, + # and returns the number of indexes it merged. + n = db.vacuum() + except Exception as e: + print(f"Error: optimization failed: {e}") + db.close() + return + after_mb = ( + os.path.getsize(db_path) / (1024 * 1024) + if db_path.exists() + else 0.0 + ) + # Same WAL caveat as optimize-storage: after a VACUUM the main file + # on disk lags until the WAL is checkpointed back (refused while a + # live gateway holds a read-mark), so stat() understates the win and + # can go negative. SQLite's page accounting is correct immediately. + logical_after = db.logical_size_bytes() + if logical_after is not None: + after_mb = logical_after / (1024 * 1024) + saved = before_mb - after_mb + print(f"Optimized {n} FTS index(es).") + print( + f"Database size: {before_mb:.1f} MB -> {after_mb:.1f} MB " + f"({_size_delta_label(saved)})" + ) + + elif action == "optimize-storage": + db_path = db.db_path + if not db.fts_optimize_available(): + print("Search index is already on the compact layout — nothing to do.") + db.close() + return + + before_bytes = os.path.getsize(db_path) if db_path.exists() else 0 + before_mb = before_bytes / (1024 * 1024) + + # Disk preflight: the rebuild adds the new index before the old is + # torn down, and the final VACUUM needs a full second copy of the + # file. Require headroom ≈ current file size to finish cleanly. + do_vacuum = not getattr(args, "no_vacuum", False) + try: + import shutil as _shutil + free_bytes = _shutil.disk_usage(db_path.parent).free + except Exception: + free_bytes = None + need_bytes = before_bytes if do_vacuum else int(before_bytes * 0.3) + print(f"Search-index optimization for {db_path}") + 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 ''})") + 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).") + db.close() + 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).") + if not getattr(args, "yes", False): + try: + resp = input("Proceed? [y/N] ").strip().lower() + except EOFError: + resp = "" + if resp not in ("y", "yes"): + print("Cancelled.") + db.close() + return + + _last = {"phase": None} + + 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) + elif phase != _last["phase"]: + label = {"teardown": "Reclaiming old index", + "vacuum": "Compacting database (VACUUM)", + "done": "Done"}.get(phase, phase) + print(f"\n {label}…", flush=True) + _last["phase"] = phase + + print("Optimizing search-index storage…") + try: + result = db.optimize_fts_storage( + progress_cb=_progress, vacuum=do_vacuum + ) + except Exception as e: + print(f"\nError: optimization failed: {e}") + print("No data was lost. Re-run to resume.") + db.close() + return + if not result.get("ok"): + print(f"\nCould not optimize: {result.get('reason', 'unknown')}") + db.close() + return + after_mb = ( + os.path.getsize(db_path) / (1024 * 1024) if db_path.exists() else 0.0 + ) + # Prefer SQLite's own page accounting over stat(). In WAL mode a + # VACUUM's rewrite sits in the -wal file until a checkpoint folds it + # back, and that checkpoint is refused while another connection (a + # live gateway) holds a read-mark — so the main file on disk still + # reads at its pre-VACUUM size and keeps growing. stat()ing it here + # reported "reclaimed -3820.1 MB" on a DB that had actually shrunk + # 60%. page_count * page_size is correct immediately. + logical_after = db.logical_size_bytes() + if logical_after is not None: + after_mb = logical_after / (1024 * 1024) + saved = before_mb - after_mb + print(f"\n✓ Search index optimized.") + print( + f" Database size: {before_mb:.1f} MB -> {after_mb:.1f} MB " + f"({_size_delta_label(saved)})" + ) + if result.get("vacuumed") is False: + print(" (VACUUM was skipped or failed — run " + "`hermes sessions optimize` later to reclaim freed space.)") + + elif action == "stats": + total = db.session_count() + msgs = db.message_count() + print(f"Total sessions: {total}") + print(f"Total messages: {msgs}") + for src in ["cli", "telegram", "discord", "whatsapp", "slack"]: + c = db.session_count(source=src) + if c > 0: + print(f" {src}: {c} sessions") + db_path = db.db_path + if db_path.exists(): + size_mb = os.path.getsize(db_path) / (1024 * 1024) + print(f"Database size: {size_mb:.1f} MB") + + else: + sessions_parser.print_help() + + db.close()