fix(routing): address review — session runtime decides, quarantine continues the chain, unclassified ids break ownership

Three defects found in review of the first head (@ehz0ah):

* The mid-request hop read the PERSISTED provider from disk. A live
  `/model xai-oauth` session over `model.provider: auto` therefore still
  reached the discovery chain and billed Nous. _try_payment_fallback now
  takes the route's main_runtime snapshot; disk is the fallback only when
  no session runtime exists.

* After a configured fallback was quarantined mid-request (401, refresh
  failed), the second pass went straight to the discovery chain, which
  the new gate refuses — so later CONFIGURED entries never ran and the
  original error was re-raised. The second pass now re-walks the task
  chain and main chain (the quarantined entry is unhealthy and skipped)
  before discovery.

* current_provider_owns_vendor dropped ids detect_vendor could not
  classify, so Bedrock's 15-id catalog (14 unclassified `us.anthropic…`)
  looked exclusively DeepSeek and `/model deepseek-v4-pro` stuck on
  Bedrock. An unclassified id now counts as evidence of a multi-vendor
  catalog: ownership requires every id to classify to the one vendor.

agent/auxiliary_client.py

@@ -3821,12 +3821,14 @@ async def _call_fallback_candidate_async(
 
 def _try_payment_fallback(
     failed_provider: str, task: str = None, reason: str = "payment error", *,
-    failed_base_url: str = "", failure_scope: Any = None,
+    failed_base_url: str = "", failure_scope: Any = None, main_runtime: Optional[Dict[str, Any]] = None,
 ) -> Tuple[Optional[Any], Optional[str], str]:
     """Try the auto-detection chain after a payment/credit or connection error, skipping the failed
     provider (and the main-provider path when it maps to the same backend). Returns (client, model, label) or (None, None, "")."""
     skip = failed_provider.lower().strip()
-    main_provider = _read_main_provider()
+    # The SESSION's provider decides whether discovery is allowed: a live `/model xai-oauth` session
+    # over a persisted ``provider: auto`` is a selection, so the disk value alone is not the answer.
+    main_provider = _normalize_main_runtime(main_runtime).get("provider") or _read_main_provider()
     if not _discovery_chain_allowed(main_provider, task):
         return None, None, ""
     skip_labels = {skip}
@@ -6934,6 +6936,27 @@ def _ladder_credential_rungs(
     return None, first_err
 
 
+def _next_fallback_after_quarantine(
+    task: Optional[str], resolved_provider: str, is_auto: bool, route: _LadderRoute,
+    failed_model: Optional[str], failure_scope: Any,
+) -> Tuple[Optional[Any], Optional[str], str]:
+    """Next candidate after a fallback entry was quarantined mid-request: remaining configured
+    entries (task chain, then main chain on auto) before the discovery chain."""
+    reason = "stale fallback credential"
+    fb = _try_configured_fallback_chain(
+        task, resolved_provider or "auto", reason=reason, failed_model=failed_model,
+        failed_base_url=route.base_info, failure_scope=failure_scope)
+    if fb[0] is None and is_auto:
+        fb = _try_main_fallback_chain(
+            task, resolved_provider or "auto", reason=reason, failed_model=failed_model,
+            failed_base_url=route.base_info, failure_scope=failure_scope)
+    if fb[0] is None:
+        fb = _try_payment_fallback(
+            resolved_provider, task, reason=reason, failed_base_url=route.base_info,
+            failure_scope=failure_scope, main_runtime=route.main_runtime)
+    return fb
+
+
 def _ladder_provider_fallback(first_err: Exception, route: _LadderRoute):
     """Last rung: other providers (per-task chain; then auto: main fallback chain + discovery
     chain, explicit: main-agent-model net). Returns the response or None.
@@ -6981,23 +7004,23 @@ def _ladder_provider_fallback(first_err: Exception, route: _LadderRoute):
         if fb_client is None:
             fb_client, fb_model, fb_label = _try_payment_fallback(
                 resolved_provider, task, reason=reason, failed_base_url=route.base_info,
-                failure_scope=_chain_failure_scope)
+                failure_scope=_chain_failure_scope, main_runtime=route.main_runtime)
     elif fb_client is None:
         fb_client, fb_model, fb_label = _try_main_agent_model_fallback(
             resolved_provider, task, reason=reason, failed_model=_chain_failed_model,
             failed_base_url=route.base_info, failure_scope=_chain_failure_scope)
     if fb_client is not None:
-        # Second pass: the candidate credential was stale and quarantined — walk the discovery
-        # chain once more (unhealthy entries are skipped).
+        # Second pass: the candidate credential was stale and quarantined — re-walk the CONFIGURED
+        # chains first (the quarantined entry is now unhealthy and skipped, so later entries get
+        # their turn), then discovery where the selection policy allows it.
         for _pass in range(2):
             _record_route_info(route.route_info, _fallback_provider_from_label(fb_label), fb_model)
             fb_resp = yield _LadderStep("fallback", (fb_client, fb_model, fb_label))
             if fb_resp is not None:
                 return fb_resp
             if _pass == 0:
-                fb_client, fb_model, fb_label = _try_payment_fallback(
-                    resolved_provider, task, reason="stale fallback credential",
-                    failed_base_url=route.base_info, failure_scope=_chain_failure_scope)
+                fb_client, fb_model, fb_label = _next_fallback_after_quarantine(
+                    task, resolved_provider, is_auto, route, _chain_failed_model, _chain_failure_scope)
                 if fb_client is None:
                     break
     # All fallback layers exhausted — one user-visible warning, then re-raise.

hermes_cli/models_detect.py

@@ -65,7 +65,9 @@ def current_provider_owns_vendor(model_name: str, current_provider: str) -> bool
     vendor = detect_vendor(model_name or "")
     if not vendor:
         return False
-    native = {detect_vendor(mid) for mid in _PROVIDER_MODELS.get(normalized, ())} - {None}
+    # An id the classifier cannot place (Bedrock ``us.anthropic.claude-…``) is evidence the
+    # provider is NOT single-vendor; only a fully classified, single-vendor catalog owns the name.
+    native = {detect_vendor(mid) for mid in _PROVIDER_MODELS.get(normalized, ())}
     return native == {vendor}
 
 

tests/agent/test_auxiliary_auto_never_guesses_provider.py

@@ -28,11 +28,29 @@ def nous_is_the_only_working_provider():
         yield
 
 
-def test_selected_main_provider_down_refuses_to_guess_another_account(nous_is_the_only_working_provider):
+@pytest.mark.parametrize("persisted_provider", ["xai-oauth", "auto"])
+def test_selected_main_provider_down_refuses_to_guess_another_account(
+        nous_is_the_only_working_provider, persisted_provider):
+    """The SESSION runtime is the selection: a live `/model xai-oauth` over a persisted
+    ``provider: auto`` must not re-open discovery through the disk value."""
     runtime = {"provider": "xai-oauth", "model": "grok-4.6", "base_url": "https://api.x.ai/v1", "api_key": "dead"}
-    with patch.object(aux, "_read_main_provider", return_value="xai-oauth"):
+    with patch.object(aux, "_read_main_provider", return_value=persisted_provider):
         assert aux._resolve_auto_route(main_runtime=runtime, task="compression") == (None, None, "")
-        assert aux._try_payment_fallback("xai-oauth", task="compression") == (None, None, "")
+        assert aux._try_payment_fallback("xai-oauth", task="compression", main_runtime=runtime) == (None, None, "")
+
+
+def test_quarantined_fallback_hands_over_to_the_next_configured_entry():
+    """A stale first fallback (401, refresh failed) is quarantined mid-request; the second
+    configured entry must still get its turn instead of the request dying on the original error."""
+    healthy = MagicMock(name="second-fallback")
+    route = aux._LadderRoute(None, "compression", "", False, "", "xai-oauth", None, None, None, None, None,
+                             {"provider": "xai-oauth"}, None)
+    with patch.object(aux, "_try_configured_fallback_chain", return_value=(healthy, "m2", "fallback_chain[1](nous)")), \
+         patch.object(aux, "_try_payment_fallback") as discovery:
+        client, model, label = aux._next_fallback_after_quarantine(
+            "compression", "auto", True, route, None, None)
+    assert client is healthy and label == "fallback_chain[1](nous)"
+    discovery.assert_not_called()
 
 
 def test_no_selected_main_provider_still_discovers(nous_is_the_only_working_provider):

tests/hermes_cli/test_models_detect_vendor_ownership.py

@@ -30,6 +30,10 @@ def test_own_vendor_id_stays_when_live_catalog_is_empty(ladder_would_hijack, pro
     assert models.detect_provider_for_model(model, provider) is None
 
 
-def test_other_vendor_id_still_remaps_to_keyed_aggregator(ladder_would_hijack):
-    assert models.detect_provider_for_model("claude-opus-4.7", "openai-codex") == (
-        "openrouter", "vendor/claude-opus-4.7")
+@pytest.mark.parametrize("provider,model", [
+    ("openai-codex", "claude-opus-4.7"),   # other vendor's id on a single-vendor provider
+    ("bedrock", "deepseek-v4-pro"),        # multi-vendor catalog whose non-deepseek ids the
+                                           # classifier cannot place: never "exclusively deepseek"
+])
+def test_non_owned_id_still_remaps_to_keyed_aggregator(ladder_would_hijack, provider, model):
+    assert models.detect_provider_for_model(model, provider) == ("openrouter", f"vendor/{model}")