From 1006faa6f89577a019db31fdef38fac545249c86 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Fri, 7 Aug 2026 07:55:00 -0700 Subject: [PATCH] feat(doctor): add opt-in `hermes doctor --live` real-call backend probes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a bounded, read-only health probe per CONFIGURED tool backend, run only when the user explicitly passes `--live` (real network calls): - Firecrawl: credit-usage metadata GET (auth check, no scrape spend) - FAL: models metadata GET (never a generation) - Browser: headless launch + about:blank + close (full cleanup) - MCP: initialize + tools/list per configured server (reuses the `hermes mcp test` machinery in mcp_config._probe_single_server) - TTS/STT: provider models/voices list GET (openai/groq/elevenlabs); local providers (edge/piper/faster-whisper/...) skipped Invariants: - Opt-in only: zero probes without --live (default False) - Bounded: sequential, per-probe timeout (doctor.live_probe_timeout, default 10s, config.yaml knob) - Never mutates state; unconfigured backends skip with a note - Failure isolation: every probe wrapped in a catch-all; a probe crash can never break the doctor run; failures append to the issues summary New: hermes_cli/doctor_live.py, tests/hermes_cli/test_doctor_live.py (23 tests, probes mocked at the HTTP/client seam). Wired: --live flag in subcommands/doctor.py; run_doctor calls maybe_run_live_checks after all static checks. Coordination: PR #70124 (--probe-routes) probes LLM routes; this flag probes TOOL backends — different surface, no code-region collision (the run_doctor hook here sits at the end-of-run summary, not the API Connectivity section #70124 extends). Inspired by: paradigmxyz/centaur tool-health-smoke (MIT/Apache-2.0); sibling: #70124 (LLM route probes — different surface) --- hermes_cli/config_defaults.py | 7 + hermes_cli/doctor.py | 8 + hermes_cli/doctor_live.py | 316 +++++++++++++++++++++++++++ hermes_cli/subcommands/doctor.py | 9 + tests/hermes_cli/test_doctor_live.py | 270 +++++++++++++++++++++++ 5 files changed, 610 insertions(+) create mode 100644 hermes_cli/doctor_live.py create mode 100644 tests/hermes_cli/test_doctor_live.py diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 8e37663caf698..0eaa2a9f05ba8 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -2799,6 +2799,13 @@ DEFAULT_CONFIG = { }, }, + # ``hermes doctor`` behaviour. + "doctor": { + # Per-probe timeout (seconds) for the opt-in `hermes doctor --live` + # real-call backend probes (Firecrawl/FAL/browser/MCP/TTS/STT). + "live_probe_timeout": 10, + }, + # ``hermes update`` behaviour. "updates": { # Pre-update safety backup — ONE consolidated mechanism, three modes: diff --git a/hermes_cli/doctor.py b/hermes_cli/doctor.py index 238d54827fd10..458db27140c91 100644 --- a/hermes_cli/doctor.py +++ b/hermes_cli/doctor.py @@ -2747,6 +2747,14 @@ def run_doctor(args): except Exception: pass + # Opt-in live backend probes run AFTER all static checks, only with + # `hermes doctor --live` (real network calls; bounded + read-only). + try: + from hermes_cli.doctor_live import maybe_run_live_checks + maybe_run_live_checks(args, manual_issues) + except Exception: + pass + print() remaining_issues = issues + manual_issues if should_fix and fixed_count > 0: diff --git a/hermes_cli/doctor_live.py b/hermes_cli/doctor_live.py new file mode 100644 index 0000000000000..4a8120ea50699 --- /dev/null +++ b/hermes_cli/doctor_live.py @@ -0,0 +1,316 @@ +"""``hermes doctor --live`` — opt-in bounded real-call tool-backend probes. + +Design invariants: + +- **Opt-in only.** These probes make real (cheap, metadata/read-only) network + calls and may spend a trivial amount of quota. They run ONLY when the user + passes ``hermes doctor --live``. +- **Bounded.** One probe per configured backend, sequential, each with a + ~10s timeout (configurable via ``doctor.live_probe_timeout`` in + config.yaml). +- **Read-only.** Metadata GETs only — no generation, no scrapes that spend + credits, no state mutation anywhere. +- **Failure-isolated.** A probe crashing must never crash the doctor run; + every probe is wrapped in a catch-all. +- **Configured-only.** Backends without credentials / config are skipped with + a note, never failed. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Callable, List, Optional + +from hermes_cli.doctor import ( + _section, + check_fail, + check_info, + check_ok, + check_warn, +) + +DEFAULT_PROBE_TIMEOUT = 10.0 + +# Metadata-only endpoints. None of these spend generation credits. +FIRECRAWL_HEALTH_URL = "https://api.firecrawl.dev/v2/team/credit-usage" +FAL_MODELS_URL = "https://fal.ai/api/models?page=1" +OPENAI_MODELS_URL = "https://api.openai.com/v1/models" +GROQ_MODELS_URL = "https://api.groq.com/openai/v1/models" +ELEVENLABS_VOICES_URL = "https://api.elevenlabs.io/v1/voices" + +# TTS/STT providers that never touch the network (nothing to probe). +_LOCAL_AUDIO_PROVIDERS = {"", "local", "edge", "neutts", "kittentts", "piper"} + + +@dataclass +class ProbeResult: + """Outcome of one backend probe.""" + + name: str + status: str # "pass" | "warn" | "fail" | "skip" + detail: str = "" + + +# --------------------------------------------------------------------------- +# Small seams (monkeypatchable in tests, and single points of control). +# --------------------------------------------------------------------------- + +def _load_config() -> dict: + try: + from hermes_cli.config import load_config + + return load_config() or {} + except Exception: + return {} + + +def _http_get(url: str, headers: Optional[dict] = None, + timeout: Optional[float] = None): + """Single HTTP GET seam for all metadata probes.""" + import httpx + + return httpx.get(url, headers=headers or {}, timeout=timeout) + + +def _browser_available() -> bool: + """Is the local browser automation backend (agent-browser) installed?""" + import shutil + + if shutil.which("agent-browser"): + return True + try: + from hermes_cli.doctor import HERMES_HOME, PROJECT_ROOT + + if (PROJECT_ROOT / "node_modules" / "agent-browser").exists(): + return True + for candidate in (HERMES_HOME / "node" / "bin", + HERMES_HOME / "node", + HERMES_HOME / "node_modules" / ".bin"): + if shutil.which("agent-browser", path=str(candidate)): + return True + except Exception: + pass + return False + + +def _launch_browser_probe(timeout: float) -> tuple: + """Launch a browser, open about:blank, close. Returns (ok, detail). + + Uses Playwright directly (what agent-browser drives underneath) so the + probe owns the full lifecycle and always cleans up. + """ + try: + from playwright.sync_api import sync_playwright + except ImportError: + return (False, "playwright not installed") + + with sync_playwright() as p: + browser = p.chromium.launch(headless=True, + timeout=timeout * 1000) + try: + page = browser.new_page() + page.goto("about:blank", timeout=timeout * 1000) + finally: + browser.close() + return (True, "launched + about:blank + closed") + + +def _probe_mcp_server(name: str, config: dict, timeout: float): + """initialize + tools/list against one configured MCP server. + + Reuses the exact machinery behind ``hermes mcp test``. + """ + from hermes_cli.mcp_config import _probe_single_server + + return _probe_single_server(name, config, connect_timeout=timeout) + + +# --------------------------------------------------------------------------- +# Per-backend probes. Each returns a ProbeResult and never raises upward +# beyond what run_live_checks' catch-all handles. +# --------------------------------------------------------------------------- + +def _classify_http(name: str, resp, key_hint: str) -> ProbeResult: + code = getattr(resp, "status_code", None) + if code is not None and 200 <= code < 300: + return ProbeResult(name, "pass", f"(HTTP {code})") + if code in (401, 403): + return ProbeResult(name, "fail", + f"(HTTP {code} — check {key_hint})") + return ProbeResult(name, "fail", f"(HTTP {code})") + + +def _probe_firecrawl(timeout: float) -> ProbeResult: + key = os.getenv("FIRECRAWL_API_KEY", "").strip() + if not key: + return ProbeResult("Firecrawl", "skip", "(not configured)") + resp = _http_get(FIRECRAWL_HEALTH_URL, + headers={"Authorization": f"Bearer {key}"}, + timeout=timeout) + return _classify_http("Firecrawl", resp, "FIRECRAWL_API_KEY") + + +def _probe_fal(timeout: float) -> ProbeResult: + key = os.getenv("FAL_KEY", "").strip() + if not key: + return ProbeResult("FAL", "skip", "(not configured)") + # Metadata GET only — never a generation call. + resp = _http_get(FAL_MODELS_URL, + headers={"Authorization": f"Key {key}"}, + timeout=timeout) + return _classify_http("FAL", resp, "FAL_KEY") + + +def _probe_browser(timeout: float) -> ProbeResult: + if not _browser_available(): + return ProbeResult("Browser", "skip", "(not configured)") + ok, detail = _launch_browser_probe(timeout) + return ProbeResult("Browser", "pass" if ok else "fail", f"({detail})") + + +def _audio_provider_probe(kind: str, provider: str, + timeout: float) -> ProbeResult: + """Shared TTS/STT metadata probe (voices/models list GET only).""" + name = kind.upper() + provider = (provider or "").strip().lower() + if provider in _LOCAL_AUDIO_PROVIDERS: + return ProbeResult(name, "skip", + f"(provider '{provider or 'local'}' — no remote " + "backend to probe)") + + probes = { + "openai": (OPENAI_MODELS_URL, "OPENAI_API_KEY", "Bearer"), + "groq": (GROQ_MODELS_URL, "GROQ_API_KEY", "Bearer"), + "elevenlabs": (ELEVENLABS_VOICES_URL, "ELEVENLABS_API_KEY", "xi"), + } + entry = probes.get(provider) + if entry is None: + return ProbeResult(name, "skip", + f"(provider '{provider}' — no live probe " + "implemented)") + url, env_var, scheme = entry + key = os.getenv(env_var, "").strip() + if not key: + return ProbeResult(name, "warn", + f"(provider '{provider}' configured but " + f"{env_var} is not set)") + if scheme == "xi": + headers = {"xi-api-key": key} + else: + headers = {"Authorization": f"Bearer {key}"} + resp = _http_get(url, headers=headers, timeout=timeout) + result = _classify_http(name, resp, env_var) + result.detail = f"({provider}) {result.detail}" + return result + + +def _probe_tts(config: dict, timeout: float) -> ProbeResult: + provider = ((config.get("tts") or {}).get("provider")) or "" + return _audio_provider_probe("tts", provider, timeout) + + +def _probe_stt(config: dict, timeout: float) -> ProbeResult: + provider = ((config.get("stt") or {}).get("provider")) or "" + return _audio_provider_probe("stt", provider, timeout) + + +# --------------------------------------------------------------------------- +# Orchestration +# --------------------------------------------------------------------------- + +def _report(result: ProbeResult, issues: List[str]) -> None: + if result.status == "pass": + check_ok(result.name, result.detail) + elif result.status == "warn": + check_warn(result.name, result.detail) + elif result.status == "fail": + check_fail(result.name, result.detail) + issues.append(f"Live probe failed: {result.name} {result.detail}") + else: # skip + check_info(f"{result.name} {result.detail} — skipped") + + +def _run_one(name: str, fn: Callable[[], ProbeResult], + issues: List[str]) -> ProbeResult: + """Run one probe with a catch-all so a crash never kills doctor.""" + try: + result = fn() + except TimeoutError as exc: + result = ProbeResult(name, "fail", f"(timed out: {exc})") + except Exception as exc: + msg = str(exc) or exc.__class__.__name__ + if "time" in msg.lower(): + result = ProbeResult(name, "fail", f"(timed out: {msg})") + else: + result = ProbeResult(name, "fail", f"({msg})") + _report(result, issues) + return result + + +def run_live_checks(issues: List[str]) -> List[ProbeResult]: + """Run one bounded, read-only probe per configured tool backend. + + Sequential by design (bounded, predictable output ordering). Appends a + remediation line to ``issues`` for each failed probe. Skipped backends + never fail and never append issues. + """ + config = _load_config() + try: + timeout = float( + (config.get("doctor") or {}).get("live_probe_timeout", + DEFAULT_PROBE_TIMEOUT)) + except (TypeError, ValueError): + timeout = DEFAULT_PROBE_TIMEOUT + timeout = max(1.0, timeout) + + _section("Live Backend Probes (opt-in, real calls)") + results: List[ProbeResult] = [] + + results.append(_run_one( + "Firecrawl", lambda: _probe_firecrawl(timeout), issues)) + results.append(_run_one( + "FAL", lambda: _probe_fal(timeout), issues)) + results.append(_run_one( + "Browser", lambda: _probe_browser(timeout), issues)) + + servers = config.get("mcp_servers") or {} + if isinstance(servers, dict) and servers: + for name in sorted(servers): + entry = servers[name] + label = f"MCP: {name}" + + def _probe(n=name, e=entry) -> ProbeResult: + if not isinstance(e, dict): + return ProbeResult(f"MCP: {n}", "skip", + "(malformed config entry)") + tools = _probe_mcp_server(n, e, timeout) + return ProbeResult(f"MCP: {n}", "pass", + f"({len(tools)} tool(s))") + + results.append(_run_one(label, _probe, issues)) + else: + results.append(ProbeResult("MCP", "skip", "(no servers configured)")) + _report(results[-1], issues) + + results.append(_run_one( + "TTS", lambda: _probe_tts(config, timeout), issues)) + results.append(_run_one( + "STT", lambda: _probe_stt(config, timeout), issues)) + + return results + + +def maybe_run_live_checks(args, issues: List[str]): + """Entry point called from ``run_doctor`` after the static checks. + + No-ops (returns None) unless the user explicitly passed ``--live``. + A crash anywhere in the live subsystem must never break doctor. + """ + if not getattr(args, "live", False): + return None + try: + return run_live_checks(issues) + except Exception as exc: # catch-all: doctor must survive + check_warn("Live backend probes crashed", f"({exc})") + return None diff --git a/hermes_cli/subcommands/doctor.py b/hermes_cli/subcommands/doctor.py index 5be37c645581e..81a257192ac37 100644 --- a/hermes_cli/subcommands/doctor.py +++ b/hermes_cli/subcommands/doctor.py @@ -22,6 +22,15 @@ def build_doctor_parser(subparsers, *, cmd_doctor: Callable) -> None: doctor_parser.add_argument( "--fix", action="store_true", help="Attempt to fix issues automatically" ) + doctor_parser.add_argument( + "--live", + action="store_true", + help=( + "Opt-in: run one bounded, read-only real-call health probe per " + "configured tool backend (Firecrawl/FAL/browser/MCP/TTS/STT) " + "after the static checks. Makes real network calls." + ), + ) doctor_parser.add_argument( "--ack", metavar="ADVISORY_ID", diff --git a/tests/hermes_cli/test_doctor_live.py b/tests/hermes_cli/test_doctor_live.py new file mode 100644 index 0000000000000..0174e619c26f8 --- /dev/null +++ b/tests/hermes_cli/test_doctor_live.py @@ -0,0 +1,270 @@ +"""Tests for ``hermes doctor --live`` — opt-in bounded real-call tool-backend probes. + +All probes are mocked at the HTTP/client layer; no real network calls are made. +""" + +from __future__ import annotations + +import argparse +from types import SimpleNamespace + +import pytest + +from hermes_cli import doctor_live +from hermes_cli.doctor_live import ( + ProbeResult, + maybe_run_live_checks, + run_live_checks, +) + + +def _args(live: bool = True) -> argparse.Namespace: + return argparse.Namespace(live=live) + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch): + """Strip backend credentials so each test opts in explicitly.""" + for var in ("FIRECRAWL_API_KEY", "FAL_KEY", "OPENAI_API_KEY", + "ELEVENLABS_API_KEY", "GROQ_API_KEY"): + monkeypatch.delenv(var, raising=False) + # Default: empty config, no MCP servers, local tts/stt. + monkeypatch.setattr(doctor_live, "_load_config", lambda: {}) + # Default: browser not installed. + monkeypatch.setattr(doctor_live, "_browser_available", lambda: False) + + +class TestLiveFlagGating: + def test_parser_has_live_flag_default_false(self): + from hermes_cli.subcommands.doctor import build_doctor_parser + + parser = argparse.ArgumentParser() + sub = parser.add_subparsers(dest="command") + build_doctor_parser(sub, cmd_doctor=lambda a: None) + args = parser.parse_args(["doctor"]) + assert args.live is False + args = parser.parse_args(["doctor", "--live"]) + assert args.live is True + + def test_no_live_flag_means_zero_probes(self, monkeypatch): + called = [] + monkeypatch.setattr( + doctor_live, "run_live_checks", + lambda *a, **k: called.append(True)) + result = maybe_run_live_checks(_args(live=False), []) + assert result is None + assert called == [] + + def test_missing_live_attr_means_zero_probes(self, monkeypatch): + called = [] + monkeypatch.setattr( + doctor_live, "run_live_checks", + lambda *a, **k: called.append(True)) + assert maybe_run_live_checks(SimpleNamespace(), []) is None + assert called == [] + + def test_live_flag_runs_checks(self, monkeypatch): + called = [] + monkeypatch.setattr( + doctor_live, "run_live_checks", + lambda issues, **k: called.append(issues) or []) + issues: list[str] = [] + maybe_run_live_checks(_args(live=True), issues) + assert called == [issues] + + def test_live_check_crash_never_propagates(self, monkeypatch, capsys): + def _boom(*a, **k): + raise RuntimeError("probe subsystem exploded") + + monkeypatch.setattr(doctor_live, "run_live_checks", _boom) + # Must not raise. + maybe_run_live_checks(_args(live=True), []) + + +class TestConfiguredOnlySelection: + def test_all_unconfigured_all_skipped(self, capsys): + results = run_live_checks([]) + assert results, "expected one result per backend" + assert all(r.status == "skip" for r in results) + # No issues appended for skips. + + def test_unconfigured_backends_do_not_touch_network(self, monkeypatch): + def _no_net(*a, **k): + raise AssertionError("HTTP call made for unconfigured backend") + + monkeypatch.setattr(doctor_live, "_http_get", _no_net) + results = run_live_checks([]) + assert all(r.status == "skip" for r in results) + + def test_firecrawl_probed_when_key_present(self, monkeypatch): + monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test") + calls = [] + + def _fake_get(url, headers=None, timeout=None): + calls.append(url) + return SimpleNamespace(status_code=200) + + monkeypatch.setattr(doctor_live, "_http_get", _fake_get) + results = {r.name: r for r in run_live_checks([])} + assert results["Firecrawl"].status == "pass" + assert any("firecrawl" in u for u in calls) + + def test_firecrawl_invalid_key_fails_and_appends_issue(self, monkeypatch): + monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-bad") + monkeypatch.setattr( + doctor_live, "_http_get", + lambda *a, **k: SimpleNamespace(status_code=401)) + issues: list[str] = [] + results = {r.name: r for r in run_live_checks(issues)} + assert results["Firecrawl"].status == "fail" + assert any("FIRECRAWL" in i or "Firecrawl" in i for i in issues) + + def test_fal_probed_when_key_present(self, monkeypatch): + monkeypatch.setenv("FAL_KEY", "fal-test") + monkeypatch.setattr( + doctor_live, "_http_get", + lambda *a, **k: SimpleNamespace(status_code=200)) + results = {r.name: r for r in run_live_checks([])} + assert results["FAL"].status == "pass" + + def test_mcp_servers_probed_per_configured_server(self, monkeypatch): + monkeypatch.setattr( + doctor_live, "_load_config", + lambda: {"mcp_servers": {"alpha": {"url": "https://x"}, + "beta": {"command": "foo"}}}) + probed = [] + monkeypatch.setattr( + doctor_live, "_probe_mcp_server", + lambda name, cfg, timeout: probed.append(name) or [("t", "d")]) + results = [r for r in run_live_checks([]) if r.name.startswith("MCP")] + assert sorted(probed) == ["alpha", "beta"] + assert len(results) == 2 + assert all(r.status == "pass" for r in results) + + def test_tts_local_provider_skipped(self, monkeypatch): + monkeypatch.setattr( + doctor_live, "_load_config", + lambda: {"tts": {"provider": "edge"}}) + results = {r.name: r for r in run_live_checks([])} + assert results["TTS"].status == "skip" + + def test_tts_openai_probed_with_key(self, monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "sk-test") + monkeypatch.setattr( + doctor_live, "_load_config", + lambda: {"tts": {"provider": "openai"}}) + monkeypatch.setattr( + doctor_live, "_http_get", + lambda *a, **k: SimpleNamespace(status_code=200)) + results = {r.name: r for r in run_live_checks([])} + assert results["TTS"].status == "pass" + + def test_stt_groq_probed_with_key(self, monkeypatch): + monkeypatch.setenv("GROQ_API_KEY", "gsk-test") + monkeypatch.setattr( + doctor_live, "_load_config", + lambda: {"stt": {"provider": "groq"}}) + monkeypatch.setattr( + doctor_live, "_http_get", + lambda *a, **k: SimpleNamespace(status_code=200)) + results = {r.name: r for r in run_live_checks([])} + assert results["STT"].status == "pass" + + def test_stt_provider_configured_but_key_missing_warns(self, monkeypatch): + monkeypatch.setattr( + doctor_live, "_load_config", + lambda: {"stt": {"provider": "groq"}}) + results = {r.name: r for r in run_live_checks([])} + assert results["STT"].status == "warn" + + def test_browser_probed_when_available(self, monkeypatch): + monkeypatch.setattr(doctor_live, "_browser_available", lambda: True) + monkeypatch.setattr( + doctor_live, "_launch_browser_probe", + lambda timeout: (True, "about:blank ok")) + results = {r.name: r for r in run_live_checks([])} + assert results["Browser"].status == "pass" + + +class TestFailureIsolation: + def test_one_probe_raising_does_not_stop_others(self, monkeypatch): + monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test") + monkeypatch.setenv("FAL_KEY", "fal-test") + + def _get(url, headers=None, timeout=None): + if "firecrawl" in url: + raise RuntimeError("connection reset") + return SimpleNamespace(status_code=200) + + monkeypatch.setattr(doctor_live, "_http_get", _get) + issues: list[str] = [] + results = {r.name: r for r in run_live_checks(issues)} + assert results["Firecrawl"].status == "fail" + assert results["FAL"].status == "pass" + + def test_mcp_probe_failure_isolated_per_server(self, monkeypatch): + monkeypatch.setattr( + doctor_live, "_load_config", + lambda: {"mcp_servers": {"bad": {"url": "https://x"}, + "good": {"url": "https://y"}}}) + + def _probe(name, cfg, timeout): + if name == "bad": + raise ConnectionError("refused") + return [("tool", "desc")] + + monkeypatch.setattr(doctor_live, "_probe_mcp_server", _probe) + results = {r.name: r for r in run_live_checks([])} + assert results["MCP: bad"].status == "fail" + assert results["MCP: good"].status == "pass" + + +class TestTimeoutHandling: + def test_timeout_reported_as_fail(self, monkeypatch): + monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test") + + def _slow(*a, **k): + raise TimeoutError("timed out") + + monkeypatch.setattr(doctor_live, "_http_get", _slow) + results = {r.name: r for r in run_live_checks([])} + assert results["Firecrawl"].status == "fail" + assert "time" in (results["Firecrawl"].detail or "").lower() + + def test_probe_timeout_bounded_and_configurable(self, monkeypatch): + monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test") + monkeypatch.setattr( + doctor_live, "_load_config", + lambda: {"doctor": {"live_probe_timeout": 3}}) + seen = {} + + def _get(url, headers=None, timeout=None): + seen["timeout"] = timeout + return SimpleNamespace(status_code=200) + + monkeypatch.setattr(doctor_live, "_http_get", _get) + run_live_checks([]) + assert seen["timeout"] == 3 + + def test_default_timeout_is_10s(self, monkeypatch): + monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-test") + seen = {} + + def _get(url, headers=None, timeout=None): + seen["timeout"] = timeout + return SimpleNamespace(status_code=200) + + monkeypatch.setattr(doctor_live, "_http_get", _get) + run_live_checks([]) + assert seen["timeout"] == 10.0 + + +class TestReadOnly: + def test_probe_result_is_plain_record(self): + r = ProbeResult(name="X", status="skip", detail="not configured") + assert (r.name, r.status, r.detail) == ("X", "skip", "not configured") + + def test_skips_never_append_issues(self, capsys): + issues: list[str] = [] + run_live_checks(issues) + assert issues == []