test: provider-catalog E2E matrix — credential routing, base URL, dialect turn, usage/cost, listing, /model switch, fallback over every discovered provider

Rows come from real plugin discovery (providers.list_providers in a child), sharded by
name hash into 3 files. Strict KNOWN entries: #121347 (xai base_url), #121359 (fallback
base_url for anthropic/openrouter), #121387 (picker listing ignores base_url), #121388
(nebius switch validation). OAuth cells switch to a strict message-gated helper.

tests/e2e/core/providers/_catalog_helpers.py

@@ -0,0 +1,266 @@
+"""Shared harness for the provider-catalog E2E matrix.
+
+The provider list is NEVER hardcoded: :func:`discover_catalog` runs the real plugin discovery
+(``providers.list_providers()``) in a clean child interpreter, so a new plugin under
+``plugins/model-providers/`` joins every matrix automatically. Each row is driven through the
+real ``python -m hermes_cli.main`` in a hermetic HOME against
+:class:`tests.fakes.providers.catalog_fake.CatalogFake`, redirected the way the product documents
+(``model.provider`` + ``model.base_url`` in config.yaml; ``/anthropic`` path for the Anthropic
+Messages dialect). Every other provider's key is present as a decoy, and all non-loopback egress
+goes through the fake's sentinel proxy, so a credential sent to the wrong host is observable.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import hashlib
+import json
+import os
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, Iterator
+import re
+
+import pytest
+import yaml
+
+from tests.fakes.providers.catalog_fake import USAGE_IN, USAGE_OUT, CatalogFake, Recorded
+
+REPO_ROOT = Path(__file__).resolve().parents[4]
+TURN_TIMEOUT = 75.0
+SHARDS = 3
+FINAL = "CATALOG-TURN-COMPLETE"
+
+# Wire dialect the fake must see for each transport the runtime can resolve.
+DIALECT_OF_API_MODE = {"chat_completions": "chat", "anthropic_messages": "anthropic", "codex_responses": "responses"}
+# Header that must carry the key, per dialect (Anthropic Messages = x-api-key; OpenAI wire = Bearer).
+AUTH_HEADER_OF_DIALECT = {"chat": "authorization", "responses": "authorization", "anthropic": "x-api-key"}
+# Where the documented base-URL override points, per dialect (Anthropic needs a ``/anthropic``
+# path: the product only trusts an Anthropic-protocol override that looks like one).
+URL_SUFFIX_OF_DIALECT = {"chat": "/v1", "responses": "/v1", "anthropic": "/anthropic"}
+
+# auth types whose transport cannot be pointed at a loopback HTTP fake by config/env alone.
+UNREDIRECTABLE_AUTH = {
+    "aws_sdk": "AWS SigV4 via boto3 default chain; no HTTP fake for bedrock-runtime in this lane",
+    "vertex": "Google ADC/OAuth2 token minting is required before any request",
+    "external_process": "speaks ACP to a local vendor CLI subprocess, not HTTP",
+    "copilot": "the GitHub token exchange URL is hardcoded (api.github.com/copilot_internal/v2/token)",
+    "oauth_device_code": "login + refresh covered by test_catalog_oauth.py",
+    "oauth_external": "login + refresh covered by test_catalog_oauth.py (where redirectable)",
+}
+# Hosts a hermetic run may reach without carrying any vendor credential.
+CREDENTIAL_FREE_HOSTS = frozenset({"models.dev:443"})
+
+_SECRET_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_ACCESS_KEY", "_KEY")
+_PASSTHROUGH = frozenset({"PATH", "LANG", "LANGUAGE", "USER", "LOGNAME", "SHELL", "TMPDIR", "TZ"})
+
+
+@dataclass(frozen=True)
+class Row:
+    name: str
+    api_mode: str
+    auth_type: str
+    key_env: str | None
+    base_url: str
+    aliases: tuple[str, ...]
+    supports_model_listing: bool
+    # The declared transport is mandated by the provider's OWN host (e.g. a Responses-native
+    # endpoint); at a foreign base URL the product speaks plain chat completions instead.
+    host_mandated: bool = False
+
+    @property
+    def dialect(self) -> str | None:
+        return DIALECT_OF_API_MODE.get(self.api_mode)
+
+    def skip_reason(self) -> str | None:
+        if self.auth_type in UNREDIRECTABLE_AUTH:
+            return f"{self.name}: {UNREDIRECTABLE_AUTH[self.auth_type]}"
+        if self.dialect is None:
+            return f"{self.name}: transport {self.api_mode!r} has no loopback dialect in CatalogFake"
+        if self.key_env is None:
+            return f"{self.name}: declares no credential env var (keyless/custom endpoint)"
+        return None
+
+
+_CATALOG: list[Row] | None = None
+
+_DISCOVER = """
+import json, providers
+from hermes_cli.providers import host_mandated_api_mode
+out = []
+for p in providers.list_providers():
+    key = next((e for e in p.env_vars if not e.endswith("_BASE_URL")), None)
+    out.append(dict(name=p.name, api_mode=p.api_mode, auth_type=p.auth_type, key_env=key,
+                    base_url=p.base_url, aliases=list(p.aliases),
+                    supports_model_listing=bool(p.supports_model_listing),
+                    host_mandated=host_mandated_api_mode(p.base_url) is not None))
+print("CATALOG=" + json.dumps(out))
+"""
+
+
+def discover_catalog() -> list[Row]:
+    """Real plugin discovery in a clean child (no user plugins, no credentials)."""
+    global _CATALOG
+    if _CATALOG is None:
+        with tempfile.TemporaryDirectory(prefix="catalog-discover-") as d:
+            env = {k: v for k, v in os.environ.items() if k in _PASSTHROUGH}
+            env.update(HOME=d, HERMES_HOME=str(Path(d) / ".hermes"), PYTHONPATH=str(REPO_ROOT))
+            proc = subprocess.run([sys.executable, "-c", _DISCOVER], env=env, capture_output=True,
+                                  text=True, timeout=120, cwd=str(REPO_ROOT))
+        line = next((ln for ln in proc.stdout.splitlines() if ln.startswith("CATALOG=")), None)
+        assert line, f"provider discovery failed: {proc.stderr[-2000:]}"
+        _CATALOG = sorted((Row(**{**r, "aliases": tuple(r["aliases"])}) for r in json.loads(line[8:])),
+                          key=lambda r: r.name)
+    return _CATALOG
+
+
+def shard_of(name: str, shards: int = SHARDS) -> int:
+    return int(hashlib.sha256(name.encode()).hexdigest()[:8], 16) % shards
+
+
+def decoy_keys(catalog: list[Row]) -> dict[str, str]:
+    """One distinct fake secret per credential env var any provider declares."""
+    return {r.key_env: f"sk-cat-{r.key_env.lower()}" for r in catalog if r.key_env}
+
+
+def hermetic_env(home: Path, extra: dict[str, str]) -> dict[str, str]:
+    env = {k: v for k, v in os.environ.items()
+           if (k in _PASSTHROUGH or k.startswith("LC_")) and not k.endswith(_SECRET_SUFFIXES)}
+    env.update({
+        "HOME": str(home), "HERMES_HOME": str(home / ".hermes"), "PYTHONPATH": str(REPO_ROOT),
+        "PYTHONUNBUFFERED": "1", "NO_COLOR": "1", "TERM": "dumb",
+        # The child's HOME is tmp_path; this is the state-db guard's documented child escape hatch.
+        "HERMES_STATE_DB_GUARD_BYPASS": "1",
+    })
+    env.update(extra)
+    return env
+
+
+def write_home(root: Path, model: dict[str, Any], extra_cfg: dict[str, Any] | None = None) -> Path:
+    home = root / "home"
+    (home / ".hermes").mkdir(parents=True, exist_ok=True)
+    cfg = {"model": {"default": "catalog-model-a", "context_length": 128000, **model},
+           "agent": {"api_max_retries": 1}, "updates": {"check": False}, **(extra_cfg or {})}
+    (home / ".hermes" / "config.yaml").write_text(yaml.safe_dump(cfg, sort_keys=False), encoding="utf-8")
+    return home
+
+
+def run_hermes(home: Path, cwd: Path, env_extra: dict[str, str], *args: str,
+               timeout: float = TURN_TIMEOUT) -> subprocess.CompletedProcess:
+    try:
+        return subprocess.run([sys.executable, "-m", "hermes_cli.main", *args], cwd=cwd,
+                              env=hermetic_env(home, env_extra), capture_output=True, text=True,
+                              timeout=timeout, stdin=subprocess.DEVNULL)
+    except subprocess.TimeoutExpired as exc:
+        out = exc.stdout.decode() if isinstance(exc.stdout, bytes) else (exc.stdout or "")
+        return subprocess.CompletedProcess(exc.cmd, -9, out, f"TIMEOUT after {timeout}s")
+
+
+def session_usage(home: Path) -> dict[str, Any] | None:
+    db = home / ".hermes" / "state.db"
+    if not db.exists():
+        return None
+    con = sqlite3.connect(f"file:{db}?mode=ro", uri=True)
+    try:
+        con.row_factory = sqlite3.Row
+        row = con.execute("SELECT input_tokens, output_tokens, estimated_cost_usd, cost_status "
+                          "FROM sessions ORDER BY started_at DESC LIMIT 1").fetchone()
+        return dict(row) if row else None
+    finally:
+        con.close()
+
+
+def credential_values(rec: Recorded, secrets: set[str]) -> dict[str, str]:
+    """header -> the secret it carries, for every header carrying any known secret."""
+    return {h: s for h, v in rec.headers.items() for s in secrets if s in v}
+
+
+@dataclass
+class TurnResult:
+    row: Row
+    rc: int
+    stdout: str
+    stderr: str
+    requests: list[Recorded]
+    egress: list[str]
+    own_key: str
+    secrets: set[str]
+    usage: dict[str, Any] | None
+    canary: str
+    wall_s: float
+    cells: dict[str, bool] = field(default_factory=dict)
+
+    def detail(self) -> str:
+        reqs = [f"{r.method} {r.path} creds={credential_values(r, self.secrets)}" for r in self.requests]
+        return (f"rc={self.rc} wall={self.wall_s}s egress={self.egress}\n  usage={self.usage}\n"
+                f"  requests={reqs}\n  stdout={self.stdout[-600:]!r}\n  stderr={self.stderr[-1200:]!r}")
+
+
+def drive_turn(row: Row, root: Path, catalog: list[Row]) -> TurnResult:
+    """One oneshot turn with one tool round trip for ``row`` against its own fake."""
+    project = root / "project"
+    project.mkdir(parents=True, exist_ok=True)
+    canary = f"CANARY-{row.name}-{os.urandom(4).hex()}"
+    (project / "canary.txt").write_text(canary + "\n", encoding="utf-8")
+    keys = decoy_keys(catalog)
+    started = time.monotonic()
+    with CatalogFake(tool_args={"path": str(project / "canary.txt")}, final_text=FINAL) as fake:
+        base = f"{fake.origin}/{row.name}{URL_SUFFIX_OF_DIALECT[row.dialect or 'chat']}"
+        home = write_home(root, {"provider": row.name, "base_url": base})
+        proc = run_hermes(home, project, {**keys, **fake.proxy_env()}, "-z", "Read canary.txt and report.")
+        requests = list(fake.requests)
+        egress = fake.egress_hosts()
+    return TurnResult(row=row, rc=proc.returncode, stdout=proc.stdout, stderr=proc.stderr, requests=requests,
+                      egress=egress, own_key=keys[row.key_env or ""], secrets=set(keys.values()),
+                      usage=session_usage(home), canary=canary, wall_s=round(time.monotonic() - started, 1))
+
+
+def provider_hosts(catalog: list[Row]) -> set[str]:
+    from urllib.parse import urlsplit
+    return {urlsplit(r.base_url).hostname or "" for r in catalog if r.base_url.startswith("https://")} - {""}
+
+
+def evaluate(t: TurnResult, catalog: list[Row]) -> dict[str, bool]:
+    """Every invariant of one row. Relationship checks only — never literals of today's output."""
+    inference = [r for r in t.requests if r.method == "POST"]
+    # A transport mandated by the provider's own host may fall back to chat at a foreign URL.
+    expected = {t.row.dialect} | ({"chat"} if t.row.host_mandated else set())
+    main = [r for r in inference if isinstance(r.body, dict) and r.body.get("tools")]
+    foreign = t.secrets - {t.own_key}
+    foreign_hosts = provider_hosts([r for r in catalog if r.name != t.row.name]) - provider_hosts([t.row])
+    return {
+        "turn_completed": t.rc == 0 and FINAL in t.stdout,
+        "reached_own_endpoint": bool(inference) and all(r.path.startswith(f"/{t.row.name}/") for r in inference),
+        "dialect_matches_transport": bool(main) and all(r.dialect in expected for r in main),
+        "tool_round_trip": any(t.canary in json.dumps(r.body) for r in main),
+        "own_key_in_auth_header": bool(inference) and all(
+            t.own_key in r.headers.get(AUTH_HEADER_OF_DIALECT.get(r.dialect, "authorization"), "") for r in inference),
+        "no_foreign_key_on_wire": not any(s in v for r in t.requests for v in r.headers.values() for s in foreign),
+        # Egress sentinel: nothing may leave for ANOTHER provider's host (CONNECT target).
+        "no_egress_to_foreign_provider_hosts": not [h for h in t.egress if h.rsplit(":", 1)[0] in foreign_hosts],
+        "usage_recorded": bool(t.usage) and (t.usage["input_tokens"] or 0) >= USAGE_IN
+        and (t.usage["output_tokens"] or 0) >= USAGE_OUT,
+        # Unknown pricing (a model no catalog prices) must be explicit, never a silent $0 estimate.
+        # (usage_recorded owns absence; this cell judges only a row that exists.)
+        "cost_not_silent_zero": not t.usage or not (
+            (t.usage.get("estimated_cost_usd") in (0, 0.0)) and t.usage.get("cost_status") not in (None, "unknown")),
+    }
+
+
+@contextlib.contextmanager
+def strict_known(pattern: str, reason: str) -> Iterator[None]:
+    """Strict run-time xfail for a filed bug: an AssertionError whose text matches ``pattern`` XFAILs
+    the cell; any other failure propagates; a clean pass FAILS, so the fix PR must drop the entry
+    (the campaign's strict-KNOWN rule, applied to cells whose assertions run after a live wait)."""
+    try:
+        yield
+    except AssertionError as exc:
+        if not re.search(pattern, str(exc)):
+            raise
+        pytest.xfail(f"{reason} [observed: {str(exc).splitlines()[0][:240]}]")
+    pytest.fail(f"KNOWN bug now fixed — drop its KNOWN entry: {reason}")

tests/e2e/core/providers/test_catalog_fallback.py

@@ -0,0 +1,92 @@
+"""fallback_providers across the discovered catalog: a dead provider falls through to the next.
+
+Every runnable provider is the PRIMARY of one row, its catalog neighbour the fallback (both at
+their own loopback fakes, both configured through the documented ``model.base_url`` /
+``fallback_providers[].base_url`` keys). The primary answers HTTP 500 to every inference request.
+The oneshot must still deliver the fallback's answer after one tool round trip, the fallback must
+be called with ITS OWN key only, and the primary's key must never reach the fallback's endpoint.
+"""
+
+from __future__ import annotations
+
+import json
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.core.providers._catalog_helpers import (
+    FINAL, URL_SUFFIX_OF_DIALECT, Row, decoy_keys, discover_catalog, run_hermes, write_home,
+)
+from tests.fakes.providers.catalog_fake import CatalogFake
+
+CATALOG = discover_catalog()
+# Rows whose primary can be redirected; xai cannot (#121347), so it is never a primary/fallback here.
+ROWS = [r for r in CATALOG if r.skip_reason() is None and r.name != "xai"]
+# Keyed on the FALLBACK provider (pairs follow catalog order, so primaries shift as plugins are
+# added). Strict: a listed cell that turns green fails the row until the entry is dropped.
+_FB_BASE_URL_IGNORED = "#121359 fallback entry ignores its base_url; goes to the vendor host"
+_FB_CELLS = ("fallback_answered", "fallback_tool_round_trip", "fallback_used_own_key")
+KNOWN: dict[tuple[str, str], str] = {
+    **{("anthropic", c): _FB_BASE_URL_IGNORED for c in _FB_CELLS},
+    **{("openrouter", c): _FB_BASE_URL_IGNORED for c in _FB_CELLS},
+}
+
+
+def _url(fake: CatalogFake, row: Row) -> str:
+    return f"{fake.origin}/{row.name}{URL_SUFFIX_OF_DIALECT[row.dialect or 'chat']}"
+
+
+def _drive(primary: Row, fallback: Row, root: Path) -> dict:
+    project = root / "project"
+    project.mkdir(parents=True, exist_ok=True)
+    canary = f"CANARY-FB-{primary.name}"
+    (project / "canary.txt").write_text(canary + "\n", encoding="utf-8")
+    keys = decoy_keys(CATALOG)
+    args = {"path": str(project / "canary.txt")}
+    with CatalogFake(fail_status=500) as dead, CatalogFake(tool_args=args, final_text=FINAL) as alive:
+        home = write_home(root, {"provider": primary.name, "base_url": _url(dead, primary)}, {
+            "fallback_providers": [{"provider": fallback.name, "model": "catalog-model-a",
+                                    "base_url": _url(alive, fallback)}]})
+        proc = run_hermes(home, project, {**keys, **dead.proxy_env()}, "-z", "Read canary.txt and report.")
+        dead_reqs, alive_reqs = dead.inference(), alive.inference()
+        egress = dead.egress_hosts()
+    pk, fk = keys[primary.key_env or ""], keys[fallback.key_env or ""]
+    return {
+        "rc": proc.returncode, "stdout": proc.stdout[-400:], "stderr": proc.stderr[-1200:],
+        "egress": egress, "dead_paths": [r.path for r in dead_reqs], "alive_paths": [r.path for r in alive_reqs],
+        "cells": {
+            "primary_tried_first": bool(dead_reqs) and any(pk in v for v in dead_reqs[0].headers.values()),
+            "fallback_answered": proc.returncode == 0 and FINAL in proc.stdout,
+            "fallback_tool_round_trip": any(canary in json.dumps(r.body) for r in alive_reqs),
+            "fallback_used_own_key": bool(alive_reqs) and all(
+                any(fk in v for v in r.headers.values()) for r in alive_reqs),
+            "primary_key_not_sent_to_fallback": pk == fk or not any(
+                pk in v for r in alive_reqs for v in r.headers.values()),
+        },
+    }
+
+
+@pytest.fixture(scope="module")
+def results(tmp_path_factory: pytest.TempPathFactory) -> dict[str, dict]:
+    pairs = {r.name: (r, ROWS[(i + 1) % len(ROWS)]) for i, r in enumerate(ROWS)}
+    with ThreadPoolExecutor(max_workers=8, thread_name_prefix="fallback") as pool:
+        futs = {n: pool.submit(_drive, p, f, Path(tmp_path_factory.mktemp(f"fb-{n}"))) for n, (p, f) in pairs.items()}
+        out = {n: f.result() for n, f in futs.items()}
+    for n, (_p, f) in pairs.items():
+        out[n]["fallback"] = f.name
+    return out
+
+
+@pytest.mark.parametrize("row", [pytest.param(r, id=r.name) for r in ROWS])
+def test_dead_primary_falls_through(row: Row, results: dict[str, dict]) -> None:
+    res = results[row.name]
+    cells = res["cells"]
+    known = {c: ref for (p, c), ref in KNOWN.items() if p == res["fallback"]}
+    fixed = sorted(c for c in known if cells.get(c))
+    assert not fixed, f"{row.name}: {fixed} now green — drop their KNOWN entries ({set(known.values())})"
+    failed = sorted(c for c, ok in cells.items() if not ok and c not in known)
+    assert not failed, f"{row.name} -> {res['fallback']}: cells red: {failed}\n" + json.dumps(
+        {k: v for k, v in res.items() if k != "cells"})[:2500]
+    if known:
+        pytest.xfail(f"{sorted(known)}: {'; '.join(sorted(set(known.values())))}")

tests/e2e/core/providers/test_catalog_listing.py

@@ -0,0 +1,120 @@
+"""Model listing and ``/model --provider`` switching per discovered provider.
+
+Each runnable provider is configured at a custom ``model.base_url`` (a loopback fake serving
+provider-unique model ids). Two child processes per row call the real product functions:
+
+* listing — ``hermes_cli.models.provider_model_ids`` (the catalog the ``/model`` picker renders)
+  must query the CONFIGURED endpoint when the provider declares a listing endpoint, and never the
+  provider's canonical host (#120844 class): a relay user's picker must list the relay's models,
+  and the relay's key must not be addressed to the vendor;
+* switch — ``hermes_cli.model_switch.switch_model(explicit_provider=<row>)`` must resolve to that
+  provider at its configured endpoint, not to an alias on another endpoint (#120295 class).
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from urllib.parse import urlsplit
+
+import pytest
+
+from tests.e2e.core.providers._catalog_helpers import (
+    URL_SUFFIX_OF_DIALECT, Row, decoy_keys, discover_catalog, hermetic_env, write_home,
+)
+from tests.fakes.providers.catalog_fake import CatalogFake
+
+CATALOG = discover_catalog()
+ROWS = [r for r in CATALOG if r.skip_reason() is None]
+_LISTING_IGNORES_BASE_URL = "#121387 picker listing ignores model.base_url; queries the vendor host"
+_LISTING_BROKEN = (
+    "ai-gateway", "alibaba", "alibaba-cn", "alibaba-coding-plan", "alibaba-coding-plan-cn",
+    "alibaba-token-plan", "alibaba-token-plan-cn", "arcee", "commandcode", "commandcode-anthropic",
+    "deepinfra", "deepseek", "fireworks", "gemini", "gmi", "huggingface", "kilocode", "kimi-coding",
+    "kimi-coding-cn", "meta-ai", "minimax", "minimax-cn", "nebius-token-factory", "novita", "nvidia",
+    "ollama-cloud", "opencode-go", "opencode-zen", "openrouter", "router", "stepfun", "upstage", "xai",
+    "xiaomi", "zai",
+)
+# Strict: a listed cell that turns green fails the row until its entry is dropped.
+KNOWN: dict[tuple[str, str], str] = {
+    **{(n, "listing_uses_configured_endpoint"): _LISTING_IGNORES_BASE_URL for n in _LISTING_BROKEN},
+    ("nebius-token-factory", "switch_resolves_requested_provider"):
+        "#121388 nebius /model switch validates against the vendor host, ignoring model.base_url",
+    ("xai", "switch_resolves_requested_provider"): "#121347 xai ignores model.base_url",
+}
+
+_LIST = r"""
+import json, sys
+from hermes_cli.models import provider_model_ids
+try:
+    print("RESULT=" + json.dumps({"ids": provider_model_ids(sys.argv[1], force_refresh=True)}))
+except Exception as exc:
+    print("RESULT=" + json.dumps({"error": f"{type(exc).__name__}: {exc}"}))
+"""
+_SWITCH = r"""
+import json, sys
+from hermes_cli.model_switch import switch_model
+from hermes_cli.providers import normalize_provider
+r = switch_model(raw_input=sys.argv[2], current_provider="custom", current_model="x", current_base_url="",
+                 current_api_key="", explicit_provider=sys.argv[1])
+print("RESULT=" + json.dumps({"ok": bool(r.success), "provider": r.target_provider, "base_url": r.base_url,
+                              "error": r.error_message, "want": normalize_provider(sys.argv[1]),
+                              "got": normalize_provider(r.target_provider or "")}))
+"""
+
+
+def _probe(row: Row, root: Path, script: str) -> dict:
+    unique = f"catalog-{row.name}-alpha"
+    keys = decoy_keys(CATALOG)
+    with CatalogFake(models=[unique, f"catalog-{row.name}-beta"]) as fake:
+        base = f"{fake.origin}/{row.name}{URL_SUFFIX_OF_DIALECT[row.dialect or 'chat']}"
+        home = write_home(root, {"provider": row.name, "base_url": base})
+        try:
+            proc = subprocess.run([sys.executable, "-c", script, row.name, unique], cwd=root, capture_output=True,
+                                  text=True, env=hermetic_env(home, {**keys, **fake.proxy_env()}), timeout=90,
+                                  stdin=subprocess.DEVNULL)
+            out, err = proc.stdout, proc.stderr
+        except subprocess.TimeoutExpired:
+            out, err = "", "TIMEOUT after 90s"
+        listings, egress = fake.listings(), fake.egress_hosts()
+    line = next((ln for ln in out.splitlines() if ln.startswith("RESULT=")), None)
+    res = json.loads(line[7:]) if line else {"error": f"probe crashed: {err[-1200:]}"}
+    return {**res, "base": base, "listing_paths": [r.path for r in listings], "egress": egress,
+            "canonical_host_hit": f"{urlsplit(row.base_url).hostname}:443" in egress}
+
+
+@pytest.fixture(scope="module")
+def probes(tmp_path_factory: pytest.TempPathFactory) -> dict[tuple[str, str], dict]:
+    jobs = [(r, kind, script) for r in ROWS for kind, script in (("list", _LIST), ("switch", _SWITCH))]
+    with ThreadPoolExecutor(max_workers=8, thread_name_prefix="listing") as pool:
+        futs = {(r.name, kind): pool.submit(_probe, r, Path(tmp_path_factory.mktemp(f"{kind}-{r.name}")), script)
+                for r, kind, script in jobs}
+        return {k: f.result() for k, f in futs.items()}
+
+
+def _cells(row: Row, ls: dict, sw: dict) -> dict[str, bool]:
+    return {
+        "listing_uses_configured_endpoint": (not row.supports_model_listing) or (
+            "error" not in ls and bool(ls["listing_paths"])
+            and all(p.startswith(f"/{row.name}/") for p in ls["listing_paths"]) and not ls["canonical_host_hit"]),
+        # Same provider (alias-normalised: ai-gateway == vercel) at the configured endpoint.
+        "switch_resolves_requested_provider": bool(sw.get("ok")) and bool(sw.get("want")) and sw.get("got") == sw["want"]
+        and str(sw.get("base_url") or "").rstrip("/") == sw["base"].rstrip("/"),
+    }
+
+
+@pytest.mark.parametrize("row", [pytest.param(r, id=r.name) for r in ROWS])
+def test_listing_and_switch(row: Row, probes: dict) -> None:
+    ls, sw = probes[(row.name, "list")], probes[(row.name, "switch")]
+    cells = _cells(row, ls, sw)
+    known = {c: ref for (p, c), ref in KNOWN.items() if p == row.name}
+    fixed = sorted(c for c in known if cells.get(c))
+    assert not fixed, f"{row.name}: {fixed} now green — drop their KNOWN entries ({set(known.values())})"
+    failed = sorted(c for c, ok in cells.items() if not ok and c not in known)
+    detail = {"list": {**ls, "ids": (ls.get("ids") or [])[:6]}, "switch": sw}
+    assert not failed, f"{row.name}: cells red: {failed}\n{json.dumps(detail, default=str)[:2500]}"
+    if known:
+        pytest.xfail(f"{sorted(known)}: {'; '.join(sorted(set(known.values())))}")

tests/e2e/core/providers/test_catalog_matrix_0.py

@@ -0,0 +1,79 @@
+"""Provider-catalog matrix, shard 0/3: one real oneshot turn per discovered provider.
+
+Rows are the providers whose name hashes to this shard (``_catalog_helpers.shard_of``), so a new
+plugin joins some shard automatically. Each row runs ``hermes -z`` against its own loopback fake
+(redirected via ``model.base_url``) with every other provider's key present as a decoy, and checks:
+the turn completes with a tool round trip through the provider's dialect; only the provider's own
+key reaches the wire, in the dialect's auth header, and only at its configured endpoint (no egress
+to any provider host); usage lands in state.db and unknown pricing is not a silent $0.
+"""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.core.providers._catalog_helpers import (
+    SHARDS, Row, TurnResult, discover_catalog, drive_turn, evaluate, shard_of,
+)
+
+SHARD = 0
+# Cells red on origin/main for a tracked open bug: (provider, cell) -> "#issue one line".
+# Strict: the row FAILS as soon as a listed cell turns green (drop the entry with the fix), and
+# any cell NOT listed still fails normally, so a known bug never masks a new regression.
+XAI_BASE_URL_IGNORED = "#121347 xai ignores model.base_url; request + key go to api.x.ai"
+KNOWN: dict[tuple[str, str], str] = {
+    ("xai", "turn_completed"): XAI_BASE_URL_IGNORED,
+    ("xai", "reached_own_endpoint"): XAI_BASE_URL_IGNORED,
+    ("xai", "dialect_matches_transport"): XAI_BASE_URL_IGNORED,
+    ("xai", "tool_round_trip"): XAI_BASE_URL_IGNORED,
+    ("xai", "own_key_in_auth_header"): XAI_BASE_URL_IGNORED,
+    ("xai", "usage_recorded"): XAI_BASE_URL_IGNORED,
+}
+
+CATALOG = discover_catalog()
+ROWS = [r for r in CATALOG if shard_of(r.name) == SHARD]
+
+
+@pytest.fixture(scope="module")
+def turns(tmp_path_factory: pytest.TempPathFactory) -> dict[str, TurnResult]:
+    """Every runnable row of the shard driven concurrently (own home, fake and process each)."""
+    runnable = [r for r in ROWS if r.skip_reason() is None]
+    roots = {r.name: Path(tmp_path_factory.mktemp(f"cat-{r.name}")) for r in runnable}
+    with ThreadPoolExecutor(max_workers=8, thread_name_prefix="catalog") as pool:
+        futs = {r.name: pool.submit(drive_turn, r, roots[r.name], CATALOG) for r in runnable}
+        return {name: f.result() for name, f in futs.items()}
+
+
+def _params() -> list:
+    out = []
+    for r in ROWS:
+        marks = [pytest.mark.skip(reason=r.skip_reason())] if r.skip_reason() else []
+        out.append(pytest.param(r, id=r.name, marks=marks))
+    return out
+
+
+@pytest.mark.parametrize("row", _params())
+def test_provider_row(row: Row, turns: dict[str, TurnResult]) -> None:
+    t = turns[row.name]
+    cells = evaluate(t, CATALOG)
+    known = {c: ref for (p, c), ref in KNOWN.items() if p == row.name}
+    fixed = sorted(c for c in known if cells.get(c))
+    assert not fixed, f"{row.name}: {fixed} now green — drop their KNOWN entries ({set(known.values())})"
+    failed = sorted(c for c, ok in cells.items() if not ok and c not in known)
+    assert not failed, f"{row.name} ({row.api_mode}/{row.auth_type}): cells red: {failed}\n{t.detail()}"
+    if known:
+        pytest.xfail(f"{sorted(known)}: {'; '.join(sorted(set(known.values())))}")
+
+
+def test_catalog_is_discovered_not_listed() -> None:
+    """Every bundled plugin dir registers a provider that some shard runs (or skips with a
+    reason): the matrix follows discovery, so a new plugin can never silently fall out of it."""
+    root = Path(__file__).resolve().parents[4] / "plugins" / "model-providers"
+    dirs = {d.name for d in root.iterdir() if (d / "__init__.py").exists()}
+    names = {r.name for r in CATALOG}
+    assert dirs, "no bundled model-provider plugins found"
+    assert dirs <= names, f"plugin dirs with no discovered profile: {sorted(dirs - names)}"
+    assert {shard_of(n) for n in names} <= set(range(SHARDS))

tests/e2e/core/providers/test_catalog_matrix_1.py

@@ -0,0 +1,60 @@
+"""Provider-catalog matrix, shard 1/3: one real oneshot turn per discovered provider.
+
+Rows are the providers whose name hashes to this shard (``_catalog_helpers.shard_of``), so a new
+plugin joins some shard automatically. Each row runs ``hermes -z`` against its own loopback fake
+(redirected via ``model.base_url``) with every other provider's key present as a decoy, and checks:
+the turn completes with a tool round trip through the provider's dialect; only the provider's own
+key reaches the wire, in the dialect's auth header, and only at its configured endpoint (no egress
+to any provider host); usage lands in state.db and unknown pricing is not a silent $0.
+"""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.core.providers._catalog_helpers import (
+    SHARDS, Row, TurnResult, discover_catalog, drive_turn, evaluate, shard_of,
+)
+
+SHARD = 1
+# Cells red on origin/main for a tracked open bug: (provider, cell) -> "#issue one line".
+# Strict: the row FAILS as soon as a listed cell turns green (drop the entry with the fix), and
+# any cell NOT listed still fails normally, so a known bug never masks a new regression.
+KNOWN: dict[tuple[str, str], str] = {}
+
+CATALOG = discover_catalog()
+ROWS = [r for r in CATALOG if shard_of(r.name) == SHARD]
+
+
+@pytest.fixture(scope="module")
+def turns(tmp_path_factory: pytest.TempPathFactory) -> dict[str, TurnResult]:
+    """Every runnable row of the shard driven concurrently (own home, fake and process each)."""
+    runnable = [r for r in ROWS if r.skip_reason() is None]
+    roots = {r.name: Path(tmp_path_factory.mktemp(f"cat-{r.name}")) for r in runnable}
+    with ThreadPoolExecutor(max_workers=8, thread_name_prefix="catalog") as pool:
+        futs = {r.name: pool.submit(drive_turn, r, roots[r.name], CATALOG) for r in runnable}
+        return {name: f.result() for name, f in futs.items()}
+
+
+def _params() -> list:
+    out = []
+    for r in ROWS:
+        marks = [pytest.mark.skip(reason=r.skip_reason())] if r.skip_reason() else []
+        out.append(pytest.param(r, id=r.name, marks=marks))
+    return out
+
+
+@pytest.mark.parametrize("row", _params())
+def test_provider_row(row: Row, turns: dict[str, TurnResult]) -> None:
+    t = turns[row.name]
+    cells = evaluate(t, CATALOG)
+    known = {c: ref for (p, c), ref in KNOWN.items() if p == row.name}
+    fixed = sorted(c for c in known if cells.get(c))
+    assert not fixed, f"{row.name}: {fixed} now green — drop their KNOWN entries ({set(known.values())})"
+    failed = sorted(c for c, ok in cells.items() if not ok and c not in known)
+    assert not failed, f"{row.name} ({row.api_mode}/{row.auth_type}): cells red: {failed}\n{t.detail()}"
+    if known:
+        pytest.xfail(f"{sorted(known)}: {'; '.join(sorted(set(known.values())))}")

tests/e2e/core/providers/test_catalog_matrix_2.py

@@ -0,0 +1,60 @@
+"""Provider-catalog matrix, shard 2/3: one real oneshot turn per discovered provider.
+
+Rows are the providers whose name hashes to this shard (``_catalog_helpers.shard_of``), so a new
+plugin joins some shard automatically. Each row runs ``hermes -z`` against its own loopback fake
+(redirected via ``model.base_url``) with every other provider's key present as a decoy, and checks:
+the turn completes with a tool round trip through the provider's dialect; only the provider's own
+key reaches the wire, in the dialect's auth header, and only at its configured endpoint (no egress
+to any provider host); usage lands in state.db and unknown pricing is not a silent $0.
+"""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.core.providers._catalog_helpers import (
+    SHARDS, Row, TurnResult, discover_catalog, drive_turn, evaluate, shard_of,
+)
+
+SHARD = 2
+# Cells red on origin/main for a tracked open bug: (provider, cell) -> "#issue one line".
+# Strict: the row FAILS as soon as a listed cell turns green (drop the entry with the fix), and
+# any cell NOT listed still fails normally, so a known bug never masks a new regression.
+KNOWN: dict[tuple[str, str], str] = {}
+
+CATALOG = discover_catalog()
+ROWS = [r for r in CATALOG if shard_of(r.name) == SHARD]
+
+
+@pytest.fixture(scope="module")
+def turns(tmp_path_factory: pytest.TempPathFactory) -> dict[str, TurnResult]:
+    """Every runnable row of the shard driven concurrently (own home, fake and process each)."""
+    runnable = [r for r in ROWS if r.skip_reason() is None]
+    roots = {r.name: Path(tmp_path_factory.mktemp(f"cat-{r.name}")) for r in runnable}
+    with ThreadPoolExecutor(max_workers=8, thread_name_prefix="catalog") as pool:
+        futs = {r.name: pool.submit(drive_turn, r, roots[r.name], CATALOG) for r in runnable}
+        return {name: f.result() for name, f in futs.items()}
+
+
+def _params() -> list:
+    out = []
+    for r in ROWS:
+        marks = [pytest.mark.skip(reason=r.skip_reason())] if r.skip_reason() else []
+        out.append(pytest.param(r, id=r.name, marks=marks))
+    return out
+
+
+@pytest.mark.parametrize("row", _params())
+def test_provider_row(row: Row, turns: dict[str, TurnResult]) -> None:
+    t = turns[row.name]
+    cells = evaluate(t, CATALOG)
+    known = {c: ref for (p, c), ref in KNOWN.items() if p == row.name}
+    fixed = sorted(c for c in known if cells.get(c))
+    assert not fixed, f"{row.name}: {fixed} now green — drop their KNOWN entries ({set(known.values())})"
+    failed = sorted(c for c, ok in cells.items() if not ok and c not in known)
+    assert not failed, f"{row.name} ({row.api_mode}/{row.auth_type}): cells red: {failed}\n{t.detail()}"
+    if known:
+        pytest.xfail(f"{sorted(known)}: {'; '.join(sorted(set(known.values())))}")

tests/e2e/core/providers/test_catalog_oauth.py

@@ -7,8 +7,8 @@ egress goes through the ``CatalogFake`` sentinel proxy, which refuses and record
 host. Cells assert user-visible outcomes: the device-code polling cadence the vendor sees, the
 reply on stdout, the bearer on the next wire request, and the tokens persisted to auth.json.
 
-Open bugs are message-gated run-time xfails (``KNOWN`` + ``known_failure``): a cell XFAILs only
-while it fails with that bug's signature and simply passes once the fix lands, in any merge order.
+Open bugs are strict, message-gated run-time xfails (``KNOWN`` + ``strict_known``): a cell XFAILs
+only while it fails with that bug's signature, and FAILS once the fix lands so the entry is dropped.
 
 Not redirectable to a loopback fake, so not covered here (explicit skips below): openai-codex and
 qwen-oauth refresh (token URLs are module constants, no env/config override) and the Copilot token
@@ -27,7 +27,7 @@ from typing import Any
 
 import pytest
 
-from tests.e2e.core._pending_fixes import known_failure
+from tests.e2e.core.providers._catalog_helpers import strict_known
 from tests.fakes.providers.catalog_fake import CatalogFake
 from tests.fakes.providers.catalog_oauth import NOUS_INVOKE_SCOPE, OAuthFake, make_jwt
 
@@ -218,7 +218,7 @@ def test_nous_device_login_slow_down_grows_interval(device_login) -> None:
 def test_nous_device_login_honors_server_interval(device_login) -> None:
     gaps = device_login["gaps"]
     assert len(gaps) == len(DEVICE_SCRIPT), f"unexpected poll count, gaps={gaps}"
-    with known_failure(*KNOWN["device_interval"]):
+    with strict_known(*KNOWN["device_interval"]):
         assert gaps[0] >= DEVICE_INTERVAL - 0.1, (
             f"device poll gap {gaps[0]:.2f}s < server interval {DEVICE_INTERVAL}s (gaps={[round(g, 2) for g in gaps]})")
 
@@ -272,7 +272,7 @@ def test_nous_inference_401_refreshes_rotates_and_retries(tmp_path, sentinel) ->
         info = _describe(proc, fake, sentinel)
         assert any(r.bearer == revoked for r in fake.inference()), f"the stale bearer was never tried\n{info}"
         fresh = _assert_rotation_persisted(home, seed, fake, info)
-        with known_failure(*KNOWN["nous_401_retry_route"]):
+        with strict_known(*KNOWN["nous_401_retry_route"]):
             assert not _vendor_egress(sentinel), (
                 f"401 recovery retry left NOUS_INFERENCE_BASE_URL: egress to {_vendor_egress(sentinel)}\n{info}")
             assert any(r.bearer == fresh for r in fake.inference()), f"no retry with the refreshed token\n{info}"