hermes: align capacity gate with auth classifier

CAPACITY_PATTERN (the gate that sets result.capacity_failure) lacked the
bare unauthorized/forbidden/401/403 signals that classify_capacity_failure
already recognizes, so an auth blip surfacing only as "403 Forbidden"
was blocked as capability instead of transient: no auto failover, no
health cooldown. Add 401|403|unauthorized|forbidden to the gate so it
matches the classifier; reason classification still distinguishes auth
from quota/rate-limit/transport.

Tests: an auto card failing with only "403 Forbidden" now fails over to
the other provider, records an auth cooldown (authenticated:false), and
classifies the fallback reason as auth in metrics; a manual card with the
same failure still fails closed as transient. Router-outage tests moved to
test_hermes_cli_router_outage.py to keep both files <500 LOC.

Based on PR #15 (fix/hermes-result-decomposition-reliability).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

services/hermes/scripts/cli_lane_config.py

@@ -55,7 +55,8 @@ WORKTREE_LOCK = threading.Lock()
 BOARD_CORRUPTION_ERRORS: dict[str, str] = {}
 LAST_ARTIFACT_GC = 0.0
 CAPACITY_PATTERN = re.compile(
-    r"(?:rate.?limit|capacity|overload|usage.?limit|quota|credit|exhaust|429|529|authentication|oauth|token.*expired)",
+    r"(?:rate.?limit|capacity|overload|usage.?limit|quota|credit|exhaust|429|529"
+    r"|authentication|unauthorized|forbidden|oauth|token.*expired|401|403)",
     re.I,
 )
 NO_CLAUDE_SESSION = "No conversation found with session ID:"

testing/tests/test_hermes_cli_quota_failover.py

@@ -196,6 +196,80 @@ def test_auto_failover_preserves_effort_and_records_cooldown(
     assert claude_health["state"] == "available"
 
 
+def test_bare_forbidden_auth_blip_fails_over_and_records_auth_cooldown(
+    tmp_path: Path, monkeypatch
+):
+    task = SimpleNamespace(
+        id="t_forbidden",
+        status="running",
+        result=None,
+        current_run_id=38,
+        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)
+    fallbacks: list = []
+    monkeypatch.setattr(
+        lanes,
+        "record_provider_fallback",
+        lambda source, target, reason: fallbacks.append((source, target, reason)),
+    )
+    # The failure surfaces only as "403 Forbidden" with none of the words the
+    # capacity gate previously keyed on (authentication/oauth/token expired).
+    reports = [
+        lanes.ProcessResult(1, "403 Forbidden", 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_forbidden")
+
+    # (a) failed over to the other provider and completed.
+    assert reports == []
+    assert calls and calls[0][0] == "complete"
+    # (c) the fallback reason classifies as auth in metrics.
+    assert fallbacks == [("codex", "claude", "auth")]
+    assert any(
+        "Provider fallback: codex -> claude after a auth failure" in item
+        for item in comments
+    )
+    # (b) an auth cooldown was recorded (authenticated:false, longer window).
+    codex_health = json.loads(health_paths["codex"].read_text())
+    assert codex_health["state"] == "unavailable"
+    assert codex_health["authenticated"] is False
+    assert codex_health["failure_reason"] == "auth"
+    assert codex_health["cooldown_until"] == codex_health["failed_at"] + 3600.0
+
+
+def test_bare_forbidden_auth_blip_still_fails_closed_on_manual_lane(
+    tmp_path: Path, monkeypatch
+):
+    calls, _comments, route_calls, runs, health_paths = _run_manual_fail_closed(
+        tmp_path, monkeypatch, "cli-codex-high", "codex", "403 Forbidden"
+    )
+
+    # Fail-closed is unchanged: the pinned provider is not switched.
+    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 "auth" in kwargs["reason"]
+    codex_health = json.loads(health_paths["codex"].read_text())
+    assert codex_health["state"] == "unavailable"
+    assert codex_health["authenticated"] is False
+
+
 def test_double_capacity_failure_blocks_transient_with_both_reasons(
     tmp_path: Path, monkeypatch
 ):
@@ -256,140 +330,6 @@ def test_double_capacity_failure_blocks_transient_with_both_reasons(
     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",

testing/tests/test_hermes_cli_router_outage.py

@@ -0,0 +1,149 @@
+"""Switchyard outage blocks a card as transient at every selection boundary."""
+
+from __future__ import annotations
+
+from testing.tests.test_hermes_cli_quota_failover import (
+    COMPLETED_RESULT,
+    _isolate_lane,
+    _lane_board,
+    _route,
+)
+from testing.tests.test_hermes_cli_support import (
+    Path,
+    SimpleNamespace,
+    lanes,
+)
+
+
+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"]