diff --git a/services/hermes/scripts/claude_oauth_broker.py b/services/hermes/scripts/claude_oauth_broker.py index a27075d4..2718f97f 100644 --- a/services/hermes/scripts/claude_oauth_broker.py +++ b/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, - "auth_method": raw.get("authMethod") if isinstance(raw, dict) else None, + "api_provider": api_provider, + "auth_method": auth_method, + "auth_probe_stale": probe_stale, "subscription_type": raw.get("subscriptionType") if isinstance(raw, dict) else None, diff --git a/testing/tests/test_hermes_claude_broker.py b/testing/tests/test_hermes_claude_broker.py index 8b019fff..1f9fb7a9 100644 --- a/testing/tests/test_hermes_claude_broker.py +++ b/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