diff --git a/hermes_cli/sizefmt.py b/hermes_cli/sizefmt.py index 0448da095e..29dddcfa85 100644 --- a/hermes_cli/sizefmt.py +++ b/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): diff --git a/hermes_cli/sqlite_runtime.py b/hermes_cli/sqlite_runtime.py index ba9f0ca23a..5e019c3172 100644 --- a/hermes_cli/sqlite_runtime.py +++ b/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 diff --git a/hermes_cli/stderr_timestamp.py b/hermes_cli/stderr_timestamp.py index c7263590d2..5ed7fe4b60 100644 --- a/hermes_cli/stderr_timestamp.py +++ b/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=`` 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) diff --git a/hermes_cli/stdio.py b/hermes_cli/stdio.py index b927404e67..75161f14d9 100644 --- a/hermes_cli/stdio.py +++ b/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]) diff --git a/hermes_cli/terminal_breadcrumbs.py b/hermes_cli/terminal_breadcrumbs.py index 67defe1481..aacbd9d96b 100644 --- a/hermes_cli/terminal_breadcrumbs.py +++ b/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 diff --git a/hermes_cli/terminal_notify.py b/hermes_cli/terminal_notify.py index 47ff111761..4b370ef8d3 100644 --- a/hermes_cli/terminal_notify.py +++ b/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 ; 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 ; BEL``): Warp's structured CLI-agent protocol (tab status + notification mailbox). +OSC 9 (``ESC ] 9 ; BEL``): Ghostty, iTerm2, Kitty and WezTerm raise an OS notification; +others drop it. OSC 777 (``ESC ] 777 ; notify ; warp://cli-agent ; 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" diff --git a/hermes_cli/timeouts.py b/hermes_cli/timeouts.py index 7480634319..71b1880c17 100644 --- a/hermes_cli/timeouts.py +++ b/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..models..`` wins over ``providers..``.""" 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