refactor(hermes_cli): compact stdio/breadcrumbs/notify/timeouts/sqlite_runtime/sizefmt helpers

hermes_cli/sizefmt.py

@@ -1,22 +1,12 @@
-"""Small shared size-formatting helpers for CLI/agent output.
-
-Sibling of ``hermes_cli.timefmt`` (same extraction rationale: a tiny purpose-named module
-lightweight consumers can import without dragging in the CLI surface). Replaces six near-identical
-private byte formatters.
-
-Two in-repo formatters intentionally do NOT delegate here:
-"""
+"""Small shared size-formatting helpers for CLI/agent output (sibling of ``hermes_cli.timefmt``:
+a tiny purpose-named module lightweight consumers can import without the CLI surface)."""
 
 from __future__ import annotations
 
 
 def format_bytes(n) -> str:
-    """1234567 -> '1.2 MB' (B/KB/MB/GB/TB; integer bytes, one decimal above).
-
-    Accepts anything ``float()`` accepts; returns ``"?"`` for None or unparseable input so display
-    call sites never raise (contract inherited from doctor's original copy — its stats dict
-    tolerates None fields).
-    """
+    """1234567 -> '1.2 MB' (B/KB/MB/GB/TB; integer bytes, one decimal above). Returns ``"?"`` for
+    None or unparseable input so display call sites never raise."""
     try:
         size = float(n)
     except (TypeError, ValueError):

hermes_cli/sqlite_runtime.py

@@ -20,9 +20,7 @@ def _version_tuple(parts: Iterable[object]) -> tuple[int, int, int]:
     return tuple(values[:3])
 
 
-def is_sqlite_wal_reset_vulnerable(
-    version_info: tuple[int, ...],
-) -> bool:
+def is_sqlite_wal_reset_vulnerable(version_info: tuple[int, ...]) -> bool:
     """Return whether *version_info* contains SQLite's WAL-reset bug."""
     info = _version_tuple(version_info)
     return not (
@@ -75,33 +73,18 @@ def isolated_interpreter_env() -> dict[str, str]:
     """Copy of ``os.environ`` with conda/uv/venv/PYTHON* overrides stripped, so a child interpreter
     reports its *own* runtime rather than the caller's."""
     env = dict(os.environ)
-    for key in (
-        "CONDA_DEFAULT_ENV",
-        "CONDA_PREFIX",
-        "PYTHONHOME",
-        "PYTHONPATH",
-        "UV_PROJECT_ENVIRONMENT",
-        "UV_PYTHON",
-        "VIRTUAL_ENV",
-    ):
+    for key in ("CONDA_DEFAULT_ENV", "CONDA_PREFIX", "PYTHONHOME", "PYTHONPATH", "UV_PROJECT_ENVIRONMENT",
+                "UV_PYTHON", "VIRTUAL_ENV"):
         env.pop(key, None)
     return env
 
 
-def probe_sqlite_runtime(
-    python: str | Path,
-    *,
-    timeout: float = 30.0,
-) -> SQLiteRuntimeInfo | None:
+def probe_sqlite_runtime(python: str | Path, *, timeout: float = 30.0) -> SQLiteRuntimeInfo | None:
     """Probe SQLite in *python*, never the caller's linked SQLite."""
     try:
         result = subprocess.run(
-            [str(python), "-I", "-c", _PROBE_SCRIPT],
-            capture_output=True,
-            text=True,
-            timeout=timeout,
-            check=False,
-            env=isolated_interpreter_env(),
+            [str(python), "-I", "-c", _PROBE_SCRIPT], capture_output=True, text=True, timeout=timeout,
+            check=False, env=isolated_interpreter_env(),
         )
     except (OSError, subprocess.TimeoutExpired):
         return None

hermes_cli/stderr_timestamp.py

@@ -15,7 +15,6 @@ from typing import BinaryIO, Sequence, TextIO
 
 EXTERNAL_SUPERVISOR_FLAG = "--external-supervisor"
 
-
 _TIMESTAMP_PREFIX = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2},\d{3}(?:\s|$)")
 
 
@@ -73,10 +72,7 @@ def _is_hermes_gateway_run_argv(command: Sequence[str]) -> bool:
     return bool(looks_like_gateway_command_line(" ".join(str(part) for part in command)))
 
 
-def _prepare_child_command(
-    command: Sequence[str],
-    environ: Mapping[str, str] | None = None,
-) -> list[str]:
+def _prepare_child_command(command: Sequence[str], environ: Mapping[str, str] | None = None) -> list[str]:
     """Return the argv to exec, upgrading stale launchd-wrapped gateway commands.
 
     launchd stamps ``XPC_SERVICE_NAME=<job label>`` only on this wrapper (its direct child; an
@@ -87,19 +83,13 @@ def _prepare_child_command(
     argv = [str(part) for part in command]
     env = os.environ if environ is None else environ
     xpc_service = str(env.get("XPC_SERVICE_NAME", "")).strip()
-    if (
-        EXTERNAL_SUPERVISOR_FLAG not in argv
-        and xpc_service and xpc_service != "0"
-        and _is_hermes_gateway_run_argv(argv)
-    ):
+    if EXTERNAL_SUPERVISOR_FLAG not in argv and xpc_service and xpc_service != "0" and _is_hermes_gateway_run_argv(argv):
         argv.append(EXTERNAL_SUPERVISOR_FLAG)
     return argv
 
 
 def _parse_args(argv: Sequence[str] | None) -> argparse.Namespace:
-    parser = argparse.ArgumentParser(
-        description="Run a command and timestamp each stderr line into a log file."
-    )
+    parser = argparse.ArgumentParser(description="Run a command and timestamp each stderr line into a log file.")
     parser.add_argument("--error-log", required=True, type=Path)
     parser.add_argument("command", nargs=argparse.REMAINDER)
     args = parser.parse_args(argv)

hermes_cli/stdio.py

@@ -1,7 +1,7 @@
 """Windows-safe stdio configuration.
 
-The fix is to force UTF-8 on the Python side and also flip the console's code page to UTF-8 (65001).
-Both matter: Python-level only helps when Python's stdout is a real TTY; code-page flipping lets
+Forces UTF-8 on the Python side and also flips the console's code page to UTF-8 (65001). Both
+matter: Python-level only helps when Python's stdout is a real TTY; code-page flipping lets
 subprocesses and child Python ``print()`` calls agree on encoding.
 """
 
@@ -22,174 +22,102 @@ def is_windows() -> bool:
 
 
 def _flip_console_code_page_to_utf8() -> None:
-    """Set the attached console's input and output code pages to UTF-8.
-
-    Uses ``SetConsoleCP`` / ``SetConsoleOutputCP`` (CP_UTF8 = 65001). Failure is silent: without an
-    attached console (redirected stdout, service, PTY-less CI) the calls return 0 and we move on.
-    """
+    """``SetConsoleCP``/``SetConsoleOutputCP`` to CP_UTF8 (65001). Silent on failure: without an
+    attached console (redirected stdout, service, PTY-less CI) the calls return 0 and we move on."""
     try:
         import ctypes
 
         kernel32 = ctypes.windll.kernel32  # type: ignore[attr-defined]
-        # Best-effort; if there's no console attached these just fail silently.
         kernel32.SetConsoleCP(65001)
         kernel32.SetConsoleOutputCP(65001)
     except Exception:
-        # ctypes import, missing kernel32, or non-Windows — any failure here
-        # is non-fatal.  We've still reconfigured Python's own streams below.
         pass
 
 
 def _reconfigure_stream(stream, *, encoding: str = "utf-8", errors: str = "replace") -> None:
-    """Reconfigure a text stream to UTF-8 in place.
-
-    Skips rather than raising when the stream isn't a ``TextIOWrapper`` (e.g. redirected to an
-    ``io.StringIO`` during tests).
-    """
+    """Reconfigure a text stream to UTF-8 in place; skips streams without ``reconfigure`` (e.g. an
+    ``io.StringIO`` substituted during tests)."""
     try:
         reconfigure = getattr(stream, "reconfigure", None)
-        if reconfigure is None:
-            return
-        reconfigure(encoding=encoding, errors=errors)
+        if reconfigure is not None:
+            reconfigure(encoding=encoding, errors=errors)
     except Exception:
         pass
 
 
 def configure_windows_stdio() -> bool:
-    """Force UTF-8 stdio on Windows.  No-op elsewhere.
+    """Force UTF-8 stdio on Windows. No-op elsewhere.
 
     Idempotent; returns ``True`` only when something actually changed. Set
     ``HERMES_DISABLE_WINDOWS_UTF8=1`` to opt out (forces the old cp1252 path for diagnosing
-    encoding bugs). Also sets a default ``EDITOR`` on Windows if none is set (see
-    ``_default_windows_editor``).
+    encoding bugs). Also sets a default ``EDITOR`` on Windows if none is set.
     """
     global _CONFIGURED
 
     if _CONFIGURED:
         return False
-    if not is_windows():
-        # Mark configured so repeated calls on POSIX are true no-ops.
-        _CONFIGURED = True
+    if not is_windows() or os.environ.get("HERMES_DISABLE_WINDOWS_UTF8") in {"1", "true", "True", "yes"}:
+        _CONFIGURED = True  # repeated calls on POSIX / opted-out are true no-ops
         return False
 
-    if os.environ.get("HERMES_DISABLE_WINDOWS_UTF8") in {"1", "true", "True", "yes"}:
-        _CONFIGURED = True
-        return False
-
-    # Encourage every child Python process spawned by the agent to also use
-    # UTF-8 for its stdio.  PYTHONIOENCODING wins over the locale-based
-    # default in subprocesses.  Don't override an explicit user setting.
+    # Make child Python processes use UTF-8 stdio too (PYTHONIOENCODING wins over the locale
+    # default; PYTHONUTF8=1 enables UTF-8 Mode, PEP 540). Never override an explicit user setting.
     os.environ.setdefault("PYTHONIOENCODING", "utf-8")
-    # PYTHONUTF8 = 1 enables UTF-8 Mode globally for any Python subprocess
-    # (PEP 540).  Again, don't override an explicit setting.
     os.environ.setdefault("PYTHONUTF8", "1")
 
-    # Set EDITOR to a working Windows default if neither EDITOR nor VISUAL
-    # is set.  prompt_toolkit's ``open_in_editor`` falls back to POSIX-only
-    # paths (``/usr/bin/nano``, ``/usr/bin/vi``) that don't exist on
-    # Windows — Ctrl+X Ctrl+E and ``/edit`` silently do nothing there
-    # otherwise.  This happens even with full Git for Windows installed,
-    # so it's not a MinGit-specific issue.
+    # prompt_toolkit's ``open_in_editor`` falls back to POSIX-only paths (/usr/bin/nano, /usr/bin/vi)
+    # that don't exist on Windows — Ctrl+X Ctrl+E and ``/edit`` silently do nothing there
+    # otherwise, even with full Git for Windows installed.
     _default_editor = _default_windows_editor()
     if _default_editor and not os.environ.get("EDITOR") and not os.environ.get("VISUAL"):
         os.environ["EDITOR"] = _default_editor
 
-    # Augment PATH with the Hermes-managed Git install directories so
-    # subprocess calls (bash, rg, grep, etc.) resolve even in sessions
-    # that started before the User PATH broadcast reached them.  When
-    # install.ps1 adds these to User PATH via SetEnvironmentVariable,
-    # already-running shells don't see the change — which means hermes
-    # launched from the install session won't find rg / bash / grep
-    # even though they're "installed".  Prepending the known paths here
-    # closes that gap.  No-op when the paths don't exist (e.g. system-Git
-    # install without Hermes-managed PortableGit).
     _augment_path_with_known_tools()
-
-    # Flip the console code page first so that any subprocess that
-    # inherits the console (e.g. a launched shell) also sees CP_UTF8.
+    # Flip the console code page first so any subprocess inheriting the console also sees CP_UTF8.
     _flip_console_code_page_to_utf8()
-
-    # Reconfigure Python's own stdio wrappers so ``print()`` calls from
-    # this process round-trip emoji / box-drawing / non-Latin text.
-    # ``errors="replace"`` means a genuinely unencodable byte sequence
-    # gets a ``?`` rather than crashing the interpreter — we prefer
-    # degraded output over a stack trace.
-    _reconfigure_stream(sys.stdout)
-    _reconfigure_stream(sys.stderr)
-    # stdin is re-configured for completeness; Hermes's interactive
-    # input path uses prompt_toolkit which manages its own encoding,
-    # but batch/pipe input benefits from UTF-8 decoding on stdin too.
-    _reconfigure_stream(sys.stdin)
-
+    # ``errors="replace"``: a genuinely unencodable sequence prints ``?`` rather than crashing the
+    # interpreter. stdin is included for batch/pipe input (prompt_toolkit manages its own encoding).
+    for stream in (sys.stdout, sys.stderr, sys.stdin):
+        _reconfigure_stream(stream)
     _CONFIGURED = True
     return True
 
 
 def _default_windows_editor() -> str:
-    """Return a Windows-appropriate default for ``$EDITOR``.
-
-    Priority order, first match wins:
-
-    1. ``notepad`` — ships with every Windows install, no deps, works as a blocking editor
-    (``subprocess.call(["notepad", file])`` blocks until the user closes the window). This is the
-    "always-works" default.
-    """
+    """Windows default for ``$EDITOR``: ``notepad`` (ships with every install, blocks until the
+    window closes). The bare name keeps prompt_toolkit's shlex split away from paths with spaces;
+    "" when even notepad is missing (WinPE, Nano Server) so prompt_toolkit's no-op applies."""
     import shutil
 
-    # notepad.exe is always in %SystemRoot%\System32 on Windows, so shutil.which
-    # will reliably find it.  Return the bare name so prompt_toolkit's shlex
-    # split doesn't trip over a path containing spaces.
-    if shutil.which("notepad"):
-        return "notepad"
-    # On the extreme off-chance notepad is missing (WinPE, Nano Server), fall
-    # back to nothing and let prompt_toolkit's silent no-op do its thing.
-    return ""
+    return "notepad" if shutil.which("notepad") else ""
 
 
 def _augment_path_with_known_tools() -> None:
-    r"""Prepend well-known Hermes-managed tool directories to os.environ['PATH'].
+    r"""Prepend Hermes-managed tool directories to ``PATH`` (no-op on POSIX / missing dirs).
 
-    Fixes the "User PATH was just updated but my process can't see it" gap on Windows. When
-    install.ps1 runs, it adds entries like ``%LOCALAPPDATA%\hermes\git\bin`` to the User PATH via
-    ``SetEnvironmentVariable(..., "User")``.
-
-    Patch-up strategy: add the known Hermes-managed tool directories to our PATH at startup so
-    subprocess calls resolve correctly. No-op on POSIX and when the directories don't exist. The
-    User PATH broadcast still happens in the background for future shells; this just smooths over
-    the first-launch gap.
+    install.ps1 adds entries like ``%LOCALAPPDATA%\hermes\git\bin`` to the User PATH via
+    ``SetEnvironmentVariable``, but already-running shells never see that broadcast, so a hermes
+    launched from the install session would not find rg / bash / grep. Prepending the known dirs
+    at startup closes that first-launch gap.
     """
     if not is_windows():
         return
-
     local_appdata = os.environ.get("LOCALAPPDATA", "")
     if not local_appdata:
         return
 
-    # Known tool dirs installed by scripts/install.ps1.  Kept in sync with
-    # the PATH entries that installer adds to User scope — the two lists
-    # should match so this prefill fully mirrors what a fresh shell would
-    # see on next launch.
+    # Kept in sync with the PATH entries scripts/install.ps1 adds to User scope. The venv Scripts
+    # dir hosts hermes.exe + pip console scripts; WinGet\Links is where ``winget install`` drops
+    # CLI shims (ripgrep lands there as rg.exe).
     candidate_dirs = [
         os.path.join(local_appdata, "hermes", "git", "cmd"),
         os.path.join(local_appdata, "hermes", "git", "bin"),
         os.path.join(local_appdata, "hermes", "git", "usr", "bin"),
-        # Hermes venv Scripts directory — host of the hermes.exe shim itself,
-        # also where any pip-installed console scripts land.  Usually already
-        # on PATH when the user invokes hermes, but harmless to include.
         os.path.join(local_appdata, "hermes", "hermes-agent", "venv", "Scripts"),
-        # WinGet packages directory — where ``winget install`` drops CLI
-        # shims by default (ripgrep lands here as rg.exe).  Covers the case
-        # of a system-Git install + ripgrep-via-winget that isn't yet on
-        # the spawning shell's PATH.
         os.path.join(local_appdata, "Microsoft", "WinGet", "Links"),
     ]
-
     existing = os.environ.get("PATH", "")
     existing_lower = {p.lower() for p in existing.split(os.pathsep) if p}
-    prepend = []
-    for d in candidate_dirs:
-        if os.path.isdir(d) and d.lower() not in existing_lower:
-            prepend.append(d)
-
+    prepend = [d for d in candidate_dirs if os.path.isdir(d) and d.lower() not in existing_lower]
     if prepend:
         os.environ["PATH"] = os.pathsep.join([*prepend, existing])

hermes_cli/terminal_breadcrumbs.py

@@ -16,20 +16,12 @@ import time
 from pathlib import Path
 from typing import Optional
 
-# Multiplexer / terminal-emulator identity env vars, checked in order when
-# no real tty path is available (e.g. stdin piped but stdout still a pty
-# owned by a known terminal).
-_TERMINAL_ENV_VARS = (
-    "ZELLIJ_PANE_ID",
-    "TMUX_PANE",
-    "KITTY_WINDOW_ID",
-    "WEZTERM_PANE",
-    "TERM_SESSION_ID",
-    "WT_SESSION",
-)
+# Multiplexer / terminal-emulator identity env vars, checked in order when no real tty path is
+# available (e.g. stdin piped but stdout still a pty owned by a known terminal).
+_TERMINAL_ENV_VARS = ("ZELLIJ_PANE_ID", "TMUX_PANE", "KITTY_WINDOW_ID", "WEZTERM_PANE", "TERM_SESSION_ID", "WT_SESSION")
 
-# Breadcrumbs older than this are pruned opportunistically on each write —
-# a pane id from a tmux server restarted last month means nothing today.
+# Breadcrumbs older than this are pruned opportunistically on each write — a pane id from a tmux
+# server restarted last month means nothing today.
 _STALE_AFTER_SECONDS = 30 * 24 * 60 * 60
 
 _SANITIZE_RE = re.compile(r"[^A-Za-z0-9._-]")
@@ -47,12 +39,8 @@ def _sanitize(raw: str) -> str:
 
 
 def get_terminal_id() -> Optional[str]:
-    """Derive a stable identity for the terminal this process runs in.
-
-    Prefers the real tty device path (stdin, then stdout), else the first present
-    multiplexer/emulator env var. Returns ``None`` when neither is available — callers must then
-    skip breadcrumbs entirely.
-    """
+    """Stable identity for this terminal: the tty device path (stdin, then stdout), else the first
+    present multiplexer/emulator env var; ``None`` when neither exists (callers skip breadcrumbs)."""
     for fd in (sys.stdin, sys.stdout):
         try:
             name = os.ttyname(fd.fileno())
@@ -105,11 +93,7 @@ def write_breadcrumb(session_id: str, cwd: Optional[str] = None) -> None:
         directory = _breadcrumbs_dir()
         directory.mkdir(parents=True, exist_ok=True)
         now = time.time()
-        payload = {
-            "session_id": session_id,
-            "cwd": cwd or os.getcwd(),
-            "ts": now,
-        }
+        payload = {"session_id": session_id, "cwd": cwd or os.getcwd(), "ts": now}
         tmp = directory / f".{terminal_id}.tmp"
         tmp.write_text(json.dumps(payload), encoding="utf-8")
         os.replace(tmp, directory / terminal_id)
@@ -119,10 +103,7 @@ def write_breadcrumb(session_id: str, cwd: Optional[str] = None) -> None:
 
 
 def read_breadcrumb() -> Optional[dict]:
-    """Return this terminal's breadcrumb payload, or ``None``.
-
-    Ignores breadcrumbs older than the staleness window. Never raises.
-    """
+    """This terminal's breadcrumb payload, or ``None`` (missing, corrupt, or stale). Never raises."""
     try:
         terminal_id = get_terminal_id()
         if not terminal_id:
@@ -154,23 +135,23 @@ def resolve_breadcrumb_session() -> Optional[str]:
     session_id = str(crumb.get("session_id") or "").strip()
     if not session_id:
         return None
-    db = None
     try:
         from hermes_state import SessionDB
 
         db = SessionDB()
+    except Exception:
+        return None
+    try:
         if not db.get_session(session_id):
             return None  # session was deleted — fall back to latest
         try:
-            session_id = db.get_compression_tip(session_id) or session_id
+            return db.get_compression_tip(session_id) or session_id
         except Exception:
-            pass
-        return session_id
+            return session_id
     except Exception:
         return None
     finally:
-        if db is not None:
-            try:
-                db.close()
-            except Exception:
-                pass
+        try:
+            db.close()
+        except Exception:
+            pass

hermes_cli/terminal_notify.py

@@ -1,8 +1,8 @@
 """Terminal-native desktop notifications: OSC 9 and Warp's OSC 777 CLI-agent protocol.
 
-- **OSC 9** (``ESC ] 9 ; <body> BEL``): Ghostty, iTerm2, Kitty and WezTerm raise an OS notification;
-terminals that don't know the sequence drop it. - **OSC 777** (``ESC ] 777 ; notify ; warp://cli-
-agent ; <json> BEL``): Warp's structured CLI-agent protocol (tab status + notification mailbox).
+OSC 9 (``ESC ] 9 ; <body> BEL``): Ghostty, iTerm2, Kitty and WezTerm raise an OS notification;
+others drop it. OSC 777 (``ESC ] 777 ; notify ; warp://cli-agent ; <json> BEL``): Warp's
+structured CLI-agent protocol (tab status + notification mailbox).
 
 Sequences are written to ``/dev/tty`` because prompt_toolkit's stdout wrapper can buffer or strip
 raw escapes; when ``/dev/tty`` can't be opened (Windows, no controlling terminal) they fall back to
@@ -18,13 +18,9 @@ import sys
 
 _C0_AND_DEL = re.compile(r"[\x00-\x1f\x7f]")
 _WARP_PROTOCOL_VERSION = 1
-# Last Warp release per channel that set WARP_CLI_AGENT_PROTOCOL_VERSION but
-# could not render structured payloads (Warp's reference agent plugin,
-# should-use-structured.sh). Bash compares these lexicographically; so do we.
-_WARP_LAST_BROKEN = {
-    "stable": "v0.2026.03.25.08.24.stable_05",
-    "preview": "v0.2026.03.25.08.24.preview_05",
-}
+# Last Warp release per channel that set WARP_CLI_AGENT_PROTOCOL_VERSION but could not render
+# structured payloads (Warp's should-use-structured.sh). Bash compares lexicographically; so do we.
+_WARP_LAST_BROKEN = {"stable": "v0.2026.03.25.08.24.stable_05", "preview": "v0.2026.03.25.08.24.preview_05"}
 
 
 def _write_tty(seq: str) -> None:
@@ -50,15 +46,10 @@ def osc9(body: str) -> str:
 def warp_supported(env=None) -> bool:
     """True when running in a Warp build that can render OSC 777 agent payloads."""
     env = os.environ if env is None else env
-    if env.get("TERM_PROGRAM") != "WarpTerminal" or not env.get("WARP_CLI_AGENT_PROTOCOL_VERSION"):
-        return False
     client = env.get("WARP_CLIENT_VERSION", "")
-    if not client:
+    if env.get("TERM_PROGRAM") != "WarpTerminal" or not env.get("WARP_CLI_AGENT_PROTOCOL_VERSION") or not client:
         return False
-    for channel, last_broken in _WARP_LAST_BROKEN.items():
-        if channel in client and client <= last_broken:
-            return False
-    return True
+    return not any(channel in client and client <= last_broken for channel, last_broken in _WARP_LAST_BROKEN.items())
 
 
 def warp_osc777(event: str, detail: str, session_id: str = "") -> str:
@@ -68,14 +59,8 @@ def warp_osc777(event: str, detail: str, session_id: str = "") -> str:
     except ValueError:
         advertised = 1
     cwd = os.getcwd()
-    payload = {
-        "v": min(advertised, _WARP_PROTOCOL_VERSION),
-        "agent": "hermes",
-        "event": event,
-        "session_id": session_id,
-        "cwd": cwd,
-        "project": os.path.basename(cwd),
-    }
+    payload = {"v": min(advertised, _WARP_PROTOCOL_VERSION), "agent": "hermes", "event": event,
+               "session_id": session_id, "cwd": cwd, "project": os.path.basename(cwd)}
     payload["summary" if event == "permission_request" else "response"] = detail[:200]
     return f"\x1b]777;notify;warp://cli-agent;{json.dumps(payload, separators=(',', ':'))}\x07"
 

hermes_cli/timeouts.py

@@ -9,53 +9,40 @@ def _coerce_timeout(raw: object) -> float | None:
     return timeout if timeout > 0 else None
 
 
-def _configured_timeout(
-    provider_id: str, model: str | None, model_key: str, provider_key: str
-) -> float | None:
+def _configured_timeout(provider_id: str, model: str | None, model_key: str, provider_key: str) -> float | None:
     """Per-model ``providers.<id>.models.<model>.<model_key>`` wins over ``providers.<id>.<provider_key>``."""
     if not provider_id:
         return None
-
     try:
         from hermes_cli.config import load_config_readonly
         config = load_config_readonly()
     except Exception:
         return None
-
     providers = config.get("providers", {}) if isinstance(config, dict) else {}
     provider_config = providers.get(provider_id, {}) if isinstance(providers, dict) else {}
     if not isinstance(provider_config, dict):
         return None
-
     model_config = _get_model_config(provider_config, model)
     if model_config is not None:
         timeout = _coerce_timeout(model_config.get(model_key))
         if timeout is not None:
             return timeout
-
     return _coerce_timeout(provider_config.get(provider_key))
 
 
-def get_provider_request_timeout(
-    provider_id: str, model: str | None = None
-) -> float | None:
+def get_provider_request_timeout(provider_id: str, model: str | None = None) -> float | None:
     """Return a configured provider request timeout in seconds, if any."""
     return _configured_timeout(provider_id, model, "timeout_seconds", "request_timeout_seconds")
 
 
-def get_provider_stale_timeout(
-    provider_id: str, model: str | None = None
-) -> float | None:
+def get_provider_stale_timeout(provider_id: str, model: str | None = None) -> float | None:
     """Return a configured non-stream stale timeout in seconds, if any."""
     return _configured_timeout(provider_id, model, "stale_timeout_seconds", "stale_timeout_seconds")
 
 
-def _get_model_config(
-    provider_config: dict[str, object], model: str | None
-) -> dict[str, object] | None:
+def _get_model_config(provider_config: dict[str, object], model: str | None) -> dict[str, object] | None:
     if not model:
         return None
-
     models = provider_config.get("models", {})
     model_config = models.get(model, {}) if isinstance(models, dict) else {}
     return model_config if isinstance(model_config, dict) else None