fix: fail-closed catalog redirects, withdraw overlay, and known-cost rank

README.md

@@ -115,7 +115,9 @@ One fused orchestration loop:
 - Deep path: a natural-language workflow is built with planner, worker, verifier, and synthesizer steps.
 - Each step has an access list, so workers see only the prior outputs intentionally exposed to them.
 - Agent definitions are data, so provider preference, exclusions, privacy constraints, and mock testing do not require code changes.
-- Provider calls are resilient: transient failures (timeouts, 429, 5xx) retry with full-jitter exponential backoff, while caller errors (4xx) fail fast. If an agent still fails, the request fails over to the next capability-matched agent in the pool, and a per-agent circuit breaker skips a persistently failing provider until it cools down. Failover is recorded in the trace (`served_agent_id`, `failover_from`).
+- Provider catalogs can overlay the seed pool after a KV credential is registered under `NVIDIA_NIM_API_KEY`, `NVIDIA_NIM_API_KEY_SUB`, `BYTEZ_API_KEY`, `OPENROUTER_API_KEY`, or `OPENAI_API_KEY`. CLI: `refresh-provider-catalog`. Serve: `--refresh-provider-catalog`. Admin: `POST /api/v1/provider_catalogs/refresh`. Refresh is fail-closed (last-known-good kept; no invented models). Bytez uses native `https://api.bytez.com/models/v2`, not OpenAI `/v1/models`. See [docs/provider_catalog.md](docs/provider_catalog.md).
+- Route selection is capability-first, then minimum **known** price (`price_per_million` / catalog prices). Unpriced models are not treated as free. A free channel with a known list/original catalog price is compared at that list price, not as cost 0.0; explicit $0 with no list price may compete as 0. Trace `selection_reason` records why a worker won. Quality/Pareto beyond tags is issue #86 and is not claimed here.
+- Provider calls are resilient: transient failures (timeouts, 429, 5xx) retry with full-jitter exponential backoff, while caller errors (4xx) fail fast. If the **already chosen** agent still fails, the request fails over to the next capability-ranked agent, and a per-agent circuit breaker skips a persistently failing provider until it cools down. Failover is resilience after the primary errors, not the selection policy (`served_agent_id`, `failover_from`).
 
 See [docs/architecture.md](docs/architecture.md) for the source-backed analysis.
 
@@ -255,6 +257,8 @@ python tests/test_paper_contracts.py
 python tests/test_admin_contract.py
 python tests/test_conventions.py
 python tests/test_api_contract.py
+python tests/test_quality_cost_auto_routing.py
+python tests/test_provider_catalog.py
 python tests/test_security_hardening.py
 python tests/test_repository_security_metadata.py
 python tests/test_product_planning_contract.py

contextual_orchestrator/__init__.py

@@ -37,17 +37,51 @@
 from .cost_router import CostRoutingCoordinator
 from .credentials import NotConfigured, get_credential, register_credential
 from .kv_config import InMemoryConfigStore, get_config_store
-from .orchestrator import ModelAgent, TaskOrchestrator, WorkflowStep, load_agents
+from .orchestrator import ModelAgent, TaskOrchestrator, WorkflowStep, known_price_rank, load_agents
+from .provider_catalog import (
+    DEFAULT_PROVIDER_ACCOUNTS,
+    CatalogHttpError,
+    CatalogModelRecord,
+    DiscoveredModel,
+    InMemoryProviderCatalogStore,
+    ProviderAccount,
+    ProviderAwareModelClient,
+    ProviderCatalogHttpClient,
+    ProviderCatalogService,
+    ProviderCatalogUnavailable,
+    bootstrap_provider_credentials,
+    build_catalog_orchestrator,
+    known_catalog_prices,
+    normalize_models_document,
+    refresh_and_overlay,
+)
 from .token_counting import HeuristicTokenCounter, build_token_counter
 
 __all__ = [
     "ModelAgent",
     "TaskOrchestrator",
     "WorkflowStep",
     "load_agents",
+    "known_price_rank",
     "get_credential",
     "register_credential",
     "NotConfigured",
+    # provider catalog (PR #574 inventory/refresh on main)
+    "DEFAULT_PROVIDER_ACCOUNTS",
+    "ProviderAccount",
+    "DiscoveredModel",
+    "CatalogModelRecord",
+    "CatalogHttpError",
+    "ProviderCatalogUnavailable",
+    "InMemoryProviderCatalogStore",
+    "ProviderCatalogHttpClient",
+    "ProviderAwareModelClient",
+    "ProviderCatalogService",
+    "bootstrap_provider_credentials",
+    "normalize_models_document",
+    "known_catalog_prices",
+    "refresh_and_overlay",
+    "build_catalog_orchestrator",
     # cost review
     "ATTRIBUTION_DIMENSIONS",
     "AttributionDimensions",

contextual_orchestrator/__main__.py

@@ -9,6 +9,12 @@
 
 from .credentials import register_credential
 from .orchestrator import ModelClient, TaskOrchestrator, load_agents
+from .provider_catalog import (
+    InMemoryProviderCatalogStore,
+    ProviderAwareModelClient,
+    ProviderCatalogService,
+    refresh_and_overlay,
+)
 from .server import SecurityConfig, serve
 
 
@@ -55,11 +61,29 @@ def _register_credential_command(argv: list[str]) -> None:
     print(json.dumps({"registered": args.name, "backend": "kv"}, ensure_ascii=False))
 
 
+def _refresh_provider_catalog_command(argv: list[str]) -> None:
+    """Refresh discovered models from registered KV credentials and print a secret-free summary."""
+    parser = argparse.ArgumentParser(
+        prog="python -m contextual_orchestrator refresh-provider-catalog",
+        description="Discover provider catalogs for registered KV credentials and overlay the worker pool.",
+    )
+    parser.add_argument("--agents", default="examples/agents.mock.json", help="Seed agent config JSON.")
+    parser.add_argument("--force-refresh", action="store_true", help="Bypass the refresh throttle.")
+    args = parser.parse_args(argv)
+    orchestrator = TaskOrchestrator(load_agents(args.agents), client=ProviderAwareModelClient())
+    service = ProviderCatalogService(store=InMemoryProviderCatalogStore())
+    summary = refresh_and_overlay(orchestrator, service, force_refresh=args.force_refresh)
+    print(json.dumps(summary, ensure_ascii=False, indent=2))
+
+
 def main() -> None:
     """Parse CLI options and run bootstrap, prompt completion, or the HTTP server."""
     if len(sys.argv) > 1 and sys.argv[1] == "register-credential":
         _register_credential_command(sys.argv[2:])
         return
+    if len(sys.argv) > 1 and sys.argv[1] == "refresh-provider-catalog":
+        _refresh_provider_catalog_command(sys.argv[2:])
+        return
 
     parser = argparse.ArgumentParser(description="Route or conduct chat requests across model agents.")
     parser.add_argument("prompt", nargs="?", help="User prompt for CLI mode.")
@@ -80,6 +104,14 @@ def main() -> None:
                         help="Base URL of a Clearfolio deployment to use as the admin document viewer (default: disabled).")
     parser.add_argument("--agents-db", default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_AGENTS_DB") or None,
                         help="Optional sqlite path so runtime agent-pool changes (add/patch/remove) survive restarts.")
+    parser.add_argument(
+        "--refresh-provider-catalog",
+        action="store_true",
+        help=(
+            "After loading seed agents, refresh the five org-credential catalogs from the KV "
+            "and overlay discovered workers. Failures are audited and do not invent models."
+        ),
+    )
     parser.add_argument("--provider-ca-bundle", default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_PROVIDER_CA_BUNDLE") or None,
                         help="Path to a CA bundle used to verify provider TLS (e.g. a corporate gateway root).")
     parser.add_argument("--insecure-skip-tls-verify", action="store_true",
@@ -94,7 +126,8 @@ def main() -> None:
                         help="Measure orchestration vs a single-worker baseline on these prompts and print the report.")
     args = parser.parse_args()
 
-    client = ModelClient(ca_bundle=args.provider_ca_bundle, verify_tls=not args.insecure_skip_tls_verify)
+    client_cls = ProviderAwareModelClient if args.refresh_provider_catalog else ModelClient
+    client = client_cls(ca_bundle=args.provider_ca_bundle, verify_tls=not args.insecure_skip_tls_verify)
     orchestrator = TaskOrchestrator(
         load_agents(args.agents),
         client=client,
@@ -104,6 +137,10 @@ def main() -> None:
         budget_max_cost_usd=args.budget_max_cost_usd,
         cache_ttl=args.cache_ttl,
     )
+    if args.refresh_provider_catalog:
+        service = ProviderCatalogService(store=InMemoryProviderCatalogStore())
+        orchestrator.catalog_service = service
+        refresh_and_overlay(orchestrator, service, require_candidates=False)
 
     if args.eval:
         print(json.dumps(orchestrator.compare_to_baseline(args.eval, mode=args.mode), ensure_ascii=False, indent=2))

contextual_orchestrator/api_contract.py

@@ -532,6 +532,33 @@
                 "responses": {"200": {"description": "Evaluation run"}},
             }
         },
+        "/api/v1/provider_catalogs/latest": {
+            "get": {
+                "operationId": "get_provider_catalog",
+                "summary": "Read the last fail-closed provider catalog refresh summary",
+                "security": [{"admin_bearer_auth": []}],
+                "responses": {"200": {"description": "Catalog refresh snapshot"}},
+            }
+        },
+        "/api/v1/provider_catalogs/refresh": {
+            "post": {
+                "operationId": "refresh_provider_catalog",
+                "summary": "Refresh provider catalogs from registered KV credentials and overlay workers",
+                "security": [{"admin_bearer_auth": []}],
+                "requestBody": {
+                    "required": False,
+                    "content": {
+                        "application/json": {
+                            "schema": {
+                                "type": "object",
+                                "properties": {"force_refresh": {"type": "boolean"}},
+                            },
+                        },
+                    },
+                },
+                "responses": {"200": {"description": "Catalog refresh summary"}},
+            }
+        },
         "/api/v1/locale_bundles/{locale_code}": {
             "get": {
                 "operationId": "get_locale_bundle",

contextual_orchestrator/orchestrator.py

@@ -10,6 +10,7 @@
 import hashlib
 import ipaddress
 import json
+import math
 import os
 from pathlib import Path
 import random
@@ -48,6 +49,22 @@ def estimate_tokens(text: str) -> int:
     return (len(text) + 3) // 4 if text else 0
 
 
+def known_price_rank(value: Any) -> tuple[int, float]:
+    """Rank valid nonnegative price evidence ahead of unpriced metadata.
+
+    The second value is negated because callers sort descending: a lower known
+    price therefore ranks ahead of a higher known price. Missing, boolean,
+    nonnumeric, negative, NaN, and infinite values are unpriced rather than free.
+    Explicit ``0`` is a valid known price.
+    """
+    if type(value) not in (int, float):
+        return (0, 0.0)
+    normalized = float(value)
+    if not math.isfinite(normalized) or normalized < 0:
+        return (0, 0.0)
+    return (1, -normalized)
+
+
 _COMMERCIAL_REPORT_CACHE: ContextVar[dict[tuple[Any, Any, Any], dict[str, Any]] | None] = ContextVar(
     "commercial_report_cache",
     default=None,
@@ -174,6 +191,8 @@ def as_dict(self) -> dict[str, Any]:
         """Return the API-safe policy snapshot for workflow records."""
         return {
             "route_p95_seconds": self.route_p95_seconds,
+            "routing_objective": "maximize_capability_then_minimize_known_cost",
+            "unpriced_model_policy": "unpriced_not_free",
             "conduct_hint_threshold": self.conduct_hint_threshold,
             "verifier_required": self.verifier_required,
             "workflow_planning": self.workflow_planning,
@@ -856,6 +875,7 @@ def __init__(
         # (zero behavior change). When set, runs/audit/analytics survive restart.
         self._store = _StateStore(state_db) if state_db else None
         self._commercial_report_cache_local = threading.local()
+        self._catalog_overlay_ids: set[str] = set()
         if self._store is not None:
             self._reload_state()
 
@@ -954,7 +974,8 @@ def stream_route(self, messages: list[ChatMessage], workflow_run_id: str | None
             "answer": answer,
             "trace": [
                 {"id": 0, "role": "worker", "agent_id": agent.id, "subtask": "Direct route (streamed)",
-                 "access": [], "output": answer}
+                 "access": [], "output": answer,
+                 "selection_reason": self._selection_reason(agent, text, "worker")}
             ],
             "policy_snapshot": self.policy.as_dict(),
             "verification": {"accepted": True, "reason": "single route path", "verifier_output": ""},
@@ -1313,6 +1334,57 @@ def add_agent(self, agent_pool_id: str, value: dict[str, Any]) -> dict[str, Any]
         )
         return self._agent_to_admin_payload(agent)
 
+    def overlay_discovered_agents(
+        self,
+        agents: list[ModelAgent],
+        prices: dict[str, float] | None = None,
+    ) -> dict[str, Any]:
+        """Merge catalog-derived workers onto the live pool without inventing models.
+
+        Seed / operator agents are kept. Discovered rows overlay by agent id.
+        A later successful catalog drops previously overlaid ids that are no
+        longer present so a withdrawn or injected model cannot persist.
+        Only finite nonnegative prices enter ``price_per_million``; missing
+        catalog prices stay unpriced rather than becoming free.
+        """
+        incoming_ids = {agent.id for agent in agents if not agent.disabled}
+        withdrawn_ids = self._catalog_overlay_ids - incoming_ids
+        existing = {agent.id: agent for agent in self.agents}
+        withdrawn_count = 0
+        for agent_id in withdrawn_ids:
+            removed = existing.pop(agent_id, None)
+            if removed is None:
+                continue
+            withdrawn_count += 1
+            if self._pool_store is not None:
+                self._pool_store.save(replace(removed, disabled=True))
+        added_ids: list[str] = []
+        for agent in agents:
+            if agent.disabled:
+                continue
+            if agent.id not in existing:
+                added_ids.append(agent.id)
+            existing[agent.id] = agent
+            if self._pool_store is not None:
+                self._pool_store.save(agent)
+        self.agents = [agent for agent in existing.values() if not agent.disabled]
+        if not self.agents:
+            raise ValueError("catalog overlay cannot remove the last enabled agent")
+        self._catalog_overlay_ids = incoming_ids
+        ingested = 0
+        for model_name, price in (prices or {}).items():
+            if known_price_rank(price)[0]:
+                self.price_per_million[model_name] = float(price)
+                ingested += 1
+        detail = {
+            "added_agent_count": len(added_ids),
+            "catalog_agent_count": len(agents),
+            "priced_model_count": ingested,
+            "withdrawn_agent_count": withdrawn_count,
+        }
+        self._append_audit_event("catalog_overlay_applied", detail)
+        return detail
+
     def remove_agent(self, agent_pool_id: str, worker_agent_id: str) -> dict[str, Any]:
         """Remove a worker agent from the pool; the pool must keep at least one enabled agent."""
         if agent_pool_id != "default":  # pragma: no cover
@@ -1351,6 +1423,7 @@ def route_once(self, messages: list[ChatMessage]) -> dict[str, Any]:
             "access": [],
             "latency_ms": round(latency_ms, 2),
             "output": answer,
+            "selection_reason": self._selection_reason(agent, text, "worker"),
         }
         if usage is not None:
             row["usage"] = usage
@@ -1525,9 +1598,37 @@ def _score_agent(self, agent: ModelAgent, role: str, lowered: str) -> tuple[int,
         return (role_score + domain_score + agent.priority, len(agent.tags), agent.id)
 
     def _ranked_agents(self, text: str, role: str) -> list[ModelAgent]:
-        """Agents sorted best-first for a role; the head is the primary, the tail are failovers."""
+        """Rank by maximum capability, then minimum trustworthy known price.
+
+        Failover still walks this list after the chosen primary errors. Price is
+        never the first key: a cheap summarizer cannot beat a coding worker on a
+        coding task. Extra catalog tags are not a capability score; they sort
+        after known cost. Unpriced models lose a cost comparison; they are not
+        free.
+        """
         lowered = text.lower()
-        return sorted(self.agents, key=lambda agent: self._score_agent(agent, role, lowered), reverse=True)
+
+        def selection_key(agent: ModelAgent) -> tuple[int, int, float, int, str]:
+            capability_score, tag_count, agent_id = self._score_agent(agent, role, lowered)
+            price_known, inverse_price = known_price_rank(self.price_per_million.get(agent.model))
+            return (capability_score, price_known, inverse_price, tag_count, agent_id)
+
+        return sorted(self.agents, key=selection_key, reverse=True)
+
+    def _selection_reason(self, agent: ModelAgent, text: str, role: str) -> dict[str, Any]:
+        """Explain why this agent won: capability first, known cost second."""
+        capability_score, tag_count, _agent_id = self._score_agent(agent, role, text.lower())
+        price = self.price_per_million.get(agent.model)
+        price_known, _inverse = known_price_rank(price)
+        return {
+            "routing_objective": "maximize_capability_then_minimize_known_cost",
+            "unpriced_model_policy": "unpriced_not_free",
+            "agent_id": agent.id,
+            "capability_score": capability_score,
+            "tag_count": tag_count,
+            "price_known": bool(price_known),
+            "price_per_million_usd": float(price) if price_known else None,
+        }
 
     def _select_agent(self, text: str, role: str) -> ModelAgent:
         selected = self._ranked_agents(text, role)[0]