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

refactor(hermes_cli): simplify 15 small modules — dead wrappers, unified helpers, dict dispatch, compact docs

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:29:34 -07:00
parent 113f04616b
commit 416a1c1acd
16 changed files with 359 additions and 817 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/default_soul.py
+15 -40
View File
@@ -1,12 +1,9 @@
"""Default SOUL.md template seeded into HERMES_HOME on first run."""
# Kept identical to agent/prompt_builder.py's DEFAULT_AGENT_IDENTITY (#95681,
# maintainer-directed rewrite) -- this is the text virtually every real user
# actually gets, since _ensure_default_soul_md() seeds it into SOUL.md on
# first run. DEFAULT_AGENT_IDENTITY only serves sessions with no SOUL.md at
# all (e.g. skip_context_files), which is not the common case. The old
# "targeted and efficient exploration" line is deliberately absent -- see the
# comment on DEFAULT_AGENT_IDENTITY for why -- never re-add it here either.
# Kept identical to agent/prompt_builder.py's DEFAULT_AGENT_IDENTITY: _ensure_default_soul_md()
# seeds this into SOUL.md on first run, so it is the text virtually every real user gets. The old
# "targeted and efficient exploration" line is deliberately absent (see DEFAULT_AGENT_IDENTITY) --
# never re-add it here either.
DEFAULT_SOUL_MD = (
"You are Hermes Agent, built by Nous Research. Be direct: match the "
"length of your reply to the weight of the ask — a one-line question "
@@ -21,16 +18,11 @@ DEFAULT_SOUL_MD = (
"default."
)
# Legacy SOUL.md boilerplate that older installers (install.sh / install.ps1 /
# docker/SOUL.md) seeded before they were switched to write DEFAULT_SOUL_MD.
# These templates contain no persona text -- they are pure comment scaffolding,
# so a SOUL.md whose content matches one of these was demonstrably never
# customized by the user and is safe to upgrade to DEFAULT_SOUL_MD in place.
#
# Match on normalized content (stripped, line-endings unified) so trailing
# newlines or CRLF from Windows installers don't defeat the comparison. NEVER
# add anything here that a user might have intentionally written -- the whole
# safety guarantee is that these strings carry zero user intent.
# Auto-seeded SOUL.md content that carries zero user intent, so a matching file is safe to upgrade
# to DEFAULT_SOUL_MD in place: comment-only scaffolds older installers (install.sh / install.ps1 /
# docker/SOUL.md) wrote, plus earlier generations of the auto-seeded default text. Compared on
# normalized content (stripped, line endings unified). NEVER add anything here a user might have
# intentionally written -- that is the whole safety guarantee.
_LEGACY_TEMPLATE_SOULS = (
(
"# Hermes Agent Persona\n"
@@ -49,9 +41,7 @@ _LEGACY_TEMPLATE_SOULS = (
"Delete the contents (or this file) to use the default personality.\n"
"-->"
),
# docker/SOUL.md and the install.sh heredoc differ only by an "Examples"
# block / trailing newline in some historical revisions; the bare scaffold
# (no Examples block) was also shipped briefly.
# Bare scaffold without the "Examples" block, shipped briefly.
(
"# Hermes Agent Persona\n"
"\n"
@@ -64,11 +54,7 @@ _LEGACY_TEMPLATE_SOULS = (
"Delete the contents (or this file) to use the default personality.\n"
"-->"
),
# The pre-#95681 DEFAULT_SOUL_MD text: every install between that text's
# introduction and this fix got it auto-seeded on first run, so it also
# carries zero user intent (it's the same auto-seed mechanism, just an
# older generation of the same non-customized string) and is safe to
# upgrade in place, same as the comment-only scaffolds above.
# The previous generation of DEFAULT_SOUL_MD (same auto-seed mechanism, older string).
(
"You are Hermes Agent, an intelligent AI assistant created by Nous "
"Research. You are helpful, knowledgeable, and direct. You assist "
@@ -80,29 +66,18 @@ _LEGACY_TEMPLATE_SOULS = (
"below. Be targeted and efficient in your exploration and "
"investigations."
),
# ASCII-dashed variant of the current DEFAULT_SOUL_MD, as seeded by
# scripts/install.ps1 (which must stay pure ASCII -- see
# tests/test_install_ps1_ascii_only.py -- so it writes "--" where the
# canonical text has an em-dash). Still pure auto-seed, zero user intent;
# upgrading it in place converges Windows installs onto the canonical
# em-dash text on first run.
# ASCII-dashed variant seeded by scripts/install.ps1 (must stay pure ASCII, see
# tests/test_install_ps1_ascii_only.py); upgrading converges Windows installs on the em-dash text.
DEFAULT_SOUL_MD.replace("\u2014", "--"),
)
def _normalize_soul(text: str) -> str:
"""Normalize SOUL.md content for legacy-template comparison."""
# Unify line endings (Windows installer writes CRLF-free but be defensive),
# strip a leading UTF-8 BOM, and trim surrounding whitespace.
"""Unify line endings, strip a leading UTF-8 BOM, trim whitespace."""
return text.replace("\r\n", "\n").replace("\r", "\n").lstrip("\ufeff").strip()
def is_legacy_template_soul(text: str) -> bool:
"""True if ``text`` is a non-customized, auto-seeded SOUL.md.
Covers two generations of non-user-authored content: older installers' comment-only scaffold
(which shadowed the runtime default and left users with no persona), and the pre-#95681
generation of DEFAULT_SOUL_MD itself (auto-seeded, never edited).
"""
"""True if ``text`` is a non-customized, auto-seeded SOUL.md (see ``_LEGACY_TEMPLATE_SOULS``)."""
normalized = _normalize_soul(text)
return any(normalized == _normalize_soul(t) for t in _LEGACY_TEMPLATE_SOULS)
hermes_cli/dep_ensure.py
+14 -25
View File
@@ -1,13 +1,8 @@
"""Lazy dependency bootstrapper for non-Python runtime deps.
Detection and prompting live here in Python — not in install.sh — because: 1. shutil.which() works
on every platform; install.sh needs bash. 2. Detection is instant; spawning bash for a "is node
installed?" check is waste. 3. Python controls the UX (rich prompts, non-interactive fallback, TTY
detection).
install.sh is still the *installation* backend because it has 1900 lines of battle-tested OS
detection and package-manager logic (apt/brew/pacman/dnf/ zypper/Termux/…). Reimplementing that in
Python would be huge duplication.
Detection and prompting live here (cross-platform ``shutil.which``, instant, Python-controlled UX);
install.sh / install.ps1 remain the *installation* backend because they hold the battle-tested OS
and package-manager logic.
"""
from __future__ import annotations
@@ -23,9 +18,8 @@ from tools.environments.local import hermes_subprocess_env
_IS_WINDOWS = platform.system() == "Windows"
_DEP_CHECKS = {
# find_node_executable() rather than a bare which(): $HERMES_HOME/node is
# not on PATH, so which() would report Node missing on an install that has
# a managed one and trigger a redundant re-install.
# find_node_executable() rather than a bare which(): $HERMES_HOME/node is not on PATH, so
# which() would report Node missing on a managed install and trigger a redundant re-install.
"node": lambda: find_node_executable("node") is not None,
"browser": lambda: (
agent_browser_runnable(shutil.which("agent-browser"))
@@ -54,9 +48,9 @@ def _has_system_browser() -> bool:
def _has_npx_agent_browser() -> bool:
"""agent-browser resolves lazily via npx on the default install (#43564), invisible to the
PATH/managed-dir probes above. Mirror tools.browser_tool.check_browser_requirements's Termux
carve-out so this check can't diverge from what browser tools actually find.
"""agent-browser resolves lazily via npx on the default install, invisible to the PATH/managed-dir
probes above. Mirror tools.browser_tool.check_browser_requirements's Termux carve-out so this
check can't diverge from what browser tools actually find.
"""
try:
from tools.browser_tool import (
@@ -67,19 +61,16 @@ def _has_npx_agent_browser() -> bool:
browser_cmd = _find_agent_browser(validate=False)
except Exception:
return False
return _is_npx_agent_browser_sentinel(browser_cmd) and not _requires_real_termux_browser_install(
browser_cmd
)
return _is_npx_agent_browser_sentinel(browser_cmd) and not _requires_real_termux_browser_install(browser_cmd)
def _has_hermes_agent_browser() -> bool:
from hermes_constants import get_hermes_home
home = get_hermes_home()
if _IS_WINDOWS:
# npm -g --prefix puts .cmd shims directly in the prefix dir on Windows
if _IS_WINDOWS: # npm -g --prefix puts .cmd shims directly in the prefix dir
return (home / "node" / "agent-browser.cmd").is_file()
# install.sh installs globally into $HERMES_HOME/node/bin/ via npm -g --prefix
# Also check legacy node_modules/.bin/ path for git-clone installs.
# install.sh installs into $HERMES_HOME/node/bin/ via npm -g --prefix; legacy git-clone
# installs used node_modules/.bin/.
return (
(home / "node" / "bin" / "agent-browser").is_file()
or (home / "node_modules" / ".bin" / "agent-browser").is_file()
@@ -106,8 +97,7 @@ def _find_install_script(
def ensure_dependency(dep: str, interactive: bool = True) -> bool:
"""Ensure a non-Python dependency is available. Returns True if available."""
check = _DEP_CHECKS.get(dep)
if check is None:
# Unknown dep — don't silently forward to install script.
if check is None: # unknown dep — don't silently forward to install script
return False
if check():
return True
@@ -144,5 +134,4 @@ def ensure_dependency(dep: str, interactive: bool = True) -> bool:
run_env = hermes_subprocess_env(inherit_credentials=False)
run_env["IS_INTERACTIVE"] = "false"
result = subprocess.run(cmd, env=run_env)
return result.returncode == 0 and check()
return subprocess.run(cmd, env=run_env).returncode == 0 and check()
hermes_cli/diagnostics_upload.py
+20 -60
View File
@@ -1,26 +1,21 @@
"""Client for uploading ``hermes debug share`` bundles to Nous-internal S3.
1. POST {NAS_BASE}/api/diagnostics/upload-url → {uploadUrl, viewUrl, id, ...} (the request body
carries ``sizeBytes``; NAS signs it into the presigned URL's ``ContentLength``, so the PUT must send
exactly that many bytes) 2. PUT <uploadUrl> (the gzipped bundle, Content-Type application/gzip)
1. POST {NAS_BASE}/api/diagnostics/upload-url → {uploadUrl, viewUrl, id, ...}. The request body
carries ``sizeBytes``; NAS signs it into the presigned URL's ``ContentLength``, so the PUT must
send exactly that many bytes.
2. PUT <uploadUrl> (the gzipped bundle, Content-Type application/gzip).
NAS is stateless — the object's existence in S3 is the only state, so there is no confirm/callback
step.
NAS is stateless — the object's existence in S3 is the only state, so there is no confirm step.
"""
import json
import os
import urllib.request
# Base URL of the Nous account service that mints the signed upload URL.
# Overridable via env so the feature can be pointed at staging / a local dev
# NAS instance during testing.
NAS_BASE = os.environ.get(
"HERMES_DIAGNOSTICS_BASE_URL", "https://portal.nousresearch.com"
)
# Overridable via env so the feature can be pointed at staging / a local dev NAS instance.
NAS_BASE = os.environ.get("HERMES_DIAGNOSTICS_BASE_URL", "https://portal.nousresearch.com")
# Network timeout for each request (seconds). The upload itself can be larger
# (a gzipped log bundle), so the PUT gets a more generous window.
# Network timeouts (seconds); the PUT carries the gzipped log bundle so it gets a more generous window.
_REQUEST_TIMEOUT = 30
_UPLOAD_TIMEOUT = 120
@@ -38,10 +33,7 @@ def _urlopen_checked(req: urllib.request.Request, *, timeout: int, what: str):
return resp.read()
def request_upload_url(
content_type: str = "application/gzip",
size_bytes: int | None = None,
) -> dict:
def request_upload_url(content_type: str = "application/gzip", size_bytes: int | None = None) -> dict:
"""Ask NAS to mint a presigned PUT URL for a diagnostics bundle.
Returns the parsed JSON, expected to carry at least ``uploadUrl``, ``viewUrl`` and ``id``.
@@ -51,69 +43,37 @@ def request_upload_url(
if size_bytes is not None:
payload["sizeBytes"] = int(size_bytes)
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(
f"{NAS_BASE}/api/diagnostics/upload-url",
data=data,
data=json.dumps(payload).encode("utf-8"),
method="POST",
headers={
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": _USER_AGENT,
},
headers={"Content-Type": "application/json", "Accept": "application/json", "User-Agent": _USER_AGENT},
)
body = _urlopen_checked(
req, timeout=_REQUEST_TIMEOUT, what="diagnostics upload-url request"
).decode("utf-8")
body = _urlopen_checked(req, timeout=_REQUEST_TIMEOUT, what="diagnostics upload-url request").decode("utf-8")
try:
result = json.loads(body)
except (ValueError, json.JSONDecodeError) as exc:
raise RuntimeError(
f"diagnostics upload-url returned non-JSON response: {body[:200]}"
) from exc
raise RuntimeError(f"diagnostics upload-url returned non-JSON response: {body[:200]}") from exc
if not isinstance(result, dict) or not result.get("uploadUrl"):
raise RuntimeError(
"diagnostics upload-url response missing 'uploadUrl': "
f"{body[:200]}"
)
raise RuntimeError(f"diagnostics upload-url response missing 'uploadUrl': {body[:200]}")
return result
def put_bundle(
upload_url: str,
data: bytes,
content_type: str = "application/gzip",
) -> None:
"""PUT the gzipped *data* bundle to a presigned *upload_url*.
def put_bundle(upload_url: str, data: bytes, content_type: str = "application/gzip") -> None:
"""PUT the gzipped *data* bundle to a presigned *upload_url*. Raises on non-2xx.
Sets the ``Content-Type`` header (must match what NAS pinned when signing the URL, otherwise S3
rejects the signature). Raises on non-2xx.
``Content-Type`` must match what NAS pinned when signing the URL, otherwise S3 rejects the signature.
"""
req = urllib.request.Request(
upload_url,
data=data,
method="PUT",
headers={
"Content-Type": content_type,
"User-Agent": _USER_AGENT,
},
upload_url, data=data, method="PUT", headers={"Content-Type": content_type, "User-Agent": _USER_AGENT}
)
_urlopen_checked(req, timeout=_UPLOAD_TIMEOUT, what="diagnostics bundle PUT")
def share_to_nous(report_bundle: bytes) -> dict:
"""Orchestrate the full Nous-S3 upload of a gzipped *report_bundle*.
Two steps: mint a presigned PUT URL (sending the exact ``sizeBytes`` NAS signs into the URL's
``ContentLength``), then PUT the bundle. NAS is stateless — the object's existence in S3 is the
only state, so there is no confirm/callback step.
"""
size_bytes = len(report_bundle)
info = request_upload_url(
content_type="application/gzip", size_bytes=size_bytes
)
"""Mint a presigned PUT URL (with the exact ``sizeBytes`` NAS signs), then PUT *report_bundle*."""
info = request_upload_url(content_type="application/gzip", size_bytes=len(report_bundle))
put_bundle(info["uploadUrl"], report_bundle, content_type="application/gzip")
return info
hermes_cli/dingtalk_auth.py
+30 -60
View File
@@ -5,23 +5,16 @@ from __future__ import annotations
import os
import sys
import time
import logging
from typing import Optional, Tuple
import requests
logger = logging.getLogger(__name__)
# ── Configuration ──────────────────────────────────────────────────────────
REGISTRATION_BASE_URL = os.environ.get(
"DINGTALK_REGISTRATION_BASE_URL", "https://oapi.dingtalk.com"
).rstrip("/")
REGISTRATION_BASE_URL = os.environ.get("DINGTALK_REGISTRATION_BASE_URL", "https://oapi.dingtalk.com").rstrip("/")
REGISTRATION_SOURCE = os.environ.get("DINGTALK_REGISTRATION_SOURCE", "openClaw")
_POLL_STATUSES = {"WAITING", "SUCCESS", "FAIL", "EXPIRED"}
_RETRY_WINDOW = 120 # seconds of transient errors / non-success statuses tolerated before giving up
# ── API helpers ────────────────────────────────────────────────────────────
class RegistrationError(Exception):
"""Raised when a DingTalk registration API call fails."""
@@ -39,22 +32,17 @@ def _api_post(path: str, payload: dict) -> dict:
errcode = data.get("errcode", -1)
if errcode != 0:
errmsg = data.get("errmsg", "unknown error")
raise RegistrationError(f"API error [{path}]: {errmsg} (errcode={errcode})")
raise RegistrationError(f"API error [{path}]: {data.get('errmsg', 'unknown error')} (errcode={errcode})")
return data
# ── Core flow ──────────────────────────────────────────────────────────────
def begin_registration() -> dict:
"""Start a device-flow registration."""
# Step 1: init → nonce
"""Start a device-flow registration: init → nonce, begin → device_code + verification URL."""
init_data = _api_post("/app/registration/init", {"source": REGISTRATION_SOURCE})
nonce = str(init_data.get("nonce", "")).strip()
if not nonce:
raise RegistrationError("init response missing nonce")
# Step 2: begin → device_code, verification_uri_complete
begin_data = _api_post("/app/registration/begin", {"nonce": nonce})
device_code = str(begin_data.get("device_code", "")).strip()
verification_uri_complete = str(begin_data.get("verification_uri_complete", "")).strip()
@@ -75,9 +63,7 @@ def poll_registration(device_code: str) -> dict:
"""Poll the registration status once."""
data = _api_post("/app/registration/poll", {"device_code": device_code})
status_raw = str(data.get("status", "")).strip().upper()
if status_raw not in {"WAITING", "SUCCESS", "FAIL", "EXPIRED"}:
status_raw = "UNKNOWN"
result = {"status": status_raw}
result = {"status": status_raw if status_raw in _POLL_STATUSES else "UNKNOWN"}
for key in ("client_id", "client_secret", "fail_reason"):
result[key] = str(data.get(key, "")).strip() or None
return result
@@ -89,19 +75,26 @@ def wait_for_registration_success(
expires_in: int = 7200,
on_waiting: Optional[callable] = None,
) -> Tuple[str, str]:
"""Block until the registration succeeds or times out."""
"""Block until the registration succeeds or times out.
Transient errors and FAIL/EXPIRED/UNKNOWN statuses are retried for ``_RETRY_WINDOW`` seconds
before being raised; a WAITING status resets that window.
"""
deadline = time.monotonic() + expires_in
retry_window = 120 # 2 minutes for transient errors
retry_start = 0.0
def _within_retry_window() -> bool:
nonlocal retry_start
if retry_start == 0:
retry_start = time.monotonic()
return time.monotonic() - retry_start < _RETRY_WINDOW
while time.monotonic() < deadline:
time.sleep(interval)
try:
result = poll_registration(device_code)
except RegistrationError:
if retry_start == 0:
retry_start = time.monotonic()
if time.monotonic() - retry_start < retry_window:
if _within_retry_window():
continue
raise
@@ -112,24 +105,17 @@ def wait_for_registration_success(
on_waiting()
continue
if status == "SUCCESS":
cid = result["client_id"]
csecret = result["client_secret"]
cid, csecret = result["client_id"], result["client_secret"]
if not cid or not csecret:
raise RegistrationError("authorization succeeded but credentials are missing")
return cid, csecret
# FAIL / EXPIRED / UNKNOWN
if retry_start == 0:
retry_start = time.monotonic()
if time.monotonic() - retry_start < retry_window:
if _within_retry_window():
continue
reason = result.get("fail_reason") or status
raise RegistrationError(f"authorization failed: {reason}")
raise RegistrationError(f"authorization failed: {result.get('fail_reason') or status}")
raise RegistrationError("authorization timed out, please retry")
# ── QR code rendering ─────────────────────────────────────────────────────
def _ensure_qrcode_installed() -> bool:
"""Try to import qrcode; if missing, auto-install it via pip/uv."""
try:
@@ -143,8 +129,7 @@ def _ensure_qrcode_installed() -> bool:
from hermes_cli.tools_config import _pip_install
try:
result = _pip_install(["-q", "qrcode"], timeout=120)
if result.returncode == 0:
if _pip_install(["-q", "qrcode"], timeout=120).returncode == 0:
import qrcode # noqa: F401,F811
return True
except (subprocess.SubprocessError, ImportError, OSError):
@@ -153,43 +138,31 @@ def _ensure_qrcode_installed() -> bool:
def render_qr_to_terminal(url: str) -> bool:
"""Render *url* as a compact QR code in the terminal."""
"""Render *url* as a compact QR code (half-block glyphs, 2 rows per character) in the terminal."""
try:
import qrcode
except ImportError:
return False
qr = qrcode.QRCode(
version=1,
error_correction=qrcode.constants.ERROR_CORRECT_L,
box_size=1,
border=1,
)
qr = qrcode.QRCode(version=1, error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=1, border=1)
qr.add_data(url)
qr.make(fit=True)
# Use half-block characters for compact rendering (2 rows per character)
matrix = qr.get_matrix()
rows = len(matrix)
lines: list[str] = []
# (top, bottom) -> ▀ ▄ █ or space
glyph = {(True, True): "\u2588", (True, False): "\u2580",
(False, True): "\u2584", (False, False): " "}
# (top, bottom) -> █ ▀ ▄ or space
glyph = {(True, True): "\u2588", (True, False): "\u2580", (False, True): "\u2584", (False, False): " "}
lines = []
for r in range(0, rows, 2):
bottom_row = matrix[r + 1] if r + 1 < rows else [False] * len(matrix[r])
lines.append(" " + "".join(
glyph[(bool(top), bool(bottom))] for top, bottom in zip(matrix[r], bottom_row)))
lines.append(" " + "".join(glyph[(bool(top), bool(bottom))] for top, bottom in zip(matrix[r], bottom_row)))
print("\n".join(lines))
return True
# ── High-level entry point for the setup wizard ───────────────────────────
def dingtalk_qr_auth() -> Optional[Tuple[str, str]]:
"""Run the interactive QR-code device-flow authorization."""
"""Run the interactive QR-code device-flow authorization (setup wizard entry point)."""
from hermes_cli.setup import print_info, print_success, print_warning, print_error
print()
@@ -205,7 +178,6 @@ def dingtalk_qr_auth() -> Optional[Tuple[str, str]]:
url = reg["verification_uri_complete"]
# Ensure qrcode library is available (auto-install if missing)
if not _ensure_qrcode_installed():
print_warning(" qrcode library install failed, will show link only.")
@@ -232,9 +204,7 @@ def dingtalk_qr_auth() -> Optional[Tuple[str, str]]:
try:
client_id, client_secret = wait_for_registration_success(
device_code=reg["device_code"],
interval=reg["interval"],
expires_in=reg["expires_in"],
device_code=reg["device_code"], interval=reg["interval"], expires_in=reg["expires_in"],
on_waiting=_on_waiting,
)
except RegistrationError as exc:
hermes_cli/fallback_cmd.py
+66 -159
View File
@@ -1,13 +1,6 @@
"""hermes fallback — manage the fallback provider chain.
"""hermes fallback — manage the fallback provider chain (tried in order when the primary fails).
Fallback providers are tried in order when the primary model fails with rate-limit, overload, or
connection errors. See: https://hermes-agent.nousresearch.com/docs/user-guide/features/fallback-
providers
Subcommands: hermes fallback [list] Show the current fallback chain (default when no subcommand)
hermes fallback add Pick provider + model via the same picker as `hermes model`, then append the
selection to the chain hermes fallback remove Pick an entry to delete from the chain hermes fallback
clear Remove all fallback entries
Subcommands: ``list`` (default), ``add`` (same picker as `hermes model`), ``remove``, ``clear``.
"""
from __future__ import annotations
@@ -16,37 +9,20 @@ from typing import Any, Dict, List, Optional
from hermes_cli.fallback_config import get_fallback_chain
# Normalized fallback chain (merges legacy ``fallback_model``); always a fresh copy.
_read_chain = get_fallback_chain
def _identity(entry: Dict[str, Any]):
"""BackendIdentity for a ``{provider, model, base_url?}`` entry."""
from agent.backend_identity import BackendIdentity
return BackendIdentity.build(
provider=entry.get("provider"),
model=entry.get("model"),
base_url=entry.get("base_url"),
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _read_chain(config: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Return the normalized fallback chain as a list of dicts.
Accepts both the new list format (``fallback_providers``) and the legacy ``fallback_model``
format. When both are present, the effective chain is merged with ``fallback_providers`` entries
kept first. The returned list is always a fresh copy — callers can mutate without touching the
config dict.
"""
return get_fallback_chain(config)
return BackendIdentity.build(provider=entry.get("provider"), model=entry.get("model"), base_url=entry.get("base_url"))
def _write_chain(config: Dict[str, Any], chain: List[Dict[str, Any]]) -> None:
"""Persist the chain to ``fallback_providers`` and clear legacy key."""
"""Persist the chain to ``fallback_providers``; drop the legacy key so there is one source of truth."""
config["fallback_providers"] = chain
# Drop the legacy single-dict key on write so there's only one source of truth.
config.pop("fallback_model", None)
@@ -74,7 +50,7 @@ def _extract_fallback_from_model_cfg(model_cfg: Any) -> Optional[Dict[str, Any]]
def _snapshot_auth_active_provider() -> Any:
"""Return the current ``active_provider`` in auth.json, or a sentinel if unavailable."""
"""Current ``active_provider`` in auth.json, or None if unavailable."""
try:
from hermes_cli.auth import _load_auth_store
return _load_auth_store().get("active_provider")
@@ -83,7 +59,10 @@ def _snapshot_auth_active_provider() -> Any:
def _restore_auth_active_provider(value: Any) -> None:
"""Write back a previously snapshotted ``active_provider`` value."""
"""Write back a previously snapshotted ``active_provider`` value.
Best-effort: if auth.json can't be restored the user re-runs `hermes model`; never fail the add.
"""
try:
from hermes_cli.auth import _auth_store_lock, _load_auth_store, _save_auth_store
with _auth_store_lock():
@@ -91,15 +70,20 @@ def _restore_auth_active_provider(value: Any) -> None:
store["active_provider"] = value
_save_auth_store(store)
except Exception:
# Best-effort — if auth.json can't be restored, the user's primary
# provider may have been deactivated by the picker. They can re-run
# `hermes model` to fix it. Don't fail the fallback add.
pass
# ---------------------------------------------------------------------------
# Subcommand handlers
# ---------------------------------------------------------------------------
def _restore_model_cfg(model_before: Any) -> None:
"""Restore ``config["model"]`` to a previously-captured snapshot."""
from hermes_cli.config import load_config, save_config
cfg = load_config()
if model_before is None:
cfg.pop("model", None)
else:
cfg["model"] = copy.deepcopy(model_before)
save_config(cfg)
def _entries(n: int) -> str:
return f"{n} {'entry' if n == 1 else 'entries'}"
@@ -119,32 +103,11 @@ def _load_chain(empty_message: str):
config = load_config()
chain = _read_chain(config)
if not chain:
print()
print(empty_message)
print()
print(f"\n{empty_message}\n")
return config, None
return config, chain
def cmd_fallback_list(args) -> None: # noqa: ARG001
"""Print the current fallback chain."""
config, chain = _load_chain(" No fallback providers configured.")
if chain is None:
print(" Add one with: hermes fallback add")
print()
return
print()
primary = _describe_primary(config)
if primary:
print(f" Primary: {primary}")
print()
_print_chain("Fallback chain", chain)
print(" Tried in order when the primary fails (rate-limit, 5xx, connection errors).")
print(" Docs: https://hermes-agent.nousresearch.com/docs/user-guide/features/fallback-providers")
print()
def _describe_primary(config: Dict[str, Any]) -> Optional[str]:
"""One-line description of the primary model for display purposes."""
model_cfg = config.get("model")
@@ -157,6 +120,22 @@ def _describe_primary(config: Dict[str, Any]) -> Optional[str]:
return None
def cmd_fallback_list(args) -> None: # noqa: ARG001
"""Print the current fallback chain."""
config, chain = _load_chain(" No fallback providers configured.")
if chain is None:
print(" Add one with: hermes fallback add\n")
return
print()
primary = _describe_primary(config)
if primary:
print(f" Primary: {primary}\n")
_print_chain("Fallback chain", chain)
print(" Tried in order when the primary fails (rate-limit, 5xx, connection errors).")
print(" Docs: https://hermes-agent.nousresearch.com/docs/user-guide/features/fallback-providers\n")
def cmd_fallback_add(args) -> None:
"""Launch the same picker as `hermes model`, then append the selection to the chain."""
from hermes_cli.main import _require_tty, select_provider_and_model
@@ -164,16 +143,13 @@ def cmd_fallback_add(args) -> None:
_require_tty("fallback add")
# Snapshot BEFORE the picker runs so we can distinguish "user actually
# picked something" from "user cancelled" by comparing before/after.
before_cfg = load_config()
model_before = copy.deepcopy(before_cfg.get("model"))
# Snapshot BEFORE the picker runs: "picked" vs "cancelled" is decided by comparing before/after,
# and the primary must be restored either way.
model_before = copy.deepcopy(load_config().get("model"))
active_provider_before = _snapshot_auth_active_provider()
print()
print(" Adding a fallback provider. The picker below is the same one used by")
print(" `hermes model` — select the provider + model you want as a fallback.")
print()
print("\n Adding a fallback provider. The picker below is the same one used by\n"
" `hermes model` — select the provider + model you want as a fallback.\n")
def _restore() -> None:
_restore_model_cfg(model_before)
@@ -181,78 +157,46 @@ def cmd_fallback_add(args) -> None:
try:
select_provider_and_model(args=args)
except SystemExit:
# Some provider flows exit on auth failure — restore state and re-raise.
except SystemExit: # some provider flows exit on auth failure — restore state and re-raise
_restore()
raise
# Read the post-picker state to see what the user selected.
after_cfg = load_config()
model_after = after_cfg.get("model")
new_entry = _extract_fallback_from_model_cfg(model_after)
if not new_entry:
# Picker didn't complete (user cancelled or flow bailed). Nothing to do.
new_entry = _extract_fallback_from_model_cfg(load_config().get("model"))
if not new_entry: # picker didn't complete (user cancelled or flow bailed)
_restore()
print()
print(" No fallback added.")
print("\n No fallback added.")
return
# Picker picked the same thing that's already the primary → nothing changed,
# and there's nothing useful to add as a fallback to itself. Identity
# semantics owned by agent.backend_identity (#54250/#57584/#62984): same
# provider+model on a DIFFERENT explicit base_url is a different backend
# (multi-endpoint pool) and is a legitimate fallback.
# Same deployment as the primary → nothing to add. Identity semantics are owned by
# agent.backend_identity: same provider+model on a DIFFERENT explicit base_url is a different
# backend (multi-endpoint pool) and a legitimate fallback.
from agent.backend_identity import same_deployment
new_ident = _identity(new_entry)
primary_entry = _extract_fallback_from_model_cfg(model_before)
if primary_entry and same_deployment(_identity(primary_entry), new_ident):
_restore()
print()
print(f" Selected model matches the current primary ({_format_entry(new_entry)}).")
print(f"\n Selected model matches the current primary ({_format_entry(new_entry)}).")
print(" A provider cannot be a fallback for itself — no change.")
return
# Reload the config with the primary restored, then append the new entry
# to ``fallback_providers``. We deliberately re-load (rather than mutating
# ``after_cfg``) because the picker may have touched other top-level keys
# (custom_providers, providers credentials) that we want to keep.
# Restore the primary, then re-load (rather than mutating the post-picker config) because the
# picker may have touched other top-level keys (custom_providers, credentials) we want to keep.
_restore()
final_cfg = load_config()
chain = _read_chain(final_cfg)
# Reject exact-duplicate fallback entries (same deployment; a different
# explicit base_url is a different endpoint and NOT a duplicate).
if any(same_deployment(_identity(existing), new_ident) for existing in chain):
print()
print(f" {_format_entry(new_entry)} is already in the fallback chain — skipped.")
print(f"\n {_format_entry(new_entry)} is already in the fallback chain — skipped.")
return
chain.append(new_entry)
_write_chain(final_cfg, chain)
save_config(final_cfg)
print()
print(f" Added fallback: {_format_entry(new_entry)}")
print(f" Chain is now {_entries(len(chain))} long.")
print()
print(f"\n Added fallback: {_format_entry(new_entry)}")
print(f" Chain is now {_entries(len(chain))} long.\n")
print(" Run `hermes fallback list` to view, or `hermes fallback remove` to delete.")
def _restore_model_cfg(model_before: Any) -> None:
"""Restore ``config["model"]`` to a previously-captured snapshot."""
from hermes_cli.config import load_config, save_config
cfg = load_config()
if model_before is None:
cfg.pop("model", None)
else:
cfg["model"] = copy.deepcopy(model_before)
save_config(cfg)
def cmd_fallback_remove(args) -> None: # noqa: ARG001
"""Pick an entry from the chain and remove it."""
from hermes_cli.config import save_config
@@ -261,27 +205,19 @@ def cmd_fallback_remove(args) -> None: # noqa: ARG001
if chain is None:
return
choices = [_format_entry(e) for e in chain] + ["Cancel"]
try:
from hermes_cli.setup import _curses_prompt_choice
idx = _curses_prompt_choice("Select a fallback to remove:", choices, 0)
except Exception:
idx = _numbered_pick("Select a fallback to remove:", choices)
# The curses menu owns its own non-TTY guard and numbered fallback; -1 means cancelled.
from hermes_cli.setup import _curses_prompt_choice
idx = _curses_prompt_choice("Select a fallback to remove:", [_format_entry(e) for e in chain] + ["Cancel"], 0)
if idx is None or idx < 0 or idx >= len(chain):
print()
print(" Cancelled — no change.")
print("\n Cancelled — no change.")
return
removed = chain.pop(idx)
_write_chain(config, chain)
save_config(config)
print()
print(f" Removed fallback: {_format_entry(removed)}")
print(f" Chain is now {_entries(len(chain))} long." if chain else " Fallback chain is now empty.")
print()
print(f"\n Removed fallback: {_format_entry(removed)}")
print(f" Chain is now {_entries(len(chain))} long.\n" if chain else " Fallback chain is now empty.\n")
def cmd_fallback_clear(args) -> None: # noqa: ARG001
@@ -297,8 +233,7 @@ def cmd_fallback_clear(args) -> None: # noqa: ARG001
try:
resp = input(" Clear all entries? [y/N]: ").strip().lower()
except (KeyboardInterrupt, EOFError):
print()
print(" Cancelled.")
print("\n Cancelled.")
return
if resp not in {"y", "yes"}:
print(" Cancelled — no change.")
@@ -306,37 +241,9 @@ def cmd_fallback_clear(args) -> None: # noqa: ARG001
_write_chain(config, [])
save_config(config)
print()
print(" Fallback chain cleared.")
print()
print("\n Fallback chain cleared.\n")
def _numbered_pick(question: str, choices: List[str]) -> Optional[int]:
"""Fallback numbered-list picker when curses is unavailable."""
print(question)
for i, c in enumerate(choices, 1):
print(f" {i}. {c}")
print()
while True:
try:
val = input(f"Choice [1-{len(choices)}]: ").strip()
if not val:
return None
idx = int(val) - 1
if 0 <= idx < len(choices):
return idx
print(f"Please enter 1-{len(choices)}")
except ValueError:
print("Please enter a number")
except (KeyboardInterrupt, EOFError):
print()
return None
# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------
def cmd_fallback(args) -> None:
"""Top-level dispatcher for ``hermes fallback [subcommand]``."""
sub = getattr(args, "fallback_command", None)
hermes_cli/fallback_config.py
+6 -23
View File
@@ -6,9 +6,7 @@ from typing import Any
def _normalized_base_url(value: Any) -> str:
if not isinstance(value, str):
return ""
return value.strip().rstrip("/")
return value.strip().rstrip("/") if isinstance(value, str) else ""
def resolve_entry_api_key(entry: dict[str, Any] | None) -> str | None:
@@ -34,13 +32,7 @@ def resolve_entry_api_key(entry: dict[str, Any] | None) -> str | None:
def _iter_fallback_entries(raw: Any) -> list[dict[str, Any]]:
if isinstance(raw, dict):
candidates = [raw]
elif isinstance(raw, list):
candidates = raw
else:
return []
candidates = [raw] if isinstance(raw, dict) else raw if isinstance(raw, list) else []
entries: list[dict[str, Any]] = []
for entry in candidates:
if not isinstance(entry, dict):
@@ -49,15 +41,10 @@ def _iter_fallback_entries(raw: Any) -> list[dict[str, Any]]:
model = str(entry.get("model") or "").strip()
if not provider or not model:
continue
normalized = dict(entry)
normalized["provider"] = provider
normalized["model"] = model
normalized = {**entry, "provider": provider, "model": model}
base_url = _normalized_base_url(entry.get("base_url"))
if base_url:
normalized["base_url"] = base_url
entries.append(normalized)
return entries
@@ -78,17 +65,13 @@ def get_fallback_chain(config: dict[str, Any] | None) -> list[dict[str, Any]]:
provider/model/base_url route as an earlier entry. The returned list always contains fresh dict
copies.
"""
config = config or {}
chain: list[dict[str, Any]] = []
seen: set[tuple[str, str, str]] = set()
for key in ("fallback_providers", "fallback_model"):
for entry in _iter_fallback_entries(config.get(key)):
identity = _entry_identity(entry)
if identity in seen:
continue
seen.add(identity)
chain.append(entry)
if identity not in seen:
seen.add(identity)
chain.append(entry)
return chain
hermes_cli/focus_view.py
+26 -51
View File
@@ -1,30 +1,24 @@
"""Focus view — a display-only reduced-output mode.
``/focus`` answers one question the existing ``/verbose`` cycle cannot: *"just show me my prompt and
the answer — and tell me what you hid."*
* turning focus ON snaps ``tool_progress_mode`` to ``"off"`` and remembers the mode the user had
configured, so the *existing* suppression path does the actual hiding; * turning focus OFF restores
that remembered mode verbatim; * on top of that, focus view adds the two things ``/verbose off``
lacks — a per-turn count of what was hidden plus a recovery hint, and a persistent ``focus`` segment
in the status bar so the reduced mode is never invisible.
``/focus`` = "just show me my prompt and the answer — and tell me what you hid". Turning focus ON
snaps ``tool_progress_mode`` to ``"off"`` and remembers the configured mode so the *existing*
suppression path does the hiding; OFF restores that mode verbatim. On top, focus adds a per-turn
hidden-line count with a recovery hint and a persistent ``focus`` status-bar segment.
"""
from __future__ import annotations
from typing import Optional
# Config key used by the sibling display toggles (/battery, /timestamps,
# /footer) — a plain boolean under ``display``.
# Config key used by the sibling display toggles (/battery, /timestamps, /footer).
FOCUS_CONFIG_KEY = "display.focus_view"
#: Tool-progress mode focus view snaps to. Deliberately the SAME value
#: ``/verbose off`` uses so both features share one suppression path.
#: Tool-progress mode focus view snaps to — the SAME value ``/verbose off`` uses so both share one
#: suppression path.
FOCUS_TOOL_PROGRESS_MODE = "off"
#: Modes in which the CLI commits a per-tool scrollback line. Mirrors the gate
#: in ``HermesCLI._on_tool_progress``; kept here so the hidden-line counter and
#: the renderer can never drift apart.
#: Modes in which the CLI commits a per-tool scrollback line. Mirrors the gate in
#: ``HermesCLI._on_tool_progress`` so the hidden-line counter and the renderer never drift apart.
TOOL_PROGRESS_VISIBLE_MODES = frozenset({"new", "all", "verbose"})
#: Valid tool-progress modes (``log`` is a gateway-only extra step).
@@ -33,10 +27,12 @@ TOOL_PROGRESS_MODES = ("off", "new", "all", "verbose")
#: Status-bar label. Short on purpose — the bar is width-constrained.
FOCUS_STATUSBAR_LABEL = "◉ focus"
_ON_WORDS = frozenset({"on", "enable", "enabled", "true", "yes", "1"})
_OFF_WORDS = frozenset({"off", "disable", "disabled", "false", "no", "0"})
_STATUS_WORDS = frozenset({"status", "show", "?"})
_TOGGLE_WORDS = frozenset({"", "toggle"})
# /focus argument words -> (action, target); bare /focus toggles like /footer, /battery, /timestamps.
_FOCUS_WORDS = {
**dict.fromkeys(("status", "show", "?"), ("status", None)),
**dict.fromkeys(("on", "enable", "enabled", "true", "yes", "1"), ("set", True)),
**dict.fromkeys(("off", "disable", "disabled", "false", "no", "0"), ("set", False)),
}
def normalize_tool_progress_mode(mode: object, default: str = "all") -> str:
@@ -46,38 +42,23 @@ def normalize_tool_progress_mode(mode: object, default: str = "all") -> str:
if mode is True:
return "all"
text = str(mode or "").strip().lower()
if text in TOOL_PROGRESS_MODES:
return text
# ``log`` is a real gateway mode; treat any other unknown value as default.
if text == "log":
return "log"
return default
# ``log`` is a real gateway mode; any other unknown value becomes the default.
return text if text in TOOL_PROGRESS_MODES or text == "log" else default
def resolve_focus_arg(arg: str, current: bool) -> tuple[str, Optional[bool]]:
"""Map a ``/focus`` argument onto an action, following the sibling toggles.
"""Map a ``/focus`` argument onto ``(action, target)``.
Returns ``(action, target)`` where ``action`` is one of ``"set"``, ``"status"`` or ``"usage"``.
``target`` is the requested enabled-state for ``"set"`` and ``None`` otherwise. Bare ``/focus``
toggles, matching ``/footer`` / ``/battery`` / ``/timestamps``.
``action`` is ``"set"``, ``"status"`` or ``"usage"``; ``target`` is the requested enabled-state
for ``"set"`` and ``None`` otherwise.
"""
text = str(arg or "").strip().lower()
if text in _STATUS_WORDS:
return "status", None
if text in _ON_WORDS:
return "set", True
if text in _OFF_WORDS:
return "set", False
if text in _TOGGLE_WORDS:
if text in ("", "toggle"):
return "set", not bool(current)
return "usage", None
return _FOCUS_WORDS.get(text, ("usage", None))
def would_display_tool_line(
mode: object,
function_name: str,
last_tool_name: Optional[str] = None,
) -> bool:
def would_display_tool_line(mode: object, function_name: str, last_tool_name: Optional[str] = None) -> bool:
"""Would the CLI have committed a scrollback line for this tool call?
Counts honestly: with ``/verbose off`` focus view hides nothing extra and must not claim
@@ -86,9 +67,7 @@ def would_display_tool_line(
if not function_name:
return False
normalized = normalize_tool_progress_mode(mode)
return normalized in TOOL_PROGRESS_VISIBLE_MODES and not (
normalized == "new" and function_name == last_tool_name
)
return normalized in TOOL_PROGRESS_VISIBLE_MODES and not (normalized == "new" and function_name == last_tool_name)
def format_hidden_line(count: int) -> Optional[str]:
@@ -111,10 +90,7 @@ def format_focus_status(enabled: bool, configured_mode: object) -> str:
"""Human-readable ``/focus status`` body (no ANSI — callers colour it)."""
mode = normalize_tool_progress_mode(configured_mode).upper()
if enabled:
return (
"Focus view: ON — only your prompt and the final response.\n"
f" /focus off restores tool progress: {mode}"
)
return f"Focus view: ON — only your prompt and the final response.\n /focus off restores tool progress: {mode}"
return f"Focus view: OFF — tool progress: {mode}"
@@ -122,5 +98,4 @@ def format_focus_toggle_message(enabled: bool, configured_mode: object) -> str:
"""Confirmation line printed when focus view is switched (no ANSI)."""
if enabled:
return "Focus view enabled — just your prompt and the final response"
mode = normalize_tool_progress_mode(configured_mode)
return f"Focus view disabled — tool progress: {mode.upper()}"
return f"Focus view disabled — tool progress: {normalize_tool_progress_mode(configured_mode).upper()}"
hermes_cli/foreign_sessions.py
+40 -97
View File
@@ -1,23 +1,22 @@
"""Import sessions from foreign coding agents (Claude Code, Codex CLI).
Sources (read-only — foreign files are never modified):
Conversion contract — imported history must satisfy the provider role-alternation invariant Hermes
enforces everywhere else:
Foreign files are only ever read. Imported history must satisfy the provider role-alternation
invariant Hermes enforces everywhere else — see ``_merge_turns``.
"""
from __future__ import annotations
import json
import os
import re
import sys
import uuid
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
# User-message texts that are really injected context wrappers, not typed
# input. Matched against the stripped start of the text.
# User-message texts that are really injected context wrappers, not typed input.
_WRAPPER_TAG_RE = re.compile(
r"^<(?:user_instructions|environment_context|recommended_plugins|"
r"skills_instructions|permissions[_-]instructions|turn_context|"
@@ -27,6 +26,7 @@ _WRAPPER_TAG_RE = re.compile(
_TITLE_MAX = 60
_SOURCE_LABELS = {"claude": "Claude Code", "codex": "Codex CLI"}
_SOURCE_DB_NAMES = {"claude": "claude-code", "codex": "codex-cli"}
@dataclass
@@ -55,7 +55,7 @@ def _read_json_lines(path: Path):
for line in f:
try:
obj = json.loads(line) if line.strip() else None
except ValueError: # JSONDecodeError subclasses ValueError
except ValueError:
continue
if isinstance(obj, dict):
yield obj
@@ -75,8 +75,8 @@ def _flatten_blocks(content: Any) -> str:
parts.append(block)
elif not isinstance(block, dict):
continue
# tool_result (tool output echoed into a user message — not typed input),
# thinking / redacted_thinking / reasoning and unknown types are skipped.
# tool_result (tool output echoed into a user message), thinking/reasoning and unknown
# block types are skipped.
elif (btype := block.get("type")) in ("text", "input_text", "output_text"):
text = block.get("text")
if isinstance(text, str) and text:
@@ -88,10 +88,6 @@ def _flatten_blocks(content: Any) -> str:
return "\n\n".join(p for p in (s.strip() for s in parts) if p)
def _is_wrapper_text(text: str) -> bool:
return bool(_WRAPPER_TAG_RE.match(text.lstrip()))
def _merge_turns(raw_turns: List[Tuple[str, str]]) -> List[Dict[str, str]]:
"""Merge consecutive same-role turns; guarantee strict alternation.
@@ -117,11 +113,20 @@ def _message_turn(role: Any, content: Any) -> Optional[Tuple[str, str]]:
if role not in ("user", "assistant"):
return None
text = _flatten_blocks(content)
if not text or (role == "user" and _is_wrapper_text(text)):
if not text or (role == "user" and _WRAPPER_TAG_RE.match(text.lstrip())):
return None
return (role, text)
def _first_user_line(turns: List[Tuple[str, str]]) -> Optional[str]:
for role, text in turns:
if role == "user":
line = text.strip().splitlines()[0].strip()
if line:
return line[:_TITLE_MAX * 2]
return None
def _parsed(turns: List[Tuple[str, str]], cwd: Optional[str], session_id: Optional[str],
title: Optional[str] = None) -> Dict[str, Any]:
return {
@@ -135,9 +140,7 @@ def _parsed(turns: List[Tuple[str, str]], cwd: Optional[str], session_id: Option
def parse_claude_session(path: Path) -> Dict[str, Any]:
"""Parse one Claude Code session JSONL into normalized turns + meta."""
turns: List[Tuple[str, str]] = []
cwd: Optional[str] = None
summary: Optional[str] = None
session_id: Optional[str] = None
cwd = summary = session_id = None
for obj in _read_json_lines(path):
otype = obj.get("type")
if otype == "summary":
@@ -145,9 +148,7 @@ def parse_claude_session(path: Path) -> Dict[str, Any]:
if isinstance(s, str) and s.strip():
summary = s.strip()
continue
if otype not in ("user", "assistant"):
continue
if obj.get("isSidechain") or obj.get("isMeta"):
if otype not in ("user", "assistant") or obj.get("isSidechain") or obj.get("isMeta"):
continue
if cwd is None and isinstance(obj.get("cwd"), str):
cwd = obj["cwd"]
@@ -165,8 +166,7 @@ def parse_claude_session(path: Path) -> Dict[str, Any]:
def parse_codex_session(path: Path) -> Dict[str, Any]:
"""Parse one Codex CLI rollout JSONL into normalized turns + meta."""
turns: List[Tuple[str, str]] = []
cwd: Optional[str] = None
session_id: Optional[str] = None
cwd = session_id = None
for obj in _read_json_lines(path):
otype = obj.get("type")
payload = obj.get("payload")
@@ -182,16 +182,14 @@ def parse_codex_session(path: Path) -> Dict[str, Any]:
if otype != "response_item":
continue
ptype = payload.get("type")
if ptype == "message":
# developer/system payloads never imported
if ptype == "message": # developer/system payloads never imported
turn = _message_turn(payload.get("role"), payload.get("content"))
if turn:
turns.append(turn)
elif ptype in ("custom_tool_call", "function_call", "local_shell_call"):
# Assistant activity; merged into neighbours later. Tool outputs / reasoning skipped.
name = payload.get("name") or payload.get("tool") or "tool"
# Attach as assistant activity; merged into neighbors later.
turns.append(("assistant", f"[ran tool: {name}]"))
# tool outputs / reasoning / web_search etc. are skipped
return _parsed(turns, cwd, session_id)
@@ -215,44 +213,16 @@ def _list_sessions(source: str, root: Optional[Path]) -> List[ForeignSession]:
continue
parsed = parse(jsonl)
if parsed["turns"]:
results.append(ForeignSession(
source=source, path=jsonl, mtime=mtime, cwd=parsed["cwd"],
title_guess=parsed["title_guess"], turn_count=len(parsed["turns"]),
session_id=parsed["session_id"],
))
results.append(ForeignSession(source, jsonl, mtime, parsed["cwd"], parsed["title_guess"],
len(parsed["turns"]), parsed["session_id"]))
results.sort(key=lambda s: s.mtime, reverse=True)
return results
def list_claude_sessions(root: Optional[Path] = None) -> List[ForeignSession]:
"""Discover Claude Code sessions under ``~/.claude/projects``."""
return _list_sessions("claude", root)
def list_codex_sessions(root: Optional[Path] = None) -> List[ForeignSession]:
"""Discover Codex CLI rollouts under ``~/.codex/sessions``."""
return _list_sessions("codex", root)
def _first_user_line(turns: List[Tuple[str, str]]) -> Optional[str]:
for role, text in turns:
if role == "user":
line = text.strip().splitlines()[0].strip()
if line:
return line[:_TITLE_MAX * 2]
return None
# ── Import ───────────────────────────────────────────────────────────────
_SOURCE_DB_NAMES = {"claude": "claude-code", "codex": "codex-cli"}
def import_foreign_session(source: str, path, db=None) -> str:
"""Import one foreign session into the Hermes SessionDB.
"""Import one foreign session into the Hermes SessionDB; returns the new Hermes session id.
Returns the new Hermes session id. The foreign file is only read. Raises ``ValueError`` on
unknown source or a session with no usable conversation turns.
Raises ``ValueError`` on unknown source or a session with no usable conversation turns.
"""
source = (source or "").strip().lower().lstrip("@")
if source not in _SOURCE_LABELS:
@@ -264,14 +234,12 @@ def import_foreign_session(source: str, path, db=None) -> str:
parsed = _SOURCES[source][3](path)
turns = parsed["turns"]
if not turns:
raise ValueError(
f"No user/assistant conversation turns found in {path}"
)
raise ValueError(f"No user/assistant conversation turns found in {path}")
first_user = _first_user_line([(t["role"], t["content"]) for t in turns]) or path.stem
if len(first_user) > _TITLE_MAX:
first_user = first_user[: _TITLE_MAX - 1] + "…"
title = f"Imported from {_SOURCE_LABELS[source]}: {first_user}"
tool = _SOURCE_DB_NAMES[source]
owns_db = db is None
if owns_db:
@@ -280,16 +248,8 @@ def import_foreign_session(source: str, path, db=None) -> str:
db = SessionDB()
try:
session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
origin = {
"imported_from": {
"tool": _SOURCE_DB_NAMES[source],
"path": str(path),
"foreign_session_id": parsed.get("session_id"),
}
}
db.create_session(
session_id, source=_SOURCE_DB_NAMES[source], cwd=parsed.get("cwd"), origin_json=json.dumps(origin)
)
origin = {"imported_from": {"tool": tool, "path": str(path), "foreign_session_id": parsed.get("session_id")}}
db.create_session(session_id, source=tool, cwd=parsed.get("cwd"), origin_json=json.dumps(origin))
for turn in turns:
db.append_message(session_id, turn["role"], turn["content"])
try:
@@ -305,14 +265,8 @@ def import_foreign_session(source: str, path, db=None) -> str:
pass
# ── Picker / CLI helpers ─────────────────────────────────────────────────
def gather_foreign_sessions(
source: Optional[str] = None,
*,
claude_root: Optional[Path] = None,
codex_root: Optional[Path] = None,
source: Optional[str] = None, *, claude_root: Optional[Path] = None, codex_root: Optional[Path] = None,
limit: int = 25,
) -> List[ForeignSession]:
"""List foreign sessions across sources, newest first."""
@@ -324,13 +278,8 @@ def gather_foreign_sessions(
return sessions[:limit] if limit else sessions
def pick_foreign_session(
source: Optional[str] = None, *, limit: int = 25
) -> Optional[ForeignSession]:
def pick_foreign_session(source: Optional[str] = None, *, limit: int = 25) -> Optional[ForeignSession]:
"""Interactive numbered picker. Returns None when nothing was chosen."""
import os
import sys
sessions = gather_foreign_sessions(source, limit=limit)
if not sessions:
where = _SOURCE_LABELS.get(source or "", "Claude Code or Codex CLI")
@@ -342,16 +291,14 @@ def pick_foreign_session(
ws = f" ({os.path.basename(s.cwd.rstrip('/')) or s.cwd})" if s.cwd else ""
print(f" {i:>2}. {when} {s.label}{ws} [{s.turn_count} turns]")
if not sys.stdin.isatty():
print(
"Non-interactive terminal — pass the file path directly:\n"
" hermes sessions import --from claude|codex <path>"
)
print("Non-interactive terminal — pass the file path directly:\n"
" hermes sessions import --from claude|codex <path>")
return None
try:
raw = input(f"Import which session? [1-{len(sessions)}, empty to cancel] ")
raw = input(f"Import which session? [1-{len(sessions)}, empty to cancel] ").strip()
except (EOFError, KeyboardInterrupt):
return None
if not (raw := raw.strip()):
if not raw:
return None
try:
idx = int(raw)
@@ -368,15 +315,12 @@ def run_sessions_import(args, db=None) -> Optional[str]:
"""`hermes sessions import` entry point. Returns new session id or None."""
source = getattr(args, "from_source", None)
path = getattr(args, "path", None)
if path:
# Report a missing file distinctly instead of the misleading
# "cannot infer source" (SES-10).
# A missing file is reported as such, not as the misleading "cannot infer source".
if not Path(path).exists():
print(f"Error: file not found: {path}")
return None
if not source:
# Guess from the path shape.
if not source: # guess from the path shape
p = str(path)
if "/.claude/" in p or p.endswith(".jsonl") and "claude" in p:
source = "claude"
@@ -391,7 +335,6 @@ def run_sessions_import(args, db=None) -> Optional[str]:
if picked is None:
return None
source, chosen_path = picked.source, picked.path
try:
session_id = import_foreign_session(source, chosen_path, db=db)
except ValueError as e:
hermes_cli/gitlock.py
+23 -47
View File
@@ -1,9 +1,8 @@
"""Stale git lock-file recovery for update/check paths.
"""Stale git lock-file and aborted-fetch pack-debris recovery for update/check paths.
A crashed or killed ``git fetch`` on a shallow clone can leave ``.git/shallow.lock`` behind. Every
later fetch then fails with::
fatal: Unable to create '/path/.git/shallow.lock': File exists.
A crashed or killed ``git fetch`` can leave ``.git/shallow.lock`` behind (every later fetch then
fails with "Unable to create '.../shallow.lock': File exists") and ``tmp_pack_*`` files under
``.git/objects/pack`` that git itself never cleans up.
"""
from __future__ import annotations
@@ -17,36 +16,26 @@ from typing import Callable, Iterable, List, Optional
logger = logging.getLogger(__name__)
# Lock files younger than this are presumed live (a fetch is in flight) and
# are never removed. git lock files are created and removed within a single
# fetch (seconds); anything older than 10 minutes is abandoned by any
# reasonable standard.
# Files younger than this are presumed live (a fetch may be in flight) and are never removed. git
# lock files live for seconds and a healthy fetch completes in minutes; 10 minutes is abandoned by
# any reasonable standard.
STALE_LOCK_MIN_AGE_SECONDS = 10 * 60
# Lock files we know how to self-heal. ``shallow.lock`` is the one observed in
# the wild (interrupted fetch on a shallow clone); the others are the same
# class of failure (interrupted git operation) and harmless to clear when
# stale. Index/HEAD locks from a live git process are protected by the
# process guard in :func:`clear_stale_git_locks`.
LOCK_NAMES = ("shallow.lock", "index.lock", "HEAD.lock", "MERGE_HEAD.lock")
# Aborted-fetch pack debris younger than this is presumed live (a fetch may
# be writing it right now) and is never removed. A healthy fetch completes in
# minutes; the same 10-minute bar the lock sweep uses is comfortably safe.
STALE_TMP_PACK_MIN_AGE_SECONDS = STALE_LOCK_MIN_AGE_SECONDS
# Temp-file prefixes git writes into .git/objects/pack during a transfer and
# renames away on success. Anything left with these names after a fetch died
# is garbage by definition — git itself never reuses or cleans them.
# ``shallow.lock`` is the one observed in the wild; the others are the same class of failure
# (interrupted git operation). Locks held by a live git process are protected by the process guard.
LOCK_NAMES = ("shallow.lock", "index.lock", "HEAD.lock", "MERGE_HEAD.lock")
# Temp-file prefixes git writes into .git/objects/pack during a transfer and renames away on
# success. Anything left with these names after a fetch died is garbage by definition.
_TMP_PACK_PREFIXES = ("tmp_pack_", "tmp_idx_", "tmp_rev_", "tmp_mtimes_")
def _git_proc_running() -> bool:
"""True when a ``git`` process is currently running.
The conservative answer on any platform we can't probe: if we can't tell, treat a lock as
possibly-live and don't remove it. This is the safety check that stops us from yanking a lock a
real fetch is holding.
This is the safety check that stops us from yanking a lock a real fetch is holding. A failed
probe logs and returns False; the age floor in the sweep still applies.
"""
try:
if os.name == "nt":
@@ -55,10 +44,7 @@ def _git_proc_running() -> bool:
capture_output=True, text=True, timeout=10,
).stdout.lower()
return "git.exe" in out
out = subprocess.run(
["pgrep", "-x", "git"], capture_output=True, text=True, timeout=10,
)
return out.returncode == 0
return subprocess.run(["pgrep", "-x", "git"], capture_output=True, text=True, timeout=10).returncode == 0
except Exception:
logger.debug("git process probe failed; assuming no git running", exc_info=True)
return False
@@ -95,13 +81,11 @@ def _sweep_stale(
def clear_stale_git_locks(repo_root: Path, *, min_age_seconds: Optional[int] = None) -> List[str]:
"""Remove abandoned ``.git`` lock files under ``repo_root``.
"""Remove abandoned ``.git`` lock files under ``repo_root``; returns the removed paths.
A lock is removed only when BOTH conditions hold:
Returns the list of removed lock file paths. Never raises: a lock we cannot stat or unlink is
skipped (a concurrently-held lock may have just been created between our age check and the
unlink — the process guard makes that window vanishingly small, and skipping is always safe).
A lock is removed only when it is older than the age floor AND no git process is running. Never
raises: a lock we cannot stat or unlink is skipped (a concurrently-held lock may have been
created between the age check and the unlink; skipping is always safe).
"""
git_dir = Path(repo_root) / ".git"
return _sweep_stale(
@@ -114,16 +98,10 @@ def clear_stale_git_locks(repo_root: Path, *, min_age_seconds: Optional[int] = N
)
def clear_stale_tmp_packs(
repo_root: Path, *, min_age_seconds: Optional[int] = None
) -> List[str]:
def clear_stale_tmp_packs(repo_root: Path, *, min_age_seconds: Optional[int] = None) -> List[str]:
"""Remove aborted-fetch temp pack files under ``.git/objects/pack``.
Every ``git fetch`` that dies mid-transfer (timeout, HTTP 429, dropped connection) leaves a
``tmp_pack_*`` (and sometimes ``tmp_idx_*``) file behind, and git never cleans them up.
Same safety contract as :func:`clear_stale_git_locks`: only files older than the age floor,
never while a git process is running, never raises. Returns the removed paths.
Same safety contract as :func:`clear_stale_git_locks`. Returns the removed paths.
"""
pack_dir = Path(repo_root) / ".git" / "objects" / "pack"
@@ -140,7 +118,5 @@ def clear_stale_tmp_packs(
min_age_seconds=min_age_seconds,
default_age=STALE_TMP_PACK_MIN_AGE_SECONDS,
skip_msg="git process running; skipping tmp-pack sweep",
log_removed=lambda p, size: logger.info(
"Removed aborted-fetch pack debris %s (%d bytes)", p, size
),
log_removed=lambda p, size: logger.info("Removed aborted-fetch pack debris %s (%d bytes)", p, size),
)
hermes_cli/gui_uninstall.py
+32 -90
View File
@@ -1,7 +1,8 @@
"""Hermes Desktop (Chat GUI) uninstaller.
This holds the desktop's own ``connection.json`` / ``updates.json`` and Chromium cache — pure GUI
state, safe to remove on a GUI uninstall.
Removes only GUI state — built Electron artifacts, the packaged app, and the desktop's own
``userData`` (connection.json / updates.json / Chromium cache). Never agent source, venv, config,
sessions or .env under ``$HERMES_HOME``.
"""
import os
@@ -14,16 +15,13 @@ from hermes_constants import get_hermes_home
from hermes_cli.colors import Colors, color
def log_info(msg: str):
print(f"{color('→', Colors.CYAN)} {msg}")
def _logger(mark: str, col: str):
return lambda msg: print(f"{color(mark, col)} {msg}")
def log_success(msg: str):
print(f"{color('✓', Colors.GREEN)} {msg}")
def log_warn(msg: str):
print(f"{color('⚠', Colors.YELLOW)} {msg}")
log_info = _logger("→", Colors.CYAN)
log_success = _logger("✓", Colors.GREEN)
log_warn = _logger("⚠", Colors.YELLOW)
def _env_dir(var: str, fallback: Path) -> Path:
@@ -32,37 +30,26 @@ def _env_dir(var: str, fallback: Path) -> Path:
return Path(value) if value else fallback
# ---------------------------------------------------------------------------
# Discovery
# ---------------------------------------------------------------------------
def _agent_root(hermes_home: Path) -> Path:
"""The agent checkout root — same layout install.sh / install.ps1 use."""
return hermes_home / "hermes-agent"
def desktop_userdata_dir() -> Path:
"""Return the Electron ``userData`` directory for the desktop app.
Mirrors Electron's ``app.getPath('userData')`` for an app named "Hermes" on each platform. This
is GUI-only state (connection.json, updates.json, Chromium cache) and never holds agent config
or sessions.
"""
"""Electron ``app.getPath('userData')`` for an app named "Hermes" on each platform (GUI-only state)."""
home = Path.home()
if sys.platform == "darwin":
return home / "Library" / "Application Support" / "Hermes"
if sys.platform == "win32":
return _env_dir("APPDATA", home / "AppData" / "Roaming") / "Hermes"
# Linux / other POSIX — XDG config home.
return _env_dir("XDG_CONFIG_HOME", home / ".config") / "Hermes"
def source_built_gui_artifacts(hermes_home: Path) -> "list[Path]":
"""GUI build artifacts produced by ``hermes desktop`` inside the checkout.
These are removable on a GUI uninstall without harming the agent: the Python agent runs from
``hermes-agent/`` source + ``venv/`` and never needs the Electron build output or node_modules.
The Python agent runs from ``hermes-agent/`` source + ``venv/`` and never needs the Electron
build output or node_modules (the workspace-root node_modules only carries Electron, ~200MB).
"""
agent_root = _agent_root(hermes_home)
desktop_dir = agent_root / "apps" / "desktop"
@@ -70,76 +57,54 @@ def source_built_gui_artifacts(hermes_home: Path) -> "list[Path]":
desktop_dir / "dist",
desktop_dir / "release",
desktop_dir / "node_modules",
# Workspace-root node_modules carries Electron (devDependency of the
# desktop workspace, ~200MB). The agent does not use any npm package,
# so this is GUI tooling — safe to drop on a GUI uninstall.
agent_root / "node_modules",
hermes_home / "desktop-build-stamp.json",
]
def packaged_gui_app_paths() -> "list[Path]":
"""Standard install locations of the packaged desktop distributable.
"""Standard install locations of the packaged desktop distributable for the current OS.
Returns every candidate for the current OS; the caller filters to those that actually exist. We
never glob system-wide — only the well-known electron-builder output locations for the "Hermes"
product.
Every candidate is returned; the caller filters to those that exist. Never globs system-wide —
only the well-known electron-builder output locations for the "Hermes" product.
"""
home = Path.home()
paths: list[Path] = []
if sys.platform == "darwin":
paths += [
Path("/Applications/Hermes.app"),
home / "Applications" / "Hermes.app",
]
paths += [Path("/Applications/Hermes.app"), home / "Applications" / "Hermes.app"]
elif sys.platform == "win32":
local_base = _env_dir("LOCALAPPDATA", home / "AppData" / "Local")
paths += [
# NSIS per-user install (perMachine=false → Programs\Hermes).
local_base / "Programs" / "Hermes",
# Older / alternate layout some builds used.
local_base / "hermes-desktop",
local_base / "Programs" / "Hermes", # NSIS per-user install (perMachine=false)
local_base / "hermes-desktop", # older / alternate layout some builds used
]
program_files = os.environ.get("ProgramFiles")
if program_files:
# NSIS per-machine fallback (needs admin to remove).
paths.append(Path(program_files) / "Hermes")
paths.append(Path(program_files) / "Hermes") # NSIS per-machine fallback (needs admin)
else:
# Linux: AppImage is a single file the user placed somewhere; we can
# only reliably clean the desktop entry + icon we know the name of.
# The AppImage itself lives wherever the user put it, so we surface a
# hint rather than guessing. deb/rpm installs are owned by the system
# package manager and must be removed via apt/dnf — see the message in
# ``uninstall_gui``.
# Linux: an AppImage lives wherever the user put it and deb/rpm files belong to the package
# manager (see the hint in ``uninstall_gui``), so only the desktop entry + hicolor icons
# ``hermes desktop`` installs are cleaned here.
from hermes_cli.linux_desktop_entry import desktop_entry_path
data_base = _env_dir("XDG_DATA_HOME", home / ".local" / "share")
paths += [
# The launcher entry `hermes desktop` installs. Its icon is
# also copied into the hicolor tree (see
# linux_desktop_entry._install_icon_to_hicolor) — remove
# every size dir the installer could have written.
desktop_entry_path(),
# Some packaged builds emit this casing.
data_base / "applications" / "Hermes.desktop",
data_base / "applications" / "Hermes.desktop", # some packaged builds emit this casing
data_base / "icons" / "hicolor" / "scalable" / "apps" / "hermes.png",
]
# Fixed-size hicolor dirs the installer may have written (resized
# panel sizes plus leftover native-size copies from older builds).
# Fixed-size hicolor dirs the installer may have written (panel sizes + older native-size copies).
for size in ("24x24", "32x32", "48x48", "256x256", "512x512", "1024x1024"):
paths.append(data_base / "icons" / "hicolor" / size / "apps" / "hermes.png")
return paths
def agent_is_installed(hermes_home: Path) -> bool:
"""Return True when a usable Python agent install exists under HERMES_HOME.
"""True when a usable Python agent install exists under HERMES_HOME (gates the desktop UI's options).
Used by the desktop UI to decide which uninstall options to offer: if the agent isn't present (a
future "lite" GUI-only client), the "remove agent" options are hidden.
Package source or a venv alone is enough — a source checkout without a venv is still "the agent is here".
"""
agent_root = _agent_root(hermes_home)
# A real install has the package source + a venv. Either signal alone is
# enough — a source checkout without a venv is still "the agent is here".
return any((agent_root / sub).is_dir() for sub in ("hermes_cli", "venv", ".venv"))
@@ -152,14 +117,9 @@ def gui_is_installed(hermes_home: Path) -> bool:
def gui_install_summary(hermes_home: "Path | None" = None) -> dict:
"""Structured snapshot of what's installed, for the desktop UI to render.
Returns JSON-serializable primitives (paths as strings, booleans for the questions the UI gates
on) so the Electron main process can forward it to the renderer via IPC.
"""
"""JSON-serializable snapshot of what's installed, for the desktop UI to render via IPC."""
home: Path = hermes_home if hermes_home is not None else get_hermes_home()
userdata = desktop_userdata_dir()
return {
"hermes_home": str(home),
"agent_installed": agent_is_installed(home),
@@ -172,11 +132,6 @@ def gui_install_summary(hermes_home: "Path | None" = None) -> dict:
}
# ---------------------------------------------------------------------------
# Removal
# ---------------------------------------------------------------------------
def _remove_path(path: Path) -> bool:
"""Remove a file or directory tree. Returns True when something was removed."""
try:
@@ -191,16 +146,9 @@ def _remove_path(path: Path) -> bool:
return False
def uninstall_gui(
hermes_home: "Path | None" = None, *, remove_userdata: bool = True
) -> "list[Path]":
"""Remove the desktop GUI's artifacts, leaving the agent + user data intact.
Never touches ``hermes-agent/hermes_cli`` (agent source), ``venv/``, or any config / sessions /
.env under ``$HERMES_HOME``.
"""
def uninstall_gui(hermes_home: "Path | None" = None, *, remove_userdata: bool = True) -> "list[Path]":
"""Remove the desktop GUI's artifacts, leaving the agent + user data intact."""
home: Path = hermes_home if hermes_home is not None else get_hermes_home()
removed: list[Path] = []
def _remove_existing(paths) -> bool:
@@ -230,18 +178,11 @@ def uninstall_gui(
if not removed:
log_info("No desktop GUI artifacts found to remove")
# Linux deb/rpm installs are owned by the package manager; we can't (and
# shouldn't) rmtree files under /usr. Surface the hint so the user can
# finish the job. AppImages live wherever the user dropped them.
if sys.platform.startswith("linux"):
# The desktop entry was removed above (it is in
# ``packaged_gui_app_paths``), but the menu caches still list it.
# Reindex so Hermes disappears from the launcher.
# The desktop entry was removed above but the menu caches still list it; reindex so Hermes
# disappears from the launcher.
try:
from hermes_cli.linux_desktop_entry import (
desktop_entry_path,
refresh_desktop_databases,
)
from hermes_cli.linux_desktop_entry import desktop_entry_path, refresh_desktop_databases
entry = desktop_entry_path()
if entry in removed:
@@ -250,6 +191,7 @@ def uninstall_gui(
except Exception as e:
log_warn(f"Could not refresh the application menu cache: {e}")
# deb/rpm files under /usr belong to the package manager; AppImages live wherever the user dropped them.
log_info(
"If you installed the desktop via a .deb / .rpm package, remove it "
"with your package manager (e.g. 'sudo apt remove hermes' or "
hermes_cli/heartbeat.py
+28 -59
View File
@@ -1,11 +1,10 @@
"""Session heartbeats — recurring re-entry prompts for the current session.
This is deliberately session-scoped and in-process (CLI process or gateway process must be running)
— the durable cross-process scheduling surface remains ``hermes cron`` / the ``cronjob`` tool, which
runs in isolated sessions.
Deliberately session-scoped and in-process (the CLI or gateway process must be running); the
durable cross-process scheduling surface remains ``hermes cron`` / the ``cronjob`` tool.
Invariants (mirrors goals.py): - Injection is a plain user message. No system-prompt mutation, no
toolset swap — prompt caching stays intact. - A real user message always wins: heartbeats only fire
Invariants (mirrors goals.py): injection is a plain user message — no system-prompt mutation, no
toolset swap, so prompt caching stays intact. A real user message always wins: heartbeats only fire
into an idle session with an empty input queue.
"""
@@ -20,9 +19,7 @@ from typing import Any, Optional
logger = logging.getLogger(__name__)
# Floor: a heartbeat that re-enters the session more often than once a
# minute is a busy-loop, not a heartbeat. (Prime-Agent uses a similar floor.)
# Floor: re-entering more often than once a minute is a busy-loop, not a heartbeat.
MIN_INTERVAL_SECONDS = 60
# How often drivers poll for due heartbeats. Not user-facing.
POLL_SECONDS = 5.0
@@ -47,12 +44,22 @@ _UNIT_SECONDS = {
"d": 86400, "day": 86400, "days": 86400,
}
# field -> (coercer, default used when the stored value is missing/falsy)
_STATE_FIELDS = {
"prompt": (str, ""),
"interval_seconds": (int, 0),
"status": (str, "active"),
"created_at": (float, 0.0),
"last_fired_at": (float, 0.0),
"fire_count": (int, 0),
}
def parse_interval(text: str) -> Optional[int]:
"""Parse ``10m`` / ``every 2h`` / ``every 90 minutes`` into seconds.
Returns None when the text is not an interval; values below ``MIN_INTERVAL_SECONDS`` return
-1 so callers can distinguish "not an interval" from "too small".
None when the text is not an interval; values below ``MIN_INTERVAL_SECONDS`` return -1 so
callers can distinguish "not an interval" from "too small".
"""
m = _INTERVAL_RE.match(text) if text else None
if not m:
@@ -87,47 +94,20 @@ class HeartbeatState:
@classmethod
def from_json(cls, raw: str) -> "HeartbeatState":
data = json.loads(raw)
# Falsy/missing values fall back to the default before type coercion.
return cls(**{
name: coerce(data.get(name) or default)
for name, (coerce, default) in _STATE_FIELDS.items()
})
return cls(**{name: coerce(data.get(name) or default) for name, (coerce, default) in _STATE_FIELDS.items()})
def is_due(self, now: Optional[float] = None) -> bool:
if self.status != "active" or not self.prompt or self.interval_seconds <= 0:
return False
now = now if now is not None else time.time()
anchor = self.last_fired_at or self.created_at
return (now - anchor) >= self.interval_seconds
return (now - (self.last_fired_at or self.created_at)) >= self.interval_seconds
def render_prompt(self) -> str:
return HEARTBEAT_PROMPT_TEMPLATE.format(
interval=format_interval(self.interval_seconds),
prompt=self.prompt,
)
# field -> (coercer, default used when the stored value is missing/falsy)
_STATE_FIELDS = {
"prompt": (str, ""),
"interval_seconds": (int, 0),
"status": (str, "active"),
"created_at": (float, 0.0),
"last_fired_at": (float, 0.0),
"fire_count": (int, 0),
}
# Persistence (SessionDB state_meta) — same pattern as goals.py
def _meta_key(session_id: str) -> str:
return f"heartbeat:{session_id}"
return HEARTBEAT_PROMPT_TEMPLATE.format(interval=format_interval(self.interval_seconds), prompt=self.prompt)
def _get_session_db() -> Optional[Any]:
# Reuse the goals module's per-HERMES_HOME cached SessionDB so both
# features share one connection instead of thrashing the file.
"""Persistence goes through the goals module's per-HERMES_HOME cached SessionDB (one shared connection)."""
try:
from hermes_cli.goals import _get_session_db as _goals_db
@@ -142,7 +122,7 @@ def load_heartbeat(session_id: str) -> Optional[HeartbeatState]:
if db is None:
return None
try:
raw = db.get_meta(_meta_key(session_id))
raw = db.get_meta(f"heartbeat:{session_id}")
except Exception as exc:
logger.debug("HeartbeatManager: get_meta failed: %s", exc)
return None
@@ -166,16 +146,13 @@ def save_heartbeat(session_id: str, state: HeartbeatState) -> None:
_warn_dropped_write("HeartbeatManager", "heartbeat", session_id)
return
try:
db.set_meta(_meta_key(session_id), state.to_json())
db.set_meta(f"heartbeat:{session_id}", state.to_json())
except Exception as exc:
logger.debug("HeartbeatManager: set_meta failed: %s", exc)
# Manager — the surface CLI + gateway talk to
class HeartbeatManager:
"""Per-session heartbeat state + due-tick decisions.
"""Per-session heartbeat state + due-tick decisions; the surface CLI + gateway talk to.
Drivers (CLI thread / gateway task) call :meth:`due_prompt` on a poll cadence while the session
is idle; a non-None return is the user-role message to inject. Firing is recorded immediately so
@@ -203,14 +180,11 @@ class HeartbeatManager:
every = format_interval(s.interval_seconds)
fired = f", fired {s.fire_count}×" if s.fire_count else ""
if s.status == "active":
anchor = s.last_fired_at or s.created_at
next_in = max(0, int(anchor + s.interval_seconds - time.time()))
next_in = max(0, int((s.last_fired_at or s.created_at) + s.interval_seconds - time.time()))
return f"♥ Heartbeat (every {every}, next in ~{next_in}s{fired}): {s.prompt}"
icon = "⏸ " if s.status == "paused" else ""
return f"{icon}Heartbeat ({s.status}, every {every}{fired}): {s.prompt}"
# --- mutation -----------------------------------------------------
def set(self, prompt: str, interval_seconds: int) -> HeartbeatState:
prompt = (prompt or "").strip()
if not prompt:
@@ -218,9 +192,7 @@ class HeartbeatManager:
interval_seconds = int(interval_seconds)
if interval_seconds < MIN_INTERVAL_SECONDS:
raise ValueError(f"interval must be at least {MIN_INTERVAL_SECONDS}s")
state = HeartbeatState(
prompt=prompt, interval_seconds=interval_seconds, status="active", created_at=time.time()
)
state = HeartbeatState(prompt=prompt, interval_seconds=interval_seconds, status="active", created_at=time.time())
self._state = state
save_heartbeat(self.session_id, state)
return state
@@ -247,8 +219,6 @@ class HeartbeatManager:
self._state = None
return True
# --- driver entry point --------------------------------------------
def due_prompt(self, now: Optional[float] = None) -> Optional[str]:
"""Return the injection prompt if the heartbeat is due, else None.
@@ -287,7 +257,6 @@ def migrate_heartbeat_to_session(old_session_id: str, new_session_id: str) -> bo
__all__ = [
"HeartbeatState", "HeartbeatManager", "parse_interval", "format_interval",
"load_heartbeat", "save_heartbeat", "migrate_heartbeat_to_session",
"HEARTBEAT_PROMPT_TEMPLATE", "MIN_INTERVAL_SECONDS", "POLL_SECONDS",
"HeartbeatState", "HeartbeatManager", "parse_interval", "format_interval", "load_heartbeat", "save_heartbeat",
"migrate_heartbeat_to_session", "HEARTBEAT_PROMPT_TEMPLATE", "MIN_INTERVAL_SECONDS", "POLL_SECONDS",
]
hermes_cli/image_provenance.py
+13 -33
View File
@@ -1,11 +1,9 @@
"""Image-authored deployment provenance for immutable Hermes runtimes.
The published image bakes ``/etc/hermes/image-provenance.json`` outside both ``$HERMES_HOME`` and
the mutable checkout. A bind-mounted checkout (including ``.git``) therefore cannot hide the build
fact, and environment or config values cannot forge it.
Absence preserves every pre-existing source/package install path. Presence fails closed: an
unreadable, non-regular, or malformed marker still means the runtime is image-managed; it is an
the mutable checkout, so a bind-mounted checkout cannot hide the build fact and env/config cannot
forge it. Absence preserves every pre-existing source/package install path. Presence fails closed:
an unreadable, non-regular, or malformed marker still means the runtime is image-managed — an
integrity defect, never permission to mutate the image in place.
"""
@@ -41,15 +39,8 @@ class ImageProvenance:
def _invalid(path: Path, reason: str) -> ImageProvenance:
return ImageProvenance(
schema=IMAGE_PROVENANCE_SCHEMA,
deployment_kind="image",
manager="unknown",
image=None,
version=None,
revision=None,
marker_path=str(path),
valid=False,
error=reason,
schema=IMAGE_PROVENANCE_SCHEMA, deployment_kind="image", manager="unknown",
image=None, version=None, revision=None, marker_path=str(path), valid=False, error=reason,
)
@@ -62,15 +53,11 @@ def _optional_string(payload: dict, name: str) -> Optional[str]:
return value.strip() or None
def read_image_provenance(
marker_path: Optional[Path] = None,
) -> Optional[ImageProvenance]:
"""Read the baked marker without consulting environment or config.
def read_image_provenance(marker_path: Optional[Path] = None) -> Optional[ImageProvenance]:
"""Read the baked marker without consulting environment or config. Never raises.
``marker_path`` is a dependency-injection seam for tests and alternate image builders. Normal
callers always use the image-owned absolute path. This function never raises.
``marker_path`` is a dependency-injection seam for tests and alternate image builders.
"""
path = IMAGE_PROVENANCE_PATH
try:
path = Path(marker_path) if marker_path is not None else path
@@ -81,8 +68,7 @@ def read_image_provenance(
marker_stat = path.lstat()
except FileNotFoundError:
return None
except BaseException as exc:
# Permission errors and other lookup failures do not prove absence.
except BaseException as exc: # permission errors and other lookup failures do not prove absence
return _invalid(path, f"marker_presence_unreadable:{type(exc).__name__}")
if not stat.S_ISREG(marker_stat.st_mode):
@@ -90,17 +76,14 @@ def read_image_provenance(
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except Exception as exc:
# The file may disappear between lstat/read; it was nevertheless
# observed present, so the decision remains fail-closed.
except Exception as exc: # may vanish between lstat/read; it was observed present, so fail closed
return _invalid(path, f"marker_unreadable:{type(exc).__name__}")
if not isinstance(payload, dict):
return _invalid(path, "marker_not_object")
schema = payload.get("schema")
# ``bool`` is an ``int`` subclass in Python. Schema ``true`` must not be
# accepted as schema 1, hence the exact type check.
# ``bool`` subclasses ``int``: schema ``true`` must not be accepted as schema 1.
if type(schema) is not int or schema != IMAGE_PROVENANCE_SCHEMA:
return _invalid(path, "unsupported_marker_schema")
if payload.get("deployment_kind") != "image":
@@ -116,9 +99,6 @@ def read_image_provenance(
return _invalid(path, f"invalid_{exc.args[0]}")
return ImageProvenance(
schema=IMAGE_PROVENANCE_SCHEMA,
deployment_kind="image",
manager=manager.strip(),
marker_path=str(path),
**optional,
schema=IMAGE_PROVENANCE_SCHEMA, deployment_kind="image", manager=manager.strip(),
marker_path=str(path), **optional,
)
hermes_cli/init_command.py
+7 -22
View File
@@ -1,15 +1,10 @@
#!/usr/bin/env python3
"""``/init`` — build the prompt that generates or updates a project AGENTS.md.
1. Inspect the project with its own read-only tools (``read_file`` / ``search_files`` on manifests,
CI configs, lockfiles, existing docs) to learn the layout, toolchain, and the exact build/test/lint
commands. 2.
"""
"""``/init`` — build the prompt that asks the agent to generate or update a project AGENTS.md."""
from __future__ import annotations
# The quality bar, embedded in every prompt so the generated file reads like a
# maintainer wrote it — concrete and command-exact, not generic advice.
import os
# Embedded in every prompt so the generated file reads like a maintainer wrote it.
_QUALITY_BAR = """\
Quality bar for the file you write (this is what separates a useful AGENTS.md
from noise):
@@ -32,17 +27,9 @@ from noise):
"Pitfalls"). Flat and scannable — no deep nesting."""
def build_init_prompt(
cwd: str,
existing_file: str | None = None,
extra: str = "",
) -> str:
"""Build the agent prompt for a ``/init`` request.
When ``existing_file`` (current ``AGENTS.md`` content) is given, the prompt switches to
update-and-merge discipline instead of fresh generation. ``extra`` is the user's free text
after ``/init`` to honor while authoring.
"""
def build_init_prompt(cwd: str, existing_file: str | None = None, extra: str = "") -> str:
"""Build the ``/init`` prompt; ``existing_file`` (current AGENTS.md) switches to merge discipline,
``extra`` is the user's free text after ``/init``."""
extra = (extra or "").strip()
parts: list[str] = [
@@ -106,8 +93,6 @@ def build_init_prompt(
def build_init_prompt_for_cwd(cwd: str | None = None, extra: str = "") -> str:
"""Convenience wrapper used by the dispatch surfaces."""
import os
resolved = os.path.abspath(cwd or os.getcwd())
existing: str | None = None
agents_path = os.path.join(resolved, "AGENTS.md")
hermes_cli/input_sanitize.py
+4 -14
View File
@@ -22,35 +22,26 @@ def strip_leaked_bracketed_paste_wrappers(text: str) -> str:
"""
if not text:
return text
text = (
text.replace("\x1b[200~", "")
.replace("\x1b[201~", "")
.replace("^[[200~", "")
.replace("^[[201~", "")
)
for wrapper in ("\x1b[200~", "\x1b[201~", "^[[200~", "^[[201~"):
text = text.replace(wrapper, "")
text = _BRACKETED_PASTE_BOUNDARY_START.sub(r"\1", text)
text = _BRACKETED_PASTE_BOUNDARY_END.sub("", text)
text = _BRACKETED_PASTE_DEGRADED_START.sub(r"\1", text)
text = _BRACKETED_PASTE_DEGRADED_END.sub("", text)
return text
return _BRACKETED_PASTE_DEGRADED_END.sub("", text)
def collapse_repeated_input_artifacts(text: str, min_repeats: int = 4) -> str:
"""Drop a trailing run of the desktop ~[[e corruption signature (#62557)."""
if not text:
return text
marker = _DESKTOP_PASTE_ARTIFACT
index = len(text)
repeat_count = 0
while index >= len(marker) and text[index - len(marker) : index] == marker:
repeat_count += 1
index -= len(marker)
if repeat_count < min_repeats:
return text
start = index
if start >= 2 and text[start - 2 : start] == "[e":
start -= 2
@@ -63,5 +54,4 @@ def sanitize_user_prompt_text(text: str) -> str:
"""Normalize user-authored prompt text before persistence or model input."""
if not isinstance(text, str) or not text:
return text
cleaned = strip_leaked_bracketed_paste_wrappers(text)
return collapse_repeated_input_artifacts(cleaned)
return collapse_repeated_input_artifacts(strip_leaked_bracketed_paste_wrappers(text))
hermes_cli/install_identity.py
+30 -31
View File
@@ -23,8 +23,7 @@ _INSTALL_ID_PUBLICATION_LOCK = threading.Lock()
@contextlib.contextmanager
def _install_id_file_lock(root: Path):
"""Serialize identity publication across processes on POSIX and Windows."""
lock_path = root / ".install_id.lock"
fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
fd = os.open(root / ".install_id.lock", os.O_RDWR | os.O_CREAT, 0o600)
windows = os.name == "nt"
try:
if windows:
@@ -71,6 +70,17 @@ def _fsync_directory(path: Path) -> None:
os.close(fd)
def _read_existing(path: Path) -> tuple[Optional[str], bool]:
"""``(valid id or None, mint?)`` — mint on a missing or malformed file, never on a read failure."""
try:
existing = path.read_text(encoding="utf-8").strip().lower()
except FileNotFoundError:
return None, True
except (OSError, UnicodeDecodeError):
return None, False
return (existing, False) if _INSTALL_ID_RE.fullmatch(existing) else (None, True)
def read_or_create_install_id(root: Path | None = None) -> Optional[str]:
"""Read or atomically mint the opaque id for the physical install.
@@ -79,14 +89,9 @@ def read_or_create_install_id(root: Path | None = None) -> Optional[str]:
"""
root = get_default_hermes_root() if root is None else root
path = root / _INSTALL_ID_FILENAME
try:
existing = path.read_text(encoding="utf-8").strip().lower()
if _INSTALL_ID_RE.fullmatch(existing):
return existing
except FileNotFoundError:
pass
except (OSError, UnicodeDecodeError):
return None
existing, mint = _read_existing(path)
if not mint:
return existing
try:
root.mkdir(parents=True, exist_ok=True)
@@ -94,18 +99,13 @@ def read_or_create_install_id(root: Path | None = None) -> Optional[str]:
return None
try:
# Windows byte-range locks can report a same-process lock conflict
# instead of waiting for another thread. Serialize threads here, then
# retain the file lock as the cross-process publication fence.
# Windows byte-range locks can report a same-process lock conflict instead of waiting for
# another thread. Serialize threads here, then retain the file lock as the cross-process
# publication fence.
with _INSTALL_ID_PUBLICATION_LOCK, _install_id_file_lock(root):
try:
existing = path.read_text(encoding="utf-8").strip().lower()
if _INSTALL_ID_RE.fullmatch(existing):
return existing
except FileNotFoundError:
pass
except (OSError, UnicodeDecodeError):
return None
existing, mint = _read_existing(path)
if not mint:
return existing
minted = uuid.uuid4().hex
fd, tmp_name = tempfile.mkstemp(dir=str(root), prefix=".install_id-")
@@ -127,22 +127,21 @@ def read_or_create_install_id(root: Path | None = None) -> Optional[str]:
return None
def get_install_id(
*,
cache: dict[str, Optional[str]] | None = None,
) -> Optional[str]:
def get_install_id(*, cache: dict[str, Optional[str]] | None = None) -> Optional[str]:
"""Return the process-cached stable id for the active Hermes root."""
root = get_default_hermes_root()
root_key = str(root)
target_cache = _INSTALL_ID_CACHE if cache is None else cache
cached = target_cache.get("value")
if cached and target_cache.get("root") in (None, root_key):
return cached
with _INSTALL_ID_LOCK:
def _cached() -> Optional[str]:
cached = target_cache.get("value")
if cached and target_cache.get("root") in (None, root_key):
return cached
return cached if cached and target_cache.get("root") in (None, root_key) else None
if value := _cached():
return value
with _INSTALL_ID_LOCK:
if value := _cached():
return value
value = read_or_create_install_id(root)
if value:
target_cache["root"] = root_key
tests/hermes_cli/test_foreign_sessions.py
+5 -6
View File
@@ -9,10 +9,9 @@ import json
import pytest
from hermes_cli.foreign_sessions import (
_list_sessions,
gather_foreign_sessions,
import_foreign_session,
list_claude_sessions,
list_codex_sessions,
parse_claude_session,
parse_codex_session,
)
@@ -171,8 +170,8 @@ def test_malformed_lines_are_skipped(tmp_path):
def test_list_sessions(tmp_path):
_write_claude_fixture(tmp_path)
_write_codex_fixture(tmp_path)
claude = list_claude_sessions(tmp_path / ".claude" / "projects")
codex = list_codex_sessions(tmp_path / ".codex" / "sessions")
claude = _list_sessions("claude", tmp_path / ".claude" / "projects")
codex = _list_sessions("codex", tmp_path / ".codex" / "sessions")
assert len(claude) == 1 and claude[0].source == "claude"
assert claude[0].turn_count == 4
assert len(codex) == 1 and codex[0].source == "codex"
@@ -186,8 +185,8 @@ def test_list_sessions(tmp_path):
def test_list_sessions_missing_roots(tmp_path):
assert list_claude_sessions(tmp_path / "nope") == []
assert list_codex_sessions(tmp_path / "nope") == []
assert _list_sessions("claude", tmp_path / "nope") == []
assert _list_sessions("codex", tmp_path / "nope") == []
# ── import into SessionDB ────────────────────────────────────────────────