Compare commits
3 Commits
df75479069
...
6df688adcc
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6df688adcc | ||
|
|
034c8372c7 | ||
|
|
62d8cd984b |
@ -88,6 +88,14 @@ data:
|
||||
auto_decompose: true
|
||||
auto_decompose_per_tick: 2
|
||||
dispatch_stale_timeout_seconds: 14400
|
||||
# Steady-state quota-aware routing for the direct CLI lane: below this
|
||||
# remaining-percent a provider stops receiving NEW cli-auto work while
|
||||
# active work finishes; lane-observed capacity/auth failures exclude a
|
||||
# provider until the cooldown elapses, its quota window resets, or a
|
||||
# fresh success is recorded.
|
||||
provider_quota_min_remaining_percent: 15
|
||||
provider_capacity_cooldown_seconds: 300
|
||||
provider_auth_cooldown_seconds: 3600
|
||||
|
||||
model_catalog:
|
||||
enabled: true
|
||||
|
||||
@ -25,7 +25,7 @@ spec:
|
||||
ai.bstein.dev/execution: Hermes Kanban with durable direct Codex and Claude Code CLI workers
|
||||
ai.bstein.dev/model-policy: Jetson-assisted AUTO routing, low through xhigh, cross-provider fallback
|
||||
ai.bstein.dev/placement: rpi5 preferred; Jetson deferred until state storage is available
|
||||
ai.bstein.dev/config-rev: "20260816-auth-health-quarantine-v6"
|
||||
ai.bstein.dev/config-rev: "20260817-cli-auto-quota-routing-v1"
|
||||
prometheus.io/scrape: "true"
|
||||
prometheus.io/path: /metrics
|
||||
prometheus.io/port: "9010"
|
||||
@ -866,6 +866,8 @@ spec:
|
||||
. /opt/data/.env
|
||||
set +a
|
||||
exec /opt/hermes/.venv/bin/python /opt/coordinator/cli_lane_runner.py
|
||||
ports:
|
||||
- {name: lane-metrics, containerPort: 9011, protocol: TCP}
|
||||
env:
|
||||
- {name: HERMES_HOME, value: /opt/data}
|
||||
- {name: HERMES_AUTH_FILE, value: /runtime-access/hermes-auth.json}
|
||||
@ -876,6 +878,8 @@ spec:
|
||||
- {name: PYTHONPATH, value: /opt/hermes}
|
||||
- {name: HERMES_CLI_LANE_CONCURRENCY, value: "2"}
|
||||
- {name: HERMES_CLI_HEALTH_MAX_AGE_SECONDS, value: "60"}
|
||||
- {name: HERMES_CLI_LANE_METRICS_PORT, value: "9011"}
|
||||
- {name: HERMES_CLI_QUOTA_METRICS_URL, value: "http://127.0.0.1:9010/metrics"}
|
||||
- {name: HERMES_AUTO_ROUTER_PROFILE, value: agent}
|
||||
- {name: PATH, value: /opt/coordinator:/opt/data/tools/bin:/opt/data/home/.local/bin:/opt/hermes/.venv/bin:/usr/local/bin:/usr/bin:/bin}
|
||||
securityContext:
|
||||
|
||||
@ -119,6 +119,7 @@ spec:
|
||||
app: server
|
||||
ports:
|
||||
- {protocol: TCP, port: 9010}
|
||||
- {protocol: TCP, port: 9011}
|
||||
# agent.hermes.bstein.dev is an owner-only engineering workstation. The
|
||||
# browser boundary remains OAuth-protected, while its workers need to reach
|
||||
# every cluster namespace, Atlas LAN service, and hosted provider endpoint.
|
||||
|
||||
@ -11,6 +11,8 @@ from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
import cli_lane_goal
|
||||
|
||||
|
||||
@ -31,6 +33,12 @@ DEFAULT_MAX_RUNTIME = 12 * 60 * 60
|
||||
HEARTBEAT_SECONDS = 20
|
||||
PROVIDER_HEALTH_MAX_AGE_SECONDS = 5 * 60
|
||||
PROVIDER_AUTH_FAILURE_MAX_AGE_SECONDS = 12 * 60 * 60
|
||||
QUOTA_MIN_REMAINING_PERCENT_DEFAULT = 15.0
|
||||
CAPACITY_COOLDOWN_SECONDS_DEFAULT = 300.0
|
||||
AUTH_COOLDOWN_SECONDS_DEFAULT = 3600.0
|
||||
QUOTA_METRICS_URL = os.environ.get(
|
||||
"HERMES_CLI_QUOTA_METRICS_URL", "http://127.0.0.1:9010/metrics"
|
||||
)
|
||||
KANBAN_STORAGE_ATTEMPTS = 5
|
||||
ARTIFACT_GC_INTERVAL_SECONDS = 5 * 60
|
||||
ARTIFACT_RETENTION_AGE_SECONDS = 30 * 24 * 60 * 60
|
||||
@ -194,3 +202,18 @@ class TerminalRecoverySnapshot:
|
||||
|
||||
def utc_now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def kanban_setting(name: str, default: float) -> float:
|
||||
"""Read one numeric routing setting from the deployed kanban config block."""
|
||||
try:
|
||||
document = yaml.safe_load(
|
||||
(DATA_ROOT / "config.yaml").read_text(encoding="utf-8")
|
||||
)
|
||||
except (OSError, yaml.YAMLError):
|
||||
return default
|
||||
kanban = document.get("kanban") if isinstance(document, dict) else None
|
||||
value = kanban.get(name) if isinstance(kanban, dict) else None
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
return default
|
||||
return float(value)
|
||||
|
||||
@ -24,6 +24,7 @@ from cli_lane_config import (
|
||||
)
|
||||
from cli_lane_execution import execute_claim
|
||||
from cli_lane_files import atomic_json
|
||||
from cli_lane_metrics import start_metrics_server
|
||||
from cli_lane_recovery import _has_pending_finalization, recover_pending_finalizations
|
||||
from cli_lane_retention import maybe_gc_lane_artifacts
|
||||
|
||||
@ -155,6 +156,7 @@ def main() -> int:
|
||||
|
||||
RESULT_SCHEMA_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
atomic_json(RESULT_SCHEMA_PATH, RESULT_SCHEMA, 0o644)
|
||||
start_metrics_server()
|
||||
capabilities = initialize_kanban_capabilities(kanban_db)
|
||||
recover_orphans()
|
||||
workers = max(1, min(int(os.environ.get("HERMES_CLI_LANE_CONCURRENCY", "4")), 8))
|
||||
|
||||
@ -15,6 +15,7 @@ from cli_lane_config import (
|
||||
DEFAULT_MAX_RUNTIME,
|
||||
TerminalFinalizationPending,
|
||||
)
|
||||
from cli_lane_failover import _routed_or_blocked, capacity_failover
|
||||
from cli_lane_files import (
|
||||
_terminal_identity,
|
||||
_terminal_path,
|
||||
@ -23,8 +24,11 @@ from cli_lane_files import (
|
||||
state_path,
|
||||
)
|
||||
from cli_lane_finalization import _finalize_terminal_record, _recover_exact_run
|
||||
from cli_lane_health import record_provider_success
|
||||
from cli_lane_metrics import record_route_decision
|
||||
from cli_lane_prompt import build_prompt, git_handoff, workspace_artifacts
|
||||
from cli_lane_provider import run_provider
|
||||
from cli_lane_quota import selection_constraint
|
||||
from cli_lane_records import _persist_candidate, _write_terminal_record
|
||||
from cli_lane_recovery import _has_pending_finalization
|
||||
from cli_lane_routing import fresh_unavailable_provider, select_route
|
||||
@ -115,21 +119,30 @@ def execute_claim(board: str, task_id: str) -> None:
|
||||
|
||||
try:
|
||||
previous_route = state.get("current_route")
|
||||
excluded_provider = (
|
||||
fresh_unavailable_provider() if assignee == "cli-auto" else None
|
||||
constraint = selection_constraint(assignee)
|
||||
unavailable_provider = (
|
||||
constraint.exclude_provider if constraint.source == "health" else None
|
||||
)
|
||||
unavailable_provider = excluded_provider
|
||||
route = select_route(
|
||||
context,
|
||||
assignee,
|
||||
exclude_provider=excluded_provider,
|
||||
exclude_reason="is unavailable according to fresh native health"
|
||||
if excluded_provider
|
||||
else None,
|
||||
route = _routed_or_blocked(
|
||||
kanban_db,
|
||||
board,
|
||||
task_id,
|
||||
run_id,
|
||||
lambda: select_route(
|
||||
context,
|
||||
assignee,
|
||||
exclude_provider=constraint.exclude_provider,
|
||||
exclude_reason=constraint.exclude_reason,
|
||||
),
|
||||
)
|
||||
if excluded_provider:
|
||||
if route is None:
|
||||
return
|
||||
for note in constraint.notes:
|
||||
comment(note)
|
||||
if constraint.exclude_provider:
|
||||
comment(
|
||||
f"Provider health guard excluded {excluded_provider} before automatic routing.",
|
||||
f"Provider routing guard ({constraint.source}) excluded "
|
||||
f"{constraint.exclude_provider} before automatic routing.",
|
||||
)
|
||||
comment(
|
||||
f"CLI route: {route.provider}/{route.model} at {route.effort}; classifier={route.classifier}; {route.reason}",
|
||||
@ -181,44 +194,34 @@ def execute_claim(board: str, task_id: str) -> None:
|
||||
goal_turn=goal_turn,
|
||||
)
|
||||
if result.capacity_failure:
|
||||
unavailable_provider = route.provider
|
||||
retry_context = (
|
||||
context
|
||||
+ "\n\nRouting boundary: the first provider failed from capacity/authentication. "
|
||||
+ "Select the alternate hosted provider at an appropriate effort."
|
||||
boundary = capacity_failover(
|
||||
kanban_db,
|
||||
board=board,
|
||||
task_id=task_id,
|
||||
run_id=run_id,
|
||||
assignee=assignee,
|
||||
comment=comment,
|
||||
context=context,
|
||||
workspace=workspace,
|
||||
state=state,
|
||||
state_file=state_file,
|
||||
log_path=log_path,
|
||||
heartbeat=heartbeat,
|
||||
deadline=deadline,
|
||||
route=route,
|
||||
result=result,
|
||||
candidate_file=candidate_file,
|
||||
goal_turn=goal_turn,
|
||||
)
|
||||
alternate = "claude" if route.provider == "codex" else "codex"
|
||||
fallback = select_route(
|
||||
retry_context,
|
||||
f"cli-{alternate}-{route.effort}",
|
||||
)
|
||||
comment(
|
||||
f"Provider fallback: {route.provider} -> {fallback.provider}; Jetson reclassified the retry boundary.",
|
||||
)
|
||||
result = run_provider(
|
||||
fallback,
|
||||
build_prompt(
|
||||
context,
|
||||
workspace,
|
||||
git_handoff(workspace, result.output),
|
||||
),
|
||||
workspace,
|
||||
state,
|
||||
state_file,
|
||||
log_path,
|
||||
heartbeat,
|
||||
max(1, int(deadline - time.monotonic())),
|
||||
)
|
||||
route = fallback
|
||||
if result.structured:
|
||||
candidate_file = _persist_candidate(
|
||||
state,
|
||||
state_file,
|
||||
dict(result.structured),
|
||||
route=route,
|
||||
returncode=result.returncode,
|
||||
goal_turn=goal_turn,
|
||||
)
|
||||
route = boundary.route
|
||||
result = boundary.result
|
||||
candidate_file = boundary.candidate_file
|
||||
if boundary.unavailable_provider is not None:
|
||||
unavailable_provider = boundary.unavailable_provider
|
||||
if boundary.blocked:
|
||||
break
|
||||
if result.returncode == 0:
|
||||
record_provider_success(route.provider)
|
||||
structured = result.structured
|
||||
if structured:
|
||||
structured = dict(structured)
|
||||
@ -329,6 +332,9 @@ def execute_claim(board: str, task_id: str) -> None:
|
||||
"accepted worker result is durably journaled but "
|
||||
f"Kanban finalization is {outcome}"
|
||||
)
|
||||
record_route_decision(
|
||||
route.provider, route.effort, route.classifier, "completed"
|
||||
)
|
||||
break
|
||||
|
||||
can_continue = (
|
||||
@ -358,14 +364,25 @@ def execute_claim(board: str, task_id: str) -> None:
|
||||
excluded = unavailable_provider
|
||||
if excluded is None and assignee == "cli-auto":
|
||||
excluded = fresh_unavailable_provider()
|
||||
next_route = select_route(
|
||||
escalation_context,
|
||||
assignee,
|
||||
exclude_provider=excluded,
|
||||
exclude_reason="is unavailable according to fresh native health"
|
||||
if excluded
|
||||
else None,
|
||||
record_route_decision(
|
||||
route.provider, route.effort, route.classifier, "goal-continued"
|
||||
)
|
||||
next_route = _routed_or_blocked(
|
||||
kanban_db,
|
||||
board,
|
||||
task_id,
|
||||
run_id,
|
||||
lambda boundary=escalation_context, blocked=excluded: select_route(
|
||||
boundary,
|
||||
assignee,
|
||||
exclude_provider=blocked,
|
||||
exclude_reason="is unavailable according to fresh native health"
|
||||
if blocked
|
||||
else None,
|
||||
),
|
||||
)
|
||||
if next_route is None:
|
||||
break
|
||||
comment(
|
||||
f"Goal route {goal_turn}/{goal_max_turns}: "
|
||||
f"{next_route.provider}/{next_route.model} at {next_route.effort}; "
|
||||
@ -400,6 +417,9 @@ def execute_claim(board: str, task_id: str) -> None:
|
||||
failure_kind = (
|
||||
"transient" if result.capacity_failure else "capability"
|
||||
)
|
||||
record_route_decision(
|
||||
route.provider, route.effort, route.classifier, f"blocked-{failure_kind}"
|
||||
)
|
||||
_board_call(
|
||||
kanban_db,
|
||||
board,
|
||||
|
||||
181
services/hermes/scripts/cli_lane_failover.py
Normal file
181
services/hermes/scripts/cli_lane_failover.py
Normal file
@ -0,0 +1,181 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Bounded, fail-closed cross-provider failover for one capacity boundary."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
from cli_lane_board import _board_call
|
||||
from cli_lane_config import ProcessResult, Route
|
||||
from cli_lane_health import classify_capacity_failure, record_provider_failure
|
||||
from cli_lane_metrics import (
|
||||
record_provider_fallback,
|
||||
record_route_decision,
|
||||
record_router_selection_failure,
|
||||
)
|
||||
from cli_lane_prompt import build_prompt, git_handoff
|
||||
from cli_lane_provider import run_provider
|
||||
from cli_lane_records import _persist_candidate
|
||||
from cli_lane_routing import select_route
|
||||
|
||||
|
||||
@dataclass
|
||||
class FailoverOutcome:
|
||||
"""Where one capacity boundary left the claim's route and result."""
|
||||
|
||||
route: Route
|
||||
result: ProcessResult
|
||||
candidate_file: Path | None
|
||||
unavailable_provider: str | None
|
||||
blocked: bool
|
||||
|
||||
|
||||
def _block_transient(
|
||||
kanban_db: Any, board: str, task_id: str, run_id: Any, reason: str
|
||||
) -> None:
|
||||
"""Park one card as retryable so it re-enters when providers recover."""
|
||||
_board_call(
|
||||
kanban_db,
|
||||
board,
|
||||
lambda fresh: kanban_db.block_task(
|
||||
fresh,
|
||||
task_id,
|
||||
reason=reason,
|
||||
kind="transient",
|
||||
expected_run_id=run_id,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _routed_or_blocked(
|
||||
kanban_db: Any,
|
||||
board: str,
|
||||
task_id: str,
|
||||
run_id: Any,
|
||||
select: Callable[[], Route],
|
||||
) -> Route | None:
|
||||
"""Turn a Switchyard outage into a transient block, never a capability one."""
|
||||
try:
|
||||
return select()
|
||||
except RuntimeError as error:
|
||||
record_router_selection_failure()
|
||||
_block_transient(
|
||||
kanban_db,
|
||||
board,
|
||||
task_id,
|
||||
run_id,
|
||||
"Switchyard route selection is unavailable; the card retries "
|
||||
f"when the router recovers: {error}",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def capacity_failover(
|
||||
kanban_db: Any,
|
||||
*,
|
||||
board: str,
|
||||
task_id: str,
|
||||
run_id: Any,
|
||||
assignee: str,
|
||||
comment: Callable[[str], None],
|
||||
context: str,
|
||||
workspace: Path,
|
||||
state: dict[str, Any],
|
||||
state_file: Path,
|
||||
log_path: Path,
|
||||
heartbeat: Callable[[str], bool],
|
||||
deadline: float,
|
||||
route: Route,
|
||||
result: ProcessResult,
|
||||
candidate_file: Path | None,
|
||||
goal_turn: int,
|
||||
) -> FailoverOutcome:
|
||||
"""Apply the capacity policy once: record, fail closed, or fail over."""
|
||||
failure_class = classify_capacity_failure(result.output)
|
||||
record_provider_failure(route.provider, failure_class)
|
||||
record_route_decision(
|
||||
route.provider, route.effort, route.classifier, "capacity-failure"
|
||||
)
|
||||
if assignee != "cli-auto":
|
||||
# Fail closed: an operator pinned this provider, so a capacity or
|
||||
# auth failure must wait for that provider, never switch away.
|
||||
_block_transient(
|
||||
kanban_db,
|
||||
board,
|
||||
task_id,
|
||||
run_id,
|
||||
f"Manually pinned provider {route.provider} hit a "
|
||||
f"{failure_class} failure; the explicit {assignee} lane never "
|
||||
"switches providers. Retry after recovery or reassign to cli-auto.",
|
||||
)
|
||||
return FailoverOutcome(route, result, candidate_file, None, True)
|
||||
retry_context = (
|
||||
context
|
||||
+ "\n\nRouting boundary: the first provider failed from capacity/authentication. "
|
||||
+ "Select the alternate hosted provider at an appropriate effort."
|
||||
)
|
||||
alternate = "claude" if route.provider == "codex" else "codex"
|
||||
fallback = _routed_or_blocked(
|
||||
kanban_db,
|
||||
board,
|
||||
task_id,
|
||||
run_id,
|
||||
lambda: select_route(retry_context, f"cli-{alternate}-{route.effort}"),
|
||||
)
|
||||
if fallback is None:
|
||||
return FailoverOutcome(route, result, candidate_file, route.provider, True)
|
||||
record_provider_fallback(route.provider, fallback.provider, failure_class)
|
||||
comment(
|
||||
f"Provider fallback: {route.provider} -> {fallback.provider} "
|
||||
f"after a {failure_class} failure; Jetson reclassified the retry boundary.",
|
||||
)
|
||||
fallback_result = run_provider(
|
||||
fallback,
|
||||
build_prompt(
|
||||
context,
|
||||
workspace,
|
||||
git_handoff(workspace, result.output),
|
||||
),
|
||||
workspace,
|
||||
state,
|
||||
state_file,
|
||||
log_path,
|
||||
heartbeat,
|
||||
max(1, int(deadline - time.monotonic())),
|
||||
)
|
||||
if fallback_result.capacity_failure:
|
||||
# Bounded failover depth: two hosted providers, never a third
|
||||
# metered or local path.
|
||||
fallback_class = classify_capacity_failure(fallback_result.output)
|
||||
record_provider_failure(fallback.provider, fallback_class)
|
||||
record_route_decision(
|
||||
fallback.provider, fallback.effort, fallback.classifier, "capacity-failure"
|
||||
)
|
||||
_block_transient(
|
||||
kanban_db,
|
||||
board,
|
||||
task_id,
|
||||
run_id,
|
||||
"Both hosted providers failed at this boundary "
|
||||
f"({route.provider}: {failure_class}; {fallback.provider}: "
|
||||
f"{fallback_class}); the lane never falls to a metered or "
|
||||
"local path. Waiting for provider recovery.",
|
||||
)
|
||||
return FailoverOutcome(
|
||||
fallback, fallback_result, candidate_file, route.provider, True
|
||||
)
|
||||
if fallback_result.structured:
|
||||
candidate_file = _persist_candidate(
|
||||
state,
|
||||
state_file,
|
||||
dict(fallback_result.structured),
|
||||
route=fallback,
|
||||
returncode=fallback_result.returncode,
|
||||
goal_turn=goal_turn,
|
||||
)
|
||||
return FailoverOutcome(
|
||||
fallback, fallback_result, candidate_file, route.provider, False
|
||||
)
|
||||
102
services/hermes/scripts/cli_lane_health.py
Normal file
102
services/hermes/scripts/cli_lane_health.py
Normal file
@ -0,0 +1,102 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Lane-observed provider health with failure classification and cooldowns."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from cli_lane_config import (
|
||||
AUTH_COOLDOWN_SECONDS_DEFAULT,
|
||||
CAPACITY_COOLDOWN_SECONDS_DEFAULT,
|
||||
PROVIDER_HEALTH_PATHS,
|
||||
kanban_setting,
|
||||
utc_now,
|
||||
)
|
||||
from cli_lane_files import load_json
|
||||
|
||||
|
||||
# Ordered from most to least specific so one output maps to one actionable
|
||||
# reason instead of the single conflated capacity signal.
|
||||
FAILURE_CLASSIFICATIONS = (
|
||||
("auth", re.compile(r"authentication|unauthorized|forbidden|oauth|token.*expired|401|403", re.I)),
|
||||
("rate-limit", re.compile(r"rate.?limit|429|529|overload", re.I)),
|
||||
("quota", re.compile(r"usage.?limit|quota|credit|exhaust|capacity", re.I)),
|
||||
)
|
||||
|
||||
|
||||
def classify_capacity_failure(output: str) -> str:
|
||||
"""Split the broad capacity regex into quota/auth/rate-limit/transport."""
|
||||
for reason, pattern in FAILURE_CLASSIFICATIONS:
|
||||
if pattern.search(output):
|
||||
return reason
|
||||
return "transport"
|
||||
|
||||
|
||||
def _write_provider_health(
|
||||
provider: str, updates: dict[str, Any], clear: tuple[str, ...] = ()
|
||||
) -> None:
|
||||
"""Merge one lane observation into the shared provider health snapshot."""
|
||||
path = PROVIDER_HEALTH_PATHS[provider]
|
||||
value = load_json(path)
|
||||
for key in clear:
|
||||
value.pop(key, None)
|
||||
value.update(updates)
|
||||
try:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
|
||||
temporary.write_text(
|
||||
json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
||||
)
|
||||
os.replace(temporary, path)
|
||||
except OSError:
|
||||
# Health snapshots are advisory; never fail the lane over one write.
|
||||
pass
|
||||
|
||||
|
||||
def record_provider_success(provider: str) -> None:
|
||||
"""Publish fresh evidence of success so the provider re-enters routing."""
|
||||
_write_provider_health(
|
||||
provider,
|
||||
{
|
||||
"source": "cli-lane-runner",
|
||||
"state": "available",
|
||||
"authenticated": True,
|
||||
"checked_at": utc_now(),
|
||||
},
|
||||
clear=("failure_reason", "failed_at", "cooldown_until"),
|
||||
)
|
||||
|
||||
|
||||
def record_provider_failure(
|
||||
provider: str, reason: str, now: float | None = None
|
||||
) -> float:
|
||||
"""Persist one lane-observed failure with its re-admission cooldown."""
|
||||
current = time.time() if now is None else now
|
||||
if reason == "auth":
|
||||
cooldown = kanban_setting(
|
||||
"provider_auth_cooldown_seconds", AUTH_COOLDOWN_SECONDS_DEFAULT
|
||||
)
|
||||
state, authenticated = "unavailable", False
|
||||
else:
|
||||
cooldown = kanban_setting(
|
||||
"provider_capacity_cooldown_seconds", CAPACITY_COOLDOWN_SECONDS_DEFAULT
|
||||
)
|
||||
state, authenticated = "capacity-limited", True
|
||||
cooldown_until = current + max(0.0, cooldown)
|
||||
_write_provider_health(
|
||||
provider,
|
||||
{
|
||||
"source": "cli-lane-runner",
|
||||
"state": state,
|
||||
"authenticated": authenticated,
|
||||
"checked_at": utc_now(),
|
||||
"failure_reason": reason,
|
||||
"failed_at": current,
|
||||
"cooldown_until": cooldown_until,
|
||||
},
|
||||
)
|
||||
return cooldown_until
|
||||
188
services/hermes/scripts/cli_lane_metrics.py
Normal file
188
services/hermes/scripts/cli_lane_metrics.py
Normal file
@ -0,0 +1,188 @@
|
||||
#!/usr/bin/env python3
|
||||
"""In-process Prometheus metrics for the direct CLI lane's routing decisions."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
|
||||
|
||||
METRIC_HELP = {
|
||||
"hermes_cli_route_decisions_total": (
|
||||
"Terminal outcome of each executed CLI lane route decision."
|
||||
),
|
||||
"hermes_cli_provider_fallbacks_total": (
|
||||
"Cross-provider failovers taken by the CLI lane, by failure reason."
|
||||
),
|
||||
"hermes_cli_router_selection_failures_total": (
|
||||
"Switchyard route selections that failed and blocked a card as transient."
|
||||
),
|
||||
"hermes_cli_quota_fetch_failures_total": (
|
||||
"Quota snapshot reads that failed; routing then fails open."
|
||||
),
|
||||
"hermes_cli_quota_remaining_percent": (
|
||||
"Binding remaining-quota percent per provider as seen by the CLI lane."
|
||||
),
|
||||
"hermes_cli_quota_reset_timestamp_seconds": (
|
||||
"Unix reset time of the binding quota window as seen by the CLI lane."
|
||||
),
|
||||
"hermes_cli_provider_soft_excluded": (
|
||||
"Whether new automatic work is currently routed away from a provider."
|
||||
),
|
||||
}
|
||||
COUNTER_NAMES = frozenset(name for name in METRIC_HELP if name.endswith("_total"))
|
||||
|
||||
|
||||
def _escape_label(value: str) -> str:
|
||||
"""Escape one Prometheus label value."""
|
||||
return value.replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"')
|
||||
|
||||
|
||||
class LaneMetricsRegistry:
|
||||
"""Thread-safe counter and gauge store rendered in Prometheus text format."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._lock = threading.Lock()
|
||||
self._values: dict[tuple[str, tuple[tuple[str, str], ...]], float] = {}
|
||||
|
||||
def _key(
|
||||
self, name: str, labels: dict[str, str] | None
|
||||
) -> tuple[str, tuple[tuple[str, str], ...]]:
|
||||
return name, tuple(sorted((labels or {}).items()))
|
||||
|
||||
def increment(
|
||||
self, name: str, labels: dict[str, str] | None = None, amount: float = 1.0
|
||||
) -> None:
|
||||
key = self._key(name, labels)
|
||||
with self._lock:
|
||||
self._values[key] = self._values.get(key, 0.0) + amount
|
||||
|
||||
def set_value(
|
||||
self, name: str, labels: dict[str, str] | None, value: float
|
||||
) -> None:
|
||||
with self._lock:
|
||||
self._values[self._key(name, labels)] = value
|
||||
|
||||
def render(self) -> bytes:
|
||||
"""Render every recorded sample with stable ordering."""
|
||||
with self._lock:
|
||||
values = dict(self._values)
|
||||
lines: list[str] = []
|
||||
for name in sorted({name for name, _ in values}):
|
||||
kind = "counter" if name in COUNTER_NAMES else "gauge"
|
||||
lines.extend((f"# HELP {name} {METRIC_HELP[name]}", f"# TYPE {name} {kind}"))
|
||||
for (sample_name, labels), value in sorted(values.items()):
|
||||
if sample_name != name:
|
||||
continue
|
||||
rendered = ",".join(
|
||||
f'{key}="{_escape_label(text)}"' for key, text in labels
|
||||
)
|
||||
body = f"{{{rendered}}}" if rendered else ""
|
||||
lines.append(f"{name}{body} {value:.12g}")
|
||||
return ("\n".join(lines) + "\n").encode("utf-8")
|
||||
|
||||
|
||||
METRICS = LaneMetricsRegistry()
|
||||
|
||||
|
||||
def record_route_decision(
|
||||
provider: str, effort: str, classifier: str, outcome: str
|
||||
) -> None:
|
||||
"""Count one executed route with its terminal outcome."""
|
||||
METRICS.increment(
|
||||
"hermes_cli_route_decisions_total",
|
||||
{
|
||||
"provider": provider,
|
||||
"effort": effort,
|
||||
"classifier": classifier,
|
||||
"outcome": outcome,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def record_provider_fallback(
|
||||
from_provider: str, to_provider: str, reason: str
|
||||
) -> None:
|
||||
"""Count one cross-provider failover with its classified reason."""
|
||||
METRICS.increment(
|
||||
"hermes_cli_provider_fallbacks_total",
|
||||
{
|
||||
"from_provider": from_provider,
|
||||
"to_provider": to_provider,
|
||||
"reason": reason,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def record_router_selection_failure() -> None:
|
||||
"""Count one Switchyard outage observed at a selection boundary."""
|
||||
METRICS.increment("hermes_cli_router_selection_failures_total")
|
||||
|
||||
|
||||
def record_quota_fetch_failure() -> None:
|
||||
"""Count one unavailable quota snapshot; routing continues fail-open."""
|
||||
METRICS.increment("hermes_cli_quota_fetch_failures_total")
|
||||
|
||||
|
||||
def record_provider_quota(
|
||||
provider: str, remaining_percent: float, reset_timestamp: float | None
|
||||
) -> None:
|
||||
"""Publish the binding quota window the lane based its routing on."""
|
||||
labels = {"provider": provider}
|
||||
METRICS.set_value(
|
||||
"hermes_cli_quota_remaining_percent", labels, remaining_percent
|
||||
)
|
||||
if reset_timestamp is not None:
|
||||
METRICS.set_value(
|
||||
"hermes_cli_quota_reset_timestamp_seconds", labels, reset_timestamp
|
||||
)
|
||||
|
||||
|
||||
def record_soft_exclusion(provider: str, excluded: bool) -> None:
|
||||
"""Publish whether new automatic work avoids this provider right now."""
|
||||
METRICS.set_value(
|
||||
"hermes_cli_provider_soft_excluded",
|
||||
{"provider": provider},
|
||||
1.0 if excluded else 0.0,
|
||||
)
|
||||
|
||||
|
||||
def start_metrics_server(port: int | None = None) -> ThreadingHTTPServer | None:
|
||||
"""Serve lane metrics on the pod-local scrape port without blocking dispatch."""
|
||||
selected_port = (
|
||||
int(os.environ.get("HERMES_CLI_LANE_METRICS_PORT", "9011"))
|
||||
if port is None
|
||||
else port
|
||||
)
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def do_GET(self) -> None: # noqa: N802
|
||||
if self.path == "/metrics":
|
||||
payload = METRICS.render()
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/plain; version=0.0.4")
|
||||
self.send_header("Content-Length", str(len(payload)))
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
elif self.path == "/healthz":
|
||||
self.send_response(200)
|
||||
self.end_headers()
|
||||
else:
|
||||
self.send_error(404)
|
||||
|
||||
def log_message(self, _format: str, *_args: object) -> None:
|
||||
return
|
||||
|
||||
try:
|
||||
server = ThreadingHTTPServer(("0.0.0.0", selected_port), Handler)
|
||||
except OSError as error:
|
||||
print(
|
||||
f"cli lane metrics server unavailable: {error}",
|
||||
file=sys.stderr,
|
||||
flush=True,
|
||||
)
|
||||
return None
|
||||
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||||
return server
|
||||
@ -197,6 +197,10 @@ def _terminate_worker_process(
|
||||
def _base_env() -> dict[str, str]:
|
||||
"""Build a worker environment while preserving Vault-backed CLI homes."""
|
||||
env = os.environ.copy()
|
||||
# Subscription CLIs only: a vendor API key in the environment could
|
||||
# silently flip a worker onto a metered path, so drop every key here.
|
||||
for secret in ("ANTHROPIC_API_KEY", "CLAUDE_API_KEY", "OPENAI_API_KEY", "API_SERVER_KEY"):
|
||||
env.pop(secret, None)
|
||||
env.update(
|
||||
{
|
||||
"HOME": str(DATA_ROOT / "home"),
|
||||
|
||||
205
services/hermes/scripts/cli_lane_quota.py
Normal file
205
services/hermes/scripts/cli_lane_quota.py
Normal file
@ -0,0 +1,205 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Quota-aware soft routing for automatic CLI lane provider selection."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import sys
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable
|
||||
|
||||
from cli_lane_config import (
|
||||
QUOTA_METRICS_URL,
|
||||
QUOTA_MIN_REMAINING_PERCENT_DEFAULT,
|
||||
kanban_setting,
|
||||
)
|
||||
from cli_lane_metrics import (
|
||||
record_provider_quota,
|
||||
record_quota_fetch_failure,
|
||||
record_soft_exclusion,
|
||||
)
|
||||
from cli_lane_routing import fresh_unavailable_provider
|
||||
|
||||
|
||||
# Exporter accounts are labeled by vendor; the lane routes by CLI provider.
|
||||
QUOTA_PROVIDER_LABELS = {"openai": "codex", "anthropic": "claude"}
|
||||
# Gate only on account-wide windows: a model-specific Claude window (for
|
||||
# example seven_day_opus) must not veto the whole provider.
|
||||
GATED_QUOTA_WINDOWS = {
|
||||
"codex": None,
|
||||
"claude": ("five_hour", "seven_day"),
|
||||
}
|
||||
_SAMPLE_PATTERN = re.compile(
|
||||
r"^atlas_ai_quota_(remaining_percent|reset_timestamp_seconds|fetch_success)"
|
||||
r"\{([^}]*)\} (-?[0-9.eE+]+)$"
|
||||
)
|
||||
_LABEL_PATTERN = re.compile(r'([a-zA-Z_][a-zA-Z0-9_]*)="([^"]*)"')
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProviderQuota:
|
||||
"""One provider's binding remaining quota as seen by the lane."""
|
||||
|
||||
remaining_percent: float
|
||||
reset_timestamp: float | None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SelectionConstraint:
|
||||
"""One provider exclusion decided at a routing boundary, with evidence."""
|
||||
|
||||
exclude_provider: str | None
|
||||
exclude_reason: str | None
|
||||
source: str | None
|
||||
notes: tuple[str, ...]
|
||||
|
||||
|
||||
def _gated_window(provider: str, labels: dict[str, str]) -> str | None:
|
||||
"""Return the window key when a sample belongs to a gated account limit."""
|
||||
if labels.get("limit") != provider:
|
||||
return None
|
||||
window = labels.get("window", "")
|
||||
allowed = GATED_QUOTA_WINDOWS[provider]
|
||||
if allowed is not None and window not in allowed:
|
||||
return None
|
||||
return window
|
||||
|
||||
|
||||
def parse_quota_metrics(text: str) -> dict[str, ProviderQuota]:
|
||||
"""Reduce exporter samples to each provider's tightest account window."""
|
||||
remaining: dict[tuple[str, str], float] = {}
|
||||
resets: dict[tuple[str, str], float] = {}
|
||||
fetch_success: dict[str, bool] = {}
|
||||
for line in text.splitlines():
|
||||
match = _SAMPLE_PATTERN.match(line.strip())
|
||||
if not match:
|
||||
continue
|
||||
metric, raw_labels, raw_value = match.groups()
|
||||
labels = dict(_LABEL_PATTERN.findall(raw_labels))
|
||||
provider = QUOTA_PROVIDER_LABELS.get(labels.get("provider", ""))
|
||||
if provider is None:
|
||||
continue
|
||||
value = float(raw_value)
|
||||
if metric == "fetch_success":
|
||||
fetch_success[provider] = value >= 1.0
|
||||
continue
|
||||
window = _gated_window(provider, labels)
|
||||
if window is None:
|
||||
continue
|
||||
if metric == "remaining_percent":
|
||||
remaining[(provider, window)] = value
|
||||
else:
|
||||
resets[(provider, window)] = value
|
||||
snapshot: dict[str, ProviderQuota] = {}
|
||||
for provider in QUOTA_PROVIDER_LABELS.values():
|
||||
windows = {
|
||||
window: value
|
||||
for (candidate, window), value in remaining.items()
|
||||
if candidate == provider
|
||||
}
|
||||
if not windows or not fetch_success.get(provider, True):
|
||||
# Failed or stale exporter fetches leave this provider ungated.
|
||||
continue
|
||||
binding = min(windows, key=lambda window: (windows[window], window))
|
||||
snapshot[provider] = ProviderQuota(
|
||||
remaining_percent=windows[binding],
|
||||
reset_timestamp=resets.get((provider, binding)),
|
||||
)
|
||||
return snapshot
|
||||
|
||||
|
||||
def fetch_quota_snapshot(
|
||||
url: str = QUOTA_METRICS_URL,
|
||||
open_url: Callable = urllib.request.urlopen,
|
||||
) -> dict[str, ProviderQuota]:
|
||||
"""Read the in-pod usage exporter; empty means the signal is unavailable."""
|
||||
try:
|
||||
with open_url(url, timeout=10) as response:
|
||||
text = response.read(1 << 20).decode("utf-8", errors="replace")
|
||||
snapshot = parse_quota_metrics(text)
|
||||
except (OSError, ValueError) as error:
|
||||
record_quota_fetch_failure()
|
||||
print(
|
||||
f"quota snapshot unavailable; routing fails open: {type(error).__name__}",
|
||||
file=sys.stderr,
|
||||
flush=True,
|
||||
)
|
||||
return {}
|
||||
if not snapshot:
|
||||
record_quota_fetch_failure()
|
||||
print(
|
||||
"quota snapshot carried no gated windows; routing fails open",
|
||||
file=sys.stderr,
|
||||
flush=True,
|
||||
)
|
||||
return snapshot
|
||||
|
||||
|
||||
def quota_soft_exclusion(
|
||||
snapshot: dict[str, ProviderQuota], threshold: float
|
||||
) -> tuple[str | None, str | None]:
|
||||
"""Choose at most one provider to soft-exclude from new automatic work."""
|
||||
below = {
|
||||
provider: quota.remaining_percent
|
||||
for provider, quota in snapshot.items()
|
||||
if quota.remaining_percent < threshold
|
||||
}
|
||||
if not below:
|
||||
return None, None
|
||||
excluded = min(below, key=lambda provider: (below[provider], provider))
|
||||
if len(below) == len(snapshot) and len(snapshot) > 1:
|
||||
preferred = next(name for name in snapshot if name != excluded)
|
||||
note = (
|
||||
"Quota degraded: every provider is below the "
|
||||
f"{threshold:.6g}% remaining threshold; preferring {preferred} "
|
||||
f"({snapshot[preferred].remaining_percent:.6g}% left) over "
|
||||
f"{excluded} ({below[excluded]:.6g}% left) for new auto work."
|
||||
)
|
||||
return excluded, note
|
||||
note = (
|
||||
f"Quota guard: {excluded} has {below[excluded]:.6g}% remaining "
|
||||
f"(threshold {threshold:.6g}%); routing new auto work to the other "
|
||||
"provider while active work finishes."
|
||||
)
|
||||
return excluded, note
|
||||
|
||||
|
||||
def selection_constraint(
|
||||
assignee: str,
|
||||
*,
|
||||
snapshot_from: Callable[[], dict[str, ProviderQuota]] | None = None,
|
||||
now: float | None = None,
|
||||
) -> SelectionConstraint:
|
||||
"""Combine provider health and quota gating for one selection boundary."""
|
||||
if assignee != "cli-auto":
|
||||
# Explicit lanes are operator decisions; they are never re-routed.
|
||||
return SelectionConstraint(None, None, None, ())
|
||||
snapshot = (snapshot_from or fetch_quota_snapshot)()
|
||||
resets: dict[str, float] = {}
|
||||
for provider, quota in snapshot.items():
|
||||
record_provider_quota(provider, quota.remaining_percent, quota.reset_timestamp)
|
||||
if quota.reset_timestamp is not None:
|
||||
resets[provider] = quota.reset_timestamp
|
||||
health_excluded = fresh_unavailable_provider(now=now, quota_resets=resets)
|
||||
if health_excluded is not None:
|
||||
return SelectionConstraint(
|
||||
health_excluded,
|
||||
"is unavailable according to fresh native health",
|
||||
"health",
|
||||
(),
|
||||
)
|
||||
threshold = kanban_setting(
|
||||
"provider_quota_min_remaining_percent", QUOTA_MIN_REMAINING_PERCENT_DEFAULT
|
||||
)
|
||||
excluded, note = quota_soft_exclusion(snapshot, threshold)
|
||||
for provider in snapshot:
|
||||
record_soft_exclusion(provider, provider == excluded)
|
||||
if excluded is None or note is None:
|
||||
return SelectionConstraint(None, None, None, ())
|
||||
return SelectionConstraint(
|
||||
excluded,
|
||||
"is below its remaining-quota routing threshold",
|
||||
"quota",
|
||||
(note,),
|
||||
)
|
||||
@ -31,12 +31,49 @@ def parse_assignee(assignee: str) -> tuple[str | None, str | None]:
|
||||
raise ValueError(f"unsupported external lane assignee: {assignee}")
|
||||
return match.group(1), match.group(2)
|
||||
|
||||
def fresh_unavailable_provider(now: float | None = None) -> str | None:
|
||||
def _health_number(value: Any) -> float | None:
|
||||
"""Return a plain numeric health field, rejecting booleans and strings."""
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
return None
|
||||
return float(value)
|
||||
|
||||
def _quota_reset_readmits(
|
||||
health: dict[str, Any], current: float, reset_timestamp: float | None
|
||||
) -> bool:
|
||||
"""A passed provider quota reset ends a capacity cooldown early."""
|
||||
if health.get("authenticated") is False:
|
||||
# An authentication failure is not repaired by a quota window reset.
|
||||
return False
|
||||
failed_at = _health_number(health.get("failed_at"))
|
||||
return (
|
||||
reset_timestamp is not None
|
||||
and failed_at is not None
|
||||
and failed_at < reset_timestamp <= current
|
||||
)
|
||||
|
||||
def fresh_unavailable_provider(
|
||||
now: float | None = None,
|
||||
quota_resets: dict[str, float] | None = None,
|
||||
) -> str | None:
|
||||
"""Return one recently proven-down provider for automatic route exclusion."""
|
||||
current = time.time() if now is None else now
|
||||
resets = quota_resets or {}
|
||||
unavailable: list[str] = []
|
||||
for provider, path in PROVIDER_HEALTH_PATHS.items():
|
||||
health = load_json(path)
|
||||
if health.get("state") not in ("unavailable", "capacity-limited"):
|
||||
continue
|
||||
cooldown_until = _health_number(health.get("cooldown_until"))
|
||||
if cooldown_until is not None:
|
||||
# Lane-recorded failures re-enter only after the full cooldown has
|
||||
# elapsed (hysteresis), a quota reset passed, or fresh success
|
||||
# rewrote the state above.
|
||||
if current >= cooldown_until or _quota_reset_readmits(
|
||||
health, current, resets.get(provider)
|
||||
):
|
||||
continue
|
||||
unavailable.append(provider)
|
||||
continue
|
||||
try:
|
||||
age = current - path.stat().st_mtime
|
||||
except OSError:
|
||||
@ -46,10 +83,7 @@ def fresh_unavailable_provider(now: float | None = None) -> str | None:
|
||||
if health.get("authenticated") is False
|
||||
else PROVIDER_HEALTH_MAX_AGE_SECONDS
|
||||
)
|
||||
if (
|
||||
0 <= age <= max_age
|
||||
and health.get("state") == "unavailable"
|
||||
):
|
||||
if 0 <= age <= max_age:
|
||||
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.
|
||||
|
||||
@ -32,6 +32,11 @@ from cli_lane_board import (
|
||||
)
|
||||
from cli_lane_config import (
|
||||
ARTIFACT_GC_INTERVAL_SECONDS,
|
||||
AUTH_COOLDOWN_SECONDS_DEFAULT,
|
||||
CAPACITY_COOLDOWN_SECONDS_DEFAULT,
|
||||
QUOTA_METRICS_URL,
|
||||
QUOTA_MIN_REMAINING_PERCENT_DEFAULT,
|
||||
kanban_setting,
|
||||
ARTIFACT_RETENTION_AGE_SECONDS,
|
||||
ARTIFACT_RETENTION_BYTES,
|
||||
ARTIFACT_RETENTION_COUNT,
|
||||
@ -67,6 +72,35 @@ from cli_lane_evidence import (
|
||||
_write_json_noreplace,
|
||||
)
|
||||
from cli_lane_execution import execute_claim
|
||||
from cli_lane_failover import (
|
||||
FailoverOutcome,
|
||||
_block_transient,
|
||||
_routed_or_blocked,
|
||||
capacity_failover,
|
||||
)
|
||||
from cli_lane_health import (
|
||||
classify_capacity_failure,
|
||||
record_provider_failure,
|
||||
record_provider_success,
|
||||
)
|
||||
from cli_lane_metrics import (
|
||||
METRICS,
|
||||
record_provider_fallback,
|
||||
record_provider_quota,
|
||||
record_quota_fetch_failure,
|
||||
record_route_decision,
|
||||
record_router_selection_failure,
|
||||
record_soft_exclusion,
|
||||
start_metrics_server,
|
||||
)
|
||||
from cli_lane_quota import (
|
||||
ProviderQuota,
|
||||
SelectionConstraint,
|
||||
fetch_quota_snapshot,
|
||||
parse_quota_metrics,
|
||||
quota_soft_exclusion,
|
||||
selection_constraint,
|
||||
)
|
||||
from cli_lane_files import (
|
||||
_candidate_path,
|
||||
_fsync_directory,
|
||||
|
||||
@ -150,3 +150,24 @@ spec:
|
||||
port: 9006
|
||||
targetPort: claude-broker
|
||||
protocol: TCP
|
||||
---
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
name: hermes-cli-lane-metrics
|
||||
namespace: hermes
|
||||
labels:
|
||||
app: hermes-agent
|
||||
annotations:
|
||||
prometheus.io/scrape: "true"
|
||||
prometheus.io/port: "9011"
|
||||
prometheus.io/path: /metrics
|
||||
spec:
|
||||
type: ClusterIP
|
||||
selector:
|
||||
app: hermes-agent
|
||||
ports:
|
||||
- name: lane-metrics
|
||||
port: 9011
|
||||
targetPort: lane-metrics
|
||||
protocol: TCP
|
||||
|
||||
@ -26,12 +26,16 @@
|
||||
"services/hermes/scripts/cli_lane_dispatch.py",
|
||||
"services/hermes/scripts/cli_lane_evidence.py",
|
||||
"services/hermes/scripts/cli_lane_execution.py",
|
||||
"services/hermes/scripts/cli_lane_failover.py",
|
||||
"services/hermes/scripts/cli_lane_files.py",
|
||||
"services/hermes/scripts/cli_lane_finalization.py",
|
||||
"services/hermes/scripts/cli_lane_goal.py",
|
||||
"services/hermes/scripts/cli_lane_health.py",
|
||||
"services/hermes/scripts/cli_lane_metrics.py",
|
||||
"services/hermes/scripts/cli_lane_prompt.py",
|
||||
"services/hermes/scripts/cli_lane_provider.py",
|
||||
"services/hermes/scripts/cli_lane_quarantine.py",
|
||||
"services/hermes/scripts/cli_lane_quota.py",
|
||||
"services/hermes/scripts/cli_lane_records.py",
|
||||
"services/hermes/scripts/cli_lane_recovery.py",
|
||||
"services/hermes/scripts/cli_lane_retention.py",
|
||||
@ -45,10 +49,10 @@
|
||||
"testing/quality_gate.py",
|
||||
"ci/tests/glue/test_ariadne_schedules.py",
|
||||
"ci/tests/glue/test_glue_metrics.py",
|
||||
"testing/tests/test_publish_test_metrics.py",
|
||||
"testing/tests/test_supply_chain_report.py",
|
||||
"testing/tests/test_semgrep_report.py",
|
||||
"testing/tests/test_quality_contract.py",
|
||||
"testing/tests/test_publish_test_metrics.py",
|
||||
"testing/tests/test_supply_chain_report.py",
|
||||
"testing/tests/test_semgrep_report.py",
|
||||
"testing/tests/test_quality_contract.py",
|
||||
"testing/tests/test_quality_gate.py"
|
||||
],
|
||||
"lint_paths": [
|
||||
@ -66,12 +70,16 @@
|
||||
"services/hermes/scripts/cli_lane_dispatch.py",
|
||||
"services/hermes/scripts/cli_lane_evidence.py",
|
||||
"services/hermes/scripts/cli_lane_execution.py",
|
||||
"services/hermes/scripts/cli_lane_failover.py",
|
||||
"services/hermes/scripts/cli_lane_files.py",
|
||||
"services/hermes/scripts/cli_lane_finalization.py",
|
||||
"services/hermes/scripts/cli_lane_goal.py",
|
||||
"services/hermes/scripts/cli_lane_health.py",
|
||||
"services/hermes/scripts/cli_lane_metrics.py",
|
||||
"services/hermes/scripts/cli_lane_prompt.py",
|
||||
"services/hermes/scripts/cli_lane_provider.py",
|
||||
"services/hermes/scripts/cli_lane_quarantine.py",
|
||||
"services/hermes/scripts/cli_lane_quota.py",
|
||||
"services/hermes/scripts/cli_lane_records.py",
|
||||
"services/hermes/scripts/cli_lane_recovery.py",
|
||||
"services/hermes/scripts/cli_lane_retention.py",
|
||||
@ -216,12 +224,16 @@
|
||||
"services/hermes/scripts/cli_lane_dispatch.py",
|
||||
"services/hermes/scripts/cli_lane_evidence.py",
|
||||
"services/hermes/scripts/cli_lane_execution.py",
|
||||
"services/hermes/scripts/cli_lane_failover.py",
|
||||
"services/hermes/scripts/cli_lane_files.py",
|
||||
"services/hermes/scripts/cli_lane_finalization.py",
|
||||
"services/hermes/scripts/cli_lane_goal.py",
|
||||
"services/hermes/scripts/cli_lane_health.py",
|
||||
"services/hermes/scripts/cli_lane_metrics.py",
|
||||
"services/hermes/scripts/cli_lane_prompt.py",
|
||||
"services/hermes/scripts/cli_lane_provider.py",
|
||||
"services/hermes/scripts/cli_lane_quarantine.py",
|
||||
"services/hermes/scripts/cli_lane_quota.py",
|
||||
"services/hermes/scripts/cli_lane_records.py",
|
||||
"services/hermes/scripts/cli_lane_recovery.py",
|
||||
"services/hermes/scripts/cli_lane_retention.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"] == [{}]
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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 = {
|
||||
|
||||
@ -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(
|
||||
|
||||
@ -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, ()
|
||||
|
||||
116
testing/tests/test_hermes_cli_lane_metrics.py
Normal file
116
testing/tests/test_hermes_cli_lane_metrics.py
Normal file
@ -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()
|
||||
200
testing/tests/test_hermes_cli_provider_health.py
Normal file
200
testing/tests/test_hermes_cli_provider_health.py
Normal file
@ -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"
|
||||
440
testing/tests/test_hermes_cli_quota_failover.py
Normal file
440
testing/tests/test_hermes_cli_quota_failover.py
Normal file
@ -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
|
||||
)
|
||||
299
testing/tests/test_hermes_cli_quota_gating.py
Normal file
299
testing/tests/test_hermes_cli_quota_gating.py
Normal file
@ -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
|
||||
Loading…
x
Reference in New Issue
Block a user