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

Merge branch 'simp/r3-17-g6-misc' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:42:50 -07:00
parent 7b4977bfa5 48ef059a6c
commit 7d23e6bb2c
6 changed files with 234 additions and 418 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/claw.py
+4 -11
View File
@@ -115,8 +115,7 @@ def _detect_openclaw_processes() -> list[str]:
ps_cmd = (
'Get-CimInstance Win32_Process -Filter "Name = \'node.exe\'" | '
'Where-Object { $_.CommandLine -match "openclaw|clawd" } | '
'Select-Object -First 1 ProcessId'
)
'Select-Object -First 1 ProcessId')
result = bounded_probe_run(["powershell", "-NoProfile", "-Command", ps_cmd], timeout=5)
pid = (result.stdout or "").strip() if result is not None else ""
if pid:
@@ -124,7 +123,6 @@ def _detect_openclaw_processes() -> list[str]:
except Exception:
pass
return found
result = _posix_probe(["systemctl", "--user", "is-active", "openclaw-gateway.service"], 5)
if result is not None and result.stdout.strip() == "active":
found.append("systemd service: openclaw-gateway.service")
@@ -160,7 +158,6 @@ def _warn_if_openclaw_running(auto_yes: bool) -> None:
def _warn_if_gateway_running(auto_yes: bool) -> None:
"""Warn if a Hermes gateway has connected platforms (token conflicts, e.g. Telegram 409)."""
from gateway.status import get_running_pid, read_runtime_status
platforms = ((read_runtime_status() or {}).get("platforms") or {}) if get_running_pid() else {}
connected = [name for name, info in platforms.items()
if isinstance(info, dict) and info.get("state") == "connected"]
@@ -299,8 +296,7 @@ def _load_migrator(script_path: Path, opts: SimpleNamespace) -> Optional[Callabl
source_root=opts.source_dir.resolve(), target_root=opts.hermes_home.resolve(),
execute=execute, workspace_target=ws_target, overwrite=opts.overwrite,
migrate_secrets=opts.migrate_secrets, output_dir=None, selected_options=selected,
preset_name=opts.preset, skill_conflict_mode=opts.skill_conflict,
).migrate()
preset_name=opts.preset, skill_conflict_mode=opts.skill_conflict).migrate()
def _preview_migration(run_migrator: Callable[[bool], dict], opts: SimpleNamespace) -> bool:
@@ -387,8 +383,7 @@ def _cmd_cleanup(args):
"Stop OpenClaw first: systemctl --user stop openclaw-gateway.service"),
"Proceed anyway?",
declined="Aborted. Stop OpenClaw first, then re-run: hermes claw cleanup",
non_tty=("Non-interactive session — aborting. Stop OpenClaw and re-run.",),
):
non_tty=("Non-interactive session — aborting. Stop OpenClaw and re-run.",)):
return
total_archived = 0
for source_dir in dirs_to_check:
@@ -406,7 +401,6 @@ def _cmd_cleanup(args):
except OSError as e:
print_error(f"Could not archive: {e}")
print_info(f"Try manually: mv {source_dir} {source_dir}.pre-migration")
print()
n = len(dirs_to_check) if dry_run else total_archived
word = "directory" if n == 1 else "directories"
@@ -495,8 +489,7 @@ def _print_migration_report(report: dict, dry_run: bool):
print()
for line in (
" ⚠ API keys were NOT migrated (secrets migration is disabled by default).",
" Your OPENROUTER_API_KEY and other provider keys must be added manually.",
):
" Your OPENROUTER_API_KEY and other provider keys must be added manually."):
print(color(line, Colors.YELLOW))
_info("", "To migrate API keys, re-run with:",
" hermes claw migrate --migrate-secrets", "", "Or add your key manually:",
hermes_cli/cli_agent_setup_mixin.py
+107 -162
View File
@@ -40,16 +40,14 @@ def _current_runtime(cli) -> dict:
"api_mode": cli.api_mode,
"command": cli.acp_command,
"args": list(cli.acp_args or []),
"credential_pool": getattr(cli, "_credential_pool", None),
}
"credential_pool": getattr(cli, "_credential_pool", None)}
def _route_signature(model, runtime: dict) -> tuple:
"""Hashable identity of (model, routing) used to detect when the agent must be rebuilt."""
return (
model, runtime.get("provider"), runtime.get("requested_provider"), runtime.get("base_url"),
runtime.get("api_mode"), runtime.get("command"), tuple(runtime.get("args") or ()),
)
runtime.get("api_mode"), runtime.get("command"), tuple(runtime.get("args") or ()))
def _keyless_custom_base(base_url) -> bool:
@@ -58,8 +56,7 @@ def _keyless_custom_base(base_url) -> bool:
return bool(
isinstance(base_url, str)
and base_url
and not base_url_host_matches(base_url, "openrouter.ai")
)
and not base_url_host_matches(base_url, "openrouter.ai"))
def _compression_descendant(session_db, session_id):
@@ -78,8 +75,7 @@ def _user_display_text(content) -> str:
return " ".join(
part.get("text", "") if part.get("type") == "text" else "[image]"
for part in content
if isinstance(part, dict) and part.get("type") in ("text", "image_url")
)
if isinstance(part, dict) and part.get("type") in ("text", "image_url"))
return "" if content is None else str(content)
@@ -103,6 +99,60 @@ _RESUME_EVENT_TEXT = {
"auto_continue": "resumed interrupted turn",
}
def _collect_resume_entries(display_history, disp: dict, clean_assistant):
"""Displayable ``(role, text)`` recap entries from stored history, truncated per the
``display.resume_*`` config; system and tool-result rows are skipped. Returns
``(entries, index of last assistant entry, its un-truncated text)``.
Stored history is untrusted for display: text is sanitized so replay can't clear the
screen, retitle the window or restyle the panel. Pure-reasoning assistant rows with no
visible output are skipped, as are tool-call-only rows when ``resume_skip_tool_only``.
"""
from tools.ansi_strip import sanitize_display_text as _sanitize_display_text
max_user_len = int(disp.get("resume_max_user_chars", 300))
max_asst_len = int(disp.get("resume_max_assistant_chars", 200))
max_asst_lines = int(disp.get("resume_max_assistant_lines", 3))
skip_tool_only = disp.get("resume_skip_tool_only", True)
entries: list = []
last_asst_idx = None
last_asst_full = None
for msg in display_history:
role = msg.get("role", "")
display_kind = msg.get("display_kind")
content = msg.get("content")
tool_calls = msg.get("tool_calls") or []
if display_kind == "hidden":
continue
if display_kind in _RESUME_EVENT_TEXT:
entries.append(("event", _RESUME_EVENT_TEXT[display_kind]))
continue
if role == "user":
text = _sanitize_display_text(_user_display_text(content))
if len(text) > max_user_len:
text = text[:max_user_len] + "..."
entries.append(("user", text))
elif role == "assistant":
text = clean_assistant("" if content is None else str(content))
parts, full_parts = [], []
if text:
full_parts.append(text)
lines = text.splitlines()
if len(lines) > max_asst_lines:
text = "\n".join(lines[:max_asst_lines]) + " ..."
if len(text) > max_asst_len:
text = text[:max_asst_len] + "..."
parts.append(text)
if tool_calls:
parts.append(_tool_calls_summary(tool_calls))
full_parts.append(parts[-1])
if not text and (skip_tool_only or not tool_calls):
continue
entries.append(("assistant", " ".join(parts)))
last_asst_idx = len(entries) - 1
last_asst_full = " ".join(full_parts)
return entries, last_asst_idx, last_asst_full
# (skin key, fallback) for recap panel colors: body text, session label, border, assistant label.
_RESUME_SKIN_COLORS = (
("banner_text", "#FFF8DC"), ("session_label", "#DAA520"), ("session_border", "#8B8682"),
@@ -110,6 +160,16 @@ _RESUME_SKIN_COLORS = (
)
def _resume_panel_colors() -> tuple:
"""Active-skin colors for ``_RESUME_SKIN_COLORS`` (fallbacks when no skin loads)."""
try:
from hermes_cli.skin_engine import get_active_skin
_skin = get_active_skin()
return tuple(_skin.get_color(key, default) for key, default in _RESUME_SKIN_COLORS)
except Exception:
return tuple(default for _, default in _RESUME_SKIN_COLORS)
class CLIAgentSetupMixin:
"""Agent construction + session-resume display methods for ``HermesCLI``."""
@@ -118,36 +178,28 @@ class CLIAgentSetupMixin:
refresh are picked up without restarting the CLI. False on auth failure."""
from cli import ChatConsole, logger
from hermes_cli.runtime_provider import resolve_runtime_provider, format_runtime_provider_error
_primary_exc = None
runtime = None
try:
runtime = resolve_runtime_provider(
requested=self.requested_provider, explicit_api_key=self._explicit_api_key,
explicit_base_url=self._explicit_base_url,
)
explicit_base_url=self._explicit_base_url)
except Exception as exc:
_primary_exc = exc
if runtime is None and _primary_exc is not None:
from hermes_cli.auth import AuthError
if isinstance(_primary_exc, AuthError):
runtime = self._resolve_fallback_runtime(_primary_exc)
if runtime is not None:
_primary_exc = None
if _primary_exc is not None:
runtime = self._resolve_fallback_runtime(_primary_exc)
if runtime is not None:
_primary_exc = None
if runtime is None:
message = format_runtime_provider_error(_primary_exc) if _primary_exc else "Provider resolution failed."
ChatConsole().print(f"[bold red]{message}[/]")
return False
api_key = runtime.get("api_key")
base_url = runtime.get("base_url")
resolved_provider = runtime.get("provider", "openrouter")
resolved_routing = (
resolved_provider, runtime.get("api_mode", self.api_mode), runtime.get("command"),
list(runtime.get("args") or []),
)
list(runtime.get("args") or []))
# A callable api_key is a bearer-token provider (Azure Entra ID): the OpenAI SDK
# invokes it per request, so skip string validation / placeholder substitution.
_is_callable_provider = callable(api_key) and not isinstance(api_key, str)
@@ -158,8 +210,7 @@ class CLIAgentSetupMixin:
logger.debug(
"No API key for custom endpoint %s (source=%s), "
"using placeholder — local servers typically ignore auth",
base_url, runtime.get("source", ""),
)
base_url, runtime.get("source", ""))
else:
_prov = (resolved_provider or self.requested_provider or "").strip()
if _prov and _prov != "auto":
@@ -173,7 +224,6 @@ class CLIAgentSetupMixin:
print("\n⚠️ Provider resolver returned an empty base URL. "
"Check your provider config or run: hermes setup")
return False
credentials_changed = api_key != self.api_key or base_url != self.base_url
routing_changed = resolved_routing != (self.provider, self.api_mode, self.acp_command, self.acp_args)
self.provider, self.api_mode, self.acp_command, self.acp_args = resolved_routing
@@ -187,8 +237,7 @@ class CLIAgentSetupMixin:
# would otherwise send the provider name as the model string -> 400).
runtime_model = runtime.get("model")
if runtime_model and isinstance(runtime_model, str) and (
not self.model or self.model == self.provider or self.model == runtime.get("name")
):
not self.model or self.model == self.provider or self.model == runtime.get("name")):
self.model = runtime_model
# Still empty (e.g. `hermes auth add` without `hermes model`): fall back to the
@@ -201,8 +250,7 @@ class CLIAgentSetupMixin:
self.model = _default
logger.info(
"No model configured — defaulting to %s for provider %s",
_default, resolved_provider,
)
_default, resolved_provider)
except Exception:
pass
@@ -213,14 +261,17 @@ class CLIAgentSetupMixin:
if (credentials_changed or routing_changed or model_changed) and self.agent is not None:
self.agent = None
self._active_agent_route_signature = None
return True
def _resolve_fallback_runtime(self, primary_exc):
"""Primary provider auth failed: try each fallback entry in order and switch the
CLI's requested_provider/model to the first that resolves. None if none do."""
"""Primary provider resolution failed: on an AuthError try each fallback entry in
order and switch the CLI's requested_provider/model to the first that resolves.
None when the error is not auth-related or no fallback resolves."""
from cli import _cprint, logger
from hermes_cli.auth import AuthError
from hermes_cli.runtime_provider import resolve_runtime_provider
if not isinstance(primary_exc, AuthError):
return None
_fb_chain = self._fallback_model if isinstance(self._fallback_model, list) else []
for _fb in _fb_chain:
_fb_provider = (_fb.get("provider") or "").strip().lower()
@@ -229,7 +280,6 @@ class CLIAgentSetupMixin:
continue
try:
from hermes_cli.fallback_config import resolve_entry_api_key
_fb_kwargs = {"requested": _fb_provider}
if _fb.get("base_url"):
_fb_kwargs["explicit_base_url"] = _fb["base_url"]
@@ -239,8 +289,7 @@ class CLIAgentSetupMixin:
runtime = resolve_runtime_provider(**_fb_kwargs)
logger.warning(
"Primary provider auth failed (%s). Falling through to fallback: %s/%s",
primary_exc, _fb_provider, _fb_model,
)
primary_exc, _fb_provider, _fb_model)
_cprint(f"⚠️ Primary auth failed — switching to fallback: {_fb_provider} / {_fb_model}")
self.requested_provider = _fb_provider
self.model = _fb_model
@@ -255,12 +304,10 @@ class CLIAgentSetupMixin:
Never prints or mutates CLI state, so the interactive first-run path can route a
keyless install into onboarding before the user types into a chat that can't work."""
from hermes_cli.runtime_provider import resolve_runtime_provider
try:
runtime = resolve_runtime_provider(
requested=self.requested_provider, explicit_api_key=self._explicit_api_key,
explicit_base_url=self._explicit_base_url,
)
explicit_base_url=self._explicit_base_url)
except Exception:
return False
if not isinstance(runtime, dict):
@@ -276,7 +323,6 @@ class CLIAgentSetupMixin:
startup, TTY). Runs the same flow as ``hermes model`` so onboarding has a single
source of truth. True when a provider was configured."""
from cli import _cprint, logger
_cprint("")
_cprint("⚕ No inference provider is configured yet — let's fix that.")
_cprint(" You'll pick a provider (Nous Portal OAuth is the fastest; "
@@ -289,7 +335,6 @@ class CLIAgentSetupMixin:
if answer in {"n", "no"}:
_cprint(" Skipped. Run 'hermes model' or 'hermes setup' any time.")
return False
try:
from hermes_cli.main import select_provider_and_model
select_provider_and_model()
@@ -316,7 +361,6 @@ class CLIAgentSetupMixin:
# Force credential re-resolution + agent rebuild on next use.
self.agent = None
self._active_agent_route_signature = None
if self._runtime_credentials_ready():
_cprint(" ✓ Provider configured — you're ready to chat.")
return True
@@ -328,17 +372,15 @@ class CLIAgentSetupMixin:
provider. With `/fast` on (service_tier == "priority") attach request_overrides;
auto/cold tiers are applied per request by agent.fast_mode instead."""
from hermes_cli.models import resolve_fast_mode_overrides
runtime = _current_runtime(self)
route = {"model": self.model, "runtime": runtime, "signature": _route_signature(self.model, runtime)}
overrides = None
if getattr(self, "service_tier", None) == "priority":
try:
overrides = resolve_fast_mode_overrides(
route["model"], provider=runtime["provider"], base_url=runtime["base_url"],
)
route["model"], provider=runtime["provider"], base_url=runtime["base_url"])
except Exception:
overrides = None
pass
route["request_overrides"] = overrides
return route
@@ -374,23 +416,21 @@ class CLIAgentSetupMixin:
print(plain, file=sys.stderr)
else:
ChatConsole().print(rich)
if not session_meta:
hint = "Use a session ID from a previous CLI run (hermes sessions list)."
if _quiet_mode:
print(f"Session not found: {self.session_id}", file=sys.stderr)
print("Use a session ID from a previous CLI run (hermes sessions list).", file=sys.stderr)
print(hint, file=sys.stderr)
else:
_cprint(f"\033[1;31mSession not found: {self.session_id}{_RST}")
_cprint(f"{_DIM}Use a session ID from a previous CLI run (hermes sessions list).{_RST}")
_cprint(f"{_DIM}{hint}{_RST}")
return False
session_meta = self._follow_compression_chain(
session_meta,
lambda rid: ChatConsole().print(
f"[dim]Session {_escape(self.session_id)} was compressed into "
f"{_escape(rid)}; resuming the descendant with your "
f"transcript.[/dim]"
),
)
f"transcript.[/dim]"))
if getattr(self, "_resume_history_error", None):
return False
# Only the TIP session's rows are loaded here (no ancestors), so use the
@@ -400,8 +440,7 @@ class CLIAgentSetupMixin:
self._resume_history_error = resume_limit_error
_say(
f"Cannot resume session: {resume_limit_error}",
f"[bold red]Cannot resume session:[/] {_escape(resume_limit_error)}",
)
f"[bold red]Cannot resume session:[/] {_escape(resume_limit_error)}")
return False
restored = self._session_db.get_messages_as_conversation(self.session_id, repair_alternation=True)
if restored:
@@ -413,8 +452,7 @@ class CLIAgentSetupMixin:
_say(
f"↻ Resumed session {self.session_id}{title_part} {counts}",
f"[bold {_accent_hex()}]↻ Resumed session[/] [bold]{_escape(self.session_id)}[/]"
f"[bold {_accent_hex()}]{_escape(title_part)}[/] {counts}",
)
f"[bold {_accent_hex()}]{_escape(title_part)}[/] {counts}")
self._restore_session_cwd(session_meta, quiet=_quiet_mode)
self._restore_session_yolo(session_meta, quiet=_quiet_mode)
self._restore_session_model(session_meta, quiet=_quiet_mode)
@@ -436,20 +474,14 @@ class CLIAgentSetupMixin:
# Join the background preloaded-skills load (--skills/-s) BEFORE the agent
# snapshots self.system_prompt below. No-op when nothing was requested.
self.finalize_preloaded_skills()
_prepare_deferred_agent_startup()
self._install_tool_callbacks()
self._ensure_tirith_security()
if not self._ensure_runtime_credentials():
return False
from hermes_cli.mcp_startup import ensure_mcp_discovery_before_agent_build
ensure_mcp_discovery_before_agent_build(
logger=logger, single_query=getattr(self, "_single_query_mode", False)
)
logger=logger, single_query=getattr(self, "_single_query_mode", False))
if self._session_db is None:
try:
from hermes_state import SessionDB
@@ -458,10 +490,8 @@ class CLIAgentSetupMixin:
logger.warning("SQLite session store not available — session will NOT be indexed: %s", e)
if (
self._resumed and self._session_db and not self.conversation_history
and not self._load_resumed_history_late()
):
and not self._load_resumed_history_late()):
return False
try:
runtime = runtime_override or _current_runtime(self)
effective_model = model_override or self.model
@@ -470,8 +500,7 @@ class CLIAgentSetupMixin:
clarify_callback = (
_single_query_clarify_callback
if getattr(self, "_single_query_mode", False)
else self._clarify_callback
)
else self._clarify_callback)
self.agent = AIAgent(
model=effective_model, api_key=runtime.get("api_key"),
base_url=runtime.get("base_url"), provider=runtime.get("provider"),
@@ -507,8 +536,7 @@ class CLIAgentSetupMixin:
stream_delta_callback=self._stream_delta if self.streaming_enabled else None,
tool_gen_callback=self._on_tool_gen_start if self.streaming_enabled else None,
notice_callback=self._on_notice, notice_clear_callback=self._on_notice_clear,
reaction_callback=self._on_reaction,
)
reaction_callback=self._on_reaction)
# Reference for atexit memory-provider shutdown: ``_run_cleanup`` in cli.py
# reads ``cli._active_agent_ref``, so this MUST write the ``cli`` module's
# global — a ``global`` statement here would bind this module's namespace.
@@ -521,7 +549,6 @@ class CLIAgentSetupMixin:
# warning shows before the first message. Idempotent + fail-open in the helper.
try:
from agent.credits_tracker import seed_credits_at_session_start
seed_credits_at_session_start(self.agent)
except Exception:
pass
@@ -544,7 +571,6 @@ class CLIAgentSetupMixin:
console = ChatConsole()
console.print(f"[bold red]Failed to initialize agent: {e}[/]")
from hermes_constants import partial_update_hint
for line in partial_update_hint(e):
console.print(line)
return False
@@ -559,7 +585,6 @@ class CLIAgentSetupMixin:
return None
from cli import logger
from hermes_state import SessionResumeTooLargeError
try:
safety_check = getattr(self._session_db, "assert_resume_safe", None)
if not callable(safety_check):
@@ -573,8 +598,7 @@ class CLIAgentSetupMixin:
except Exception as exc:
logger.warning(
"Resume safety check failed for %s (proceeding without guard): %s",
self.session_id, exc,
)
self.session_id, exc)
return None
def _preload_resumed_session(self) -> bool:
@@ -584,35 +608,28 @@ class CLIAgentSetupMixin:
from cli import _accent_hex
if not self._resumed or not self._session_db:
return False
session_meta = self._session_db.get_session(self.session_id)
if not session_meta:
self._console_print(f"[bold red]Session not found: {self.session_id}[/]")
self._console_print("[dim]Use a session ID from a previous CLI run (hermes sessions list).[/]")
return False
session_meta = self._follow_compression_chain(
session_meta,
lambda rid: self._console_print(
f"[dim]Session {self.session_id} was compressed into "
f"{rid}; resuming the descendant with your transcript.[/]"
),
)
f"{rid}; resuming the descendant with your transcript.[/]"))
resume_limit_error = self._resume_history_limit_error()
if resume_limit_error:
self._resume_history_error = resume_limit_error
self._console_print(f"[bold red]Cannot resume session:[/] {resume_limit_error}")
return False
restored, display_history = self._session_db.get_resume_conversations(self.session_id)
accent_color = _accent_hex()
if not restored:
self._console_print(
f"[{accent_color}]Session {self.session_id} found but has no "
f"messages. Starting fresh.[/]"
)
f"messages. Starting fresh.[/]")
return False
restored = [m for m in restored if m.get("role") != "session_meta"]
self.conversation_history = restored
self._resume_display_history = [m for m in display_history if m.get("role") != "session_meta"]
@@ -626,12 +643,10 @@ class CLIAgentSetupMixin:
f"[{accent_color}]↻ Resumed session [bold]{self.session_id}[/bold]"
f"{title_part} "
f"({msg_count} user message{'s' if msg_count != 1 else ''}, "
f"{len(restored)} total messages)[/]"
)
f"{len(restored)} total messages)[/]")
self._restore_session_cwd(session_meta)
self._restore_session_yolo(session_meta)
self._restore_session_model(session_meta)
self._reopen_session()
return True
@@ -643,88 +658,22 @@ class CLIAgentSetupMixin:
display_history = getattr(self, "_resume_display_history", self.conversation_history)
if not display_history or self.resume_display == "minimal":
return
_disp = CLI_CONFIG.get("display", {})
MAX_DISPLAY_EXCHANGES = int(_disp.get("resume_exchanges", 10))
MAX_USER_LEN = int(_disp.get("resume_max_user_chars", 300))
MAX_ASST_LEN = int(_disp.get("resume_max_assistant_chars", 200))
MAX_ASST_LINES = int(_disp.get("resume_max_assistant_lines", 3))
SKIP_TOOL_ONLY = _disp.get("resume_skip_tool_only", True)
# Collect displayable (role, text) entries; system and tool-result rows are skipped.
entries = []
_last_asst_idx = None # index of last assistant entry
_last_asst_full = None # un-truncated display text for last assistant
for msg in display_history:
role = msg.get("role", "")
display_kind = msg.get("display_kind")
content = msg.get("content")
tool_calls = msg.get("tool_calls") or []
if display_kind == "hidden":
continue
if display_kind in _RESUME_EVENT_TEXT:
entries.append(("event", _RESUME_EVENT_TEXT[display_kind]))
continue
if role in ("system", "tool"):
continue
# Stored history is untrusted for display: strip escape sequences/control
# chars so replay can't clear the screen, retitle the window or restyle the panel.
if role == "user":
text = _sanitize_display_text(_user_display_text(content))
if len(text) > MAX_USER_LEN:
text = text[:MAX_USER_LEN] + "..."
entries.append(("user", text))
elif role == "assistant":
text = "" if content is None else str(content)
text = _sanitize_display_text(_strip_reasoning_tags(text))
parts = []
full_parts = [] # un-truncated version
if text:
full_parts.append(text)
lines = text.splitlines()
if len(lines) > MAX_ASST_LINES:
text = "\n".join(lines[:MAX_ASST_LINES]) + " ..."
if len(text) > MAX_ASST_LEN:
text = text[:MAX_ASST_LEN] + "..."
parts.append(text)
if tool_calls:
parts.append(_tool_calls_summary(tool_calls))
full_parts.append(parts[-1])
# Skip pure-reasoning messages with no visible output, and tool-call-only
# entries when SKIP_TOOL_ONLY is enabled.
if not text and (SKIP_TOOL_ONLY or not tool_calls):
continue
entries.append(("assistant", " ".join(parts)))
_last_asst_idx = len(entries) - 1
_last_asst_full = " ".join(full_parts)
entries, _last_asst_idx, _last_asst_full = _collect_resume_entries(
display_history, _disp, lambda t: _sanitize_display_text(_strip_reasoning_tags(t)))
if not entries:
return
skipped = max(0, len(entries) - MAX_DISPLAY_EXCHANGES * 2)
skipped = max(0, len(entries) - int(_disp.get("resume_exchanges", 10)) * 2)
entries = entries[skipped:]
# Show the last assistant entry in full so the user sees where they left off.
if _last_asst_idx is not None and _last_asst_full:
adj_idx = _last_asst_idx - skipped
if 0 <= adj_idx < len(entries):
entries[adj_idx] = ("assistant_last", _last_asst_full)
from rich.panel import Panel
from rich.text import Text
try:
from hermes_cli.skin_engine import get_active_skin
_skin = get_active_skin()
_history_text_c, _session_label_c, _session_border_c, _assistant_label_c = (
_skin.get_color(key, default) for key, default in _RESUME_SKIN_COLORS
)
except Exception:
_history_text_c, _session_label_c, _session_border_c, _assistant_label_c = (
default for _, default in _RESUME_SKIN_COLORS
)
_history_text_c, _session_label_c, _session_border_c, _assistant_label_c = (
_resume_panel_colors())
# role -> (label, label style, body style, continuation indent)
role_styles = {
@@ -732,11 +681,9 @@ class CLIAgentSetupMixin:
"assistant": (" ◆ Hermes: ", f"dim bold {_assistant_label_c}", "dim", " " * 12),
"assistant_last": (" ◆ Hermes: ", f"bold {_assistant_label_c}", "", " " * 12), # full, non-dim
}
lines = Text()
if skipped:
lines.append(f" ... {skipped} earlier messages ...\n\n", style="dim italic")
for i, (role, text) in enumerate(entries):
if role == "event":
lines.append(f" ◈ {text}\n", style="dim italic")
@@ -749,11 +696,9 @@ class CLIAgentSetupMixin:
lines.append(f"{indent}{ml}\n", style=body_style)
if i < len(entries) - 1:
lines.append("") # small gap
panel = Panel(
lines, title=f"[dim {_session_label_c}]Previous Conversation[/]",
border_style=f"dim {_session_border_c}", padding=(0, 1), style=_history_text_c,
)
border_style=f"dim {_session_border_c}", padding=(0, 1), style=_history_text_c)
_record_output_history_entry(lambda: self._render_resume_history_panel_lines(panel))
with _suspend_output_history():
self._console_print(panel)
hermes_cli/cli_loops_mixin.py
+42 -79
View File
@@ -21,6 +21,15 @@ def _preview(payload: str) -> str:
return f"{payload[:80]}{'...' if len(payload) > 80 else ''}"
def _print_decision_message(decision: dict) -> bool:
"""Print a manager decision's ``message`` (if any) via _cprint; True when one was printed."""
from cli import _cprint
msg = decision.get("message") or ""
if msg:
_cprint(f" {msg}")
return bool(msg)
class CLILoopsMixin:
"""Simple slash-command wrappers plus goal/heartbeat/loop manager hooks for the interactive CLI"""
@@ -49,8 +58,7 @@ class CLILoopsMixin:
def _cmd_clear(self, cmd_original: str):
from cli import (
ChatConsole, _build_compact_banner, _clear_output_history, _cprint,
build_welcome_banner, get_tool_definitions,
)
build_welcome_banner, get_tool_definitions)
if self._confirm_destructive_slash(
"clear",
"This clears the screen and starts a new session.\n"
@@ -78,14 +86,14 @@ class CLILoopsMixin:
cc.print(_build_compact_banner())
else:
tools = get_tool_definitions(enabled_toolsets=self.enabled_toolsets, quiet_mode=True)
agent = getattr(self, "agent", None)
ctx_len = None
if hasattr(self, 'agent') and self.agent and hasattr(self.agent, 'context_compressor'):
ctx_len = self.agent.context_compressor.context_length
if agent and hasattr(agent, "context_compressor"):
ctx_len = agent.context_compressor.context_length
build_welcome_banner(
console=cc, model=self.model, cwd=os.getenv("TERMINAL_CWD", os.getcwd()),
tools=tools, enabled_toolsets=self.enabled_toolsets, session_id=self.session_id,
context_length=ctx_len, provider=self.provider,
)
context_length=ctx_len, provider=self.provider)
_cprint(_FRESH_START)
self._print_random_tip()
@@ -179,8 +187,7 @@ class CLILoopsMixin:
_undo_desc = (
"This removes the last user/assistant exchange from history."
if _undo_n == 1
else f"This removes the last {_undo_n} user turns from history."
)
else f"This removes the last {_undo_n} user turns from history.")
if self._confirm_destructive_slash("undo", _undo_desc, cmd_original=cmd_original) is None:
return True # confirmation cancelled — command handled, keep REPL alive
self.undo_last(_undo_n)
@@ -191,11 +198,9 @@ class CLILoopsMixin:
def _cmd_egress(self, cmd_original: str):
from hermes_cli.slash_exec import CommandContext, execute_command
self._console_print(
execute_command("egress", CommandContext(surface="cli")).text,
highlight=False, markup=False,
)
highlight=False, markup=False)
def _cmd_statusbar(self, cmd_original: str):
self._status_bar_visible = not self._status_bar_visible
@@ -208,7 +213,6 @@ class CLILoopsMixin:
def _cmd_version(self, cmd_original: str):
from hermes_cli.main import _print_version_info
_print_version_info(check_updates=True)
def _cmd_reload(self, cmd_original: str):
@@ -227,9 +231,7 @@ class CLILoopsMixin:
# installed-but-not-enabled plugins show up; the plugin manager only knows
# *loaded* plugins and made fresh installs look like "nothing installed".
from hermes_cli.plugins_cmd import (
_discover_all_plugins, _get_disabled_set, _get_enabled_set, _plugin_status
)
_discover_all_plugins, _get_disabled_set, _get_enabled_set, _plugin_status)
entries = _discover_all_plugins()
enabled = _get_enabled_set()
disabled = _get_disabled_set()
@@ -238,7 +240,6 @@ class CLILoopsMixin:
# on one line (full catalog behind `hermes plugins list`).
user_entries = [e for e in entries if e[3] != "bundled"]
bundled_count = len(entries) - len(user_entries)
if not user_entries:
print("No user plugins installed.")
print(" Install one: hermes plugins install owner/repo")
@@ -246,22 +247,18 @@ class CLILoopsMixin:
if bundled_count:
print(f" ({bundled_count} bundled plugins available — see: hermes plugins list)")
return
# Loaded-plugin details (tools/hooks/commands counts, errors) by name.
loaded: dict = {}
try:
try: # loaded-plugin details (tools/hooks/commands counts, errors) by name
from hermes_cli.plugins import get_plugin_manager
for p in get_plugin_manager().list_plugins():
loaded[p["name"]] = p
loaded = {p["name"]: p for p in get_plugin_manager().list_plugins()}
except Exception:
loaded = {}
print(f"User plugins ({len(user_entries)}):")
for name, version, _desc, source, _dir, key in sorted(user_entries):
state = _plugin_status(name, enabled, disabled, key=key)
glyph = {"enabled": "✓", "disabled": "✗"}.get(state, "○")
ver = f" v{version}" if version else ""
info = loaded.get(name) or {}
bits = [f"{info[k]} {k}" for k in ("tools", "hooks", "commands") if info.get(k)]
glyph = {"enabled": "✓", "disabled": "✗"}.get(state, "○")
ver = f" v{version}" if version else ""
detail = f" ({', '.join(bits)})" if bits else ""
label = "" if state == "enabled" else f" [{state}]"
error = f" — {info['error']}" if info.get("error") else ""
@@ -362,18 +359,13 @@ class CLILoopsMixin:
return self._session_bound_manager("_loop_manager", "loop manager", load)
def _start_heartbeat_watchdog(self):
"""Start the idle-poll thread that fires due heartbeats.
Same pattern as the wake-word watchdog: a daemon thread polls a few times a
minute; when the session is idle (no agent running, empty input queue) and the
heartbeat is due, its prompt is injected into ``_pending_input`` as a normal user
turn. Missed ticks coalesce — the anchor resets on fire, so a busy hour yields ONE
heartbeat turn. Idempotent; safe to call on every /heartbeat set.
"""
"""Start the idle-poll daemon that injects a due heartbeat prompt into
``_pending_input`` as a normal user turn when the session is idle. Missed ticks
coalesce (the anchor resets on fire, so a busy hour yields ONE heartbeat turn).
Idempotent; safe to call on every /heartbeat set."""
if getattr(self, "_heartbeat_watchdog_started", False):
return
self._heartbeat_watchdog_started = True
from hermes_cli.heartbeat import POLL_SECONDS
def _loop():
@@ -388,8 +380,7 @@ class CLILoopsMixin:
self._agent_running
or getattr(self, "_voice_recording", False)
or getattr(self, "_voice_processing", False)
or not self._pending_input.empty()
)
or not self._pending_input.empty())
if busy:
continue
prompt = mgr.due_prompt()
@@ -399,7 +390,6 @@ class CLILoopsMixin:
logging.debug("heartbeat watchdog tick failed: %s", exc)
finally:
self._heartbeat_watchdog_started = False
threading.Thread(target=_loop, daemon=True, name="heartbeat-watchdog").start()
def _maybe_fire_loop_tick(self) -> None:
@@ -426,12 +416,10 @@ class CLILoopsMixin:
return
try:
from hermes_cli.loops import goal_blocks_loop_tick
if goal_blocks_loop_tick(mgr.session_id):
return
except Exception:
pass
wakeup = mgr.fire_tick()
if not wakeup:
return
@@ -453,10 +441,7 @@ class CLILoopsMixin:
# judge; complete them immediately (caps and scheduling still apply).
if wakeup.lstrip().startswith("/"):
try:
decision = mgr.complete_tick("")
msg = decision.get("message") or ""
if msg:
_cprint(f" {msg}")
_print_decision_message(mgr.complete_tick(""))
except Exception:
pass
@@ -470,8 +455,7 @@ class CLILoopsMixin:
parts = [
p.get("text", "")
for p in content
if isinstance(p, dict) and p.get("type") in {"text", "output_text"}
]
if isinstance(p, dict) and p.get("type") in {"text", "output_text"}]
return "\n".join(t for t in parts if t)
return str(content or "")
except Exception:
@@ -502,27 +486,19 @@ class CLILoopsMixin:
pass
_cprint(
f" {_DIM}⏸ Loop paused — wakeup turn was interrupted. "
f"Use /loop resume to continue, or /loop stop to end it.{_RST}"
)
f"Use /loop resume to continue, or /loop stop to end it.{_RST}")
return
decision = mgr.complete_tick(self._last_assistant_response_text())
msg = decision.get("message") or ""
if msg:
_cprint(f" {msg}")
elif decision.get("status") == "active" and mgr.state is not None:
if (not _print_decision_message(decision) and decision.get("status") == "active"
and mgr.state is not None):
_cprint(f" {_DIM}↻ Loop: {mgr.state.remaining_label()}.{_RST}")
def _maybe_continue_goal_after_turn(self) -> None:
"""Hook run after every CLI turn: judge the goal and maybe re-queue a continuation.
Safe when no goal is set. Preemption is automatic: a real user message already in
``_pending_input`` skips judging (the user's turn goes first; we re-judge after).
A user-cancelled turn (Ctrl+C) AUTO-PAUSES the goal instead of judging — the judge
would run on partial output, almost always say "continue", and re-queue exactly
what the user cancelled; pausing is observable and recoverable via ``/goal
resume``. The empty-response skip mirrors ``_handle_message`` in ``gateway/run.py``.
"""
"""Post-turn hook: judge the goal and maybe re-queue a continuation. A real user
message already queued preempts judging (re-judged after their turn). Ctrl+C
AUTO-PAUSES instead of judging — the judge on partial output nearly always says
"continue" and would re-queue exactly what was cancelled; pausing is recoverable
via ``/goal resume``. Empty-response skip mirrors ``gateway/run.py``."""
from cli import _DIM, _RST, _cprint, _looks_like_slash_command
mgr = self._get_goal_manager()
if mgr is None or not mgr.is_active():
@@ -536,22 +512,16 @@ class CLILoopsMixin:
try:
pending = getattr(self, "_pending_input", None)
if pending is not None and not pending.empty():
has_real_message = False
try:
for entry in list(pending.queue):
if isinstance(entry, tuple) and entry:
entry = entry[0]
if isinstance(entry, str) and _looks_like_slash_command(entry):
continue
has_real_message = True
break
entries = [e[0] if isinstance(e, tuple) and e else e for e in list(pending.queue)]
has_real_message = any(
not (isinstance(e, str) and _looks_like_slash_command(e)) for e in entries)
except Exception:
has_real_message = True # can't introspect — defer to be safe
if has_real_message:
return
except Exception:
pass
if getattr(self, "_last_turn_interrupted", False):
try:
mgr.pause(reason="user-interrupted (Ctrl+C)")
@@ -559,8 +529,7 @@ class CLILoopsMixin:
logging.debug("goal pause-on-interrupt failed: %s", exc)
_cprint(
f" {_DIM}⏸ Goal paused — turn was interrupted. "
f"Use /goal resume to continue, or /goal clear to stop.{_RST}"
)
f"Use /goal resume to continue, or /goal clear to stop.{_RST}")
return
# Empty/whitespace responses are almost always transient failures (API error,
@@ -568,20 +537,14 @@ class CLILoopsMixin:
last_response = self._last_assistant_response_text()
if not last_response.strip():
return
try:
from hermes_cli.goals import gather_background_processes as _gather_bg
_bg_procs = _gather_bg()
except Exception:
_bg_procs = None
decision = mgr.evaluate_after_turn(
last_response, user_initiated=True, background_processes=_bg_procs
)
msg = decision.get("message") or ""
if msg:
_cprint(f" {msg}")
last_response, user_initiated=True, background_processes=_bg_procs)
_print_decision_message(decision)
if decision.get("should_continue"):
prompt = decision.get("continuation_prompt")
if prompt:
hermes_cli/cli_terminal_mixin.py
+53 -101
View File
@@ -65,8 +65,7 @@ class CLITerminalMixin:
pass
logger.warning(
"Terminal I/O broken%s — freezing UI paints to avoid redraw storm (#81521)",
f" ({reason})" if reason else "",
)
f" ({reason})" if reason else "")
def _app_invalidate(self, app, where: str, *, swallow: bool) -> None:
"""``app.invalidate()``: EIO freezes paints, other OSErrors re-raise, and any
@@ -83,17 +82,13 @@ class CLITerminalMixin:
raise
def _invalidate(self, min_interval: float = 0.25) -> None:
"""Throttled UI repaint for high-frequency background updates.
"""Throttled UI repaint for high-frequency background updates (spinner frames,
streaming flushes): the throttle prevents blinking on slow/SSH links and the
resize-recovery guard keeps footer chrome out of scrollback mid-SIGWINCH.
For spinner frames, streaming token flushes and other repaints that fire many
times per second: the throttle prevents blinking on slow/SSH links, and the
resize-recovery guard avoids stamping footer/status-bar chrome into scrollback
while a SIGWINCH reflow is in flight.
Do NOT use for user-blocking modal prompts (approval / clarify / sudo): those
must paint immediately via ``_paint_now``. Sent through this throttle, an
unrelated repaint within the 250ms window — or an in-flight resize — silently
drops the modal's entry paint, so it never renders and times out unseen (#41098).
NOT for user-blocking modals (approval / clarify / sudo) — use ``_paint_now``: a
throttled or resize-gated entry paint is silently dropped, so the prompt never
renders and times out unseen (#41098).
"""
if getattr(self, "_terminal_io_broken", False):
return
@@ -139,14 +134,10 @@ class CLITerminalMixin:
self._app_invalidate(app, "force_full_redraw", swallow=True)
def _schedule_focus_regain_redraw(self, min_interval: float = 1.0) -> None:
"""Repaint after a terminal focus-in report (``CSI I``), rate-limited.
Terminals with focus tracking (Ghostty, iTerm2, xterm, muxes toggling DECSET
1004) emit ``\\x1b[I`` when the Hermes tab becomes visible again; emulators may
coalesce hidden-tab output, so on regain the incremental diff stacks on stale
content (#60920 focus-regain variant, #25337). Self-gating — terminals without
focus tracking never emit it. Rate-limited so a burst (rapid Alt+Tab, pane hops)
repaints at most once per ``min_interval`` seconds.
"""Repaint after a terminal focus-in report (``CSI I``), at most once per
``min_interval`` (rapid Alt+Tab / pane-hop bursts). Emulators with focus tracking
(DECSET 1004) may coalesce hidden-tab output, so on regain the incremental diff
stacks on stale content (#60920, #25337); terminals without it never emit ``CSI I``.
"""
now = time.monotonic()
if now - getattr(self, "_last_focus_regain_redraw", 0.0) < min_interval:
@@ -172,18 +163,12 @@ class CLITerminalMixin:
return bool(raw)
def _recover_terminal_after_interrupt(self) -> None:
"""Recover the terminal after an interrupted agent turn (#33271).
An in-flight ``CSI 6n`` cursor query may answer (``ESC[<row>;<col>R``) after the
input parser tore down: the reply leaks as literal text and the VT100 parser can
stall mid-escape, so the terminal looks frozen. ``flush_stdin()`` drains stray
bytes (no-op on non-TTY), then ``_force_full_redraw()`` repaints cleanly; each
step self-guards so one failing never blocks the other. A dead PTY (EIO) skips
the redraw — painting a broken fd is the #81521 redraw storm.
Do NOT clear output history here: the interruption marker is printed under
``_suspend_output_history`` in chat(), so the replay reproduces the response
without duplicating the marker (#60920).
"""Recover the terminal after an interrupted agent turn (#33271): an in-flight
``CSI 6n`` reply arriving after the input parser tore down leaks as literal text
and can stall the VT100 parser mid-escape. Drain stdin, then full redraw; each step
self-guards. A dead PTY (EIO) skips the redraw (#81521 redraw storm). Never clear
output history here — the interruption marker is printed under
``_suspend_output_history``, so replay already excludes it (#60920).
"""
if getattr(self, "_terminal_io_broken", False):
return
@@ -222,30 +207,20 @@ class CLITerminalMixin:
def _recover_after_resize(self, app, original_on_resize) -> None:
"""Recover a resized classic CLI without desynchronizing cursor state.
Unlike ``_force_full_redraw`` this never clears scrollback: the startup banner
lives there and ``_replay_output_history`` cannot reconstruct it. prompt_toolkit's
own resize path runs with its renderer cursor cache intact — its
``Application._on_resize()`` erases via the cached cursor position, and resetting
the renderer first loses the origin and strands stale prompt glyphs.
``_status_bar_suppressed_after_resize`` hides the dynamic status bar / input rules
while the reflow settles: on column shrink the terminal reflows already-painted
rows into scrollback before prompt_toolkit erases them, so a fresh bar looks
duplicated (#19280, #22976). Suppression alone cannot erase an already-reflowed
OLD bar (``renderer.erase()`` does ``cursor_up(_cursor_pos.y)`` with the y cached
at the OLD width, undershooting the reflowed rows), so on an OBSERVED width
change we wipe the viewport (CSI 2J — banner-safe; 3J only when
``display.cli_rebuild_scrollback_on_redraw``) and replay the transcript before
delegating. Same-width SIGWINCH (tmux attach, GNOME tab bar, focus signals) and
the first signal without a baseline are left untouched: a 2J+replay against
preserved scrollback duplicates everything in ``_OUTPUT_HISTORY`` (#65293);
``_install_resize_recovery`` seeds the baseline so an initial maximize still
counts. The stale-previous_screen crash on tmux attach is handled by
``_hermes_call_output_screen_diff``'s retry (#83874).
Suppression is transient: a debounced timer clears it and repaints once the
reflow settles, so the bar returns during idle (previously it stayed hidden
until the next submitted input). The next-submit clear remains as a fast path.
Never clears scrollback (the startup banner lives there and replay cannot rebuild
it) and never resets the renderer before prompt_toolkit's own ``_on_resize``,
which erases via the cached cursor position. The status bar / input rules are
suppressed while the reflow settles: on column shrink the terminal reflows
already-painted rows into scrollback first, so a fresh bar looks duplicated
(#19280, #22976). Suppression cannot erase the already-reflowed OLD bar
(``renderer.erase()`` uses ``_cursor_pos.y`` cached at the OLD width), so on an
OBSERVED width change we wipe the viewport (CSI 2J, banner-safe; 3J only via
``display.cli_rebuild_scrollback_on_redraw``) and replay the transcript first.
Same-width SIGWINCH (tmux attach, GNOME tab bar, focus) and the first signal
without a seeded baseline are left alone — 2J+replay against preserved scrollback
duplicates ``_OUTPUT_HISTORY`` (#65293). tmux-attach's stale previous_screen is
handled by ``_hermes_call_output_screen_diff`` (#83874). Suppression is cleared by
a debounced timer so the bar returns during idle; next-submit stays a fast path.
"""
from cli import _replay_output_history
self._status_bar_suppressed_after_resize = True
@@ -258,8 +233,7 @@ class CLITerminalMixin:
if width_changed:
try:
self._clear_prompt_toolkit_screen(
app, rebuild_scrollback=self._redraw_rebuilds_scrollback()
)
app, rebuild_scrollback=self._redraw_rebuilds_scrollback())
_replay_output_history()
except Exception:
pass
@@ -298,10 +272,8 @@ class CLITerminalMixin:
app.invalidate()
except Exception:
pass
self._restart_debounce_timer(
"_status_bar_unsuppress_timer", delay, lambda _t: _run_on_app_loop(app, _clear)
)
"_status_bar_unsuppress_timer", delay, lambda _t: _run_on_app_loop(app, _clear))
except Exception:
# Fail open: never leave the bar stuck hidden.
self._status_bar_suppressed_after_resize = False
@@ -322,9 +294,7 @@ class CLITerminalMixin:
self._resize_recovery_timer = None
self._resize_recovery_pending = False
self._recover_after_resize(app, original_on_resize)
_run_on_app_loop(app, _run_recovery)
with lock:
self._resize_recovery_pending = True
self._restart_debounce_timer("_resize_recovery_timer", delay, _timer_fired)
@@ -334,38 +304,29 @@ class CLITerminalMixin:
def _install_resize_recovery(self, app) -> None:
"""Route ``app._on_resize`` through the debounced ghost-clearing recovery
(#5474/#49120) and seed the width baseline for width-change detection.
The seed keeps the session's FIRST SIGWINCH honest (#65293): without it a benign
signal is indistinguishable from a real resize. It reads ``app.output`` directly,
NOT ``_get_tui_terminal_width``: before ``app.run()`` ``get_app()`` is the
DummyApplication whose DummyOutput reports a hardcoded 80 columns, and seeding
that fake width would make the first real signal look like a change.
``app.output`` is what the running resize handler measures, so both are comparable.
(#5474/#49120) and seed the width baseline so the FIRST SIGWINCH can tell a benign
signal from a real resize (#65293). Reads ``app.output`` directly, NOT
``_get_tui_terminal_width``: before ``app.run()`` the DummyApplication's output
reports a hardcoded 80 columns, which would make the first real signal look like a
width change. ``app.output`` is what the running resize handler measures.
"""
width = None
try:
width = app.output.get_size().columns
except Exception:
width = None
if not width or width <= 0:
for probe in (lambda: app.output.get_size().columns,
lambda: shutil.get_terminal_size((80, 24)).columns):
try:
width = shutil.get_terminal_size((80, 24)).columns
width = probe()
except Exception:
width = None
if width and width > 0:
break
self._last_resize_width = width
original_on_resize = app._on_resize
def _resize_clear_ghosts():
self._schedule_resize_recovery(app, original_on_resize)
app._on_resize = _resize_clear_ghosts
app._on_resize = lambda: self._schedule_resize_recovery(app, original_on_resize)
def _try_attach_clipboard_image(self) -> bool:
"""Save a clipboard image to ~/.hermes/images/ and attach it; True if attached."""
from cli import datetime
from hermes_cli.clipboard import save_clipboard_image
self._image_counter += 1
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
img_path = get_hermes_home() / "images" / f"clip_{ts}_{self._image_counter}.png"
@@ -393,14 +354,12 @@ class CLITerminalMixin:
"""Best-effort reset when leaked mouse reports indicate mode drift."""
from cli import (
CLI_CONFIG, _DIM, _RST, _TERMINAL_INPUT_MODE_RESET_SEQ,
_cli_multiline_shortcuts_enabled, _cprint, _enable_extended_enter_keys, logger,
)
_cli_multiline_shortcuts_enabled, _cprint, _enable_extended_enter_keys, logger)
now = time.monotonic()
# Rate-limit to avoid thrashing if a terminal floods reports.
if now - self._last_input_mode_recovery < 0.5:
return
self._last_input_mode_recovery = now
app = getattr(self, "_app", None)
output = getattr(app, "output", None) if app else None
try:
@@ -415,24 +374,19 @@ class CLITerminalMixin:
_enable_extended_enter_keys(output)
except Exception:
pass
logger.warning("Recovered terminal input modes after leak: %s", reason)
if not self._input_mode_recovery_notice_shown:
self._input_mode_recovery_notice_shown = True
_cprint(
f" {_DIM}Recovered terminal input modes after leaked mouse reports. "
f"If this repeats, run /new or restart this tab.{_RST}"
)
f"If this repeats, run /new or restart this tab.{_RST}")
def _check_termios_drift(self) -> None:
"""Watchdog: heal the tty if it drifted back to cooked mode.
See ``_heal_cooked_mode_drift``: a lost ``run_in_terminal`` cooked→raw restore
leaves the tty line-buffering while prompt_toolkit believes it owns raw mode —
the CLI looks dead but the process is healthy. Called from ``process_loop``'s
idle branch so it self-heals within ~1s of idling. Skipped while a
``run_in_terminal`` window legitimately holds cooked mode, while the agent is
running (approval/sudo prompts manipulate the tty), and on Windows (no termios).
"""Idle watchdog: heal a tty that drifted back to cooked mode (a lost
``run_in_terminal`` cooked→raw restore leaves it line-buffering while prompt_toolkit
believes it owns raw mode — CLI looks dead, process is healthy). Skipped while
``run_in_terminal`` legitimately holds cooked mode, while the agent runs
(approval/sudo prompts touch the tty), and on Windows (no termios).
"""
from cli import _DIM, _RST, _cprint, _heal_cooked_mode_drift, logger
if os.name == "nt":
@@ -455,8 +409,7 @@ class CLITerminalMixin:
if _heal_cooked_mode_drift(fd):
logger.warning(
"Healed cooked-mode termios drift on stdin — a "
"run_in_terminal cooked→raw restore was lost."
)
"run_in_terminal cooked→raw restore was lost.")
try:
self._invalidate() # so the prompt is visibly alive again
except Exception:
@@ -465,5 +418,4 @@ class CLITerminalMixin:
self._termios_drift_notice_shown = True
_cprint(
f" {_DIM}Recovered terminal from cooked-mode drift "
f"(input should respond normally again).{_RST}"
)
f"(input should respond normally again).{_RST}")
hermes_cli/clipboard.py
+2 -5
View File
@@ -177,8 +177,7 @@ _PS_IMAGE_STRATEGIES = (
"if ($null -eq $img) { exit 1 }"
"$ms = New-Object System.IO.MemoryStream;"
"$img.Save($ms, [System.Drawing.Imaging.ImageFormat]::Png);"
"[System.Convert]::ToBase64String($ms.ToArray())",
),
"[System.Convert]::ToBase64String($ms.ToArray())"),
( # Get-Clipboard -Format Image (System.Drawing.Image or WPF BitmapSource)
"try { "
"$img = Get-Clipboard -Format Image -ErrorAction Stop;"
@@ -199,8 +198,7 @@ _PS_IMAGE_STRATEGIES = (
"$enc.Save($ms)"
"} else { exit 2 }"
"[System.Convert]::ToBase64String($ms.ToArray())"
"} catch { exit 1 }",
),
"} catch { exit 1 }"),
( # copied image *file* (Explorer file drop)
_PS_FILEDROP_HIT
+ "if ($null -ne $hit) { 'True' } else { 'False' }"
@@ -322,7 +320,6 @@ def _convert_to_png(path: Path) -> bool:
pass
except Exception as e:
logger.debug("Pillow BMP→PNG conversion failed: %s", e)
tmp = path.with_suffix(".bmp")
try:
path.rename(tmp)
hermes_cli/codex_runtime_plugin_migration.py
+26 -60
View File
@@ -15,8 +15,7 @@ logger = logging.getLogger(__name__)
# Marker comments wrapping the managed section so re-runs can detect what's ours and what's
# user-edited. Both must appear or strip is a no-op.
MIGRATION_MARKER = (
"# managed by hermes-agent — `hermes codex-runtime migrate` regenerates this section"
)
"# managed by hermes-agent — `hermes codex-runtime migrate` regenerates this section")
MIGRATION_END_MARKER = "# end hermes-agent managed section"
@@ -65,8 +64,7 @@ _KNOWN_HERMES_KEYS = {
"command", "args", "env", "cwd",
"url", "headers", "transport",
"timeout", "connect_timeout",
"enabled", "description",
}
"enabled", "description"}
_KEYS_DROPPED_WITH_WARNING = {"sampling"}
# (hermes key, codex key, skip note) — timeouts are emitted as floats or skipped when non-numeric.
@@ -91,7 +89,6 @@ def _translate_one_server(name: str, hermes_cfg: dict) -> tuple[Optional[dict],
"""
if not isinstance(hermes_cfg, dict):
return None, []
skipped: list[str] = []
out: dict[str, Any] = {}
if hermes_cfg.get("command"):
@@ -112,7 +109,6 @@ def _translate_one_server(name: str, hermes_cfg: dict) -> tuple[Optional[dict],
skipped.append("transport=sse (codex auto-negotiates)")
else:
return None, ["no command or url field"]
for hermes_key, codex_key, note in _TIMEOUT_KEYS:
if hermes_key in hermes_cfg:
try:
@@ -121,7 +117,6 @@ def _translate_one_server(name: str, hermes_cfg: dict) -> tuple[Optional[dict],
skipped.append(note)
if hermes_cfg.get("enabled") is False:
out["enabled"] = False
for key in hermes_cfg:
if key in _KEYS_DROPPED_WITH_WARNING:
skipped.append(f"{key} (no codex equivalent)")
@@ -171,8 +166,7 @@ def _quote_key(key: str) -> str:
def render_codex_toml_section(
servers: dict[str, dict], plugins: Optional[list[dict]] = None,
default_permission_profile: Optional[str] = None,
) -> str:
default_permission_profile: Optional[str] = None) -> str:
"""Render the managed [mcp_servers.<n>] / [plugins.<id>] / default_permissions block.
``default_permission_profile`` (e.g. "workspace-write", "read-only", "full-access") is written
@@ -185,7 +179,6 @@ def render_codex_toml_section(
out.append("# (no MCP servers, plugins, or permissions configured by Hermes)")
out.append(MIGRATION_END_MARKER)
return "\n".join(out) + "\n"
if default_permission_profile:
profile = default_permission_profile
normalized = profile if profile.startswith(":") else f":{profile}"
@@ -211,13 +204,11 @@ def _insert_managed_block_at_top_level(user_text: str, managed_block: str) -> st
"""
if not user_text.strip():
return managed_block
lines = user_text.splitlines(keepends=True)
first_table_idx = next((i for i, ln in enumerate(lines) if ln.lstrip().startswith("[")), None)
if first_table_idx is None:
prefix = user_text.rstrip("\n")
return f"{prefix}\n\n{managed_block}" if prefix else managed_block
prefix = "".join(lines[:first_table_idx]).rstrip("\n")
suffix = "".join(lines[first_table_idx:]).lstrip("\n")
return f"{prefix}\n\n{managed_block}\n{suffix}" if prefix else f"{managed_block}\n{suffix}"
@@ -287,35 +278,27 @@ def _strip_existing_managed_block(toml_text: str) -> str:
def _query_codex_plugins(
codex_home: Optional[Path] = None, timeout: float = 8.0
) -> tuple[list[dict], Optional[str]]:
"""Query codex's `plugin/list` for installed curated plugins.
Spawns ``codex app-server`` briefly, sends initialize + plugin/list and keeps installed=true.
Returns ``(plugins, error)``; any failure (not installed, RPC error, timeout) yields ``([],
error)`` and is non-fatal — MCP servers and permissions still write through.
Plugins codex reports as unavailable (broken install, missing OAuth, removed from marketplace)
are skipped: we write codex's config.toml directly, so a broken plugin would surface as a
codex error on the user's first turn. ``enabled`` is carried forward as codex reports it.
codex_home: Optional[Path] = None, timeout: float = 8.0) -> tuple[list[dict], Optional[str]]:
"""Spawn ``codex app-server`` briefly and return ``(installed plugins, error)`` from
``plugin/list``. Any failure yields ``([], error)`` and is non-fatal (servers and
permissions still write). Plugins codex reports unavailable (broken install, missing OAuth,
delisted) are skipped — we write config.toml directly, so they would surface as a codex
error on the first turn. ``enabled`` is carried forward as reported.
"""
try:
from agent.transports.codex_app_server import CodexAppServerClient
except Exception as exc:
return [], f"transport unavailable: {exc}"
try:
with CodexAppServerClient(codex_home=str(codex_home) if codex_home else None) as client:
client.initialize(client_name="hermes-migration")
resp = client.request("plugin/list", {}, timeout=timeout)
except Exception as exc:
return [], f"plugin/list query failed: {exc}"
marketplaces = resp.get("marketplaces") or []
if not isinstance(marketplaces, list):
return [], "plugin/list response missing 'marketplaces'"
out: list[dict] = []
seen: set[tuple[str, str]] = set()
out: dict[tuple[str, str], dict] = {} # (name, marketplace) -> entry; first wins
for marketplace in marketplaces:
if not isinstance(marketplace, dict):
continue
@@ -330,12 +313,11 @@ def _query_codex_plugins(
availability)
continue
name = str(plugin.get("name") or "")
if not name or (name, market_name) in seen:
continue
seen.add((name, market_name))
out.append({"name": name, "marketplace": market_name,
"enabled": bool(plugin.get("enabled", True))})
return out, None
if name:
out.setdefault((name, market_name), {
"name": name, "marketplace": market_name,
"enabled": bool(plugin.get("enabled", True))})
return list(out.values()), None
# pytest tempdir shapes: ``pytest-of-<user>/pytest-<n>/``, macOS ``/private/var/folders/…/T``.
@@ -353,20 +335,16 @@ def _looks_like_test_tempdir(path: str) -> bool:
def _build_hermes_tools_mcp_entry() -> dict:
"""Build the codex stdio entry that launches Hermes' own tool surface as an MCP server, so the
codex subprocess can call back for browser/web/delegate_task/vision/memory/skills tools.
"""Codex stdio entry launching Hermes' own tool surface as an MCP server (browser/web/
delegate_task/vision/memory/skills call-backs).
HERMES_HOME passes through IF SET so the subprocess sees the same config/auth/sessions DB as
the parent CLI. Read from os.environ (not get_hermes_home()) on purpose: when unset, the codex
subprocess should inherit whatever HERMES_HOME its launcher sets at runtime (systemd unit,
gateway, kanban dispatcher, custom shell) rather than a migrate-time default burned into
config.toml, which would pin it to the wrong profile. The pytest-tempdir guard stops a sibling
test's ``monkeypatch.setenv("HERMES_HOME", tmp_path)`` from leaking a transient dir into the
user's real config.toml. PYTHONPATH passes through so a worktree-launched hermes finds the
branch's modules instead of the installed package.
HERMES_HOME passes through only IF SET, read from os.environ (not get_hermes_home()): when
unset the codex subprocess must inherit its launcher's runtime HERMES_HOME (systemd, gateway,
kanban), not a migrate-time default burned into config.toml that pins the wrong profile. The
pytest-tempdir guard keeps a sibling test's monkeypatched HERMES_HOME out of the user's real
config. PYTHONPATH passes through so a worktree-launched hermes finds the branch's modules.
"""
import sys
env: dict[str, str] = {}
hermes_home = os.environ.get("HERMES_HOME") or ""
if hermes_home and not _looks_like_test_tempdir(hermes_home):
@@ -376,22 +354,19 @@ def _build_hermes_tools_mcp_entry() -> dict:
# Quiet mode + redaction defaults so the MCP wire stays clean.
env["HERMES_QUIET"] = "1"
env["HERMES_REDACT_SECRETS"] = env.get("HERMES_REDACT_SECRETS", "true")
return {
"command": sys.executable,
"args": ["-m", "agent.transports.hermes_tools_mcp_server"],
"env": env,
# Generous timeouts — browser_navigate or delegate_task can take a while.
"startup_timeout_sec": 30.0,
"tool_timeout_sec": 600.0,
}
"tool_timeout_sec": 600.0}
def _write_atomic(target: Path, text: str) -> None:
"""Write via a same-directory temp file + rename (atomic on POSIX, ReplaceFile on Windows) so a
crash mid-write never leaves a half-written config.toml that codex would refuse to load."""
import tempfile
tmp_fd, tmp_path_str = tempfile.mkstemp(prefix=".config.toml.", dir=str(target.parent))
tmp_path = Path(tmp_path_str)
try:
@@ -400,8 +375,7 @@ def _write_atomic(target: Path, text: str) -> None:
tmp_path.replace(target)
except Exception:
try:
if tmp_path.exists():
tmp_path.unlink()
tmp_path.unlink(missing_ok=True)
except Exception:
pass
raise
@@ -410,8 +384,7 @@ def _write_atomic(target: Path, text: str) -> None:
def migrate(
hermes_config: dict, *, codex_home: Optional[Path] = None, dry_run: bool = False,
discover_plugins: bool = True, default_permission_profile: Optional[str] = ":workspace",
expose_hermes_tools: bool = True,
) -> MigrationReport:
expose_hermes_tools: bool = True) -> MigrationReport:
"""Translate Hermes mcp_servers config + Codex curated plugins into ~/.codex/config.toml.
``discover_plugins`` spawns the live codex CLI (set False in tests); discovery is best-effort
@@ -425,12 +398,10 @@ def migrate(
codex_home = codex_home or Path.home() / ".codex"
target = codex_home / "config.toml"
report.target_path = target
hermes_servers = (hermes_config or {}).get("mcp_servers") or {}
if not isinstance(hermes_servers, dict):
report.errors.append("mcp_servers in Hermes config is not a dict; cannot migrate.")
return report
translated: dict[str, dict] = {}
for raw_name, cfg in hermes_servers.items():
name = str(raw_name)
@@ -443,7 +414,6 @@ def migrate(
if skipped:
report.skipped_keys_per_server[name] = skipped
report.migrated.append(name)
plugins: list[dict] = []
plugin_query_succeeded = False
if discover_plugins and not dry_run:
@@ -454,17 +424,14 @@ def migrate(
# re-render and may strip pre-existing tables outside the managed block.
plugin_query_succeeded = not plugin_err
report.migrated_plugins += [f"{p['name']}@{p['marketplace']}" for p in plugins]
if default_permission_profile:
report.wrote_permissions_default = default_permission_profile
if expose_hermes_tools:
translated["hermes-tools"] = _build_hermes_tools_mcp_entry()
if "hermes-tools" not in report.migrated:
report.migrated.append("hermes-tools")
managed_block = render_codex_toml_section(
translated, plugins=plugins, default_permission_profile=default_permission_profile,
)
translated, plugins=plugins, default_permission_profile=default_permission_profile)
new_text = managed_block
if target.exists():
try:
@@ -476,7 +443,6 @@ def migrate(
if plugin_query_succeeded:
without_managed = _strip_unmanaged_plugin_tables(without_managed)
new_text = _insert_managed_block_at_top_level(without_managed, managed_block)
if dry_run:
return report
try: