hermes: stabilize Claude access health probes

services/hermes/scripts/claude_oauth_broker.py

@@ -67,6 +67,9 @@ _auth_probe_value: dict[str, Any] = {}
 HEALTH_POLL_SECONDS: Final = max(
     30, int(os.environ.get("HERMES_CLAUDE_HEALTH_POLL_SECONDS", "60"))
 )
+AUTH_PROBE_GRACE_SECONDS: Final = max(
+    60, int(os.environ.get("HERMES_CLAUDE_AUTH_PROBE_GRACE_SECONDS", "300"))
+)
 
 
 def _read_secret(file_env_name: str) -> str:
@@ -265,7 +268,8 @@ def _subscription_health(force: bool = False) -> dict[str, Any]:
     with _auth_probe_lock:
         if not force and _auth_probe_value and now - _auth_probe_at < 60:
             return dict(_auth_probe_value)
-        checked_at = datetime.now(timezone.utc).isoformat()
+        checked_time = datetime.now(timezone.utc)
+        checked_at = checked_time.isoformat()
         try:
             completed = subprocess.run(
                 [CLAUDE_BIN, "auth", "status"],
@@ -279,6 +283,10 @@ def _subscription_health(force: bool = False) -> dict[str, Any]:
         except (OSError, subprocess.SubprocessError, ValueError, json.JSONDecodeError):
             completed = None
             raw = {}
+        previous = _previous_health()
+        probe_conclusive = isinstance(raw, dict) and isinstance(
+            raw.get("loggedIn"), bool
+        )
         authenticated = bool(
             completed
             and completed.returncode == 0
@@ -286,14 +294,42 @@ def _subscription_health(force: bool = False) -> dict[str, Any]:
             and raw.get("loggedIn") is True
             and raw.get("apiProvider") == "firstParty"
         )
-        previous = _previous_health()
+        api_provider = raw.get("apiProvider") if isinstance(raw, dict) else None
+        auth_method = raw.get("authMethod") if isinstance(raw, dict) else None
+        # A concurrent inference can make Claude's secondary auth-status
+        # process time out or return no JSON. A real, recent inference is
+        # stronger evidence than that inconclusive probe. Preserve the last
+        # proven-good auth state briefly, but never mask an explicit logged-out
+        # response from the CLI.
+        recent_success = False
+        try:
+            last_success = datetime.fromisoformat(
+                str(previous.get("last_success_at") or "").replace("Z", "+00:00")
+            )
+            if last_success.tzinfo is None:
+                last_success = last_success.replace(tzinfo=timezone.utc)
+            age = (checked_time - last_success.astimezone(timezone.utc)).total_seconds()
+            recent_success = 0 <= age <= AUTH_PROBE_GRACE_SECONDS
+        except (TypeError, ValueError):
+            pass
+        probe_stale = bool(
+            not probe_conclusive
+            and recent_success
+            and previous.get("authenticated") is True
+            and previous.get("api_provider") == "firstParty"
+        )
+        if probe_stale:
+            authenticated = True
+            api_provider = previous.get("api_provider")
+            auth_method = previous.get("auth_method")
         value = {
             "transport": "claude-code-cli-subscription",
             "state": "available" if authenticated else "unavailable",
             "checked_at": checked_at,
             "authenticated": authenticated,
-            "api_provider": raw.get("apiProvider") if isinstance(raw, dict) else None,
+            "api_provider": api_provider,
-            "auth_method": raw.get("authMethod") if isinstance(raw, dict) else None,
+            "auth_method": auth_method,
+            "auth_probe_stale": probe_stale,
             "subscription_type": raw.get("subscriptionType")
             if isinstance(raw, dict)
             else None,

testing/tests/test_hermes_claude_broker.py

@@ -3,10 +3,12 @@
 from __future__ import annotations
 
 import importlib.util
+import json
 import sys
+from datetime import datetime, timezone
 from io import BytesIO
 from pathlib import Path
-from types import ModuleType
+from types import ModuleType, SimpleNamespace
 
 import pytest
 
@@ -119,3 +121,65 @@ def test_json_response_tolerates_disconnected_client(monkeypatch, disconnect) ->
     handler.wfile = Disconnected()
 
     handler._json(200, {"ok": True})
+
+
+def test_transient_auth_probe_keeps_recent_proven_access(monkeypatch, tmp_path) -> None:
+    module = _module(monkeypatch)
+    module.HEALTH_PATH = tmp_path / "claude.json"
+    module.HEALTH_PATH.write_text(
+        json.dumps(
+            {
+                "authenticated": True,
+                "api_provider": "firstParty",
+                "auth_method": "oauth_token",
+                "last_success_at": datetime.now(timezone.utc).isoformat(),
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *_args, **_kwargs: SimpleNamespace(
+            returncode=1, stdout="", stderr="transient timeout"
+        ),
+    )
+
+    health = module._subscription_health(force=True)
+
+    assert health["authenticated"] is True
+    assert health["state"] == "available"
+    assert health["api_provider"] == "firstParty"
+    assert health["auth_method"] == "oauth_token"
+    assert health["auth_probe_stale"] is True
+
+
+def test_explicit_logged_out_probe_clears_recent_access(monkeypatch, tmp_path) -> None:
+    module = _module(monkeypatch)
+    module.HEALTH_PATH = tmp_path / "claude.json"
+    module.HEALTH_PATH.write_text(
+        json.dumps(
+            {
+                "authenticated": True,
+                "api_provider": "firstParty",
+                "auth_method": "oauth_token",
+                "last_success_at": datetime.now(timezone.utc).isoformat(),
+            }
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(
+        module.subprocess,
+        "run",
+        lambda *_args, **_kwargs: SimpleNamespace(
+            returncode=1,
+            stdout=json.dumps({"loggedIn": False}),
+            stderr="",
+        ),
+    )
+
+    health = module._subscription_health(force=True)
+
+    assert health["authenticated"] is False
+    assert health["state"] == "unavailable"
+    assert health["auth_probe_stale"] is False