diff --git a/hermes_cli/default_soul.py b/hermes_cli/default_soul.py index 1479eb4080..992a66b464 100644 --- a/hermes_cli/default_soul.py +++ b/hermes_cli/default_soul.py @@ -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) diff --git a/hermes_cli/dep_ensure.py b/hermes_cli/dep_ensure.py index bf5549e7fa..0c24a41371 100644 --- a/hermes_cli/dep_ensure.py +++ b/hermes_cli/dep_ensure.py @@ -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() diff --git a/hermes_cli/diagnostics_upload.py b/hermes_cli/diagnostics_upload.py index 172e62bedf..e8f3635ceb 100644 --- a/hermes_cli/diagnostics_upload.py +++ b/hermes_cli/diagnostics_upload.py @@ -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 (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 (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 diff --git a/hermes_cli/dingtalk_auth.py b/hermes_cli/dingtalk_auth.py index 90f2906211..2961ab1eac 100644 --- a/hermes_cli/dingtalk_auth.py +++ b/hermes_cli/dingtalk_auth.py @@ -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: diff --git a/hermes_cli/fallback_cmd.py b/hermes_cli/fallback_cmd.py index 3b00430937..6efd7a0bf5 100644 --- a/hermes_cli/fallback_cmd.py +++ b/hermes_cli/fallback_cmd.py @@ -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) diff --git a/hermes_cli/fallback_config.py b/hermes_cli/fallback_config.py index b80e5ca4bd..56f8b77331 100644 --- a/hermes_cli/fallback_config.py +++ b/hermes_cli/fallback_config.py @@ -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 diff --git a/hermes_cli/focus_view.py b/hermes_cli/focus_view.py index b615736966..5b7b6b91cb 100644 --- a/hermes_cli/focus_view.py +++ b/hermes_cli/focus_view.py @@ -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()}" diff --git a/hermes_cli/foreign_sessions.py b/hermes_cli/foreign_sessions.py index bc258d643b..66a3bb6109 100644 --- a/hermes_cli/foreign_sessions.py +++ b/hermes_cli/foreign_sessions.py @@ -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 " - ) + print("Non-interactive terminal — pass the file path directly:\n" + " hermes sessions import --from claude|codex ") 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: diff --git a/hermes_cli/gitlock.py b/hermes_cli/gitlock.py index f6246e6fa4..3ec27f80b0 100644 --- a/hermes_cli/gitlock.py +++ b/hermes_cli/gitlock.py @@ -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), ) diff --git a/hermes_cli/gui_uninstall.py b/hermes_cli/gui_uninstall.py index 52b8c1e608..ab6efca862 100644 --- a/hermes_cli/gui_uninstall.py +++ b/hermes_cli/gui_uninstall.py @@ -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 " diff --git a/hermes_cli/heartbeat.py b/hermes_cli/heartbeat.py index fdb1d4bb4e..526f4a9930 100644 --- a/hermes_cli/heartbeat.py +++ b/hermes_cli/heartbeat.py @@ -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", ] diff --git a/hermes_cli/image_provenance.py b/hermes_cli/image_provenance.py index 17399bfcef..3cc075197c 100644 --- a/hermes_cli/image_provenance.py +++ b/hermes_cli/image_provenance.py @@ -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, ) diff --git a/hermes_cli/init_command.py b/hermes_cli/init_command.py index 7f299a8319..d0090bd25e 100644 --- a/hermes_cli/init_command.py +++ b/hermes_cli/init_command.py @@ -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") diff --git a/hermes_cli/input_sanitize.py b/hermes_cli/input_sanitize.py index e69ece09ae..fd26e5a2a7 100644 --- a/hermes_cli/input_sanitize.py +++ b/hermes_cli/input_sanitize.py @@ -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)) diff --git a/hermes_cli/install_identity.py b/hermes_cli/install_identity.py index a1e0502d90..a39151264e 100644 --- a/hermes_cli/install_identity.py +++ b/hermes_cli/install_identity.py @@ -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 diff --git a/tests/hermes_cli/test_foreign_sessions.py b/tests/hermes_cli/test_foreign_sessions.py index 941455af92..a522068971 100644 --- a/tests/hermes_cli/test_foreign_sessions.py +++ b/tests/hermes_cli/test_foreign_sessions.py @@ -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 ────────────────────────────────────────────────