From 0e5190cb9577b89667eeadb49f1ca7f1cf0cbb5f Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:27:46 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20console=5Fengine=20?= =?UTF-8?q?=E2=80=94=20merge=20command=20tables,=20closing()=20for=20Sessi?= =?UTF-8?q?onDB,=20inline=20single-use=20helpers,=20compact=20layout?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/console_engine.py | 533 ++++++++++++----------------------- 1 file changed, 181 insertions(+), 352 deletions(-) diff --git a/hermes_cli/console_engine.py b/hermes_cli/console_engine.py index ae892f2eb6..a95ffd9b67 100644 --- a/hermes_cli/console_engine.py +++ b/hermes_cli/console_engine.py @@ -10,6 +10,7 @@ import importlib import io import json import sys +from contextlib import closing from dataclasses import dataclass from pathlib import Path from typing import Callable, Iterable, Literal, NoReturn, Sequence @@ -48,45 +49,36 @@ class _ArgumentParser(argparse.ArgumentParser): def _capture_output(fn: Callable[[], object]) -> str: - stdout = io.StringIO() - stderr = io.StringIO() - code = 0 - message = "" + stdout, stderr = io.StringIO(), io.StringIO() + code, message = 0, "" with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): try: result = fn() if isinstance(result, int) and result: raise SystemExit(result) except SystemExit as exc: - # sys.exit("msg") / raise SystemExit("msg") is the standard non-zero-exit idiom: - # exc.code is the message string, not an int. int() would raise ValueError here, - # which escapes execute()'s ConsoleCommandError handler and crashes the REPL. + # sys.exit("msg") carries the message as exc.code (not an int); int() on it would + # raise ValueError past execute()'s handler and crash the REPL. if isinstance(exc.code, str): - message = exc.code - code = 1 + message, code = exc.code, 1 else: code = int(exc.code or 0) except ConsoleCommandError: raise except RuntimeError as exc: - # Fail-closed config write guards raise RuntimeError (e.g. - # require_readable_config_before_write refusing an unparseable - # config.yaml). Convert to a console error instead of letting it - # escape execute() and kill the REPL / websocket session. - message = str(exc) - code = 1 + # Fail-closed config write guards raise RuntimeError; surface it as a console error + # instead of killing the REPL / websocket session. + message, code = str(exc), 1 text = stdout.getvalue() + stderr.getvalue() if code: - raise ConsoleCommandError(message.strip() or text.strip() or f"Command exited with status {code}") + raise ConsoleCommandError( + message.strip() or text.strip() or f"Command exited with status {code}") return text.rstrip() def _is_status_footer_rule(line: str) -> bool: stripped = _strip_ansi(line).strip() - if len(stripped) < 8: - return False - normalized = stripped.replace("\u2500", "-") - return set(normalized) <= {"-"} + return len(stripped) >= 8 and set(stripped.replace("\u2500", "-")) <= {"-"} def _strip_console_status_footer(text: str) -> str: @@ -95,15 +87,9 @@ def _strip_console_status_footer(text: str) -> str: lines.pop() if len(lines) < 2: return text.rstrip() - - last = _strip_ansi(lines[-1]).strip() - prev = _strip_ansi(lines[-2]).strip() - if not ( - prev.startswith("Run 'hermes doctor'") - and last.startswith("Run 'hermes setup'") - ): + last, prev = (_strip_ansi(lines[i]).strip() for i in (-1, -2)) + if not (prev.startswith("Run 'hermes doctor'") and last.startswith("Run 'hermes setup'")): return text.rstrip() - lines = lines[:-2] while lines and not _strip_ansi(lines[-1]).strip(): lines.pop() @@ -114,43 +100,38 @@ def _strip_console_status_footer(text: str) -> str: def _table_summary(summary: str, *, limit: int = 76) -> str: summary = " ".join(summary.split()) - if len(summary) <= limit: - return summary - return f"{summary[: limit - 3].rstrip()}..." + return summary if len(summary) <= limit else f"{summary[: limit - 3].rstrip()}..." def _split_line(line: str) -> list[str]: - try: - # Windows-safe splitter: plain shlex posix=True eats backslashes, so - # `sessions export C:\Users\me\out.jsonl` silently became a mangled - # relative filename in the cwd (#83934). - from hermes_cli._subprocess_compat import split_command_line + # Windows-safe splitter: plain shlex posix=True eats backslashes in paths. + from hermes_cli._subprocess_compat import split_command_line + try: return split_command_line(line) except ValueError as exc: raise ConsoleCommandError(f"Could not parse command: {exc}") from exc +_SHELL_TOKENS = {"|", "||", "&", "&&", ";", ">", ">>", "<", "<<", "2>", "2>>"} + + def _contains_shell_syntax(line: str, tokens: Sequence[str]) -> bool: - if "$(" in line or "`" in line: - return True - shell_tokens = {"|", "||", "&", "&&", ";", ">", ">>", "<", "<<", "2>", "2>>"} - if any(token in shell_tokens for token in tokens): - return True - return any(ch in line for ch in "|<>;") + return ( + "$(" in line or "`" in line + or any(token in _SHELL_TOKENS for token in tokens) + or any(ch in line for ch in "|<>;")) def _format_sessions(sessions: Sequence[dict]) -> str: if not sessions: return "No sessions found." - lines = [f"{'ID':<32} {'Source':<12} {'Msgs':>5} Title / Preview"] - lines.append("-" * 82) + lines = [f"{'ID':<32} {'Source':<12} {'Msgs':>5} Title / Preview", "-" * 82] for session in sessions: sid = str(session.get("id") or "")[:32] source = str(session.get("source") or "-")[:12] messages = session.get("message_count") or 0 - title = session.get("title") or session.get("preview") or "" - title = str(title).replace("\n", " ")[:60] + title = str(session.get("title") or session.get("preview") or "").replace("\n", " ")[:60] lines.append(f"{sid:<32} {source:<12} {messages:>5} {title}") return "\n".join(lines) @@ -159,53 +140,37 @@ def _format_job(job: dict, action: str) -> str: from cron.jobs import effective_job_state job_id = job.get("id") or job.get("job_id") or "?" - name = job.get("name") or "(unnamed)" - state = effective_job_state(job) - return f"{action} job: {name} ({job_id}) [{state}]" - - -def _parser_root() -> tuple[_ArgumentParser, argparse._SubParsersAction]: - parser = _ArgumentParser(prog="hermes", add_help=False) - subparsers = parser.add_subparsers(dest="_console_command") - return parser, subparsers - - -def _subparser_actions(parser: argparse.ArgumentParser) -> list[argparse._SubParsersAction]: - return [ - action - for action in parser._actions - if isinstance(action, argparse._SubParsersAction) - ] - - -def _choice_help(action: argparse._SubParsersAction, name: str) -> str: - for choice in action._choices_actions: - if getattr(choice, "dest", None) == name or getattr(choice, "metavar", None) == name: - help_text = getattr(choice, "help", None) - if help_text and help_text is not argparse.SUPPRESS: - return str(help_text) - return "" + return f"{action} job: {job.get('name') or '(unnamed)'} ({job_id}) [{effective_job_state(job)}]" def _clean_summary(text: str | None) -> str: if not text or text is argparse.SUPPRESS: return "" summary = " ".join(str(text).split()) - if not summary or summary.startswith("Run `hermes "): - return "" - return summary + return "" if summary.startswith("Run `hermes ") else summary + + +def _choice_help(action: argparse._SubParsersAction, name: str) -> str: + for choice in action._choices_actions: + if name in (getattr(choice, "dest", None), getattr(choice, "metavar", None)): + help_text = getattr(choice, "help", None) + if help_text and help_text is not argparse.SUPPRESS: + return str(help_text) + return "" def _summaries_from_parser(parser: argparse.ArgumentParser) -> dict[tuple[str, ...], str]: + """``{subcommand path: help}`` over the whole tree; choice help beats child description.""" summaries: dict[tuple[str, ...], str] = {} def walk(current: argparse.ArgumentParser, path: tuple[str, ...]) -> None: - for action in _subparser_actions(current): + for action in current._actions: + if not isinstance(action, argparse._SubParsersAction): + continue for name, child in action.choices.items(): child_path = (*path, name) summary = _clean_summary(_choice_help(action, name)) or _clean_summary( - child.description - ) + child.description) if summary: summaries.setdefault(child_path, summary) walk(child, child_path) @@ -222,11 +187,9 @@ def _noop_console_command(_args: argparse.Namespace) -> None: class _CliSurface: """How a CLI subcommand module hangs its argparse tree off a root subparsers action. - ``kind`` selects the wiring convention: - * ``extracted`` — ``builder(subparsers, =fn)`` (hermes_cli.subcommands.*; fn from hermes_cli.main) - * ``registered`` — ``register(subparsers.add_parser(root))``; optional module-level ``handler`` as func - * ``builder`` — ``top = builder(subparsers)``; func = ``handler`` from hermes_cli.main - * ``adder`` — ``add(subparsers)`` wires its own func + ``kind``: ``extracted`` = ``builder(subparsers, =fn)`` (fn from hermes_cli.main); + ``registered`` = ``register(subparsers.add_parser(root))`` with optional module ``handler``; + ``builder`` = ``top = builder(subparsers)``, func from hermes_cli.main; ``adder`` wires itself. """ kind: Literal["extracted", "registered", "builder", "adder"] @@ -235,37 +198,32 @@ class _CliSurface: handler: str | None = None def build(self, root: str, *, live: bool) -> _ArgumentParser: - """Build a throwaway parser. ``live=False`` wires no-op handlers (summary extraction only).""" - parser, subparsers = _parser_root() + """Build a throwaway parser; ``live=False`` wires no-op handlers (summary extraction).""" + parser = _ArgumentParser(prog="hermes", add_help=False) + subparsers = parser.add_subparsers(dest="_console_command") module = importlib.import_module(self.module) entry = getattr(module, self.builder) + main_handler = ( + (lambda: getattr(importlib.import_module("hermes_cli.main"), self.handler)) + if live else (lambda: _noop_console_command)) if self.kind == "extracted": - fn = ( - getattr(importlib.import_module("hermes_cli.main"), self.handler) - if live - else _noop_console_command - ) - entry(subparsers, **{self.handler: fn}) + entry(subparsers, **{self.handler: main_handler()}) elif self.kind == "registered": top_parser = subparsers.add_parser(root) entry(top_parser) if live and self.handler: top_parser.set_defaults(func=getattr(module, self.handler)) elif self.kind == "builder": - main_module = importlib.import_module("hermes_cli.main") if live else None top_parser = entry(subparsers) if live: - top_parser.set_defaults(func=getattr(main_module, self.handler)) + top_parser.set_defaults(func=main_handler()) else: entry(subparsers) return parser -# The CLI surface these helpers reflect is process-static: they import a -# subcommand module and build a throwaway argparse tree purely to extract help -# summaries. Nothing about the result changes across engine instances, but the -# dashboard opens a fresh HermesConsoleEngine per /api/console connection, so -# without memoization every reconnect re-imports + re-parses the whole surface. +# Memoized: the surface is process-static, but the dashboard opens a fresh engine per +# /api/console connection and would otherwise re-import + re-parse it on every reconnect. @functools.lru_cache(maxsize=None) def _surface_summaries(surface: _CliSurface, root: str) -> dict[tuple[str, ...], str]: try: @@ -274,30 +232,26 @@ def _surface_summaries(surface: _CliSurface, root: str) -> dict[tuple[str, ...], return {} -def _invoke_namespace(args: argparse.Namespace) -> object: - func = getattr(args, "func", None) - if not callable(func): - raise ConsoleCommandError("No handler is available for that console command.") - return func(args) - - def _dispatch( surface: _CliSurface, root: str, fixed: Sequence[str], args: Sequence[str], - namespace_update: Callable[[argparse.Namespace], None] | None = None, -) -> str: - parser = surface.build(root, live=True) - namespace = parser.parse_args([root, *fixed, *args]) + namespace_update: Callable[[argparse.Namespace], None] | None = None) -> str: + namespace = surface.build(root, live=True).parse_args([root, *fixed, *args]) if namespace_update: namespace_update(namespace) - return _capture_output(lambda: _invoke_namespace(namespace)) + func = getattr(namespace, "func", None) + if not callable(func): + raise ConsoleCommandError("No handler is available for that console command.") + return _capture_output(lambda: func(namespace)) def _paths(spec: str) -> list[tuple[str, ...]]: - """``"list, snapshot export"`` -> ``[("list",), ("snapshot", "export")]``; ``"."`` is the bare root.""" - return [() if item.strip() == "." else tuple(item.split()) for item in spec.split(",") if item.strip()] + """``"list, snapshot export"`` -> ``[("list",), ("snapshot", "export")]``; ``"."`` is root.""" + return [ + () if item.strip() == "." else tuple(item.split()) + for item in spec.split(",") if item.strip()] def _sub(module: str, builder: str, handler: str) -> _CliSurface: @@ -321,58 +275,49 @@ _CLI_FAMILIES: dict[str, tuple[_CliSurface, str, str]] = { "tools": ( _sub("tools", "build_tools_parser", "cmd_tools"), "list, enable, disable, post-setup", - "enable, disable, post-setup", - ), + "enable, disable, post-setup"), "plugins": ( _sub("plugins", "build_plugins_parser", "cmd_plugins"), "list, enable, disable, install, update, remove", - "enable, disable, install, update, remove", - ), + "enable, disable, install, update, remove"), "skills": ( _sub("skills", "build_skills_parser", "cmd_skills"), "browse, search, inspect, list, check, list-modified, diff, install, update, audit, " "uninstall, reset, opt-in, opt-out, repair-official, snapshot export, snapshot import, " "tap list, tap add, tap remove", "install, update, audit, uninstall, reset, opt-in, opt-out, repair-official, " - "snapshot export, snapshot import, tap add, tap remove", - ), + "snapshot export, snapshot import, tap add, tap remove"), "mcp": ( _sub("mcp", "build_mcp_parser", "cmd_mcp"), "list, catalog, test, add, remove, install, login, reauth, configure, picker", - "add, remove, install, login, reauth, configure, picker", - ), - "memory": (_sub("memory", "build_memory_parser", "cmd_memory"), "status, off, reset", "off, reset"), + "add, remove, install, login, reauth, configure, picker"), + "memory": ( + _sub("memory", "build_memory_parser", "cmd_memory"), "status, off, reset", "off, reset"), "auth": ( _sub("auth", "build_auth_parser", "cmd_auth"), "list, status, reset, add, remove, logout, spotify status, spotify login, spotify logout", - "reset, add, remove, logout, spotify login, spotify logout", - ), + "reset, add, remove, logout, spotify login, spotify logout"), "pairing": ( _sub("pairing", "build_pairing_parser", "cmd_pairing"), "list, approve, revoke, clear-pending", - "approve, revoke, clear-pending", - ), + "approve, revoke, clear-pending"), "webhook": ( _sub("webhook", "build_webhook_parser", "cmd_webhook"), "list, subscribe, remove, test", - "subscribe, remove", - ), + "subscribe, remove"), "hooks": ( _sub("hooks", "build_hooks_parser", "cmd_hooks"), "list, test, doctor, revoke", - "test, doctor, revoke", - ), + "test, doctor, revoke"), "slack": (_sub("slack", "build_slack_parser", "cmd_slack"), "manifest", ""), "profile": ( _sub("profile", "build_profile_parser", "cmd_profile"), "list, show, info, create, use, describe, rename, delete, export, import, install, update", - "create, use, describe, rename, delete, export, import, install, update", - ), + "create, use, describe, rename, delete, export, import, install, update"), "cron": ( _sub("cron", "build_cron_parser", "cmd_cron"), "create, edit, remove, tick", - "create, edit, remove, tick", - ), + "create, edit, remove, tick"), "portal": (_CliSurface("adder", "hermes_cli.portal_cli", "add_parser"), "info, tools", ""), "project": ( _CliSurface("builder", "hermes_cli.projects_cmd", "build_parser", "cmd_project"), @@ -388,30 +333,24 @@ _CLI_FAMILIES: dict[str, tuple[_CliSurface, str, str]] = { "promote, archive, stats, runs, heartbeat, assignments, context", "init, boards create, boards rm, boards switch, boards rename, boards set-workdir, " "create, assign, reclaim, reassign, link, unlink, claim, comment, complete, edit, " - "block, schedule, unblock, promote, archive", - ), + "block, schedule, unblock, promote, archive"), "bundles": ( _reg("bundles", "bundles_command"), "list, show, create, delete, reload", - "create, delete, reload", - ), + "create, delete, reload"), "checkpoints": ( _reg("checkpoints"), "status, list, prune, clear, clear-legacy", - "prune, clear, clear-legacy", - ), + "prune, clear, clear-legacy"), "curator": ( _reg("curator"), "status, run, pause, resume, pin, unpin, restore, list-archived, archive, prune, " "backup, rollback", - "run, pause, resume, pin, unpin, restore, archive, prune, backup, rollback", - ), + "run, pause, resume, pin, unpin, restore, archive, prune, backup, rollback"), "pets": ( _reg("pets"), "list, install, select, show, off, scale, remove, doctor", - "install, select, off, scale, remove", - ), -} + "install, select, off, scale, remove")} # Only extracted/registered families skip nested prompts after console confirmation # (builder/adder families never did). @@ -421,11 +360,7 @@ _SEND_SURFACE = _CliSurface("adder", "hermes_cli.send_cmd", "register_send_subpa def _register_command_family( - engine: "HermesConsoleEngine", - root: str, - surface: _CliSurface, - paths: str, - mutating: str, + engine: "HermesConsoleEngine", root: str, surface: _CliSurface, paths: str, mutating: str ) -> None: summaries = _surface_summaries(surface, root) mutating_paths = set(_paths(mutating)) @@ -443,14 +378,12 @@ def _register_command_family( summaries.get(full_path) or f"Run `hermes {usage}`.", handler, mutating=child_path in mutating_paths, - confirmation=f"Run `hermes {usage}`?", - ) + confirmation=f"Run `hermes {usage}`?") _BLOCKED_TOP = frozenset( "acp chat claw completion dashboard desktop fallback gateway gui login logout model moa " - "oneshot proxy serve setup uninstall update whatsapp whatsapp-cloud".split() -) + "oneshot proxy serve setup uninstall update whatsapp whatsapp-cloud".split()) _BLOCKED_PAIRS = { ("config", "edit"): "`config edit` opens an editor and is not available in Hermes Console.", @@ -486,37 +419,28 @@ class HermesConsoleEngine: raw_line = line.strip() if not raw_line: return ConsoleResult("ok") - try: tokens = _split_line(raw_line) if tokens and tokens[0] == "hermes": tokens = tokens[1:] if not tokens: return ConsoleResult("ok", output=self.help_text()) - if _contains_shell_syntax(raw_line, tokens): raise ConsoleCommandError( "Hermes Console does not run shell syntax. Use one supported " - "Hermes command at a time." - ) - + "Hermes command at a time.") builtin = self._execute_builtin(tokens) if builtin is not None: if raw_line not in {"history", "clear"}: self.history.append(raw_line) return builtin - command, args = self._resolve_command(tokens) if command.mutating and not confirmed: return ConsoleResult( "confirm_required", command=raw_line, - confirmation_message=command.confirmation - or f"Run `{command.usage}`?", - ) - - output = command.handler(self, args).rstrip() - output = self._cap_output(output) + confirmation_message=command.confirmation or f"Run `{command.usage}`?") + output = self._cap_output(command.handler(self, args).rstrip()) self.history.append(raw_line) return ConsoleResult("ok", output=output, command=raw_line) except ConsoleCommandError as exc: @@ -524,32 +448,23 @@ class HermesConsoleEngine: def help_text(self, subject: str | None = None) -> str: if subject: - tokens = subject.split() - command, _args = self._resolve_command(tokens) + command, _args = self._resolve_command(subject.split()) return f"{command.usage}\n{command.summary}" - - lines = [ - "Hermes Console", - "", - "Supported commands:", - ] + lines = ["Hermes Console", "", "Supported commands:"] for command in sorted(self.commands.values(), key=lambda c: c.usage): marker = " *" if command.mutating else " " lines.append(f"{marker} {command.usage:<32} {_table_summary(command.summary)}") - lines.extend( - [ - "", - "* requires confirmation", - "Built-ins: help, help , history, clear, exit, quit", - ] - ) + lines += [ + "", + "* requires confirmation", + "Built-ins: help, help , history, clear, exit, quit"] return "\n".join(lines) def _register_defaults(self) -> None: - for path, usage, summary, handler in _READONLY_COMMANDS: - self.register(path, usage, summary, handler) - for path, usage, summary, handler, confirmation in _MUTATING_COMMANDS: - self.register(path, usage, summary, handler, mutating=True, confirmation=confirmation) + for path, usage, summary, handler, confirmation in _BUILTIN_COMMANDS: + self.register( + path, usage, summary, handler, + mutating=bool(confirmation), confirmation=confirmation) for root, (surface, paths, mutating) in _CLI_FAMILIES.items(): _register_command_family(self, root, surface, paths, mutating) self.register( @@ -558,8 +473,7 @@ class HermesConsoleEngine: "Send a message to a configured platform.", lambda _engine, args: _dispatch(_SEND_SURFACE, "send", (), args), mutating=True, - confirmation="Send this message?", - ) + confirmation="Send this message?") def register( self, @@ -569,17 +483,9 @@ class HermesConsoleEngine: handler: Callable[["HermesConsoleEngine", list[str]], str], *, mutating: bool = False, - confirmation: str = "", - ) -> None: + confirmation: str = "") -> None: key = tuple(path) - self.commands[key] = ConsoleCommand( - path=key, - usage=usage, - summary=summary, - handler=handler, - mutating=mutating, - confirmation=confirmation, - ) + self.commands[key] = ConsoleCommand(key, usage, summary, handler, mutating, confirmation) def _execute_builtin(self, tokens: list[str]) -> ConsoleResult | None: head = tokens[0] @@ -602,13 +508,10 @@ class HermesConsoleEngine: rejected = self._rejection_for(tokens) if rejected: raise ConsoleCommandError(rejected) - for size in range(min(len(tokens), 3), 0, -1): - key = tuple(tokens[:size]) - command = self.commands.get(key) + command = self.commands.get(tuple(tokens[:size])) if command: return command, list(tokens[size:]) - available = [" ".join(path) for path in self.commands] probe = " ".join(tokens[:2]) if len(tokens) > 1 else tokens[0] suggestions = difflib.get_close_matches(probe, available, n=3, cutoff=0.45) @@ -637,39 +540,30 @@ def _expect_no_args(args: Sequence[str], usage: str) -> None: def _apply_confirmed_defaults(args: argparse.Namespace) -> None: """Skip nested prompts after the console-level confirmation has happened.""" - - for attr in ("yes",): - if hasattr(args, attr): - setattr(args, attr, True) + if hasattr(args, "yes"): + args.yes = True if getattr(args, "_console_command", None) == "import": - setattr(args, "force", True) - # Every checkpoints subcommand the console registers as mutating gates its - # own confirmation on --force, so all three belong here. `prune` reaches - # _confirm() for its orphan preview, and the console never redirects stdin. + args.force = True + # Every mutating checkpoints subcommand gates its own confirmation on --force; `prune` + # reaches _confirm() for its orphan preview, and the console never redirects stdin. if getattr(args, "checkpoints_command", None) in {"prune", "clear", "clear-legacy"}: - setattr(args, "force", True) + args.force = True if ( getattr(args, "plugins_action", None) == "install" and not getattr(args, "enable", False) - and not getattr(args, "no_enable", False) - ): - setattr(args, "no_enable", True) + and not getattr(args, "no_enable", False)): + args.no_enable = True if getattr(args, "auth_action", None) == "add": auth_type = getattr(args, "auth_type", None) if auth_type in {"api-key", "api_key"} and not getattr(args, "api_key", None): - raise ConsoleCommandError("auth add --type api-key requires --api-key in Hermes Console.") + raise ConsoleCommandError( + "auth add --type api-key requires --api-key in Hermes Console.") if getattr(args, "import_name", None) is not None: - # profile import has no prompt flag; leave it alone. - return - if getattr(args, "skills_action", None) in { - "install", - "reset", - "opt-out", - "repair-official", - }: - setattr(args, "yes", True) + return # profile import has no prompt flag; leave it alone. + if getattr(args, "skills_action", None) in {"install", "reset", "opt-out", "repair-official"}: + args.yes = True if getattr(args, "memory_command", None) == "reset": - setattr(args, "yes", True) + args.yes = True def _version(_engine: HermesConsoleEngine, args: list[str]) -> str: @@ -704,29 +598,26 @@ def _logs(_engine: HermesConsoleEngine, args: list[str]) -> str: parser = _ArgumentParser(prog="logs", add_help=False) parser.add_argument("log_name", nargs="?", default="agent") parser.add_argument("-n", "--lines", type=int, default=50) - parser.add_argument("--level") - parser.add_argument("--session") - parser.add_argument("--since") - parser.add_argument("--component") + for flag in ("--level", "--session", "--since", "--component"): + parser.add_argument(flag) ns = parser.parse_args(args) if ns.lines < 1 or ns.lines > 500: raise ConsoleCommandError("logs --lines must be between 1 and 500") - from hermes_cli.logs import list_logs, tail_log if ns.log_name == "list": return _capture_output(list_logs) return _capture_output( lambda: tail_log( - ns.log_name, - num_lines=ns.lines, - follow=False, - level=ns.level, - session=ns.session, - since=ns.since, - component=ns.component, - ) - ) + ns.log_name, num_lines=ns.lines, follow=False, level=ns.level, + session=ns.session, since=ns.since, component=ns.component)) + + +def _session_db(): + """``with _session_db() as db:`` — SessionDB closed on exit.""" + from hermes_state import SessionDB + + return closing(SessionDB()) def _sessions_list(_engine: HermesConsoleEngine, args: list[str]) -> str: @@ -735,42 +626,26 @@ def _sessions_list(_engine: HermesConsoleEngine, args: list[str]) -> str: ns = parser.parse_args(args) if ns.limit < 1 or ns.limit > 200: raise ConsoleCommandError("sessions list --limit must be between 1 and 200") - - from hermes_state import SessionDB - - db = SessionDB() - try: + with _session_db() as db: sessions = db.list_sessions_rich( - exclude_sources=["kanban", "tool"], - limit=ns.limit, - order_by_last_active=True, - ) - finally: - db.close() + exclude_sources=["kanban", "tool"], limit=ns.limit, order_by_last_active=True) return _format_sessions(sessions) def _sessions_stats(_engine: HermesConsoleEngine, args: list[str]) -> str: _expect_no_args(args, "sessions stats") - from hermes_state import SessionDB - - db = SessionDB() - try: + with _session_db() as db: total = db.session_count() listable = db.session_count(exclude_children=True, exclude_sources=["kanban", "tool"]) - messages = db.message_count() lines = [ f"Total sessions: {total}", f"Listable sessions: {listable}", - f"Total messages: {messages}", - ] + f"Total messages: {db.message_count()}"] for source in ["cli", "tui", "telegram", "discord", "slack", "cron"]: count = db.session_count(source=source) if count: lines.append(f" {source}: {count}") return "\n".join(lines) - finally: - db.close() def _config_show(_engine: HermesConsoleEngine, args: list[str]) -> str: @@ -790,11 +665,9 @@ def _config_path(_engine: HermesConsoleEngine, args: list[str]) -> str: def _config_set(_engine: HermesConsoleEngine, args: list[str]) -> str: if len(args) < 2: raise ConsoleCommandError("Usage: config set ") - key = args[0] - value = " ".join(args[1:]) from hermes_cli.config import set_config_value - return _capture_output(lambda: set_config_value(key, value)) + return _capture_output(lambda: set_config_value(args[0], " ".join(args[1:]))) def _config_migrate(_engine: HermesConsoleEngine, args: list[str]) -> str: @@ -808,8 +681,7 @@ def _config_migrate(_engine: HermesConsoleEngine, args: list[str]) -> str: print("Configuration updated.") else: print("Configuration is up to date.") - warnings = results.get("warnings") or [] - for warning in warnings: + for warning in results.get("warnings") or []: print(f"Warning: {warning}") return _capture_output(_run) @@ -823,19 +695,12 @@ def _sessions_export(_engine: HermesConsoleEngine, args: list[str]) -> str: ns = parser.parse_args(args) def _run() -> None: - from hermes_state import ( - SessionDB, - SessionExportTooLargeError, - resolved_max_export_messages, - ) + from hermes_state import SessionExportTooLargeError, resolved_max_export_messages - db = SessionDB() - try: + with _session_db() as db: def _guard_exports(session_ids: list[str]) -> None: - # Per-session budget: each session is checked independently - # against the configured limit, so a full-DB backup of many - # small sessions never trips the guard — only an individual - # runaway transcript does. 0 disables the guard. + # Per-session budget: a full-DB backup of many small sessions never trips the + # guard, only an individual runaway transcript does. 0 disables it. limit = resolved_max_export_messages() if limit <= 0: return @@ -861,15 +726,10 @@ def _sessions_export(_engine: HermesConsoleEngine, args: list[str]) -> str: raise ConsoleCommandError(f"Session '{ns.session_id}' not found.") rows = [data] else: - session_ids = [ - session["id"] - for session in db.search_sessions(source=ns.source, limit=100000) - ] - _guard_exports(session_ids) + found = db.search_sessions(source=ns.source, limit=100000) + _guard_exports([session["id"] for session in found]) rows = db.export_all(source=ns.source) - - lines = [json.dumps(row, ensure_ascii=False) for row in rows] - text = "\n".join(lines) + text = "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) if text: text += "\n" if ns.output == "-": @@ -877,8 +737,6 @@ def _sessions_export(_engine: HermesConsoleEngine, args: list[str]) -> str: else: Path(ns.output).expanduser().write_text(text, encoding="utf-8") print(f"Exported {len(rows)} session(s) to {ns.output}") - finally: - db.close() return _capture_output(_run) @@ -890,10 +748,7 @@ def _sessions_rename(_engine: HermesConsoleEngine, args: list[str]) -> str: ns = parser.parse_args(args) def _run() -> None: - from hermes_state import SessionDB - - db = SessionDB() - try: + with _session_db() as db: resolved_session_id = db.resolve_session_id(ns.session_id) if not resolved_session_id: raise ConsoleCommandError(f"Session '{ns.session_id}' not found.") @@ -901,8 +756,6 @@ def _sessions_rename(_engine: HermesConsoleEngine, args: list[str]) -> str: if not db.set_session_title(resolved_session_id, title): raise ConsoleCommandError(f"Session '{ns.session_id}' not found.") print(f"Session '{resolved_session_id}' renamed to: {title}") - finally: - db.close() return _capture_output(_run) @@ -911,14 +764,8 @@ def _sessions_optimize(_engine: HermesConsoleEngine, args: list[str]) -> str: _expect_no_args(args, "sessions optimize") def _run() -> None: - from hermes_state import SessionDB - - db = SessionDB() - try: - count = db.vacuum() - print(f"Optimized {count} FTS index(es).") - finally: - db.close() + with _session_db() as db: + print(f"Optimized {db.vacuum()} FTS index(es).") return _capture_output(_run) @@ -944,13 +791,12 @@ def _sessions_repair(_engine: HermesConsoleEngine, args: list[str]) -> str: if ns.check_only: return report = repair_state_db_schema(db_path, backup=not ns.no_backup) - if report.get("repaired"): - if report.get("backup_path"): - print(f"backup: {report['backup_path']}") - print(f"strategy: {report.get('strategy')}") - print("Repaired session database.") - return - raise ConsoleCommandError(f"Repair failed: {report.get('error')}") + if not report.get("repaired"): + raise ConsoleCommandError(f"Repair failed: {report.get('error')}") + if report.get("backup_path"): + print(f"backup: {report['backup_path']}") + print(f"strategy: {report.get('strategy')}") + print("Repaired session database.") return _capture_output(_run) @@ -996,8 +842,7 @@ def _cron_pause(_engine: HermesConsoleEngine, args: list[str]) -> str: return _cron_job_action( args, "cron pause ", "Paused", - lambda ref: pause_job(ref, reason="paused from hermes console"), - ) + lambda ref: pause_job(ref, reason="paused from hermes console")) def _cron_resume(_engine: HermesConsoleEngine, args: list[str]) -> str: @@ -1028,60 +873,51 @@ def _cron_run(_engine: HermesConsoleEngine, args: list[str]) -> str: return _cron_job_action(args, "cron run ", "Triggered", trigger_job) -_READONLY_COMMANDS = ( - (("status",), "status", "Show Hermes component status.", _status), - (("version",), "version", "Show Hermes version information.", _version), - (("doctor",), "doctor", "Run diagnostics without auto-fix.", _doctor), - (("logs",), "logs [name] [-n N]", "Show recent Hermes logs.", _logs), - (("sessions", "list"), "sessions list [--limit N]", "List recent sessions.", _sessions_list), - (("sessions", "stats"), "sessions stats", "Show session store statistics.", _sessions_stats), - (("config", "show"), "config show", "Show current configuration.", _config_show), - (("config", "path"), "config path", "Print config.yaml path.", _config_path), - (("cron", "list"), "cron list [--all]", "List scheduled jobs.", _cron_list), - (("cron", "status"), "cron status", "Show cron scheduler status.", _cron_status), - (("profile",), "profile", "Show active profile status.", _profile_status), -) - -# (path, usage, summary, handler, confirmation prompt) -_MUTATING_COMMANDS = ( +# (path, usage, summary, handler, confirmation prompt) — a non-empty prompt marks it mutating. +_BUILTIN_COMMANDS = ( + (("status",), "status", "Show Hermes component status.", _status, ""), + (("version",), "version", "Show Hermes version information.", _version, ""), + (("doctor",), "doctor", "Run diagnostics without auto-fix.", _doctor, ""), + (("logs",), "logs [name] [-n N]", "Show recent Hermes logs.", _logs, ""), + (("sessions", "list"), "sessions list [--limit N]", "List recent sessions.", _sessions_list, + ""), + (("sessions", "stats"), "sessions stats", "Show session store statistics.", _sessions_stats, + ""), + (("config", "show"), "config show", "Show current configuration.", _config_show, ""), + (("config", "path"), "config path", "Print config.yaml path.", _config_path, ""), + (("cron", "list"), "cron list [--all]", "List scheduled jobs.", _cron_list, ""), + (("cron", "status"), "cron status", "Show cron scheduler status.", _cron_status, ""), + (("profile",), "profile", "Show active profile status.", _profile_status, ""), (("config", "set"), "config set ", "Set a configuration value.", _config_set, "Update Hermes configuration?"), - (("cron", "pause"), "cron pause ", "Pause a scheduled job.", _cron_pause, "Pause this cron job?"), - (("cron", "resume"), "cron resume ", "Resume a paused cron job.", _cron_resume, "Resume this cron job?"), + (("cron", "pause"), "cron pause ", "Pause a scheduled job.", _cron_pause, + "Pause this cron job?"), + (("cron", "resume"), "cron resume ", "Resume a paused cron job.", _cron_resume, + "Resume this cron job?"), (("cron", "run"), "cron run ", "Run a job on the next scheduler tick.", _cron_run, "Trigger this cron job?"), (("config", "migrate"), "config migrate", "Update config with new options.", _config_migrate, "Update Hermes configuration with missing defaults?"), (("sessions", "export"), "sessions export [--source SOURCE] [--session-id ID]", "Export sessions to JSONL.", _sessions_export, "Export session data?"), - (("sessions", "rename"), "sessions rename ", "Rename a session.", _sessions_rename, - "Rename this session?"), - (("sessions", "optimize"), "sessions optimize", "Optimize the session store.", _sessions_optimize, - "Optimize the session database?"), + (("sessions", "rename"), "sessions rename <session> <title>", "Rename a session.", + _sessions_rename, "Rename this session?"), + (("sessions", "optimize"), "sessions optimize", "Optimize the session store.", + _sessions_optimize, "Optimize the session database?"), (("sessions", "repair"), "sessions repair [--check-only] [--no-backup]", - "Repair a malformed session database schema.", _sessions_repair, "Repair the session database?"), -) + "Repair a malformed session database schema.", _sessions_repair, + "Repair the session database?")) def run_console_repl( - *, - stdin=None, - stdout=None, - stderr=None, - interactive: bool | None = None, -) -> int: + *, stdin=None, stdout=None, stderr=None, interactive: bool | None = None) -> int: """Run the local ``hermes console`` REPL.""" - - stdin = stdin or sys.stdin - stdout = stdout or sys.stdout - stderr = stderr or sys.stderr + stdin, stdout, stderr = stdin or sys.stdin, stdout or sys.stdout, stderr or sys.stderr if interactive is None: interactive = bool(getattr(stdin, "isatty", lambda: False)()) - engine = HermesConsoleEngine() if interactive: print("Hermes Console. Type `help` for commands, `exit` to quit.", file=stdout) - while True: if interactive: print("hermes> ", end="", file=stdout, flush=True) @@ -1090,24 +926,17 @@ def run_console_repl( if interactive: print(file=stdout) return 0 - result = engine.execute(line) if result.status == "confirm_required": if not interactive: - print( - f"Confirmation required: {result.confirmation_message}", - file=stderr, - ) + print(f"Confirmation required: {result.confirmation_message}", file=stderr) return 1 print(f"{result.confirmation_message} [y/N] ", end="", file=stdout, flush=True) - answer = stdin.readline() - if answer.strip().lower() not in {"y", "yes"}: + if stdin.readline().strip().lower() not in {"y", "yes"}: print("Cancelled.", file=stdout) continue result = engine.execute(result.command, confirmed=True) - if result.output: - stream = stderr if result.status == "error" else stdout - print(result.output, file=stream) + print(result.output, file=stderr if result.status == "error" else stdout) if result.status == "exit": return 0