From 18eeeabb6221e2a92386a4db0542fa403ecde0e5 Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 16 Aug 2026 09:19:23 -0300 Subject: [PATCH] hermes: avoid unavailable worker providers --- .../hermes/scripts/claude_oauth_broker.py | 11 +++ services/hermes/scripts/cli_lane_runner.py | 71 +++++++++++++++++-- testing/tests/test_hermes_cli_lanes.py | 45 ++++++++++++ 3 files changed, 122 insertions(+), 5 deletions(-) diff --git a/services/hermes/scripts/claude_oauth_broker.py b/services/hermes/scripts/claude_oauth_broker.py index 4e0dd1a3..a27075d4 100644 --- a/services/hermes/scripts/claude_oauth_broker.py +++ b/services/hermes/scripts/claude_oauth_broker.py @@ -64,6 +64,9 @@ _slots = threading.BoundedSemaphore(MAX_CONCURRENCY) _auth_probe_lock = threading.Lock() _auth_probe_at = 0.0 _auth_probe_value: dict[str, Any] = {} +HEALTH_POLL_SECONDS: Final = max( + 30, int(os.environ.get("HERMES_CLAUDE_HEALTH_POLL_SECONDS", "60")) +) def _read_secret(file_env_name: str) -> str: @@ -313,6 +316,13 @@ def _subscription_health(force: bool = False) -> dict[str, Any]: return dict(value) +def _health_polling_loop() -> None: + """Keep native auth health fresh even when no request reaches the broker.""" + while True: + time.sleep(HEALTH_POLL_SECONDS) + _subscription_health(force=True) + + def _invoke(payload: dict[str, Any]) -> tuple[dict[str, Any], dict[str, int], str]: """Run one native, first-party Claude Code subscription request.""" requested = payload.get("model") @@ -645,4 +655,5 @@ class Server(ThreadingHTTPServer): if __name__ == "__main__": _subscription_health(force=True) + threading.Thread(target=_health_polling_loop, daemon=True).start() Server((HOST, PORT), Handler).serve_forever() diff --git a/services/hermes/scripts/cli_lane_runner.py b/services/hermes/scripts/cli_lane_runner.py index 05b617cb..1c9410f6 100644 --- a/services/hermes/scripts/cli_lane_runner.py +++ b/services/hermes/scripts/cli_lane_runner.py @@ -37,6 +37,11 @@ EXTERNAL_PREFIX = "cli-" DEFAULT_CLAIM_TTL = 7 * 24 * 60 * 60 DEFAULT_MAX_RUNTIME = 12 * 60 * 60 HEARTBEAT_SECONDS = 20 +PROVIDER_HEALTH_MAX_AGE_SECONDS = 5 * 60 +PROVIDER_HEALTH_PATHS = { + "codex": DATA_ROOT / "provider-health/codex.json", + "claude": DATA_ROOT / "provider-health/claude.json", +} WORKTREE_LOCK = threading.Lock() BOARD_CORRUPTION_ERRORS: dict[str, str] = {} CAPACITY_PATTERN = re.compile( @@ -119,6 +124,26 @@ def parse_assignee(assignee: str) -> tuple[str | None, str | None]: return match.group(1), match.group(2) +def fresh_unavailable_provider(now: float | None = None) -> str | None: + """Return one recently proven-down provider for automatic route exclusion.""" + current = time.time() if now is None else now + unavailable: list[str] = [] + for provider, path in PROVIDER_HEALTH_PATHS.items(): + health = load_json(path) + try: + age = current - path.stat().st_mtime + except OSError: + continue + if ( + 0 <= age <= PROVIDER_HEALTH_MAX_AGE_SECONDS + and health.get("state") == "unavailable" + ): + unavailable.append(provider) + # If both providers are down, keep the normal attempt/fallback path so the + # card records authoritative current errors instead of trusting snapshots. + return unavailable[0] if len(unavailable) == 1 else None + + def _decode_worker_target(value: str) -> tuple[str, str, str]: """Decode the selected model header emitted by a worker decision route.""" parts = value.split("/", 3) @@ -135,6 +160,7 @@ def select_route( assignee: str, *, exclude_provider: str | None = None, + exclude_reason: str | None = None, switchyard_url: str = SWITCHYARD_URL, open_request: Callable[..., Any] = urllib.request.urlopen, ) -> Route: @@ -149,9 +175,10 @@ def select_route( source = "switchyard-classifier" context = prompt if exclude_provider: + reason = exclude_reason or "failed or exhausted capacity at this boundary" context += ( - f"\n\nRouting constraint: the {exclude_provider} provider failed or " - "exhausted capacity at this boundary. Do not select it." + f"\n\nRouting constraint: the {exclude_provider} provider {reason}. " + "Do not select it." ) payload = json.dumps( { @@ -192,8 +219,25 @@ def select_route( except (AttributeError, IndexError, KeyError, RuntimeError, TypeError, ValueError, json.JSONDecodeError): pass if exclude_provider and provider == exclude_provider: - raise RuntimeError( - f"Switchyard selected excluded provider {exclude_provider} for retry" + alternate = "claude" if provider == "codex" else "codex" + guarded = select_route( + prompt, + f"cli-{alternate}-{effort}", + switchyard_url=switchyard_url, + open_request=open_request, + ) + return Route( + provider=guarded.provider, + model=guarded.model, + effort=guarded.effort, + profile=guarded.profile, + classifier=f"{source}-health-guard", + reason=( + f"Switchyard selected excluded {exclude_provider}; preserved " + f"its {effort} effort on healthy-provider route. {guarded.reason}" + ), + latency_ms=int((time.monotonic() - started) * 1000), + fallback_chain=(), ) return Route( provider=provider, @@ -681,7 +725,24 @@ def execute_claim(board: str, task_id: str) -> None: try: previous_route = state.get("current_route") - route = select_route(context, assignee) + excluded_provider = ( + fresh_unavailable_provider() if assignee == "cli-auto" else None + ) + route = select_route( + context, + assignee, + exclude_provider=excluded_provider, + exclude_reason="is unavailable according to fresh native health" + if excluded_provider + else None, + ) + if excluded_provider: + kanban_db.add_comment( + conn, + task_id, + "cli-lane-runner", + f"Provider health guard excluded {excluded_provider} before automatic routing.", + ) kanban_db.add_comment( conn, task_id, diff --git a/testing/tests/test_hermes_cli_lanes.py b/testing/tests/test_hermes_cli_lanes.py index 4123553d..164c04a0 100644 --- a/testing/tests/test_hermes_cli_lanes.py +++ b/testing/tests/test_hermes_cli_lanes.py @@ -151,6 +151,51 @@ def test_cross_provider_retry_passes_failed_provider_to_switchyard(): assert "codex provider failed or exhausted capacity" in content +def test_classifier_cannot_select_a_freshly_excluded_provider(): + payloads = [] + + def route(request, timeout): + payload = json.loads(request.data) + payloads.append(payload) + if len(payloads) == 1: + return _SwitchyardResponse("worker/claude/opus/xhigh") + return _SwitchyardResponse("worker/codex/sol/xhigh", "healthy route") + + selected = lanes.select_route( + "Perform a consequential review.", + "cli-auto", + exclude_provider="claude", + exclude_reason="is unavailable according to fresh native health", + open_request=route, + ) + + assert selected.provider == "codex" + assert selected.effort == "xhigh" + assert selected.classifier == "switchyard-classifier-health-guard" + assert payloads[0]["model"] == "atlas/worker/auto/maximum" + assert payloads[1]["model"] == "atlas/worker/manual/codex/xhigh" + assert "claude provider is unavailable" in payloads[0]["messages"][0]["content"] + + +def test_fresh_native_health_excludes_only_one_proven_down_provider( + tmp_path: Path, monkeypatch +): + paths = { + "codex": tmp_path / "codex.json", + "claude": tmp_path / "claude.json", + } + paths["codex"].write_text('{"state":"available"}\n', encoding="utf-8") + paths["claude"].write_text('{"state":"unavailable"}\n', encoding="utf-8") + monkeypatch.setattr(lanes, "PROVIDER_HEALTH_PATHS", paths) + + now = max(path.stat().st_mtime for path in paths.values()) + assert lanes.fresh_unavailable_provider(now=now) == "claude" + + paths["codex"].write_text('{"state":"unavailable"}\n', encoding="utf-8") + now = max(path.stat().st_mtime for path in paths.values()) + assert lanes.fresh_unavailable_provider(now=now) is None + + def test_worker_environment_preserves_vault_backed_cli_homes(monkeypatch): """Kanban workers must not fall back to credential-free persistent homes.""" monkeypatch.setenv("CODEX_HOME", "/runtime-access/codex")