From 6df688adcc212988a281e5bb8858a0531c638971 Mon Sep 17 00:00:00 2001 From: jenkins Date: Mon, 17 Aug 2026 20:31:12 -0300 Subject: [PATCH] hermes: test quota gating, cooldown recovery, and failover policy Deterministic coverage for the quota-aware lane: threshold boundaries (14.9/15/15.1), both-below preference, fetch-failure fail-open, cooldown elapsed-vs-not hysteresis, quota-reset recovery (never for auth), explicit fail-closed in both directions, bounded double-failure block, failure-reason classification, metrics emission, and worker env key stripping. Based on PR #15 (fix/hermes-result-decomposition-reliability). Co-Authored-By: Claude Opus 4.8 --- testing/tests/test_hermes_agent_security.py | 5 +- .../tests/test_hermes_cli_dispatch_runtime.py | 4 + .../tests/test_hermes_cli_execution_edges.py | 6 +- testing/tests/test_hermes_cli_fallback.py | 2 +- .../tests/test_hermes_cli_goal_execution.py | 2 +- testing/tests/test_hermes_cli_lane_metrics.py | 116 +++++ .../tests/test_hermes_cli_provider_health.py | 200 ++++++++ .../tests/test_hermes_cli_quota_failover.py | 440 ++++++++++++++++++ testing/tests/test_hermes_cli_quota_gating.py | 299 ++++++++++++ 9 files changed, 1068 insertions(+), 6 deletions(-) create mode 100644 testing/tests/test_hermes_cli_lane_metrics.py create mode 100644 testing/tests/test_hermes_cli_provider_health.py create mode 100644 testing/tests/test_hermes_cli_quota_failover.py create mode 100644 testing/tests/test_hermes_cli_quota_gating.py diff --git a/testing/tests/test_hermes_agent_security.py b/testing/tests/test_hermes_agent_security.py index cf23a2d0..1d713203 100644 --- a/testing/tests/test_hermes_agent_security.py +++ b/testing/tests/test_hermes_agent_security.py @@ -179,7 +179,10 @@ def test_agent_network_boundary_allows_only_authenticated_and_metrics_surfaces() "podSelector": {"matchLabels": {"app": "server"}}, } ], - "ports": [{"protocol": "TCP", "port": 9010}], + "ports": [ + {"protocol": "TCP", "port": 9010}, + {"protocol": "TCP", "port": 9011}, + ], }, ] assert isolation["spec"]["egress"] == [{}] diff --git a/testing/tests/test_hermes_cli_dispatch_runtime.py b/testing/tests/test_hermes_cli_dispatch_runtime.py index 06873f0a..3ffc0b39 100644 --- a/testing/tests/test_hermes_cli_dispatch_runtime.py +++ b/testing/tests/test_hermes_cli_dispatch_runtime.py @@ -161,6 +161,7 @@ def test_ready_dispatch_loop_submits_and_reaps_failed_workers( monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "schema.json") monkeypatch.setattr(lanes, "recover_orphans", lambda: None) + monkeypatch.setattr(lanes, "start_metrics_server", lambda port=None: None) ready = lanes.KanbanCapabilities(True, True, True) monkeypatch.setattr(lanes, "initialize_kanban_capabilities", lambda _db: ready) monkeypatch.setattr(lanes, "refresh_kanban_capabilities", lambda _db: ready) @@ -196,6 +197,7 @@ def test_deferred_dispatch_health_never_claims_new_work( monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "schema.json") monkeypatch.setattr(lanes, "recover_orphans", lambda: None) + monkeypatch.setattr(lanes, "start_metrics_server", lambda port=None: None) deferred = lanes.KanbanCapabilities(True, False, False) monkeypatch.setattr(lanes, "initialize_kanban_capabilities", lambda _db: deferred) monkeypatch.setattr(lanes, "refresh_kanban_capabilities", lambda _db: deferred) @@ -251,6 +253,7 @@ def test_two_loop_api_transition_refreshes_dispatch_and_health( monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "schema.json") recoveries = [] monkeypatch.setattr(lanes, "recover_orphans", lambda: recoveries.append("recover")) + monkeypatch.setattr(lanes, "start_metrics_server", lambda port=None: None) monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0) monkeypatch.setattr(lanes, "maybe_gc_lane_artifacts", lambda: 0) claim_states = [] @@ -369,6 +372,7 @@ def test_dispatch_loop_survives_board_registry_scan_failure( monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) monkeypatch.setattr(lanes, "RESULT_SCHEMA_PATH", tmp_path / "schema.json") monkeypatch.setattr(lanes, "recover_orphans", lambda: None) + monkeypatch.setattr(lanes, "start_metrics_server", lambda port=None: None) ready = lanes.KanbanCapabilities(True, True, True) monkeypatch.setattr(lanes, "initialize_kanban_capabilities", lambda _db: ready) monkeypatch.setattr(lanes, "refresh_kanban_capabilities", lambda _db: ready) diff --git a/testing/tests/test_hermes_cli_execution_edges.py b/testing/tests/test_hermes_cli_execution_edges.py index 7873c86c..d1ffbda8 100644 --- a/testing/tests/test_hermes_cli_execution_edges.py +++ b/testing/tests/test_hermes_cli_execution_edges.py @@ -58,7 +58,7 @@ def test_transient_callback_storage_errors_do_not_kill_provider( monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") monkeypatch.setattr(lanes, "KANBAN_STORAGE_ATTEMPTS", 1) - monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda: "claude") + monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda *_args, **_kwargs: "claude") route = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ()) monkeypatch.setattr(lanes, "select_route", lambda *_args, **_kwargs: route) @@ -246,7 +246,7 @@ def test_capacity_fallback_without_evidence_blocks_with_provider_output( ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") - monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda: None) + monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda *_args, **_kwargs: None) codex = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ()) claude = lanes.Route("claude", "fable", "high", "p", "c", "r", 1, ()) monkeypatch.setattr( @@ -298,7 +298,7 @@ def test_worker_blockers_become_the_block_reason_without_goal_mode( ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db)) monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes") - monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda: None) + monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda *_args, **_kwargs: None) route = lanes.Route("codex", "gpt", "high", "p", "c", "r", 1, ()) monkeypatch.setattr(lanes, "select_route", lambda *_args, **_kwargs: route) structured = { diff --git a/testing/tests/test_hermes_cli_fallback.py b/testing/tests/test_hermes_cli_fallback.py index e2a8141e..e8e4bc1d 100644 --- a/testing/tests/test_hermes_cli_fallback.py +++ b/testing/tests/test_hermes_cli_fallback.py @@ -73,7 +73,7 @@ def test_goal_card_continues_after_local_judge_rejects_progress( return codex_xhigh monkeypatch.setattr(lanes, "select_route", select_route) - monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda: None) + monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda *_args, **_kwargs: None) reports = [ lanes.ProcessResult(1, "authentication expired", None, True), lanes.ProcessResult( diff --git a/testing/tests/test_hermes_cli_goal_execution.py b/testing/tests/test_hermes_cli_goal_execution.py index 749a3d12..b8f1b1a5 100644 --- a/testing/tests/test_hermes_cli_goal_execution.py +++ b/testing/tests/test_hermes_cli_goal_execution.py @@ -191,7 +191,7 @@ def test_goal_loop_recovers_from_corrupt_rejection_history( monkeypatch.setattr( lanes, "fresh_unavailable_provider", - lambda: probes.append("health") and None, + lambda *_args, **_kwargs: probes.append("health") and None, ) route = lanes.Route( "codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, () diff --git a/testing/tests/test_hermes_cli_lane_metrics.py b/testing/tests/test_hermes_cli_lane_metrics.py new file mode 100644 index 00000000..ff832065 --- /dev/null +++ b/testing/tests/test_hermes_cli_lane_metrics.py @@ -0,0 +1,116 @@ +"""Lane metrics registry semantics and the pod-local scrape endpoint.""" + +from __future__ import annotations + +import urllib.error +import urllib.request + +from testing.tests.test_hermes_cli_support import lanes + + +def test_registry_renders_counters_and_gauges_with_stable_labels(): + registry = lanes.LaneMetricsRegistry() + registry.increment( + "hermes_cli_route_decisions_total", + {"provider": "codex", "effort": "high", "classifier": "c", "outcome": "completed"}, + ) + registry.increment( + "hermes_cli_route_decisions_total", + {"provider": "codex", "effort": "high", "classifier": "c", "outcome": "completed"}, + ) + registry.increment("hermes_cli_quota_fetch_failures_total") + registry.set_value( + "hermes_cli_quota_remaining_percent", {"provider": 'we"ird\n\\'}, 12.5 + ) + rendered = registry.render().decode("utf-8") + + assert "# TYPE hermes_cli_route_decisions_total counter" in rendered + assert "# TYPE hermes_cli_quota_remaining_percent gauge" in rendered + assert ( + 'hermes_cli_route_decisions_total{classifier="c",effort="high",' + 'outcome="completed",provider="codex"} 2' in rendered + ) + assert "hermes_cli_quota_fetch_failures_total 1" in rendered + assert ( + 'hermes_cli_quota_remaining_percent{provider="we\\"ird\\n\\\\"} 12.5' + in rendered + ) + + +def test_recording_helpers_feed_the_shared_registry(): + lanes.record_route_decision("codex", "xhigh", "metrics-test", "completed") + lanes.record_provider_fallback("codex", "claude", "rate-limit") + lanes.record_router_selection_failure() + lanes.record_quota_fetch_failure() + lanes.record_provider_quota("metrics-test-provider", 42.0, 777.0) + lanes.record_provider_quota("metrics-test-no-reset", 41.0, None) + lanes.record_soft_exclusion("metrics-test-provider", True) + rendered = lanes.METRICS.render().decode("utf-8") + + assert ( + 'hermes_cli_route_decisions_total{classifier="metrics-test",' + 'effort="xhigh",outcome="completed",provider="codex"} 1' in rendered + ) + assert ( + 'hermes_cli_provider_fallbacks_total{from_provider="codex",' + 'reason="rate-limit",to_provider="claude"}' in rendered + ) + assert "hermes_cli_router_selection_failures_total" in rendered + assert "hermes_cli_quota_fetch_failures_total" in rendered + assert ( + 'hermes_cli_quota_remaining_percent{provider="metrics-test-provider"} 42' + in rendered + ) + assert ( + 'hermes_cli_quota_reset_timestamp_seconds{provider="metrics-test-provider"} 777' + in rendered + ) + assert ( + 'hermes_cli_quota_reset_timestamp_seconds{provider="metrics-test-no-reset"}' + not in rendered + ) + assert ( + 'hermes_cli_provider_soft_excluded{provider="metrics-test-provider"} 1' + in rendered + ) + + +def test_metrics_server_serves_scrapes_health_and_rejects_unknown_paths(): + server = lanes.start_metrics_server(0) + assert server is not None + try: + port = server.server_address[1] + lanes.record_route_decision("codex", "low", "scrape-test", "completed") + with urllib.request.urlopen(f"http://127.0.0.1:{port}/metrics") as response: + assert response.status == 200 + assert b'classifier="scrape-test"' in response.read() + with urllib.request.urlopen(f"http://127.0.0.1:{port}/healthz") as response: + assert response.status == 200 + try: + urllib.request.urlopen(f"http://127.0.0.1:{port}/nope") + raise AssertionError("unknown paths must 404") + except urllib.error.HTTPError as error: + assert error.code == 404 + finally: + server.shutdown() + server.server_close() + + +def test_metrics_server_reads_port_from_environment(monkeypatch): + monkeypatch.setenv("HERMES_CLI_LANE_METRICS_PORT", "0") + server = lanes.start_metrics_server() + assert server is not None + server.shutdown() + server.server_close() + + +def test_metrics_server_declines_conflicting_port(capsys): + first = lanes.start_metrics_server(0) + assert first is not None + try: + second = lanes.start_metrics_server(first.server_address[1]) + assert second is None + assert "metrics server unavailable" in capsys.readouterr().err + finally: + first.shutdown() + first.server_close() diff --git a/testing/tests/test_hermes_cli_provider_health.py b/testing/tests/test_hermes_cli_provider_health.py new file mode 100644 index 00000000..01bb810e --- /dev/null +++ b/testing/tests/test_hermes_cli_provider_health.py @@ -0,0 +1,200 @@ +"""Lane-side provider health writes, cooldown hysteresis, and recovery.""" + +from __future__ import annotations + +from testing.tests.test_hermes_cli_support import ( + Path, + json, + lanes, + pytest, +) + + +@pytest.fixture +def health_paths(tmp_path: Path, monkeypatch) -> dict[str, Path]: + paths = { + "codex": tmp_path / "health/codex.json", + "claude": tmp_path / "health/claude.json", + } + monkeypatch.setattr(lanes, "PROVIDER_HEALTH_PATHS", paths) + monkeypatch.setattr(lanes, "DATA_ROOT", tmp_path) + return paths + + +@pytest.mark.parametrize( + ("output", "expected"), + [ + ("You have hit your usage limit for this billing cycle.", "quota"), + ("credit balance exhausted", "quota"), + ("HTTP 429: too many requests", "rate-limit"), + ("overloaded_error (529)", "rate-limit"), + ("OAuth token expired; run codex login", "auth"), + ("401 Unauthorized", "auth"), + ("stream disconnected before completion", "transport"), + ], +) +def test_capacity_failures_are_classified_actionably(output: str, expected: str): + assert lanes.classify_capacity_failure(output) == expected + + +def test_capacity_failure_records_cooldown_with_defaults(health_paths): + until = lanes.record_provider_failure("codex", "quota", now=1000.0) + + value = json.loads(health_paths["codex"].read_text()) + assert until == 1300.0 + assert value["state"] == "capacity-limited" + assert value["authenticated"] is True + assert value["failure_reason"] == "quota" + assert value["failed_at"] == 1000.0 + assert value["cooldown_until"] == 1300.0 + assert value["source"] == "cli-lane-runner" + + +def test_auth_failure_uses_longer_cooldown_and_marks_unauthenticated(health_paths): + until = lanes.record_provider_failure("claude", "auth", now=1000.0) + + value = json.loads(health_paths["claude"].read_text()) + assert until == 4600.0 + assert value["state"] == "unavailable" + assert value["authenticated"] is False + + +def test_cooldowns_are_configurable_through_the_kanban_block( + health_paths, tmp_path: Path +): + (tmp_path / "config.yaml").write_text( + "kanban:\n" + " provider_capacity_cooldown_seconds: 60\n" + " provider_auth_cooldown_seconds: 120\n" + ) + + assert lanes.record_provider_failure("codex", "rate-limit", now=1000.0) == 1060.0 + assert lanes.record_provider_failure("codex", "auth", now=1000.0) == 1120.0 + + +def test_lane_writes_merge_with_broker_health_snapshots(health_paths): + health_paths["codex"].parent.mkdir(parents=True) + health_paths["codex"].write_text( + json.dumps({"transport": "codex-chatgpt-subscription", "state": "available"}) + ) + + lanes.record_provider_failure("codex", "quota", now=1000.0) + + value = json.loads(health_paths["codex"].read_text()) + assert value["transport"] == "codex-chatgpt-subscription" + assert value["state"] == "capacity-limited" + + +def test_success_clears_failure_evidence_and_readmits(health_paths): + lanes.record_provider_failure("codex", "quota", now=1000.0) + lanes.record_provider_success("codex") + + value = json.loads(health_paths["codex"].read_text()) + assert value["state"] == "available" + assert value["authenticated"] is True + for stale in ("failure_reason", "failed_at", "cooldown_until"): + assert stale not in value + assert lanes.fresh_unavailable_provider(now=1001.0) is None + + +def test_health_write_failures_never_break_the_lane(health_paths, tmp_path: Path): + blocker = tmp_path / "health" + blocker.write_text("a file where the directory must go") + + lanes.record_provider_failure("codex", "quota", now=1000.0) + + +def test_broker_reported_capacity_limited_state_is_excluded(health_paths): + health_paths["codex"].parent.mkdir(parents=True) + health_paths["codex"].write_text('{"state":"capacity-limited"}\n') + + now = health_paths["codex"].stat().st_mtime + assert lanes.fresh_unavailable_provider(now=now) == "codex" + + +def test_cooldown_must_fully_elapse_before_readmission(health_paths): + lanes.record_provider_failure("codex", "quota", now=1000.0) + + assert lanes.fresh_unavailable_provider(now=1299.9) == "codex" + assert lanes.fresh_unavailable_provider(now=1300.0) is None + + +def test_quota_reset_readmits_capacity_limited_provider_early(health_paths): + lanes.record_provider_failure("codex", "quota", now=1000.0) + + # Reset before the failure proves nothing; a future reset has not passed. + assert ( + lanes.fresh_unavailable_provider(now=1100.0, quota_resets={"codex": 900.0}) + == "codex" + ) + assert ( + lanes.fresh_unavailable_provider(now=1100.0, quota_resets={"codex": 1150.0}) + == "codex" + ) + # A reset between the failure and now means the window rolled over. + assert ( + lanes.fresh_unavailable_provider(now=1200.0, quota_resets={"codex": 1150.0}) + is None + ) + + +def test_quota_reset_never_readmits_auth_failures(health_paths): + lanes.record_provider_failure("codex", "auth", now=1000.0) + + assert ( + lanes.fresh_unavailable_provider(now=1200.0, quota_resets={"codex": 1150.0}) + == "codex" + ) + + +def test_both_providers_in_cooldown_defer_to_live_attempts(health_paths): + lanes.record_provider_failure("codex", "quota", now=1000.0) + lanes.record_provider_failure("claude", "quota", now=1000.0) + + assert lanes.fresh_unavailable_provider(now=1100.0) is None + + +def test_vanished_health_file_is_not_excluded(health_paths, monkeypatch): + monkeypatch.setattr(lanes, "load_json", lambda _path: {"state": "unavailable"}) + + assert lanes.fresh_unavailable_provider(now=1000.0) is None + + +def test_malformed_cooldown_falls_back_to_freshness_window(health_paths): + health_paths["claude"].parent.mkdir(parents=True) + health_paths["claude"].write_text( + '{"state":"capacity-limited","cooldown_until":true}\n' + ) + + now = health_paths["claude"].stat().st_mtime + assert lanes.fresh_unavailable_provider(now=now) == "claude" + assert lanes.fresh_unavailable_provider(now=now + 10 * 60) is None + + +def test_health_number_rejects_booleans_and_strings(): + assert lanes._health_number(True) is None + assert lanes._health_number("5") is None + assert lanes._health_number(5) == 5.0 + + +def test_worker_environment_strips_metered_api_keys(monkeypatch): + for name in ( + "ANTHROPIC_API_KEY", + "CLAUDE_API_KEY", + "OPENAI_API_KEY", + "API_SERVER_KEY", + ): + monkeypatch.setenv(name, "must-not-leak") + monkeypatch.setenv("CODEX_HOME", "/runtime-access/codex") + + env = lanes._base_env() + + for name in ( + "ANTHROPIC_API_KEY", + "CLAUDE_API_KEY", + "OPENAI_API_KEY", + "API_SERVER_KEY", + ): + assert name not in env + assert env["CODEX_HOME"] == "/runtime-access/codex" + assert env["GIT_TERMINAL_PROMPT"] == "0" diff --git a/testing/tests/test_hermes_cli_quota_failover.py b/testing/tests/test_hermes_cli_quota_failover.py new file mode 100644 index 00000000..3cc54850 --- /dev/null +++ b/testing/tests/test_hermes_cli_quota_failover.py @@ -0,0 +1,440 @@ +"""Fail-closed manual lanes, bounded failover, and router-outage blocking.""" + +from __future__ import annotations + +from testing.tests.test_hermes_cli_support import ( + Path, + SimpleNamespace, + json, + lanes, + nullcontext, + sys, +) + + +COMPLETED_RESULT = { + "status": "completed", + "summary": "Work finished with evidence.", + "changed_files": ["src/a.py"], + "tests_run": ["pytest full: passed"], + "artifacts": [], + "findings": [], + "blockers": [], +} + + +class _Connection: + def close(self): + return None + + +def _lane_board(tmp_path: Path, task, comments: list, calls: list) -> SimpleNamespace: + return SimpleNamespace( + scoped_current_board=lambda _board: nullcontext(), + connect=lambda board: _Connection(), + get_task=lambda _conn, _task_id: task, + worker_log_path=lambda _task_id, board: tmp_path / "worker.log", + _resolve_worktree_workspace=lambda _task, board: (tmp_path, f"wt/{task.id}"), + set_branch_name=lambda *_args: None, + set_workspace_path=lambda *_args: None, + build_worker_context=lambda *_args: "Finish the assigned card.", + heartbeat_worker=lambda *_args, **_kwargs: True, + add_comment=lambda _conn, _task_id, _author, body: comments.append(body), + complete_task=lambda *_args, **kwargs: ( + calls.append(("complete", kwargs)) or True + ), + block_task=lambda *_args, **kwargs: calls.append(("block", kwargs)), + ) + + +def _isolate_lane(tmp_path: Path, monkeypatch, board) -> dict[str, Path]: + monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=board)) + monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes") + monkeypatch.setattr(lanes, "DATA_ROOT", tmp_path) + health_paths = { + "codex": tmp_path / "provider-health/codex.json", + "claude": tmp_path / "provider-health/claude.json", + } + monkeypatch.setattr(lanes, "PROVIDER_HEALTH_PATHS", health_paths) + monkeypatch.setattr(lanes, "fetch_quota_snapshot", lambda *_a, **_k: {}) + return health_paths + + +def _route(provider: str, effort: str): + model = "gpt-5.6-sol" if provider == "codex" else "claude-opus-5" + return lanes.Route( + provider, model, effort, f"{provider}-{effort}", "switchyard-manual", "r", 1, () + ) + + +def _run_manual_fail_closed(tmp_path, monkeypatch, assignee, provider, output): + task = SimpleNamespace( + id="t_manual", + status="running", + result=None, + current_run_id=31, + assignee=assignee, + max_runtime_seconds=60, + ) + comments: list = [] + calls: list = [] + board = _lane_board(tmp_path, task, comments, calls) + health_paths = _isolate_lane(tmp_path, monkeypatch, board) + route_calls: list = [] + monkeypatch.setattr( + lanes, + "select_route", + lambda _prompt, lane, **kwargs: ( + route_calls.append((lane, kwargs)) or _route(provider, "high") + ), + ) + runs: list = [] + monkeypatch.setattr( + lanes, + "run_provider", + lambda *_args, **_kwargs: ( + runs.append(provider) or lanes.ProcessResult(1, output, None, True) + ), + ) + + lanes.execute_claim("cassandra", "t_manual") + return calls, comments, route_calls, runs, health_paths + + +def test_manual_codex_assignee_fails_closed_without_provider_switch( + tmp_path: Path, monkeypatch +): + calls, _comments, route_calls, runs, health_paths = _run_manual_fail_closed( + tmp_path, monkeypatch, "cli-codex-high", "codex", "HTTP 429 rate limit hit" + ) + + assert runs == ["codex"] + assert [lane for lane, _ in route_calls] == ["cli-codex-high"] + kind, kwargs = calls[-1] + assert kind == "block" and kwargs["kind"] == "transient" + assert "Manually pinned provider codex" in kwargs["reason"] + assert "cli-codex-high" in kwargs["reason"] + assert "rate-limit" in kwargs["reason"] + health = json.loads(health_paths["codex"].read_text()) + assert health["state"] == "capacity-limited" + assert not health_paths["claude"].exists() + + +def test_manual_claude_assignee_fails_closed_without_provider_switch( + tmp_path: Path, monkeypatch +): + calls, _comments, route_calls, runs, health_paths = _run_manual_fail_closed( + tmp_path, monkeypatch, "cli-claude-low", "claude", "usage limit exhausted" + ) + + assert runs == ["claude"] + assert [lane for lane, _ in route_calls] == ["cli-claude-low"] + kind, kwargs = calls[-1] + assert kind == "block" and kwargs["kind"] == "transient" + assert "Manually pinned provider claude" in kwargs["reason"] + assert "quota" in kwargs["reason"] + health = json.loads(health_paths["claude"].read_text()) + assert health["state"] == "capacity-limited" + assert not health_paths["codex"].exists() + + +def test_auto_failover_preserves_effort_and_records_cooldown( + tmp_path: Path, monkeypatch +): + task = SimpleNamespace( + id="t_auto", + status="running", + result=None, + current_run_id=32, + assignee="cli-auto", + max_runtime_seconds=120, + ) + comments: list = [] + calls: list = [] + board = _lane_board(tmp_path, task, comments, calls) + health_paths = _isolate_lane(tmp_path, monkeypatch, board) + route_calls: list = [] + + def select_route(_prompt, lane, **kwargs): + route_calls.append((lane, kwargs)) + if lane == "cli-auto": + return lanes.Route( + "codex", "gpt-5.6-terra", "xhigh", "codex-xhigh", + "switchyard-classifier", "vote", 1, (), + ) + assert lane == "cli-claude-xhigh" + return _route("claude", "xhigh") + + monkeypatch.setattr(lanes, "select_route", select_route) + fallbacks: list = [] + monkeypatch.setattr( + lanes, + "record_provider_fallback", + lambda source, target, reason: fallbacks.append((source, target, reason)), + ) + reports = [ + lanes.ProcessResult(1, "You have hit your usage limit.", None, True), + lanes.ProcessResult(0, "done", dict(COMPLETED_RESULT), False), + ] + monkeypatch.setattr( + lanes, "run_provider", lambda *_args, **_kwargs: reports.pop(0) + ) + + lanes.execute_claim("cassandra", "t_auto") + + assert reports == [] + assert calls and calls[0][0] == "complete" + assert fallbacks == [("codex", "claude", "quota")] + assert any( + "Provider fallback: codex -> claude after a quota failure" in item + for item in comments + ) + codex_health = json.loads(health_paths["codex"].read_text()) + assert codex_health["state"] == "capacity-limited" + assert codex_health["cooldown_until"] == codex_health["failed_at"] + 300.0 + claude_health = json.loads(health_paths["claude"].read_text()) + assert claude_health["state"] == "available" + + +def test_double_capacity_failure_blocks_transient_with_both_reasons( + tmp_path: Path, monkeypatch +): + task = SimpleNamespace( + id="t_double", + status="running", + result=None, + current_run_id=33, + assignee="cli-auto", + max_runtime_seconds=120, + ) + comments: list = [] + calls: list = [] + board = _lane_board(tmp_path, task, comments, calls) + health_paths = _isolate_lane(tmp_path, monkeypatch, board) + + def select_route(_prompt, lane, **_kwargs): + return ( + _route("codex", "high") if lane == "cli-auto" else _route("claude", "high") + ) + + monkeypatch.setattr(lanes, "select_route", select_route) + decisions: list = [] + monkeypatch.setattr( + lanes, + "record_route_decision", + lambda provider, effort, classifier, outcome: decisions.append( + (provider, effort, outcome) + ), + ) + reports = [ + lanes.ProcessResult(1, "usage limit reached for this window", None, True), + lanes.ProcessResult(1, "invalid oauth token, authentication failed", None, True), + ] + runs: list = [] + monkeypatch.setattr( + lanes, + "run_provider", + lambda *_args, **_kwargs: runs.append(1) or reports.pop(0), + ) + + lanes.execute_claim("cassandra", "t_double") + + assert len(runs) == 2 + kind, kwargs = calls[-1] + assert kind == "block" and kwargs["kind"] == "transient" + assert "Both hosted providers failed" in kwargs["reason"] + assert "codex: quota" in kwargs["reason"] + assert "claude: auth" in kwargs["reason"] + assert "never falls to a metered or local path" in kwargs["reason"] + assert decisions == [ + ("codex", "high", "capacity-failure"), + ("claude", "high", "capacity-failure"), + ] + assert json.loads(health_paths["codex"].read_text())["state"] == "capacity-limited" + claude_health = json.loads(health_paths["claude"].read_text()) + assert claude_health["state"] == "unavailable" + assert claude_health["authenticated"] is False + + +def test_router_outage_blocks_transient_before_any_provider_run( + tmp_path: Path, monkeypatch +): + task = SimpleNamespace( + id="t_router", + status="running", + result=None, + current_run_id=34, + assignee="cli-auto", + max_runtime_seconds=60, + ) + comments: list = [] + calls: list = [] + board = _lane_board(tmp_path, task, comments, calls) + _isolate_lane(tmp_path, monkeypatch, board) + monkeypatch.setattr( + lanes, + "select_route", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + RuntimeError("Switchyard worker routing failed: refused") + ), + ) + outages: list = [] + monkeypatch.setattr( + lanes, "record_router_selection_failure", lambda: outages.append(1) + ) + monkeypatch.setattr( + lanes, + "run_provider", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("no provider may run without a route") + ), + ) + + lanes.execute_claim("cassandra", "t_router") + + assert outages == [1] + kind, kwargs = calls[-1] + assert kind == "block" and kwargs["kind"] == "transient" + assert "Switchyard route selection is unavailable" in kwargs["reason"] + + +def test_router_outage_during_fallback_selection_blocks_transient( + tmp_path: Path, monkeypatch +): + task = SimpleNamespace( + id="t_router_fb", + status="running", + result=None, + current_run_id=35, + assignee="cli-auto", + max_runtime_seconds=60, + ) + comments: list = [] + calls: list = [] + board = _lane_board(tmp_path, task, comments, calls) + _isolate_lane(tmp_path, monkeypatch, board) + selections: list = [] + + def select_route(_prompt, lane, **_kwargs): + selections.append(lane) + if lane == "cli-auto": + return _route("codex", "medium") + raise RuntimeError("Switchyard worker routing failed: refused") + + monkeypatch.setattr(lanes, "select_route", select_route) + monkeypatch.setattr( + lanes, + "run_provider", + lambda *_args, **_kwargs: lanes.ProcessResult(1, "quota exceeded", None, True), + ) + + lanes.execute_claim("cassandra", "t_router_fb") + + assert selections == ["cli-auto", "cli-claude-medium"] + kind, kwargs = calls[-1] + assert kind == "block" and kwargs["kind"] == "transient" + assert "Switchyard route selection is unavailable" in kwargs["reason"] + + +def test_goal_loop_router_outage_blocks_transient(tmp_path: Path, monkeypatch): + task = SimpleNamespace( + id="t_goal_router", + status="running", + result=None, + current_run_id=36, + assignee="cli-auto", + max_runtime_seconds=300, + goal_mode=True, + goal_max_turns=3, + ) + comments: list = [] + calls: list = [] + board = _lane_board(tmp_path, task, comments, calls) + _isolate_lane(tmp_path, monkeypatch, board) + selections: list = [] + + def select_route(_prompt, lane, **_kwargs): + selections.append(lane) + if len(selections) == 1: + return _route("codex", "high") + raise RuntimeError("Switchyard worker routing failed: refused") + + monkeypatch.setattr(lanes, "select_route", select_route) + monkeypatch.setattr( + lanes, + "run_provider", + lambda *_args, **_kwargs: lanes.ProcessResult( + 0, "turn", dict(COMPLETED_RESULT), False + ), + ) + monkeypatch.setattr( + lanes.cli_lane_goal, + "judge_goal_completion", + lambda *_args, **_kwargs: (False, "verification is still missing"), + ) + continued: list = [] + monkeypatch.setattr( + lanes, + "record_route_decision", + lambda provider, _effort, _classifier, outcome: continued.append( + (provider, outcome) + ), + ) + + lanes.execute_claim("cassandra", "t_goal_router") + + assert selections == ["cli-auto", "cli-auto"] + assert ("codex", "goal-continued") in continued + kind, kwargs = calls[-1] + assert kind == "block" and kwargs["kind"] == "transient" + assert "Switchyard route selection is unavailable" in kwargs["reason"] + + +def test_soft_quota_exclusion_steers_new_auto_work(tmp_path: Path, monkeypatch): + task = SimpleNamespace( + id="t_quota", + status="running", + result=None, + current_run_id=37, + assignee="cli-auto", + max_runtime_seconds=60, + ) + comments: list = [] + calls: list = [] + board = _lane_board(tmp_path, task, comments, calls) + _isolate_lane(tmp_path, monkeypatch, board) + monkeypatch.setattr( + lanes, + "fetch_quota_snapshot", + lambda *_a, **_k: { + "codex": lanes.ProviderQuota(10.0, None), + "claude": lanes.ProviderQuota(60.0, None), + }, + ) + route_calls: list = [] + monkeypatch.setattr( + lanes, + "select_route", + lambda _prompt, lane, **kwargs: ( + route_calls.append((lane, kwargs)) or _route("claude", "high") + ), + ) + monkeypatch.setattr( + lanes, + "run_provider", + lambda *_args, **_kwargs: lanes.ProcessResult( + 0, "done", dict(COMPLETED_RESULT), False + ), + ) + + lanes.execute_claim("cassandra", "t_quota") + + assert calls and calls[0][0] == "complete" + lane, kwargs = route_calls[0] + assert lane == "cli-auto" + assert kwargs["exclude_provider"] == "codex" + assert kwargs["exclude_reason"] == "is below its remaining-quota routing threshold" + assert any("Quota guard: codex" in item for item in comments) + assert any( + "Provider routing guard (quota) excluded codex" in item for item in comments + ) diff --git a/testing/tests/test_hermes_cli_quota_gating.py b/testing/tests/test_hermes_cli_quota_gating.py new file mode 100644 index 00000000..c3ce7fa7 --- /dev/null +++ b/testing/tests/test_hermes_cli_quota_gating.py @@ -0,0 +1,299 @@ +"""Quota-aware soft routing thresholds, parsing, and manifest wiring.""" + +from __future__ import annotations + +from testing.tests.test_hermes_cli_support import ( + HERMES, + Path, + lanes, + yaml, +) + + +EXPORTER_TEXT = """\ +# HELP atlas_ai_quota_remaining_percent Remaining percentage. +# TYPE atlas_ai_quota_remaining_percent gauge +atlas_ai_quota_remaining_percent{limit="codex",provider="openai",window="five_hour"} 37.5 +atlas_ai_quota_remaining_percent{limit="codex",provider="openai",window="seven_day"} 80 +atlas_ai_quota_remaining_percent{limit="bonus-credits",provider="openai",window="five_hour"} 1 +atlas_ai_quota_remaining_percent{limit="claude",provider="anthropic",window="five_hour"} 62 +atlas_ai_quota_remaining_percent{limit="claude",provider="anthropic",window="seven_day"} 40 +atlas_ai_quota_remaining_percent{limit="claude",provider="anthropic",window="seven_day_opus"} 1 +atlas_ai_quota_remaining_percent{limit="other",provider="somebody",window="five_hour"} 2 +atlas_ai_quota_reset_timestamp_seconds{limit="codex",provider="openai",window="five_hour"} 111 +atlas_ai_quota_reset_timestamp_seconds{limit="codex",provider="openai",window="seven_day"} 222 +atlas_ai_quota_reset_timestamp_seconds{limit="claude",provider="anthropic",window="seven_day"} 444 +atlas_ai_quota_fetch_success{provider="openai"} 1 +atlas_ai_quota_fetch_success{provider="anthropic"} 1 +not a metric line at all +""" + + +class _MetricsResponse: + def __init__(self, text: str): + self._payload = text.encode("utf-8") + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def read(self, _limit: int) -> bytes: + return self._payload + + +def test_parse_reduces_each_provider_to_its_binding_account_window(): + snapshot = lanes.parse_quota_metrics(EXPORTER_TEXT) + + assert snapshot["codex"].remaining_percent == 37.5 + assert snapshot["codex"].reset_timestamp == 111 + # The model-specific seven_day_opus window must not veto claude; the + # binding account window is seven_day at 40% with its own reset. + assert snapshot["claude"].remaining_percent == 40 + assert snapshot["claude"].reset_timestamp == 444 + + +def test_parse_drops_provider_after_failed_exporter_fetch(): + text = EXPORTER_TEXT.replace( + 'atlas_ai_quota_fetch_success{provider="openai"} 1', + 'atlas_ai_quota_fetch_success{provider="openai"} 0', + ) + snapshot = lanes.parse_quota_metrics(text) + + assert "codex" not in snapshot + assert snapshot["claude"].remaining_percent == 40 + + +def test_parse_reports_no_reset_when_binding_window_lacks_one(): + text = ( + 'atlas_ai_quota_remaining_percent{limit="claude",provider="anthropic",window="five_hour"} 9\n' + ) + snapshot = lanes.parse_quota_metrics(text) + + assert snapshot == {"claude": lanes.ProviderQuota(9.0, None)} + + +def test_fetch_snapshot_fails_open_on_transport_error(monkeypatch, capsys): + failures = [] + monkeypatch.setattr(lanes, "record_quota_fetch_failure", lambda: failures.append(1)) + + def refused(_url, timeout): + raise OSError("connection refused") + + assert lanes.fetch_quota_snapshot(open_url=refused) == {} + assert failures == [1] + assert "fails open" in capsys.readouterr().err + + +def test_fetch_snapshot_counts_empty_signal_and_parses_live_payload( + monkeypatch, capsys +): + failures = [] + monkeypatch.setattr(lanes, "record_quota_fetch_failure", lambda: failures.append(1)) + + empty = lanes.fetch_quota_snapshot( + open_url=lambda _url, timeout: _MetricsResponse("") + ) + assert empty == {} + assert failures == [1] + assert "no gated windows" in capsys.readouterr().err + + live = lanes.fetch_quota_snapshot( + open_url=lambda _url, timeout: _MetricsResponse(EXPORTER_TEXT) + ) + assert live["codex"].remaining_percent == 37.5 + assert failures == [1] + + +def test_threshold_boundary_is_strictly_below(): + for remaining, expected in ((14.9, "codex"), (15.0, None), (15.1, None)): + snapshot = { + "codex": lanes.ProviderQuota(remaining, None), + "claude": lanes.ProviderQuota(80.0, None), + } + excluded, note = lanes.quota_soft_exclusion(snapshot, 15.0) + assert excluded == expected + if expected is None: + assert note is None + else: + assert "14.9% remaining" in note + assert "threshold 15%" in note + + +def test_both_providers_below_threshold_prefers_more_remaining(): + snapshot = { + "codex": lanes.ProviderQuota(3.0, None), + "claude": lanes.ProviderQuota(9.0, None), + } + excluded, note = lanes.quota_soft_exclusion(snapshot, 15.0) + + assert excluded == "codex" + assert "every provider is below" in note + assert "preferring claude (9% left) over codex (3% left)" in note + + +def test_both_below_tie_excludes_deterministically(): + snapshot = { + "codex": lanes.ProviderQuota(5.0, None), + "claude": lanes.ProviderQuota(5.0, None), + } + excluded, _note = lanes.quota_soft_exclusion(snapshot, 15.0) + + assert excluded == "claude" + + +def test_single_known_provider_below_threshold_is_excluded(): + snapshot = {"claude": lanes.ProviderQuota(2.0, None)} + excluded, note = lanes.quota_soft_exclusion(snapshot, 15.0) + + assert excluded == "claude" + assert "routing new auto work" in note + + +def test_kanban_setting_reads_deployed_config_and_fails_to_default( + tmp_path: Path, monkeypatch +): + monkeypatch.setattr(lanes, "DATA_ROOT", tmp_path) + assert lanes.kanban_setting("provider_quota_min_remaining_percent", 15.0) == 15.0 + + config = tmp_path / "config.yaml" + config.write_text("kanban:\n provider_quota_min_remaining_percent: 22\n") + assert lanes.kanban_setting("provider_quota_min_remaining_percent", 15.0) == 22.0 + + config.write_text("kanban: 5\n") + assert lanes.kanban_setting("provider_quota_min_remaining_percent", 15.0) == 15.0 + + config.write_text("just a string\n") + assert lanes.kanban_setting("provider_quota_min_remaining_percent", 15.0) == 15.0 + + config.write_text("kanban:\n provider_quota_min_remaining_percent: true\n") + assert lanes.kanban_setting("provider_quota_min_remaining_percent", 15.0) == 15.0 + + config.write_text("kanban:\n provider_quota_min_remaining_percent: '20'\n") + assert lanes.kanban_setting("provider_quota_min_remaining_percent", 15.0) == 15.0 + + config.write_text("kanban: [\n") + assert lanes.kanban_setting("provider_quota_min_remaining_percent", 15.0) == 15.0 + + +def test_selection_constraint_never_gates_explicit_lanes(): + def forbidden(): + raise AssertionError("explicit lanes must not read quota state") + + constraint = lanes.selection_constraint("cli-codex-high", snapshot_from=forbidden) + + assert constraint == lanes.SelectionConstraint(None, None, None, ()) + + +def test_selection_constraint_health_exclusion_wins_and_carries_resets(monkeypatch): + observed = {} + + def health(now=None, quota_resets=None): + observed["resets"] = quota_resets + return "codex" + + monkeypatch.setattr(lanes, "fresh_unavailable_provider", health) + snapshot = {"claude": lanes.ProviderQuota(50.0, 1234.0)} + constraint = lanes.selection_constraint( + "cli-auto", snapshot_from=lambda: snapshot + ) + + assert constraint.exclude_provider == "codex" + assert constraint.source == "health" + assert constraint.notes == () + assert observed["resets"] == {"claude": 1234.0} + + +def test_selection_constraint_quota_gate_publishes_gauges( + tmp_path: Path, monkeypatch +): + monkeypatch.setattr(lanes, "DATA_ROOT", tmp_path) + monkeypatch.setattr( + lanes, "fresh_unavailable_provider", lambda *_args, **_kwargs: None + ) + snapshot = { + "codex": lanes.ProviderQuota(10.0, 555.0), + "claude": lanes.ProviderQuota(60.0, None), + } + constraint = lanes.selection_constraint("cli-auto", snapshot_from=lambda: snapshot) + + assert constraint.exclude_provider == "codex" + assert constraint.source == "quota" + assert constraint.exclude_reason == "is below its remaining-quota routing threshold" + assert len(constraint.notes) == 1 and "Quota guard: codex" in constraint.notes[0] + rendered = lanes.METRICS.render().decode("utf-8") + assert 'hermes_cli_quota_remaining_percent{provider="codex"} 10' in rendered + assert 'hermes_cli_quota_reset_timestamp_seconds{provider="codex"} 555' in rendered + assert 'hermes_cli_provider_soft_excluded{provider="codex"} 1' in rendered + assert 'hermes_cli_provider_soft_excluded{provider="claude"} 0' in rendered + + +def test_selection_constraint_fails_open_without_quota_signal( + tmp_path: Path, monkeypatch +): + monkeypatch.setattr(lanes, "DATA_ROOT", tmp_path) + monkeypatch.setattr( + lanes, "fresh_unavailable_provider", lambda *_args, **_kwargs: None + ) + constraint = lanes.selection_constraint("cli-auto", snapshot_from=dict) + + assert constraint == lanes.SelectionConstraint(None, None, None, ()) + + +def _agent_config_kanban() -> dict: + document = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text()) + return yaml.safe_load(document["data"]["config.yaml"])["kanban"] + + +def test_configmap_declares_quota_routing_defaults(): + kanban = _agent_config_kanban() + + assert kanban["provider_quota_min_remaining_percent"] == 15 + assert kanban["provider_capacity_cooldown_seconds"] == 300 + assert kanban["provider_auth_cooldown_seconds"] == 3600 + + +def test_lane_metrics_are_wired_for_scraping(): + deployment = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text()) + containers = deployment["spec"]["template"]["spec"]["containers"] + lane = next(item for item in containers if item["name"] == "cli-lane-runner") + env = {entry["name"]: entry.get("value") for entry in lane["env"]} + + assert {"name": "lane-metrics", "containerPort": 9011, "protocol": "TCP"} in lane[ + "ports" + ] + assert env["HERMES_CLI_LANE_METRICS_PORT"] == "9011" + assert env["HERMES_CLI_QUOTA_METRICS_URL"] == "http://127.0.0.1:9010/metrics" + + services = { + item["metadata"]["name"]: item + for item in yaml.safe_load_all((HERMES / "service.yaml").read_text()) + if item + } + metrics_service = services["hermes-cli-lane-metrics"] + annotations = metrics_service["metadata"]["annotations"] + assert annotations["prometheus.io/scrape"] == "true" + assert annotations["prometheus.io/port"] == "9011" + assert metrics_service["spec"]["selector"] == {"app": "hermes-agent"} + assert metrics_service["spec"]["ports"][0]["targetPort"] == "lane-metrics" + + policies = { + item["metadata"]["name"]: item + for item in yaml.safe_load_all((HERMES / "networkpolicy.yaml").read_text()) + if item + } + agent_ingress = policies["hermes-agent-isolation"]["spec"]["ingress"] + monitoring_ports = [ + port["port"] + for rule in agent_ingress + for port in rule.get("ports", []) + if any( + source.get("namespaceSelector", {}) + .get("matchLabels", {}) + .get("kubernetes.io/metadata.name") + == "monitoring" + for source in rule.get("from", []) + ) + ] + assert 9010 in monitoring_ports and 9011 in monitoring_ports