From fc51206750bb0627fda92c558bb2087cc85528ec Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 21:20:42 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20second=20pass=20on=20?= =?UTF-8?q?the=2015=20small=20modules=20=E2=80=94=20compact=20bodies,=20dr?= =?UTF-8?q?op=20redundant=20locals/blanks,=20tighten=20module=20docs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/dep_ensure.py | 38 ++++----------- hermes_cli/diagnostics_upload.py | 20 ++------ hermes_cli/dingtalk_auth.py | 61 ++++++----------------- hermes_cli/fallback_cmd.py | 59 +++++----------------- hermes_cli/fallback_config.py | 7 +-- hermes_cli/focus_view.py | 29 ++++------- hermes_cli/foreign_sessions.py | 82 ++++++++++--------------------- hermes_cli/gitlock.py | 40 +++++---------- hermes_cli/gui_uninstall.py | 84 +++++++++++--------------------- hermes_cli/heartbeat.py | 46 ++++++----------- hermes_cli/image_provenance.py | 41 +++++----------- hermes_cli/init_command.py | 22 ++------- hermes_cli/input_sanitize.py | 11 ++--- hermes_cli/install_identity.py | 13 +---- 14 files changed, 159 insertions(+), 394 deletions(-) diff --git a/hermes_cli/dep_ensure.py b/hermes_cli/dep_ensure.py index 0c24a41371..a99296ee9b 100644 --- a/hermes_cli/dep_ensure.py +++ b/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() diff --git a/hermes_cli/diagnostics_upload.py b/hermes_cli/diagnostics_upload.py index e8f3635ceb..47d3d34afb 100644 --- a/hermes_cli/diagnostics_upload.py +++ b/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 (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 (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 diff --git a/hermes_cli/dingtalk_auth.py b/hermes_cli/dingtalk_auth.py index 2961ab1eac..96e89e0adf 100644 --- a/hermes_cli/dingtalk_auth.py +++ b/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 diff --git a/hermes_cli/fallback_cmd.py b/hermes_cli/fallback_cmd.py index 6efd7a0bf5..b006226e63 100644 --- a/hermes_cli/fallback_cmd.py +++ b/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, } diff --git a/hermes_cli/fallback_config.py b/hermes_cli/fallback_config.py index 56f8b77331..a3e90b569c 100644 --- a/hermes_cli/fallback_config.py +++ b/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 diff --git a/hermes_cli/focus_view.py b/hermes_cli/focus_view.py index 5b7b6b91cb..4d0d862cdf 100644 --- a/hermes_cli/focus_view.py +++ b/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: diff --git a/hermes_cli/foreign_sessions.py b/hermes_cli/foreign_sessions.py index 66a3bb6109..c87c61489f 100644 --- a/hermes_cli/foreign_sessions.py +++ b/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 ") 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" diff --git a/hermes_cli/gitlock.py b/hermes_cli/gitlock.py index 3ec27f80b0..9108ea94cb 100644 --- a/hermes_cli/gitlock.py +++ b/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), ) diff --git a/hermes_cli/gui_uninstall.py b/hermes_cli/gui_uninstall.py index ab6efca862..bee8e52b96 100644 --- a/hermes_cli/gui_uninstall.py +++ b/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 diff --git a/hermes_cli/heartbeat.py b/hermes_cli/heartbeat.py index 526f4a9930..4c0f57e176 100644 --- a/hermes_cli/heartbeat.py +++ b/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: diff --git a/hermes_cli/image_provenance.py b/hermes_cli/image_provenance.py index 3cc075197c..07f751a885 100644 --- a/hermes_cli/image_provenance.py +++ b/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) diff --git a/hermes_cli/init_command.py b/hermes_cli/init_command.py index d0090bd25e..8b8fe8aa13 100644 --- a/hermes_cli/init_command.py +++ b/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) diff --git a/hermes_cli/input_sanitize.py b/hermes_cli/input_sanitize.py index fd26e5a2a7..56afe5e638 100644 --- a/hermes_cli/input_sanitize.py +++ b/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: diff --git a/hermes_cli/install_identity.py b/hermes_cli/install_identity.py index a39151264e..8b8a07d0e8 100644 --- a/hermes_cli/install_identity.py +++ b/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: