refactor(hermes_cli): second pass on the 15 small modules — compact bodies, drop redundant locals/blanks, tighten module docs

hermes_cli/dep_ensure.py

@@ -1,9 +1,5 @@
-"""Lazy dependency bootstrapper for non-Python runtime deps.
-
-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.
-"""
+"""Lazy dependency bootstrapper for non-Python runtime deps. Detection and prompting live here (cross-platform,
+instant, Python-controlled UX); install.sh / install.ps1 remain the *installation* backend."""
 from __future__ import annotations
 
 import platform
@@ -53,11 +49,7 @@ def _has_npx_agent_browser() -> bool:
     check can't diverge from what browser tools actually find.
     """
     try:
-        from tools.browser_tool import (
-            _find_agent_browser,
-            _is_npx_agent_browser_sentinel,
-            _requires_real_termux_browser_install,
-        )
+        from tools.browser_tool import _find_agent_browser, _is_npx_agent_browser_sentinel, _requires_real_termux_browser_install
         browser_cmd = _find_agent_browser(validate=False)
     except Exception:
         return False
@@ -65,7 +57,7 @@ def _has_npx_agent_browser() -> bool:
 
 
 def _has_hermes_agent_browser() -> bool:
-    from hermes_constants import get_hermes_home
+    from hermes_constants import get_hermes_home  # late import: tests patch hermes_constants.get_hermes_home
     home = get_hermes_home()
     if _IS_WINDOWS:  # npm -g --prefix puts .cmd shims directly in the prefix dir
         return (home / "node" / "agent-browser.cmd").is_file()
@@ -77,9 +69,7 @@ def _has_hermes_agent_browser() -> bool:
     )
 
 
-def _find_install_script(
-    package_dir: Path | None = None, repo_root: Path | None = None
-) -> tuple[Path | None, str | None]:
+def _find_install_script(package_dir: Path | None = None, repo_root: Path | None = None) -> tuple[Path | None, str | None]:
     """Locate the install script — bundled in wheel or in git checkout."""
     package_dir = package_dir or Path(__file__).parent
     repo_root = repo_root or package_dir.parent
@@ -97,11 +87,8 @@ 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
-        return False
-    if check():
-        return True
-
+    if check is None or check():  # unknown dep — don't silently forward to install script
+        return check is not None
     script, shell = _find_install_script()
     desc = _DEP_DESCRIPTIONS.get(dep, dep)
     if script is None:
@@ -109,7 +96,6 @@ def ensure_dependency(dep: str, interactive: bool = True) -> bool:
             print(f"  {desc} is not installed and no install script was found.")
             print(f"  Install {dep} manually and try again.")
         return False
-
     if interactive and sys.stdin.isatty():
         try:
             reply = input(f"{desc} is not installed. Install now? [Y/n] ").strip().lower()
@@ -117,7 +103,6 @@ def ensure_dependency(dep: str, interactive: bool = True) -> bool:
             return False
         if reply not in ("", "y", "yes"):
             return False
-
     if shell == "powershell":
         from hermes_constants import get_hermes_home
         ps_bin = shutil.which("powershell") or shutil.which("pwsh")
@@ -125,13 +110,8 @@ def ensure_dependency(dep: str, interactive: bool = True) -> bool:
             if interactive:
                 print("  PowerShell not found. Install PowerShell or run install.ps1 manually.")
             return False
-        cmd = [
-            ps_bin, "-ExecutionPolicy", "Bypass", "-File", str(script),
-            "-Ensure", dep, "-HermesHome", str(get_hermes_home()),
-        ]
+        cmd = [ps_bin, "-ExecutionPolicy", "Bypass", "-File", str(script), "-Ensure", dep, "-HermesHome", str(get_hermes_home())]
     else:
         cmd = ["bash", str(script), "--ensure", dep]
-
-    run_env = hermes_subprocess_env(inherit_credentials=False)
-    run_env["IS_INTERACTIVE"] = "false"
+    run_env = {**hermes_subprocess_env(inherit_credentials=False), "IS_INTERACTIVE": "false"}
     return subprocess.run(cmd, env=run_env).returncode == 0 and check()

hermes_cli/diagnostics_upload.py

@@ -1,11 +1,8 @@
 """Client for uploading ``hermes debug share`` bundles to Nous-internal S3.
 
-1. POST {NAS_BASE}/api/diagnostics/upload-url → {uploadUrl, viewUrl, id, ...}. The request body
-   carries ``sizeBytes``; NAS signs it into the presigned URL's ``ContentLength``, so the PUT must
-   send exactly that many bytes.
-2. PUT <uploadUrl> (the gzipped bundle, Content-Type application/gzip).
-
-NAS is stateless — the object's existence in S3 is the only state, so there is no confirm step.
+1. POST {NAS_BASE}/api/diagnostics/upload-url → {uploadUrl, viewUrl, id, ...}; the body carries ``sizeBytes``,
+   which NAS signs into the presigned URL's ``ContentLength``, so the PUT must send exactly that many bytes.
+2. PUT <uploadUrl> (gzipped bundle, Content-Type application/gzip). NAS is stateless — no confirm step.
 """
 
 import json
@@ -14,11 +11,8 @@ import urllib.request
 
 # 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 timeouts (seconds); the PUT carries the gzipped log bundle so it gets a more generous window.
 _REQUEST_TIMEOUT = 30
-_UPLOAD_TIMEOUT = 120
-
+_UPLOAD_TIMEOUT = 120  # the PUT carries the gzipped log bundle, so a more generous window
 _USER_AGENT = "hermes-agent/debug-share"
 
 
@@ -26,8 +20,7 @@ def _urlopen_checked(req: urllib.request.Request, *, timeout: int, what: str):
     """Open *req*; raise ``RuntimeError`` on non-2xx and return the response body bytes."""
     with urllib.request.urlopen(req, timeout=timeout) as resp:
         status = getattr(resp, "status", None)
-        if status is None:
-            status = resp.getcode()
+        status = resp.getcode() if status is None else status
         if not (200 <= status < 300):
             raise RuntimeError(f"{what} failed: HTTP {status}")
         return resp.read()
@@ -42,7 +35,6 @@ def request_upload_url(content_type: str = "application/gzip", size_bytes: int |
     payload: dict = {"contentType": content_type}
     if size_bytes is not None:
         payload["sizeBytes"] = int(size_bytes)
-
     req = urllib.request.Request(
         f"{NAS_BASE}/api/diagnostics/upload-url",
         data=json.dumps(payload).encode("utf-8"),
@@ -50,12 +42,10 @@ def request_upload_url(content_type: str = "application/gzip", size_bytes: int |
         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")
-
     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
-
     if not isinstance(result, dict) or not result.get("uploadUrl"):
         raise RuntimeError(f"diagnostics upload-url response missing 'uploadUrl': {body[:200]}")
     return result

hermes_cli/dingtalk_auth.py

@@ -11,7 +11,6 @@ import requests
 
 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
 
@@ -29,7 +28,6 @@ def _api_post(path: str, payload: dict) -> dict:
         data = resp.json()
     except requests.RequestException as exc:
         raise RegistrationError(f"Network error calling {url}: {exc}") from exc
-
     errcode = data.get("errcode", -1)
     if errcode != 0:
         raise RegistrationError(f"API error [{path}]: {data.get('errmsg', 'unknown error')} (errcode={errcode})")
@@ -42,21 +40,14 @@ def begin_registration() -> dict:
     nonce = str(init_data.get("nonce", "")).strip()
     if not nonce:
         raise RegistrationError("init response missing nonce")
-
     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()
-    if not device_code:
-        raise RegistrationError("begin response missing device_code")
-    if not verification_uri_complete:
-        raise RegistrationError("begin response missing verification_uri_complete")
-
-    return {
-        "device_code": device_code,
-        "verification_uri_complete": verification_uri_complete,
-        "expires_in": int(begin_data.get("expires_in", 7200)),
-        "interval": max(int(begin_data.get("interval", 3)), 2),
-    }
+    reg = {key: str(begin_data.get(key, "")).strip() for key in ("device_code", "verification_uri_complete")}
+    for key, value in reg.items():
+        if not value:
+            raise RegistrationError(f"begin response missing {key}")
+    reg["expires_in"] = int(begin_data.get("expires_in", 7200))
+    reg["interval"] = max(int(begin_data.get("interval", 3)), 2)
+    return reg
 
 
 def poll_registration(device_code: str) -> dict:
@@ -64,16 +55,12 @@ def poll_registration(device_code: str) -> dict:
     data = _api_post("/app/registration/poll", {"device_code": device_code})
     status_raw = str(data.get("status", "")).strip().upper()
     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
+    result.update({key: str(data.get(key, "")).strip() or None for key in ("client_id", "client_secret", "fail_reason")})
     return result
 
 
 def wait_for_registration_success(
-    device_code: str,
-    interval: int = 3,
-    expires_in: int = 7200,
-    on_waiting: Optional[callable] = None,
+    device_code: str, interval: int = 3, expires_in: int = 7200, on_waiting: Optional[callable] = None,
 ) -> Tuple[str, str]:
     """Block until the registration succeeds or times out.
 
@@ -87,8 +74,8 @@ def wait_for_registration_success(
         nonlocal retry_start
         if retry_start == 0:
             retry_start = time.monotonic()
-        return time.monotonic() - retry_start < _RETRY_WINDOW
 
+        return time.monotonic() - retry_start < _RETRY_WINDOW
     while time.monotonic() < deadline:
         time.sleep(interval)
         try:
@@ -97,7 +84,6 @@ def wait_for_registration_success(
             if _within_retry_window():
                 continue
             raise
-
         status = result["status"]
         if status == "WAITING":
             retry_start = 0
@@ -112,7 +98,6 @@ def wait_for_registration_success(
         if _within_retry_window():
             continue
         raise RegistrationError(f"authorization failed: {result.get('fail_reason') or status}")
-
     raise RegistrationError("authorization timed out, please retry")
 
 
@@ -123,11 +108,8 @@ def _ensure_qrcode_installed() -> bool:
         return True
     except ImportError:
         pass
-
     import subprocess
-
     from hermes_cli.tools_config import _pip_install
-
     try:
         if _pip_install(["-q", "qrcode"], timeout=120).returncode == 0:
             import qrcode  # noqa: F401,F811
@@ -143,20 +125,15 @@ def render_qr_to_terminal(url: str) -> bool:
         import qrcode
     except ImportError:
         return False
-
     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)
-
     matrix = qr.get_matrix()
-    rows = len(matrix)
-    # (top, bottom) -> █ ▀ ▄ or space
-    glyph = {(True, True): "\u2588", (True, False): "\u2580", (False, True): "\u2584", (False, False): " "}
+    glyph = {(True, True): "\u2588", (True, False): "\u2580", (False, True): "\u2584", (False, False): " "}  # █ ▀ ▄ space
     lines = []
-    for r in range(0, rows, 2):
-        bottom_row = matrix[r + 1] if r + 1 < rows else [False] * len(matrix[r])
+    for r in range(0, len(matrix), 2):
+        bottom_row = matrix[r + 1] if r + 1 < len(matrix) else [False] * len(matrix[r])
         lines.append("    " + "".join(glyph[(bool(top), bool(bottom))] for top, bottom in zip(matrix[r], bottom_row)))
-
     print("\n".join(lines))
     return True
 
@@ -164,35 +141,27 @@ def render_qr_to_terminal(url: str) -> bool:
 def dingtalk_qr_auth() -> Optional[Tuple[str, str]]:
     """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()
     print_info("  Initializing DingTalk device authorization...")
     print_info("  Note: the scan page is branded 'OpenClaw' — DingTalk's")
     print_info("        ecosystem onboarding bridge. Safe to use.")
-
     try:
         reg = begin_registration()
     except RegistrationError as exc:
         print_error(f"  Authorization init failed: {exc}")
         return None
-
     url = reg["verification_uri_complete"]
-
     if not _ensure_qrcode_installed():
         print_warning("  qrcode library install failed, will show link only.")
-
     print()
     print_info("  Please scan the QR code below with DingTalk to authorize:")
     print()
-
     if not render_qr_to_terminal(url):
         print_warning("  QR code render failed, please open the link below to authorize:")
-
     print()
     print_info(f"  Or open this link manually: {url}")
     print()
     print_info("  Waiting for QR scan authorization... (timeout: 2 hours)")
-
     dot_count = 0
 
     def _on_waiting():
@@ -200,8 +169,8 @@ def dingtalk_qr_auth() -> Optional[Tuple[str, str]]:
         dot_count += 1
         if dot_count % 10 == 0:
             sys.stdout.write(".")
-            sys.stdout.flush()
 
+            sys.stdout.flush()
     try:
         client_id, client_secret = wait_for_registration_success(
             device_code=reg["device_code"], interval=reg["interval"], expires_in=reg["expires_in"],
@@ -211,10 +180,8 @@ def dingtalk_qr_auth() -> Optional[Tuple[str, str]]:
         print()
         print_error(f"  Authorization failed: {exc}")
         return None
-
     print()
     print_success("  QR scan authorization successful!")
     print_success(f"  Client ID:     {client_id}")
     print_success(f"  Client Secret: {client_secret[:8]}{'*' * (len(client_secret) - 8)}")
-
     return client_id, client_secret

hermes_cli/fallback_cmd.py

@@ -16,7 +16,6 @@ _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"))
 
 
@@ -37,15 +36,11 @@ def _extract_fallback_from_model_cfg(model_cfg: Any) -> Optional[Dict[str, Any]]
     if not isinstance(model_cfg, dict):
         return None
     provider = (model_cfg.get("provider") or "").strip()
-    # The picker writes the selected model to ``model.default``.
-    model = (model_cfg.get("default") or model_cfg.get("model") or "").strip()
+    model = (model_cfg.get("default") or model_cfg.get("model") or "").strip()  # the picker writes ``model.default``
     if not provider or not model:
         return None
     entry: Dict[str, Any] = {"provider": provider, "model": model}
-    for key in ("base_url", "api_mode"):
-        value = (model_cfg.get(key) or "").strip()
-        if value:
-            entry[key] = value
+    entry.update({key: value for key in ("base_url", "api_mode") if (value := (model_cfg.get(key) or "").strip())})
     return entry
 
 
@@ -59,16 +54,11 @@ def _snapshot_auth_active_provider() -> Any:
 
 
 def _restore_auth_active_provider(value: Any) -> None:
-    """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.
-    """
+    """Write back a snapshotted ``active_provider``; best-effort (user re-runs `hermes model`), never fails the add."""
     try:
         from hermes_cli.auth import _auth_store_lock, _load_auth_store, _save_auth_store
         with _auth_store_lock():
-            store = _load_auth_store()
-            store["active_provider"] = value
-            _save_auth_store(store)
+            _save_auth_store({**_load_auth_store(), "active_provider": value})
     except Exception:
         pass
 
@@ -76,11 +66,9 @@ def _restore_auth_active_provider(value: Any) -> None:
 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.pop("model", None)
+    if model_before is not None:
         cfg["model"] = copy.deepcopy(model_before)
     save_config(cfg)
 
@@ -91,21 +79,17 @@ def _entries(n: int) -> str:
 
 def _print_chain(heading: str, chain: List[Dict[str, Any]]) -> None:
     print(f"  {heading} ({_entries(len(chain))}):")
-    for i, entry in enumerate(chain, 1):
-        print(f"    {i}. {_format_entry(entry)}")
-    print()
+    print("".join(f"    {i}. {_format_entry(entry)}\n" for i, entry in enumerate(chain, 1)))
 
 
 def _load_chain(empty_message: str):
     """Load config + chain; print ``empty_message`` block and return ``(config, None)`` when empty."""
     from hermes_cli.config import load_config
-
     config = load_config()
     chain = _read_chain(config)
     if not chain:
         print(f"\n{empty_message}\n")
-        return config, None
-    return config, chain
+    return config, chain or None
 
 
 def _describe_primary(config: Dict[str, Any]) -> Optional[str]:
@@ -115,9 +99,7 @@ def _describe_primary(config: Dict[str, Any]) -> Optional[str]:
         provider = (model_cfg.get("provider") or "?").strip() or "?"
         model = (model_cfg.get("default") or model_cfg.get("model") or "?").strip() or "?"
         return f"{model}  (via {provider})"
-    if isinstance(model_cfg, str) and model_cfg.strip():
-        return model_cfg.strip()
-    return None
+    return model_cfg.strip() or None if isinstance(model_cfg, str) else None
 
 
 def cmd_fallback_list(args) -> None:  # noqa: ARG001
@@ -126,10 +108,8 @@ def cmd_fallback_list(args) -> None:  # noqa: ARG001
     if chain is None:
         print("  Add one with:  hermes fallback add\n")
         return
-
     print()
-    primary = _describe_primary(config)
-    if primary:
+    if primary := _describe_primary(config):
         print(f"  Primary:   {primary}\n")
     _print_chain("Fallback chain", chain)
     print("  Tried in order when the primary fails (rate-limit, 5xx, connection errors).")
@@ -140,27 +120,24 @@ 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
     from hermes_cli.config import load_config, save_config
-
     _require_tty("fallback add")
 
     # 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("\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)
-        _restore_auth_active_provider(active_provider_before)
 
+        _restore_auth_active_provider(active_provider_before)
     try:
         select_provider_and_model(args=args)
     except SystemExit:  # some provider flows exit on auth failure — restore state and re-raise
         _restore()
         raise
-
     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()
@@ -171,7 +148,6 @@ def cmd_fallback_add(args) -> None:
     # 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):
@@ -188,7 +164,6 @@ def cmd_fallback_add(args) -> None:
     if any(same_deployment(_identity(existing), new_ident) for existing in chain):
         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)
@@ -200,19 +175,16 @@ def cmd_fallback_add(args) -> None:
 def cmd_fallback_remove(args) -> None:  # noqa: ARG001
     """Pick an entry from the chain and remove it."""
     from hermes_cli.config import save_config
-
     config, chain = _load_chain("  No fallback providers configured — nothing to remove.")
     if chain is None:
         return
 
     # 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("\n  Cancelled — no change.")
         return
-
     removed = chain.pop(idx)
     _write_chain(config, chain)
     save_config(config)
@@ -223,11 +195,9 @@ def cmd_fallback_remove(args) -> None:  # noqa: ARG001
 def cmd_fallback_clear(args) -> None:  # noqa: ARG001
     """Remove all fallback entries (with confirmation)."""
     from hermes_cli.config import save_config
-
     config, chain = _load_chain("  No fallback providers configured — nothing to clear.")
     if chain is None:
         return
-
     print()
     _print_chain("Current fallback chain", chain)
     try:
@@ -238,7 +208,6 @@ def cmd_fallback_clear(args) -> None:  # noqa: ARG001
     if resp not in {"y", "yes"}:
         print("  Cancelled — no change.")
         return
-
     _write_chain(config, [])
     save_config(config)
     print("\n  Fallback chain cleared.\n")
@@ -256,8 +225,6 @@ def cmd_fallback(args) -> None:
 
 
 _SUBCOMMANDS = {
-    None: cmd_fallback_list, "": cmd_fallback_list, "list": cmd_fallback_list, "ls": cmd_fallback_list,
-    "add": cmd_fallback_add,
-    "remove": cmd_fallback_remove, "rm": cmd_fallback_remove,
-    "clear": cmd_fallback_clear,
+    **dict.fromkeys((None, "", "list", "ls"), cmd_fallback_list), "add": cmd_fallback_add,
+    **dict.fromkeys(("remove", "rm"), cmd_fallback_remove), "clear": cmd_fallback_clear,
 }

hermes_cli/fallback_config.py

@@ -20,13 +20,10 @@ def resolve_entry_api_key(entry: dict[str, Any] | None) -> str | None:
     """
     if not isinstance(entry, dict):
         return None
-    inline = str(entry.get("api_key") or "").strip()
-    if inline:
+    if inline := str(entry.get("api_key") or "").strip():
         return inline
-    key_env = str(entry.get("key_env") or entry.get("api_key_env") or "").strip()
-    if key_env:
+    if key_env := str(entry.get("key_env") or entry.get("api_key_env") or "").strip():
         from agent.secret_scope import get_secret
-
         return (get_secret(key_env) or "").strip() or None
     return None
 

hermes_cli/focus_view.py

@@ -1,31 +1,22 @@
-"""Focus view — a display-only reduced-output mode.
+"""Focus view — display-only reduced output: "just my prompt and the answer, and tell me what you hid".
 
-``/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.
+ON snaps ``tool_progress_mode`` to ``"off"`` and remembers the configured mode so the *existing* suppression
+path does the hiding; OFF restores it verbatim. 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).
-FOCUS_CONFIG_KEY = "display.focus_view"
-
-#: Tool-progress mode focus view snaps to — the SAME value ``/verbose off`` uses so both share one
-#: suppression path.
+FOCUS_CONFIG_KEY = "display.focus_view"  # plain boolean under ``display``, like /battery /timestamps /footer
+#: The SAME value ``/verbose off`` uses so both features 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`` 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).
-TOOL_PROGRESS_MODES = ("off", "new", "all", "verbose")
-
-#: Status-bar label. Short on purpose — the bar is width-constrained.
-FOCUS_STATUSBAR_LABEL = "◉ focus"
+TOOL_PROGRESS_MODES = ("off", "new", "all", "verbose")  # ``log`` is a gateway-only extra step
+FOCUS_STATUSBAR_LABEL = "◉ focus"  # short on purpose — the bar is width-constrained
 
 # /focus argument words -> (action, target); bare /focus toggles like /footer, /battery, /timestamps.
 _FOCUS_WORDS = {
@@ -76,9 +67,7 @@ def format_hidden_line(count: int) -> Optional[str]:
         n = int(count)
     except (TypeError, ValueError):
         return None
-    if n <= 0:
-        return None
-    return f"⋯ {n} {'tool line' if n == 1 else 'tool lines'} hidden · /focus off to show"
+    return f"⋯ {n} {'tool line' if n == 1 else 'tool lines'} hidden · /focus off to show" if n > 0 else None
 
 
 def focus_statusbar_segment(enabled: bool) -> str:

hermes_cli/foreign_sessions.py

@@ -1,11 +1,9 @@
-"""Import sessions from foreign coding agents (Claude Code, Codex CLI).
-
-Foreign files are only ever read. Imported history must satisfy the provider role-alternation
-invariant Hermes enforces everywhere else — see ``_merge_turns``.
-"""
+"""Import sessions from foreign coding agents (Claude Code, Codex CLI). Foreign files are only ever read;
+imported history must satisfy the provider role-alternation invariant (see ``_merge_turns``)."""
 
 from __future__ import annotations
 
+import contextlib
 import json
 import os
 import re
@@ -78,8 +76,7 @@ def _flatten_blocks(content: Any) -> str:
         # 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:
+            if isinstance(text := block.get("text"), str) and text:
                 parts.append(text)
         elif btype == "tool_use":  # Claude Code assistant block
             parts.append(f"[ran tool: {block.get('name') or 'tool'}]")
@@ -96,8 +93,7 @@ def _merge_turns(raw_turns: List[Tuple[str, str]]) -> List[Dict[str, str]]:
     """
     merged: List[Dict[str, str]] = []
     for role, text in raw_turns:
-        text = text.strip()
-        if not text:
+        if not (text := text.strip()):
             continue
         if merged and merged[-1]["role"] == role:
             merged[-1]["content"] += "\n\n" + text
@@ -113,28 +109,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 _WRAPPER_TAG_RE.match(text.lstrip())):
-        return None
-    return (role, text)
+    return None if not text or (role == "user" and _WRAPPER_TAG_RE.match(text.lstrip())) else (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]
+        if role == "user" and (line := text.strip().splitlines()[0].strip()):
+            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 {
-        "turns": _merge_turns(turns),
-        "cwd": cwd,
-        "title_guess": title or _first_user_line(turns),
-        "session_id": session_id,
-    }
+    return {"turns": _merge_turns(turns), "cwd": cwd, "title_guess": title or _first_user_line(turns),
+            "session_id": session_id}
 
 
 def parse_claude_session(path: Path) -> Dict[str, Any]:
@@ -155,10 +143,7 @@ def parse_claude_session(path: Path) -> Dict[str, Any]:
         if session_id is None and isinstance(obj.get("sessionId"), str):
             session_id = obj["sessionId"]
         message = obj.get("message")
-        if not isinstance(message, dict):
-            continue
-        turn = _message_turn(message.get("role"), message.get("content"))
-        if turn:
+        if isinstance(message, dict) and (turn := _message_turn(message.get("role"), message.get("content"))):
             turns.append(turn)
     return _parsed(turns, cwd, session_id, summary)
 
@@ -183,8 +168,7 @@ def parse_codex_session(path: Path) -> Dict[str, Any]:
             continue
         ptype = payload.get("type")
         if ptype == "message":  # developer/system payloads never imported
-            turn = _message_turn(payload.get("role"), payload.get("content"))
-            if turn:
+            if turn := _message_turn(payload.get("role"), payload.get("content")):
                 turns.append(turn)
         elif ptype in ("custom_tool_call", "function_call", "local_shell_call"):
             # Assistant activity; merged into neighbours later. Tool outputs / reasoning skipped.
@@ -204,9 +188,7 @@ def _list_sessions(source: str, root: Optional[Path]) -> List[ForeignSession]:
     default_root, pattern, recursive, parse = _SOURCES[source]
     root = Path(root) if root else Path.home().joinpath(*default_root)
     results: List[ForeignSession] = []
-    if not root.is_dir():
-        return results
-    for jsonl in sorted(root.rglob(pattern) if recursive else root.glob(pattern)):
+    for jsonl in sorted((root.rglob(pattern) if recursive else root.glob(pattern)) if root.is_dir() else ()):
         try:
             mtime = jsonl.stat().st_mtime
         except OSError:
@@ -230,7 +212,6 @@ def import_foreign_session(source: str, path, db=None) -> str:
     path = Path(path).expanduser()
     if not path.is_file():
         raise ValueError(f"Session file not found: {path}")
-
     parsed = _SOURCES[source][3](path)
     turns = parsed["turns"]
     if not turns:
@@ -238,13 +219,10 @@ def import_foreign_session(source: str, path, db=None) -> str:
     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:
         from hermes_state import SessionDB
-
         db = SessionDB()
     try:
         session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"
@@ -252,17 +230,13 @@ def import_foreign_session(source: str, path, db=None) -> str:
         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:
-            db.set_session_title(session_id, title)
-        except Exception:
-            pass  # title is cosmetic; the import itself succeeded
+        with contextlib.suppress(Exception):  # title is cosmetic; the import itself succeeded
+            db.set_session_title(session_id, f"Imported from {_SOURCE_LABELS[source]}: {first_user}")
         return session_id
     finally:
         if owns_db:
-            try:
+            with contextlib.suppress(Exception):
                 db.close()
-            except Exception:
-                pass
 
 
 def gather_foreign_sessions(
@@ -270,10 +244,8 @@ def gather_foreign_sessions(
     limit: int = 25,
 ) -> List[ForeignSession]:
     """List foreign sessions across sources, newest first."""
-    sessions: List[ForeignSession] = []
-    for name, root in (("claude", claude_root), ("codex", codex_root)):
-        if source in (None, name):
-            sessions.extend(_list_sessions(name, root))
+    sessions = [s for name, root in (("claude", claude_root), ("codex", codex_root)) if source in (None, name)
+                for s in _list_sessions(name, root)]
     sessions.sort(key=lambda s: s.mtime, reverse=True)
     return sessions[:limit] if limit else sessions
 
@@ -287,28 +259,26 @@ def pick_foreign_session(source: Optional[str] = None, *, limit: int = 25) -> Op
         return None
     print("Foreign sessions (newest first):")
     for i, s in enumerate(sessions, 1):
-        when = datetime.fromtimestamp(s.mtime).strftime("%Y-%m-%d %H:%M")
         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]")
+        print(f"  {i:>2}. {datetime.fromtimestamp(s.mtime):%Y-%m-%d %H:%M}  {s.label}{ws}  [{s.turn_count} turns]")
     if not sys.stdin.isatty():
         print("Non-interactive terminal — pass the file path directly:\n"
               "  hermes sessions import --from claude|codex <path>")
         return None
     try:
         raw = input(f"Import which session? [1-{len(sessions)}, empty to cancel] ").strip()
+        idx = int(raw) if raw else None
     except (EOFError, KeyboardInterrupt):
         return None
-    if not raw:
-        return None
-    try:
-        idx = int(raw)
     except ValueError:
         print(f"Not a number: {raw}")
         return None
-    if not 1 <= idx <= len(sessions):
-        print(f"Out of range: {idx}")
+    if idx is None:
         return None
-    return sessions[idx - 1]
+    if 1 <= idx <= len(sessions):
+        return sessions[idx - 1]
+    print(f"Out of range: {idx}")
+    return None
 
 
 def run_sessions_import(args, db=None) -> Optional[str]:
@@ -320,7 +290,7 @@ def run_sessions_import(args, db=None) -> Optional[str]:
         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; a codex match wins over a claude match
             p = str(path)
             if "/.claude/" in p or p.endswith(".jsonl") and "claude" in p:
                 source = "claude"

hermes_cli/gitlock.py

@@ -1,8 +1,7 @@
 """Stale git lock-file and aborted-fetch pack-debris recovery for update/check paths.
 
-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.
+A killed ``git fetch`` can leave ``.git/shallow.lock`` behind (every later fetch fails with "Unable to
+create '.../shallow.lock': File exists") and ``tmp_pack_*`` files git itself never cleans up.
 """
 
 from __future__ import annotations
@@ -16,18 +15,15 @@ from typing import Callable, Iterable, List, Optional
 
 logger = logging.getLogger(__name__)
 
-# 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.
+# Files younger than this are presumed live (a fetch may be in flight) and are never removed. Lock
+# files live for seconds and a healthy fetch completes in minutes; 10 minutes is abandoned.
 STALE_LOCK_MIN_AGE_SECONDS = 10 * 60
 STALE_TMP_PACK_MIN_AGE_SECONDS = STALE_LOCK_MIN_AGE_SECONDS
-
 # ``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.
+# 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_")
 
 
@@ -51,13 +47,8 @@ def _git_proc_running() -> bool:
 
 
 def _sweep_stale(
-    directory: Path,
-    candidates: Callable[[], Iterable[Path]],
-    *,
-    min_age_seconds: Optional[int],
-    default_age: int,
-    skip_msg: str,
-    log_removed: Callable[[Path, int], None],
+    directory: Path, candidates: Callable[[], Iterable[Path]], *, min_age_seconds: Optional[int], default_age: int,
+    skip_msg: str, log_removed: Callable[[Path, int], None],
 ) -> List[str]:
     """Shared guard + age-floor sweep. Never raises; skips anything it cannot stat/unlink."""
     if not directory.is_dir():
@@ -89,10 +80,8 @@ def clear_stale_git_locks(repo_root: Path, *, min_age_seconds: Optional[int] = N
     """
     git_dir = Path(repo_root) / ".git"
     return _sweep_stale(
-        git_dir,
-        lambda: [git_dir / name for name in LOCK_NAMES],
-        min_age_seconds=min_age_seconds,
-        default_age=STALE_LOCK_MIN_AGE_SECONDS,
+        git_dir, lambda: [git_dir / name for name in LOCK_NAMES],
+        min_age_seconds=min_age_seconds, default_age=STALE_LOCK_MIN_AGE_SECONDS,
         skip_msg="git process running; skipping stale-lock sweep",
         log_removed=lambda p, _size: logger.info("Removed stale git lock %s", p),
     )
@@ -107,16 +96,13 @@ def clear_stale_tmp_packs(repo_root: Path, *, min_age_seconds: Optional[int] = N
 
     def _candidates():
         try:
-            entries = list(pack_dir.iterdir())
+            return [e for e in pack_dir.iterdir() if e.name.startswith(_TMP_PACK_PREFIXES)]
         except OSError:
-            return []
-        return [e for e in entries if e.name.startswith(_TMP_PACK_PREFIXES)]
 
+            return []
     return _sweep_stale(
-        pack_dir,
-        _candidates,
-        min_age_seconds=min_age_seconds,
-        default_age=STALE_TMP_PACK_MIN_AGE_SECONDS,
+        pack_dir, _candidates,
+        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),
     )

hermes_cli/gui_uninstall.py

@@ -1,9 +1,5 @@
-"""Hermes Desktop (Chat GUI) uninstaller.
-
-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``.
-"""
+"""Hermes Desktop (Chat GUI) uninstaller: removes only GUI state — built Electron artifacts, the packaged
+app, and the desktop's own ``userData`` — never agent source, venv, config, sessions or .env."""
 
 import os
 import shutil
@@ -53,13 +49,8 @@ def source_built_gui_artifacts(hermes_home: Path) -> "list[Path]":
     """
     agent_root = _agent_root(hermes_home)
     desktop_dir = agent_root / "apps" / "desktop"
-    return [
-        desktop_dir / "dist",
-        desktop_dir / "release",
-        desktop_dir / "node_modules",
-        agent_root / "node_modules",
-        hermes_home / "desktop-build-stamp.json",
-    ]
+    return [desktop_dir / "dist", desktop_dir / "release", desktop_dir / "node_modules",
+            agent_root / "node_modules", hermes_home / "desktop-build-stamp.json"]
 
 
 def packaged_gui_app_paths() -> "list[Path]":
@@ -69,34 +60,26 @@ def packaged_gui_app_paths() -> "list[Path]":
     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"]
-    elif sys.platform == "win32":
+        return [Path("/Applications/Hermes.app"), home / "Applications" / "Hermes.app"]
+    if sys.platform == "win32":
         local_base = _env_dir("LOCALAPPDATA", home / "AppData" / "Local")
-        paths += [
-            local_base / "Programs" / "Hermes",  # NSIS per-user install (perMachine=false)
-            local_base / "hermes-desktop",  # older / alternate layout some builds used
-        ]
+        # NSIS per-user install (perMachine=false), an older/alternate layout, NSIS per-machine (needs admin).
+        paths = [local_base / "Programs" / "Hermes", local_base / "hermes-desktop"]
         program_files = os.environ.get("ProgramFiles")
         if program_files:
-            paths.append(Path(program_files) / "Hermes")  # NSIS per-machine fallback (needs admin)
-    else:
-        # 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 += [
-            desktop_entry_path(),
-            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 (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
+            paths.append(Path(program_files) / "Hermes")
+        return paths
+    # 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 are cleaned here.
+    from hermes_cli.linux_desktop_entry import desktop_entry_path
+    data_base = _env_dir("XDG_DATA_HOME", home / ".local" / "share")
+    icons = data_base / "icons" / "hicolor"
+    # "scalable" plus every fixed-size dir the installer may have written (panel sizes + older native copies).
+    return [desktop_entry_path(), data_base / "applications" / "Hermes.desktop"] + [
+        icons / size / "apps" / "hermes.png"
+        for size in ("scalable", "24x24", "32x32", "48x48", "256x256", "512x512", "1024x1024")
+    ]
 
 
 def agent_is_installed(hermes_home: Path) -> bool:
@@ -104,16 +87,12 @@ def agent_is_installed(hermes_home: Path) -> bool:
 
     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)
-    return any((agent_root / sub).is_dir() for sub in ("hermes_cli", "venv", ".venv"))
+    return any((_agent_root(hermes_home) / sub).is_dir() for sub in ("hermes_cli", "venv", ".venv"))
 
 
 def gui_is_installed(hermes_home: Path) -> bool:
     """Return True when any desktop GUI artifact exists (built or packaged)."""
-    return any(
-        p.exists()
-        for p in (*source_built_gui_artifacts(hermes_home), *packaged_gui_app_paths(), desktop_userdata_dir())
-    )
+    return any(p.exists() for p in (*source_built_gui_artifacts(hermes_home), *packaged_gui_app_paths(), desktop_userdata_dir()))
 
 
 def gui_install_summary(hermes_home: "Path | None" = None) -> dict:
@@ -137,13 +116,14 @@ def _remove_path(path: Path) -> bool:
     try:
         if path.is_symlink() or path.is_file():
             path.unlink()
-            return True
-        if path.is_dir():
+        elif path.is_dir():
             shutil.rmtree(path)
-            return True
+        else:
+            return False
+        return True
     except Exception as e:
         log_warn(f"Could not remove {path}: {e}")
-    return False
+        return False
 
 
 def uninstall_gui(hermes_home: "Path | None" = None, *, remove_userdata: bool = True) -> "list[Path]":
@@ -160,43 +140,35 @@ def uninstall_gui(hermes_home: "Path | None" = None, *, remove_userdata: bool =
                 if _remove_path(path):
                     log_success(f"Removed {path}")
                     removed.append(path)
-        return found
 
+        return found
     log_info("Removing built GUI artifacts (renderer, release, node_modules)...")
     _remove_existing(source_built_gui_artifacts(home))
-
     log_info("Removing installed desktop app...")
     if not _remove_existing(packaged_gui_app_paths()):
         log_info("No packaged desktop app found in standard locations")
-
     if remove_userdata:
         userdata = desktop_userdata_dir()
         if userdata.exists():
             log_info("Removing desktop app data (Electron userData)...")
             _remove_existing([userdata])
-
     if not removed:
         log_info("No desktop GUI artifacts found to remove")
-
     if sys.platform.startswith("linux"):
         # 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
-
             entry = desktop_entry_path()
             if entry in removed:
                 for tool in refresh_desktop_databases(entry.parent):
                     log_success(f"Refreshed the application menu cache ({tool})")
         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 "
             "'sudo dnf remove hermes'). AppImage builds are a single file you "
             "can delete from wherever you saved it."
         )
-
     return removed

hermes_cli/heartbeat.py

@@ -1,11 +1,9 @@
 """Session heartbeats — recurring re-entry prompts for the current session.
 
-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, so prompt caching stays intact. A real user message always wins: heartbeats only fire
-into an idle session with an empty input queue.
+Deliberately session-scoped and in-process (CLI or gateway must be running); durable cross-process
+scheduling remains ``hermes cron``. Invariants (mirrors goals.py): injection is a plain user message —
+no system-prompt mutation or toolset swap, so prompt caching stays intact — and a real user message
+always wins: heartbeats only fire into an idle session with an empty input queue.
 """
 
 from __future__ import annotations
@@ -19,10 +17,8 @@ from typing import Any, Optional
 
 logger = logging.getLogger(__name__)
 
-# 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
+MIN_INTERVAL_SECONDS = 60  # floor: re-entering more often than once a minute is a busy-loop, not a heartbeat
+POLL_SECONDS = 5.0  # how often drivers poll for due heartbeats; not user-facing
 
 HEARTBEAT_PROMPT_TEMPLATE = (
     "[Heartbeat — recurring instruction, fires every {interval}]\n"
@@ -33,25 +29,20 @@ HEARTBEAT_PROMPT_TEMPLATE = (
 )
 
 _INTERVAL_RE = re.compile(
-    r"^\s*(?:every\s+)?(\d+(?:\.\d+)?)\s*(s|sec|secs|seconds?|m|min|mins|minutes?|h|hr|hrs|hours?|d|days?)\s*$",
-    re.IGNORECASE,
+    r"^\s*(?:every\s+)?(\d+(?:\.\d+)?)\s*(s|sec|secs|seconds?|m|min|mins|minutes?|h|hr|hrs|hours?|d|days?)\s*$", re.IGNORECASE
 )
 
 _UNIT_SECONDS = {
-    "s": 1, "sec": 1, "secs": 1, "second": 1, "seconds": 1,
-    "m": 60, "min": 60, "mins": 60, "minute": 60, "minutes": 60,
-    "h": 3600, "hr": 3600, "hrs": 3600, "hour": 3600, "hours": 3600,
-    "d": 86400, "day": 86400, "days": 86400,
+    **dict.fromkeys(("s", "sec", "secs", "second", "seconds"), 1),
+    **dict.fromkeys(("m", "min", "mins", "minute", "minutes"), 60),
+    **dict.fromkeys(("h", "hr", "hrs", "hour", "hours"), 3600),
+    **dict.fromkeys(("d", "day", "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),
+    "prompt": (str, ""), "interval_seconds": (int, 0), "status": (str, "active"),
+    "created_at": (float, 0.0), "last_fired_at": (float, 0.0), "fire_count": (int, 0),
 }
 
 
@@ -71,10 +62,8 @@ def parse_interval(text: str) -> Optional[int]:
 def format_interval(seconds: int) -> str:
     """Human-readable interval (``600`` → ``10m``)."""
     seconds = int(seconds)
-    for unit, suffix in ((86400, "d"), (3600, "h"), (60, "m")):
-        if seconds % unit == 0:
-            return f"{seconds // unit}{suffix}"
-    return f"{seconds}s"
+    units = ((86400, "d"), (3600, "h"), (60, "m"))
+    return next((f"{seconds // unit}{suffix}" for unit, suffix in units if seconds % unit == 0), f"{seconds}s")
 
 
 @dataclass
@@ -99,8 +88,7 @@ class HeartbeatState:
     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()
-        return (now - (self.last_fired_at or self.created_at)) >= self.interval_seconds
+        return ((time.time() if now is None else 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)
@@ -110,7 +98,6 @@ def _get_session_db() -> Optional[Any]:
     """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
-
         return _goals_db()
     except Exception as exc:  # pragma: no cover
         logger.debug("HeartbeatManager: SessionDB bootstrap failed (%s)", exc)
@@ -142,7 +129,6 @@ def save_heartbeat(session_id: str, state: HeartbeatState) -> None:
     db = _get_session_db()
     if db is None:
         from hermes_cli.goals import _warn_dropped_write
-
         _warn_dropped_write("HeartbeatManager", "heartbeat", session_id)
         return
     try:

hermes_cli/image_provenance.py

@@ -1,10 +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, 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.
+The image bakes ``/etc/hermes/image-provenance.json`` outside ``$HERMES_HOME`` and the checkout, so a
+bind-mounted checkout cannot hide the build fact and env/config cannot forge it. Absence preserves every
+source/package install path. Presence fails closed: an unreadable, non-regular, or malformed marker still
+means image-managed — an integrity defect, never permission to mutate the image in place.
 """
 
 from __future__ import annotations
@@ -26,10 +25,10 @@ class ImageProvenance:
     schema: int
     deployment_kind: str
     manager: str
-    image: Optional[str]
-    version: Optional[str]
-    revision: Optional[str]
-    marker_path: str
+    image: Optional[str] = None
+    version: Optional[str] = None
+    revision: Optional[str] = None
+    marker_path: str = ""
     valid: bool = True
     error: Optional[str] = None
 
@@ -38,19 +37,14 @@ 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,
-    )
+    return ImageProvenance(IMAGE_PROVENANCE_SCHEMA, "image", "unknown", marker_path=str(path), valid=False, error=reason)
 
 
 def _optional_string(payload: dict, name: str) -> Optional[str]:
     value = payload.get(name)
-    if value is None:
-        return None
-    if not isinstance(value, str):
+    if value is not None and not isinstance(value, str):
         raise TypeError(name)
-    return value.strip() or None
+    return (value.strip() or None) if value is not None else None
 
 
 def read_image_provenance(marker_path: Optional[Path] = None) -> Optional[ImageProvenance]:
@@ -63,42 +57,31 @@ def read_image_provenance(marker_path: Optional[Path] = None) -> Optional[ImageP
         path = Path(marker_path) if marker_path is not None else path
     except BaseException as exc:
         return _invalid(path, f"marker_presence_unreadable:{type(exc).__name__}")
-
     try:
         marker_stat = path.lstat()
     except FileNotFoundError:
         return None
     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):
         return _invalid(path, "marker_not_regular_file")
-
     try:
         payload = json.loads(path.read_text(encoding="utf-8"))
     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`` 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":
         return _invalid(path, "invalid_deployment_kind")
-
     manager = payload.get("manager")
     if not isinstance(manager, str) or not manager.strip():
         return _invalid(path, "missing_manager")
-
     try:
         optional = {name: _optional_string(payload, name) for name in ("image", "version", "revision")}
     except TypeError as exc:
         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,
-    )
+    return ImageProvenance(IMAGE_PROVENANCE_SCHEMA, "image", manager.strip(), marker_path=str(path), **optional)

hermes_cli/init_command.py

@@ -31,15 +31,11 @@ def build_init_prompt(cwd: str, existing_file: str | None = None, extra: 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()
-
+    update = existing_file is not None
     parts: list[str] = [
         "[/init] The user wants you to "
-        + (
-            "UPDATE the existing AGENTS.md project-instructions file"
-            if existing_file is not None
-            else "generate an AGENTS.md project-instructions file"
-        )
-        + f" for the project at: {cwd}\n",
+        + ("UPDATE the existing" if update else "generate an")
+        + f" AGENTS.md project-instructions file for the project at: {cwd}\n",
         "AGENTS.md is the instruction file coding agents (Hermes included) "
         "load as project context every session. It should teach an agent how "
         "to work in THIS repo: what the project is, how to set up, the exact "
@@ -54,17 +50,12 @@ def build_init_prompt(cwd: str, existing_file: str | None = None, extra: str = "
         "don't guess them.\n"
         "2. Write the file to "
         f"{cwd.rstrip('/')}/AGENTS.md with `write_file`"
-        + (
-            " — but this is an UPDATE, so follow the merge discipline below."
-            if existing_file is not None
-            else "."
-        )
+        + (" — but this is an UPDATE, so follow the merge discipline below." if update else ".")
         + "\n"
         "3. Confirm to the user the exact path you wrote and summarize in one "
         "or two lines what the file covers.\n",
     ]
-
-    if existing_file is not None:
+    if update:
         parts.append(
             "MERGE DISCIPLINE — an AGENTS.md already exists (its current "
             "content is below). Do NOT overwrite or regenerate it from "
@@ -79,15 +70,12 @@ def build_init_prompt(cwd: str, existing_file: str | None = None, extra: str = "
             f"{existing_file}\n"
             "EXISTING_AGENTS_MD\n"
         )
-
     parts.append(_QUALITY_BAR)
-
     if extra:
         parts.append(
             "\nUSER NOTES — honor these while authoring (they override the "
             f"defaults above where they conflict):\n{extra}"
         )
-
     return "\n".join(parts)
 
 

hermes_cli/input_sanitize.py

@@ -42,12 +42,11 @@ def collapse_repeated_input_artifacts(text: str, min_repeats: int = 4) -> str:
         index -= len(marker)
     if repeat_count < min_repeats:
         return text
-    start = index
-    if start >= 2 and text[start - 2 : start] == "[e":
-        start -= 2
-    elif start >= 1 and text[start - 1] == "[":
-        start -= 1
-    return text[:start]
+    if index >= 2 and text[index - 2 : index] == "[e":
+        index -= 2
+    elif index >= 1 and text[index - 1] == "[":
+        index -= 1
+    return text[:index]
 
 
 def sanitize_user_prompt_text(text: str) -> str:

hermes_cli/install_identity.py

@@ -16,8 +16,7 @@ from hermes_constants import get_default_hermes_root
 _INSTALL_ID_FILENAME = "install_id"
 _INSTALL_ID_RE = re.compile(r"^[0-9a-f]{32}$")
 _INSTALL_ID_CACHE: dict[str, Optional[str]] = {"root": None, "value": None}
-_INSTALL_ID_LOCK = threading.Lock()
-_INSTALL_ID_PUBLICATION_LOCK = threading.Lock()
+_INSTALL_ID_LOCK, _INSTALL_ID_PUBLICATION_LOCK = threading.Lock(), threading.Lock()
 
 
 @contextlib.contextmanager
@@ -28,7 +27,6 @@ def _install_id_file_lock(root: Path):
     try:
         if windows:
             import msvcrt
-
             if os.fstat(fd).st_size == 0:
                 os.write(fd, b"\0")
                 os.fsync(fd)
@@ -36,19 +34,16 @@ def _install_id_file_lock(root: Path):
             msvcrt.locking(fd, msvcrt.LK_LOCK, 1)
         else:
             import fcntl
-
             fcntl.flock(fd, fcntl.LOCK_EX)
         yield
     finally:
         try:
             if windows:
                 import msvcrt
-
                 os.lseek(fd, 0, os.SEEK_SET)
                 msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
             else:
                 import fcntl
-
                 fcntl.flock(fd, fcntl.LOCK_UN)
         finally:
             os.close(fd)
@@ -92,12 +87,10 @@ def read_or_create_install_id(root: Path | None = None) -> Optional[str]:
     existing, mint = _read_existing(path)
     if not mint:
         return existing
-
     try:
         root.mkdir(parents=True, exist_ok=True)
     except OSError:
         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
@@ -106,7 +99,6 @@ def read_or_create_install_id(root: Path | None = None) -> Optional[str]:
             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-")
             try:
@@ -120,7 +112,6 @@ def read_or_create_install_id(root: Path | None = None) -> Optional[str]:
                 with contextlib.suppress(OSError):
                     os.unlink(tmp_name)
                 raise
-
             committed = path.read_text(encoding="utf-8").strip().lower()
             return committed if _INSTALL_ID_RE.fullmatch(committed) else None
     except OSError:
@@ -135,8 +126,8 @@ def get_install_id(*, cache: dict[str, Optional[str]] | None = None) -> Optional
 
     def _cached() -> Optional[str]:
         cached = target_cache.get("value")
-        return cached if cached and target_cache.get("root") in (None, root_key) else None
 
+        return cached if cached and target_cache.get("root") in (None, root_key) else None
     if value := _cached():
         return value
     with _INSTALL_ID_LOCK: