test: provider-catalog E2E: merge-order-safe KNOWN gates, exact-path fake, bounded vendor stalls

Review fixes on the provider-catalog matrix:

- KNOWN bugs go through tests.e2e.core._pending_fixes.known_failure again (strict_known and the
  "now green -> drop KNOWN" asserts are gone), so a fix PR landing first simply turns its cells
  green. Each KNOWN is gated on the bug's OWN observed signature via a dedicated CatalogGap raised
  only at the gated assertion: xai = no inference at the fake + api.x.ai CONNECT; listing = vendor
  host hit with no probe error; nebius switch = vendor-host validation error; fallback = fallback
  fake untouched + its vendor host CONNECTed; OAuth patterns anchored. Unlisted red cells still fail.
- CatalogFake answers only exact routes (configured base path + dialect endpoint / listing path),
  404 otherwise; reached_own_endpoint and the listing cells assert the exact path.
- Turns run with agent.auto_recovery_cycles: 0 (the documented post-exhaustion ladder parked the
  xai/fallback rows for minutes: 14 CONNECTs over 118 s, bounded by design, not a retry bug) and a
  watchdog kills a child 8 s after a vendor-host CONNECT with no inference at its fake.
- An unknown api_mode fails the row instead of skipping; only explicit auth types and the named
  keyless provider skip.
- Usage cell asserts the exact sum the fake reported for answered main-turn calls.
- Listing 404 / hang degradation cells for the rows that already list from the configured endpoint.
- Shard body moved into the helper; fallback key check uses the dialect's auth header; the three
  unconditional-skip OAuth params dropped (kept under NOT COVERED).

tests/e2e/core/providers/_catalog_helpers.py

@@ -16,35 +16,47 @@ import contextlib
 import hashlib
 import json
 import os
+import re
 import sqlite3
 import subprocess
 import sys
 import tempfile
 import time
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Iterator
-import re
+from typing import Any, Iterable
+from urllib.parse import urlsplit
 
 import pytest
 import yaml
 
-from tests.fakes.providers.catalog_fake import USAGE_IN, USAGE_OUT, CatalogFake, Recorded
+from tests.e2e.core._pending_fixes import known_failure
+from tests.fakes.providers.catalog_fake import USAGE_IN, USAGE_OUT, CatalogFake, Recorded, bare_path
 
 REPO_ROOT = Path(__file__).resolve().parents[4]
 TURN_TIMEOUT = 75.0
+# A child that CONNECTs to a provider's real host and then sends nothing to its loopback fake for
+# this long is talking to the vendor, not the fake: kill it and judge what it did. Healthy rows
+# (zai, deepinfra, router) CONNECT their vendor host at startup and reach the fake <1 s later.
+VENDOR_CONNECT_GRACE_S = 8.0
 SHARDS = 3
 FINAL = "CATALOG-TURN-COMPLETE"
 
-# Wire dialect the fake must see for each transport the runtime can resolve.
+# Wire dialect the fake must see for each transport the runtime can resolve. A transport missing
+# here FAILS the matrix row (teach CatalogFake the dialect); it never becomes a skip.
 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"}
+# Inference endpoint appended to the configured base URL, per dialect (the Anthropic SDK appends
+# ``/v1/messages`` to an unversioned base; the OpenAI SDK appends to a ``/v1`` base).
+ENDPOINT_OF_DIALECT = {"chat": "/chat/completions", "responses": "/responses", "anthropic": "/v1/messages"}
 
-# auth types whose transport cannot be pointed at a loopback HTTP fake by config/env alone.
+# The ONLY reasons a discovered provider may leave the matrix: auth types whose transport cannot
+# be pointed at a loopback HTTP fake by config/env alone, and named keyless providers.
 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",
@@ -53,6 +65,9 @@ UNREDIRECTABLE_AUTH = {
     "oauth_device_code": "login + refresh covered by test_catalog_oauth.py",
     "oauth_external": "login + refresh covered by test_catalog_oauth.py (where redirectable)",
 }
+KEYLESS_PROVIDERS = {
+    "custom": "user-defined endpoint keyed by config api_key; covered by test_chat_custom_endpoint.py",
+}
 # Hosts a hermetic run may reach without carrying any vendor credential.
 CREDENTIAL_FREE_HOSTS = frozenset({"models.dev:443"})
 
@@ -60,6 +75,29 @@ _SECRET_SUFFIXES = ("_API_KEY", "_TOKEN", "_SECRET", "_ACCESS_KEY", "_KEY")
 _PASSTHROUGH = frozenset({"PATH", "LANG", "LANGUAGE", "USER", "LOGNAME", "SHELL", "TMPDIR", "TZ"})
 
 
+class CatalogGap(AssertionError):
+    """Raised ONLY by a gated final assertion, so ``known_failure(raises=CatalogGap)`` can never
+    swallow a harness failure (timeout, boot crash, precondition assert)."""
+
+
+def gate(ok: bool, message: str) -> None:
+    if not ok:
+        raise CatalogGap(message)
+
+
+@dataclass(frozen=True)
+class Known:
+    """An open bug, gated on its OWN observed signature (a regex on the gated message)."""
+
+    pattern: str
+    reason: str
+    cells: frozenset[str] = frozenset()
+
+
+def known_gate(known: Known | None) -> contextlib.AbstractContextManager:
+    return known_failure(known.pattern, known.reason, raises=CatalogGap) if known else contextlib.nullcontext()
+
+
 @dataclass(frozen=True)
 class Row:
     name: str
@@ -80,12 +118,34 @@ class Row:
     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)"
+        if self.name in KEYLESS_PROVIDERS:
+            return f"{self.name}: {KEYLESS_PROVIDERS[self.name]}"
         return None
 
+    @property
+    def base_path(self) -> str:
+        """Path of the base URL every file configures for this row on its loopback fake."""
+        return f"/{self.name}{URL_SUFFIX_OF_DIALECT[self.dialect or 'chat']}"
+
+    def routes(self) -> dict[str, str]:
+        """Exact path -> dialect the fake answers for this row; everything else is a 404."""
+        assert self.dialect, f"{self.name}: transport {self.api_mode!r} has no loopback dialect"
+        p = self.base_path
+        out = {p + ENDPOINT_OF_DIALECT[self.dialect]: self.dialect}
+        if self.dialect == "responses" or self.host_mandated:
+            out[p + ENDPOINT_OF_DIALECT["chat"]] = "chat"
+        # Listing: an unversioned canonical base (api.anthropic.com, …/anthropic) lists at
+        # ``/v1/models`` under the Anthropic override; a versioned one (…/v1) at ``/models``.
+        versioned = re.search(r"/v\d+[a-z0-9]*/?$", urlsplit(self.base_url).path or "")
+        out[p + ("/v1/models" if self.dialect == "anthropic" and not versioned else "/models")] = "listing"
+        return out
+
+    def listing_route(self) -> str:
+        return next(p for p, d in self.routes().items() if d == "listing")
+
+    def vendor_host(self) -> str:
+        return f"{urlsplit(self.base_url).hostname}:443" if self.base_url.startswith("https://") else ""
+
 
 _CATALOG: list[Row] | None = None
 
@@ -145,20 +205,50 @@ def write_home(root: Path, model: dict[str, Any], extra_cfg: dict[str, Any] | No
     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 {})}
+           # auto_recovery_cycles: 0 — the documented post-exhaustion ladder (15/30/60/60/60 s) would
+           # otherwise park a transport-failed turn for minutes; one bounded attempt is the contract here.
+           "agent": {"api_max_retries": 1, "auto_recovery_cycles": 0}, "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 _vendor_stall(watch: Iterable[CatalogFake], sentinel: CatalogFake, vendor_hosts: frozenset[str]) -> str:
+    """Why the child should be killed now, or "" — it CONNECTed to a provider's real host and
+    no watched fake has seen an inference request since, for ``VENDOR_CONNECT_GRACE_S``."""
+    hits = [r for r in list(sentinel.egress) if r.path in vendor_hosts]
+    if not hits:
+        return ""
+    since = hits[0].t
+    if any(r.method == "POST" and r.t >= since for f in watch for r in list(f.requests)):
+        return ""
+    if time.time() - since < VENDOR_CONNECT_GRACE_S:
+        return ""
+    return f"killed {VENDOR_CONNECT_GRACE_S}s after CONNECT {hits[0].path} with no inference at the fake"
+
+
+def run_hermes(home: Path, cwd: Path, env_extra: dict[str, str], *args: str, timeout: float = TURN_TIMEOUT,
+               sentinel: CatalogFake | None = None, watch: Iterable[CatalogFake] = (),
+               vendor_hosts: frozenset[str] = frozenset()) -> subprocess.CompletedProcess:
+    """Run the real CLI; a child that stalls on a vendor host (see :func:`_vendor_stall`) or
+    outlives ``timeout`` is killed (rc -9) and the reason lands in stderr."""
+    watch = tuple(watch)
+    with tempfile.TemporaryFile("w+", encoding="utf-8") as out, tempfile.TemporaryFile("w+", encoding="utf-8") as err:
+        proc = subprocess.Popen([sys.executable, "-m", "hermes_cli.main", *args], cwd=cwd,
+                                env=hermetic_env(home, env_extra), stdout=out, stderr=err, text=True,
+                                stdin=subprocess.DEVNULL)
+        deadline, why = time.monotonic() + timeout, ""
+        while proc.poll() is None:
+            why = (f"TIMEOUT after {timeout}s" if time.monotonic() > deadline else
+                   _vendor_stall(watch, sentinel, vendor_hosts) if sentinel is not None else "")
+            if why:
+                proc.kill()
+                break
+            time.sleep(0.2)
+        rc = proc.wait()
+        out.seek(0)
+        err.seek(0)
+        return subprocess.CompletedProcess(proc.args, -9 if why else rc, out.read(), err.read() + (f"\n{why}" if why else ""))
 
 
 def session_usage(home: Path) -> dict[str, Any] | None:
@@ -195,9 +285,14 @@ class TurnResult:
     wall_s: float
     cells: dict[str, bool] = field(default_factory=dict)
 
+    def signature(self) -> str:
+        """Observed facts a KNOWN bug's pattern keys on (one line, first in every gated message)."""
+        n = sum(1 for r in self.requests if r.method == "POST")
+        return f"fake_inference={n} egress={self.egress}"
+
     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"
+        reqs = [f"{r.method} {r.path} [{r.dialect}] creds={credential_values(r, self.secrets)}" for r in self.requests]
+        return (f"rc={self.rc} wall={self.wall_s}s\n  usage={self.usage}\n"
                 f"  requests={reqs}\n  stdout={self.stdout[-600:]!r}\n  stderr={self.stderr[-1200:]!r}")
 
 
@@ -209,10 +304,10 @@ def drive_turn(row: Row, root: Path, catalog: list[Row]) -> TurnResult:
     (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.")
+    with CatalogFake(tool_args={"path": str(project / "canary.txt")}, final_text=FINAL, routes=row.routes()) as fake:
+        home = write_home(root, {"provider": row.name, "base_url": fake.origin + row.base_path})
+        proc = run_hermes(home, project, {**keys, **fake.proxy_env()}, "-z", "Read canary.txt and report.",
+                          sentinel=fake, watch=[fake], vendor_hosts=frozenset(provider_hosts(catalog)))
         requests = list(fake.requests)
         egress = fake.egress_hosts()
     return TurnResult(row=row, rc=proc.returncode, stdout=proc.stdout, stderr=proc.stderr, requests=requests,
@@ -221,8 +316,8 @@ def drive_turn(row: Row, root: Path, catalog: list[Row]) -> TurnResult:
 
 
 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://")} - {""}
+    """``host:443`` of every provider's canonical HTTPS endpoint (the egress sentinel's CONNECT form)."""
+    return {r.vendor_host() for r in catalog} - {""}
 
 
 def evaluate(t: TurnResult, catalog: list[Row]) -> dict[str, bool]:
@@ -231,36 +326,73 @@ def evaluate(t: TurnResult, catalog: list[Row]) -> dict[str, bool]:
     # 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")]
+    answered = [r for r in main if r.dialect != "unknown"]
     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),
+        # The fake answers ONLY the exact configured base path + dialect endpoint (else 404,
+        # dialect "unknown"), so a mangled path under the right prefix is red here.
+        "reached_own_endpoint": bool(inference) and all(r.dialect != "unknown" 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,
+        "no_egress_to_foreign_provider_hosts": not [h for h in t.egress if h in foreign_hosts],
+        # The session row charges exactly what the fake reported for every main-turn call it
+        # answered (auxiliary title calls are not charged to the session row).
+        "usage_matches_wire": bool(answered) and bool(t.usage)
+        and t.usage["input_tokens"] == USAGE_IN * len(answered) and t.usage["output_tokens"] == USAGE_OUT * len(answered),
         # 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.)
+        # (usage_matches_wire 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}")
+# --- matrix shards ------------------------------------------------------------------------------
+
+# Open bugs per provider row: the listed cells may be red ONLY while the gated message carries the
+# bug's own signature; any other red cell fails the row, and a fixed bug simply passes.
+MATRIX_KNOWN: dict[str, Known] = {
+    "xai": Known(
+        pattern=r"fake_inference=0 egress=\[[^\]]*'api\.x\.ai:443'",
+        reason="#121347 xai ignores model.base_url; request + key go to api.x.ai",
+        cells=frozenset({"turn_completed", "reached_own_endpoint", "dialect_matches_transport", "tool_round_trip",
+                         "own_key_in_auth_header", "usage_matches_wire"})),
+}
+
+
+def shard_rows(shard: int) -> list[Row]:
+    return [r for r in discover_catalog() if shard_of(r.name) == shard]
+
+
+def shard_params(shard: int) -> list:
+    """Every discovered row of the shard; only explicit auth types / keyless names skip."""
+    return [pytest.param(r, id=r.name, marks=[pytest.mark.skip(reason=r.skip_reason())] if r.skip_reason() else [])
+            for r in shard_rows(shard)]
+
+
+def drive_shard(shard: int, tmp_path_factory: pytest.TempPathFactory) -> dict[str, TurnResult]:
+    """Every runnable row of the shard driven concurrently (own home, fake and process each)."""
+    catalog = discover_catalog()
+    runnable = [r for r in shard_rows(shard) if r.skip_reason() is None and r.dialect]
+    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 check_row(row: Row, turns: dict[str, TurnResult]) -> None:
+    assert row.dialect, (f"{row.name}: transport {row.api_mode!r} has no loopback dialect in CatalogFake — "
+                         f"add it to DIALECT_OF_API_MODE (a provider may not leave the matrix as a skip)")
+    t = turns[row.name]
+    cells = evaluate(t, discover_catalog())
+    failed = sorted(c for c, ok in cells.items() if not ok)
+    known = MATRIX_KNOWN.get(row.name)
+    unlisted = [c for c in failed if not known or c not in known.cells]
+    assert not unlisted, f"{row.name} ({row.api_mode}/{row.auth_type}): cells red: {unlisted}\n{t.signature()}\n{t.detail()}"
+    with known_gate(known):
+        gate(not failed, f"{t.signature()}\n{row.name}: cells red: {failed}\n{t.detail()}")

tests/e2e/core/providers/test_catalog_fallback.py

@@ -16,25 +16,26 @@ 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,
+    AUTH_HEADER_OF_DIALECT, FINAL, Known, Row, decoy_keys, discover_catalog, gate, known_gate, 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"]
+# Rows whose primary can be redirected; xai cannot (#121347), so it is never a primary/fallback
+# here. Rows with no loopback dialect FAIL in the matrix shards.
+ROWS = [r for r in CATALOG if r.skip_reason() is None and r.dialect 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},
-}
+# added). Signature: the fallback's own fake got nothing while its vendor host was CONNECTed.
+_FB_CELLS = frozenset({"fallback_answered", "fallback_tool_round_trip", "fallback_used_own_key"})
 
 
-def _url(fake: CatalogFake, row: Row) -> str:
-    return f"{fake.origin}/{row.name}{URL_SUFFIX_OF_DIALECT[row.dialect or 'chat']}"
+def _fb_known(name: str) -> Known:
+    host = next(r.vendor_host() for r in CATALOG if r.name == name)
+    return Known(rf"^alive_inference=0 egress=\[[^\]]*'{host.replace('.', r'[.]')}'",
+                 "#121359 fallback entry ignores its base_url; goes to the vendor host", _FB_CELLS)
+
+
+KNOWN: dict[str, Known] = {n: _fb_known(n) for n in ("anthropic", "openrouter")}
 
 
 def _drive(primary: Row, fallback: Row, root: Path) -> dict:
@@ -44,23 +45,28 @@ def _drive(primary: Row, fallback: Row, root: Path) -> dict:
     (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)}, {
+    with CatalogFake(fail_status=500, routes=primary.routes()) as dead, \
+            CatalogFake(tool_args=args, final_text=FINAL, routes=fallback.routes()) as alive:
+        home = write_home(root, {"provider": primary.name, "base_url": dead.origin + primary.base_path}, {
             "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.")
+                                    "base_url": alive.origin + fallback.base_path}]})
+        proc = run_hermes(home, project, {**keys, **dead.proxy_env()}, "-z", "Read canary.txt and report.",
+                          sentinel=dead, watch=[alive], vendor_hosts=frozenset({fallback.vendor_host()} - {""}))
         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],
+        "rc": proc.returncode, "stdout": proc.stdout[-400:], "stderr": proc.stderr[-1200:], "egress": egress,
+        "alive_inference": len(alive_reqs), "dead_paths": [f"{r.path} [{r.dialect}]" for r in dead_reqs],
+        "alive_paths": [f"{r.path} [{r.dialect}]" 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()),
+            "primary_tried_first": bool(dead_reqs) and dead_reqs[0].dialect != "unknown" and pk in dead_reqs[0].headers.get(
+                AUTH_HEADER_OF_DIALECT[dead_reqs[0].dialect], ""),
             "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),
+            # Every fallback call hit its exact configured route with ITS key in the dialect's header.
             "fallback_used_own_key": bool(alive_reqs) and all(
-                any(fk in v for v in r.headers.values()) for r in alive_reqs),
+                r.dialect != "unknown" and fk in r.headers.get(AUTH_HEADER_OF_DIALECT[r.dialect], "") 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()),
         },
@@ -82,11 +88,11 @@ def results(tmp_path_factory: pytest.TempPathFactory) -> dict[str, dict]:
 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())))}")
+    failed = sorted(c for c, ok in cells.items() if not ok)
+    known = KNOWN.get(res["fallback"])
+    unlisted = [c for c in failed if not known or c not in known.cells]
+    facts = json.dumps({k: v for k, v in res.items() if k != "cells"})[:2500]
+    assert not unlisted, f"{row.name} -> {res['fallback']}: cells red: {unlisted}\n{facts}"
+    with known_gate(known):
+        gate(not failed, f"alive_inference={res['alive_inference']} egress={res['egress']}\n"
+                         f"{row.name} -> {res['fallback']}: cells red: {failed}\n{facts}")

tests/e2e/core/providers/test_catalog_listing.py

@@ -1,14 +1,17 @@
 """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:
+provider-unique model ids at the EXACT listing path under that base; anything else 404s). 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;
+  must query the CONFIGURED endpoint at its exact listing path and render the relay's models, never
+  the provider's canonical host (#120844 class): 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).
+  provider at its configured endpoint, not to an alias on another endpoint (#120295 class);
+* listing degradation — for the rows that already list from the configured endpoint, a 404 and a
+  hanging listing must still return (bounded) without crashing and without falling back to the
+  vendor host.
 """
 
 from __future__ import annotations
@@ -16,20 +19,25 @@ from __future__ import annotations
 import json
 import subprocess
 import sys
+import time
 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,
+    Known, Row, decoy_keys, discover_catalog, gate, hermetic_env, known_gate, write_home,
 )
-from tests.fakes.providers.catalog_fake import CatalogFake
+from tests.fakes.providers.catalog_fake import CatalogFake, bare_path
 
 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"
+# Rows with no loopback dialect FAIL in the matrix shards; they cannot be configured here.
+ROWS = [r for r in CATALOG if r.skip_reason() is None and r.dialect]
+PROBE_TIMEOUT = 90.0
+# A hanging listing must give up well before the fake would release it.
+HANG_S = 60.0
+HANG_BOUND_S = 40.0
+
 _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",
@@ -38,13 +46,19 @@ _LISTING_BROKEN = (
     "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",
+# Signature: the picker went to the vendor host and returned without any probe error (it renders
+# the vendor/static list instead of the relay's).
+LISTING_KNOWN: dict[str, Known] = {n: Known(
+    r"^canonical_host_hit=True probe_error=None",
+    "#121387 picker listing ignores model.base_url; queries the vendor host") for n in _LISTING_BROKEN}
+SWITCH_KNOWN: dict[str, Known] = {
+    "nebius-token-factory": Known(
+        r"^ok=False .*not found in this provider's catalog.*'api\.tokenfactory\.nebius\.com:443'",
+        "#121388 nebius /model switch validates against the vendor host, ignoring model.base_url"),
+    "xai": Known(r"^ok=True got='xai' base_url='https://api\.x\.ai/v1'", "#121347 xai ignores model.base_url"),
 }
+# Rows that list from the configured endpoint today: they also carry the 404 / hang cells.
+CONTROL_ROWS = [r for r in ROWS if r.supports_model_listing and r.name not in LISTING_KNOWN]
 
 _LIST = r"""
 import json, sys
@@ -64,57 +78,81 @@ print("RESULT=" + json.dumps({"ok": bool(r.success), "provider": r.target_provid
                               "error": r.error_message, "want": normalize_provider(sys.argv[1]),
                               "got": normalize_provider(r.target_provider or "")}))
 """
+_KINDS = {"list": (_LIST, {}), "switch": (_SWITCH, {}),
+          "list404": (_LIST, {"models_status": 404}), "listhang": (_LIST, {"models_hang_s": HANG_S})}
 
 
-def _probe(row: Row, root: Path, script: str) -> dict:
+def _probe(row: Row, root: Path, kind: str) -> dict:
+    script, fake_kwargs = _KINDS[kind]
     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']}"
+    with CatalogFake(models=[unique, f"catalog-{row.name}-beta"], routes=row.routes(), **fake_kwargs) as fake:
+        base = fake.origin + row.base_path
         home = write_home(root, {"provider": row.name, "base_url": base})
+        started = time.monotonic()
         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)
+                                  text=True, env=hermetic_env(home, {**keys, **fake.proxy_env()}),
+                                  timeout=PROBE_TIMEOUT, stdin=subprocess.DEVNULL)
             out, err = proc.stdout, proc.stderr
         except subprocess.TimeoutExpired:
-            out, err = "", "TIMEOUT after 90s"
+            out, err = "", f"TIMEOUT after {PROBE_TIMEOUT}s"
+        wall = round(time.monotonic() - started, 1)
         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}
+    return {**res, "unique": unique, "base": base, "wall_s": wall, "egress": egress,
+            "listing_paths": [r.path for r in listings], "canonical_host_hit": row.vendor_host() 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))]
+    jobs = [(r, k) for r in ROWS for k in ("list", "switch")] + [
+        (r, k) for r in CONTROL_ROWS for k in ("list404", "listhang")]
     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()}
+        futs = {(r.name, k): pool.submit(_probe, r, Path(tmp_path_factory.mktemp(f"{k}-{r.name}")), k)
+                for r, k in jobs}
+        return {key: f.result() for key, 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("/"),
-    }
+def _listing_facts(ls: dict) -> str:
+    return (f"canonical_host_hit={ls['canonical_host_hit']} probe_error={ls.get('error')!r} "
+            f"paths={ls['listing_paths']} egress={ls['egress']} wall={ls['wall_s']}s ids={(ls.get('ids') or [])[:6]}")
 
 
 @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())))}")
+def test_listing_uses_configured_endpoint(row: Row, probes: dict) -> None:
+    ls = probes[(row.name, "list")]
+    route = row.listing_route()
+    ok = (not row.supports_model_listing) or (
+        "error" not in ls and bool(ls["listing_paths"]) and all(bare_path(p) == route for p in ls["listing_paths"])
+        and not ls["canonical_host_hit"] and ls["unique"] in (ls.get("ids") or []))
+    with known_gate(LISTING_KNOWN.get(row.name)):
+        gate(ok, f"{_listing_facts(ls)}\n{row.name}: picker did not list the configured endpoint (want {route})")
+
+
+@pytest.mark.parametrize("row", [pytest.param(r, id=r.name) for r in ROWS])
+def test_switch_resolves_requested_provider(row: Row, probes: dict) -> None:
+    sw = probes[(row.name, "switch")]
+    # Same provider (alias-normalised: ai-gateway == vercel) at the configured endpoint.
+    ok = 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("/")
+    facts = (f"ok={sw.get('ok')} got={sw.get('got')!r} base_url={sw.get('base_url')!r} error={sw.get('error')!r} "
+             f"egress={sw['egress']}")
+    with known_gate(SWITCH_KNOWN.get(row.name)):
+        gate(ok, f"{facts}\n{row.name}: switch did not resolve to {sw.get('want')!r} at {sw['base']}")
+
+
+@pytest.mark.parametrize("kind", ["list404", "listhang"])
+@pytest.mark.parametrize("row", [pytest.param(r, id=r.name) for r in CONTROL_ROWS])
+def test_listing_failure_degrades_without_vendor_host(row: Row, kind: str, probes: dict) -> None:
+    """A 404 or a hanging listing at the configured endpoint: the picker still returns (within
+    ``HANG_BOUND_S``), asked the exact listing path, never crashed and never fell back to the vendor."""
+    ls = probes[(row.name, kind)]
+    route = row.listing_route()
+    assert "error" not in ls, f"{row.name} [{kind}]: listing crashed\n{_listing_facts(ls)}"
+    assert ls["wall_s"] < HANG_BOUND_S, f"{row.name} [{kind}]: listing blocked the picker\n{_listing_facts(ls)}"
+    assert ls["listing_paths"] and all(bare_path(p) == route for p in ls["listing_paths"]), (
+        f"{row.name} [{kind}]: never asked {route}\n{_listing_facts(ls)}")
+    assert not ls["canonical_host_hit"], f"{row.name} [{kind}]: fell back to the vendor host\n{_listing_facts(ls)}"
+    assert ls["unique"] not in (ls.get("ids") or []), f"{row.name} [{kind}]: fake ids without a listing?"

tests/e2e/core/providers/test_catalog_matrix_0.py

@@ -2,70 +2,35 @@
 
 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.
+(redirected via ``model.base_url``; the fake answers only the exact configured path) 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 another provider's host); the session's
+usage equals what the fake reported and unknown pricing is not a silent $0. Open bugs are gated on
+their own signature in ``_catalog_helpers.MATRIX_KNOWN``.
 """
 
 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,
+    SHARDS, Row, TurnResult, check_row, discover_catalog, drive_shard, shard_of, shard_params,
 )
 
 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()}
+    return drive_shard(SHARD, tmp_path_factory)
 
 
-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())
+@pytest.mark.parametrize("row", shard_params(SHARD))
 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())))}")
+    check_row(row, turns)
 
 
 def test_catalog_is_discovered_not_listed() -> None:
@@ -73,7 +38,7 @@ def test_catalog_is_discovered_not_listed() -> None:
     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}
+    names = {r.name for r in discover_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

@@ -2,59 +2,28 @@
 
 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.
+(redirected via ``model.base_url``; the fake answers only the exact configured path) 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 another provider's host); the session's
+usage equals what the fake reported and unknown pricing is not a silent $0. Open bugs are gated on
+their own signature in ``_catalog_helpers.MATRIX_KNOWN``.
 """
 
 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,
-)
+from tests.e2e.core.providers._catalog_helpers import Row, TurnResult, check_row, drive_shard, shard_params
 
 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()}
+    return drive_shard(SHARD, tmp_path_factory)
 
 
-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())
+@pytest.mark.parametrize("row", shard_params(SHARD))
 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())))}")
+    check_row(row, turns)

tests/e2e/core/providers/test_catalog_matrix_2.py

@@ -2,59 +2,28 @@
 
 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.
+(redirected via ``model.base_url``; the fake answers only the exact configured path) 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 another provider's host); the session's
+usage equals what the fake reported and unknown pricing is not a silent $0. Open bugs are gated on
+their own signature in ``_catalog_helpers.MATRIX_KNOWN``.
 """
 
 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,
-)
+from tests.e2e.core.providers._catalog_helpers import Row, TurnResult, check_row, drive_shard, shard_params
 
 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()}
+    return drive_shard(SHARD, tmp_path_factory)
 
 
-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())
+@pytest.mark.parametrize("row", shard_params(SHARD))
 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())))}")
+    check_row(row, turns)

tests/e2e/core/providers/test_catalog_oauth.py

@@ -7,12 +7,13 @@ 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 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.
+Open bugs are merge-order-safe, message-gated run-time xfails (``KNOWN`` +
+``tests.e2e.core._pending_fixes.known_failure``): a cell XFAILs only while its gated assertion
+fails with that bug's signature, any other failure stays red, and a fixed bug simply passes.
 
-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
-exchange (hardcoded api.github.com).
+NOT COVERED (not redirectable to a loopback fake): openai-codex and qwen-oauth refresh (token URLs
+are module constants, no env/config override) and the Copilot token exchange (hardcoded
+api.github.com).
 """
 
 from __future__ import annotations
@@ -27,7 +28,7 @@ from typing import Any
 
 import pytest
 
-from tests.e2e.core.providers._catalog_helpers import strict_known
+from tests.e2e.core.providers._catalog_helpers import Known, gate, known_gate
 from tests.fakes.providers.catalog_fake import CatalogFake
 from tests.fakes.providers.catalog_oauth import NOUS_INVOKE_SCOPE, OAuthFake, make_jwt
 
@@ -39,13 +40,13 @@ TURN_TIMEOUT = 120.0
 # Public, credential-free model-metadata catalog (pricing/context lookups); never carries a vendor token.
 CREDENTIAL_FREE_HOSTS = frozenset({"models.dev:443"})
 
-# (pattern matched against the failing assertion text, reason). Delete an entry when its fix lands.
-KNOWN: dict[str, tuple[str, str]] = {
-    "device_interval": (
-        r"device poll gap .* < server interval",
+# Signature regex on the gated assertion's own text. Delete an entry when its fix lands.
+KNOWN: dict[str, Known] = {
+    "device_interval": Known(
+        r"^device poll gap \d+\.\d+s < server interval",
         "#121163 Nous device-code login polls at 1s, ignoring the server's interval"),
-    "nous_401_retry_route": (
-        r"401 recovery retry left NOUS_INFERENCE_BASE_URL",
+    "nous_401_retry_route": Known(
+        r"^401 recovery retry left NOUS_INFERENCE_BASE_URL: egress to \[[^\]]*'inference-api\.nousresearch\.com:443'",
         "#121323 Nous 401 pool recovery retries on the stored production host, "
         "dropping the NOUS_INFERENCE_BASE_URL override"),
 }
@@ -218,9 +219,9 @@ 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 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]})")
+    with known_gate(KNOWN["device_interval"]):
+        gate(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]})")
 
 
 # --- Nous token refresh ----------------------------------------------------------------------
@@ -272,11 +273,11 @@ 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 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}"
-            assert proc.returncode == 0 and fake.reply in proc.stdout, f"turn failed after refresh\n{info}"
+        with known_gate(KNOWN["nous_401_retry_route"]):
+            gate(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}"
+        assert proc.returncode == 0 and fake.reply in proc.stdout, f"turn failed after refresh\n{info}"
 
 
 # --- MiniMax OAuth refresh -------------------------------------------------------------------
@@ -305,20 +306,3 @@ def test_minimax_oauth_expired_token_refreshes_and_persists(tmp_path, sentinel)
         assert state.get("access_token") == rotated["access_token"], f"MiniMax access token not persisted\n{info}"
         assert _row(home.auth(), "openrouter", "or-1")["access_token"] == OPENROUTER_ROW["access_token"]
         assert not _vendor_egress(sentinel), f"turn leaked egress: {_vendor_egress(sentinel)}"
-
-
-# --- not redirectable ------------------------------------------------------------------------
-
-_UNREDIRECTABLE = {
-    "openai-codex": "token refresh URL is the module constant CODEX_OAUTH_TOKEN_URL (auth.openai.com); "
-                    "no env/config override, so a refresh cannot reach a loopback fake",
-    "qwen-oauth": "token refresh URL is the module constant QWEN_OAUTH_TOKEN_URL (chat.qwen.ai); "
-                  "no env/config override",
-    "copilot": "token exchange URL is hardcoded to api.github.com",
-}
-
-
-@pytest.mark.parametrize("provider", sorted(_UNREDIRECTABLE))
-def test_oauth_refresh_not_redirectable(provider: str) -> None:
-    pytest.skip(f"{provider}: {_UNREDIRECTABLE[provider]}")
-

tests/fakes/providers/catalog_fake.py

@@ -1,13 +1,17 @@
 """Multi-dialect recording loopback provider for the provider-catalog E2E matrix.
 
 One real HTTP server on 127.0.0.1 that answers every wire dialect the bundled
-model-provider plugins speak, routed by request path:
+model-provider plugins speak, at EXACT paths only (``routes``: path -> dialect, built by the
+caller from the base URL it configured; anything else is a 404, so a request that lands under
+the right prefix but at the wrong path is visible, never silently answered):
 
-* ``POST …/chat/completions`` — OpenAI Chat Completions (JSON or SSE)
-* ``POST …/messages``         — Anthropic Messages (JSON or SSE)
-* ``POST …/responses``        — OpenAI Responses (SSE event stream)
-* ``GET  …/models``           — model listing (OpenAI ``data`` shape), or a
-  scripted 404 / hang so picker fallbacks can be exercised
+* ``chat``      — OpenAI Chat Completions (JSON or SSE)
+* ``anthropic`` — Anthropic Messages (JSON or SSE)
+* ``responses`` — OpenAI Responses (SSE event stream)
+* ``listing``   — ``GET`` model listing (OpenAI ``data`` shape), or a scripted
+  status / hang so picker fallbacks can be exercised
+
+Without ``routes`` the server answers nothing (egress sentinel only).
 
 Every request (method, path, headers, body) is recorded, so a test can assert on
 exactly which credential reached which host in which header.
@@ -40,19 +44,13 @@ class Recorded:
     path: str
     headers: dict[str, str]
     body: Any
+    # Dialect of the exact route the request hit, or "unknown" (answered 404).
+    dialect: str = "unknown"
     t: float = field(default_factory=time.time)
 
-    @property
-    def dialect(self) -> str:
-        return path_dialect(self.path) if self.method == "POST" else "listing"
 
-
-def path_dialect(path: str) -> str:
-    p = path.split("?", 1)[0].rstrip("/")
-    for suffix, name in (("/chat/completions", "chat"), ("/messages", "anthropic"), ("/responses", "responses")):
-        if p.endswith(suffix):
-            return name
-    return "unknown"
+def bare_path(path: str) -> str:
+    return path.split("?", 1)[0]
 
 
 class CatalogFake:
@@ -68,7 +66,9 @@ class CatalogFake:
         models_status: int = 200,
         models_hang_s: float = 0.0,
         fail_status: int | None = None,
+        routes: dict[str, str] | None = None,
     ) -> None:
+        self.routes = dict(routes or {})
         self.tool_name = tool_name
         self.tool_args = tool_args or {}
         self.final_text = final_text
@@ -298,9 +298,11 @@ def _handler_for(fake: CatalogFake) -> type[BaseHTTPRequestHandler]:
         def do_GET(self) -> None:  # noqa: N802
             if self._refuse_egress("GET"):
                 return
-            fake._record(Recorded("GET", self.path, self._headers(), None))
-            if not self.path.split("?", 1)[0].rstrip("/").endswith("/models"):
-                self._json(404, {"error": {"message": "not found"}})
+            dialect = fake.routes.get(bare_path(self.path), "unknown")
+            dialect = dialect if dialect == "listing" else "unknown"
+            fake._record(Recorded("GET", self.path, self._headers(), None, dialect))
+            if dialect != "listing":
+                self._json(404, {"error": {"message": f"not found: {self.path}"}})
                 return
             if fake.models_hang_s:
                 fake._stop.wait(fake.models_hang_s)
@@ -322,8 +324,9 @@ def _handler_for(fake: CatalogFake) -> type[BaseHTTPRequestHandler]:
                 body = json.loads(raw or b"{}")
             except json.JSONDecodeError:
                 body = {"_raw": raw.decode("utf-8", "replace")}
-            fake._record(Recorded("POST", self.path, self._headers(), body))
-            dialect = path_dialect(self.path)
+            dialect = fake.routes.get(bare_path(self.path), "unknown")
+            dialect = dialect if dialect in ("chat", "anthropic", "responses") else "unknown"
+            fake._record(Recorded("POST", self.path, self._headers(), body, dialect))
             if dialect == "unknown":
                 self._json(404, {"error": {"message": f"unsupported path {self.path}"}})
                 return