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

refactor(hermes_cli): console_engine — merge command tables, closing() for SessionDB, inline single-use helpers, compact layout

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:27:46 -07:00
parent 113f04616b
commit 0e5190cb95
1 changed files with 181 additions and 352 deletions
Download Patch File Download Diff File Expand all files Collapse all files

533
hermes_cli/console_engine.py
View File

@@ -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, <handler>=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, <handler>=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 <command>, history, clear, exit, quit",
]
)
lines += [
"",
"* requires confirmation",
"Built-ins: help, help <command>, 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> <value>")
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 <job>", "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 <job>", "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 <key> <value>", "Set a configuration value.", _config_set,
"Update Hermes configuration?"),
(("cron", "pause"), "cron pause <job>", "Pause a scheduled job.", _cron_pause, "Pause this cron job?"),
(("cron", "resume"), "cron resume <job>", "Resume a paused cron job.", _cron_resume, "Resume this cron job?"),
(("cron", "pause"), "cron pause <job>", "Pause a scheduled job.", _cron_pause,
"Pause this cron job?"),
(("cron", "resume"), "cron resume <job>", "Resume a paused cron job.", _cron_resume,
"Resume this cron job?"),
(("cron", "run"), "cron run <job>", "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 <output> [--source SOURCE] [--session-id ID]",
"Export sessions to JSONL.", _sessions_export, "Export session data?"),
(("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", "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