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 <noreply@anthropic.com>

services/hermes/agent-configmap.yaml

@@ -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

services/hermes/agent-deployment.yaml

@@ -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:

services/hermes/networkpolicy.yaml

@@ -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.

services/hermes/scripts/cli_lane_config.py

@@ -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)

services/hermes/scripts/cli_lane_dispatch.py

@@ -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))

services/hermes/scripts/cli_lane_execution.py

@@ -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,

services/hermes/scripts/cli_lane_failover.py

@@ -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
+    )

services/hermes/scripts/cli_lane_health.py

@@ -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

services/hermes/scripts/cli_lane_metrics.py

@@ -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

services/hermes/scripts/cli_lane_provider.py

@@ -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"),

services/hermes/scripts/cli_lane_quota.py

@@ -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,),
+    )

services/hermes/scripts/cli_lane_routing.py

@@ -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.

services/hermes/scripts/cli_lane_runner.py

@@ -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,

services/hermes/service.yaml

@@ -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

testing/quality_contract.json

@@ -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",

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"] == [{}]

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)

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 = {

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(

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, ()

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()

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"

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
+    )

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