feat: model auto-discovery for org-wide performance routing

README.md

@@ -76,6 +76,7 @@ Non-mock providers must use `https://` URLs and a **resolvable KV credential** 
 
 One public interface:
 
+- `/v1/models` lists `contextual-orchestrator` plus the current worker ids. Live discovery runs only from these KV-registered names: `NVIDIA_NIM_API_KEY`, `NVIDIA_NIM_API_KEY_SUB`, `BYTEZ_API_KEY`, `OPENROUTER_API_KEY`, and `OPENAI_API_KEY`. When none of those names is registered, discovery is skipped and the seed/mock pool is kept. The two NIM nemotron ids are a floor only after discovery runs and every catalog result is empty or fails, and only when a NIM credential is registered. See [docs/model_discovery.md](docs/model_discovery.md).
 - `/v1/chat/completions` accepts normal chat messages, and `"stream": true` returns an OpenAI-compatible `text/event-stream` of `chat.completion.chunk` deltas terminated by `data: [DONE]`. In **route** mode the worker's tokens are streamed live as they arrive from the provider (real token streaming); in **conduct** mode the multi-step answer is produced then framed as deltas (a workflow can't honestly token-stream a synthesizer that hasn't run yet).
 - `TaskOrchestrator.complete()` decides whether to route to one worker or run a short workflow.
 - `TaskOrchestrator.compare_to_baseline(prompts, mode)` (CLI `--eval PROMPT...`) measures the orchestration engine against a single-worker baseline — per-prompt and aggregate latency plus a structural coverage delta (contributing steps + verifier-pass presence). It is a measured tradeoff report, not a human-quality claim.
@@ -252,6 +253,10 @@ python -m pip install --require-hashes -r requirements.lock
 python -m pip install --no-deps -e .
 python tests/test_self_check.py
 python tests/test_paper_contracts.py
+python tests/test_model_discovery.py
+python tests/test_original_list_price.py
+python tests/test_models_list.py
+python tests/test_compute_allocation.py
 python tests/test_admin_contract.py
 python tests/test_conventions.py
 python tests/test_api_contract.py

conductor/product.md

@@ -24,7 +24,8 @@ Provide one API and one domain model:
 - route simple work to one selected worker;
 - conduct complex work through planner, worker, verifier, and synthesizer steps;
 - keep worker visibility explicit with access lists;
-- make the agent pool configurable data.
+- make the agent pool configurable data, composed by live auto-discovery
+  from registered provider keys (two NIM ids are a floor only).
 - expose an admin console for operators to inspect agents, policy, workflow trace, and audit state.
 
 ## Source-backed Product Bets

contextual_orchestrator/__init__.py

@@ -37,7 +37,16 @@
 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 .model_discovery import (
+    DISCOVERY_CREDENTIAL_NAMES,
+    FLOOR_DEFAULT_MODEL_ID,
+    FLOOR_SMALL_MODEL_ID,
+    apply_discovered_pool,
+    discover_model_catalog,
+    list_served_models,
+)
+from .orchestrator import ModelAgent, WorkflowStep, load_agents, known_agent_comparison_cost
+from .passthrough_failover import ResilientTaskOrchestrator as TaskOrchestrator
 from .token_counting import HeuristicTokenCounter, build_token_counter
 
 __all__ = [
@@ -48,6 +57,13 @@
     "get_credential",
     "register_credential",
     "NotConfigured",
+    "DISCOVERY_CREDENTIAL_NAMES",
+    "FLOOR_DEFAULT_MODEL_ID",
+    "FLOOR_SMALL_MODEL_ID",
+    "apply_discovered_pool",
+    "discover_model_catalog",
+    "list_served_models",
+    "known_agent_comparison_cost",
     # cost review
     "ATTRIBUTION_DIMENSIONS",
     "AttributionDimensions",

contextual_orchestrator/__main__.py

@@ -8,7 +8,9 @@
 import sys
 
 from .credentials import register_credential
-from .orchestrator import ModelClient, TaskOrchestrator, load_agents
+from .model_discovery import apply_discovered_pool
+from .orchestrator import ModelClient, load_agents
+from .passthrough_failover import ResilientTaskOrchestrator
 from .server import SecurityConfig, serve
 
 
@@ -95,7 +97,7 @@ def main() -> None:
     args = parser.parse_args()
 
     client = ModelClient(ca_bundle=args.provider_ca_bundle, verify_tls=not args.insecure_skip_tls_verify)
-    orchestrator = TaskOrchestrator(
+    orchestrator = ResilientTaskOrchestrator(
         load_agents(args.agents),
         client=client,
         state_db=args.state_db,
@@ -104,6 +106,12 @@ def main() -> None:
         budget_max_cost_usd=args.budget_max_cost_usd,
         cache_ttl=args.cache_ttl,
     )
+    # Product auto-discovery: when any of the five provider keys is in the KV,
+    # replace the seed pool with the live catalog. Catalog HTTPS reuses this
+    # ModelClient's TLS settings. NIM floor only if every fetch is empty and a
+    # NIM credential is registered. Unregistered keys are skipped — never
+    # os.getenv as a "key exists" signal.
+    apply_discovered_pool(orchestrator)
 
     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

@@ -17,6 +17,30 @@
         }
     },
     "paths": {
+        "/v1/models": {
+            "get": {
+                "operationId": "list_served_models",
+                "summary": "List the gateway model plus discovered or floor worker ids",
+                "security": [{"inference_bearer_auth": []}],
+                "responses": {"200": {"description": "OpenAI-compatible model list"}},
+            }
+        },
+        "/api/v1/provider_catalogs": {
+            "get": {
+                "operationId": "get_provider_catalogs",
+                "summary": "Read the last secret-redacted discovery snapshot",
+                "security": [{"admin_bearer_auth": []}],
+                "responses": {"200": {"description": "Provider catalog snapshot"}},
+            }
+        },
+        "/api/v1/provider_catalogs/refresh": {
+            "post": {
+                "operationId": "refresh_provider_catalogs",
+                "summary": "Re-run live discovery from KV-registered provider keys",
+                "security": [{"admin_bearer_auth": []}],
+                "responses": {"200": {"description": "Refreshed catalog snapshot"}},
+            }
+        },
         "/api/v1/agent_pools": {
             "get": {
                 "operationId": "list_agent_pools",

contextual_orchestrator/batch_routing.py

@@ -126,9 +126,10 @@ def cheapest_upstream(
 
     Cost-optimising upstream selection for load balancing: given candidate
     provider/model pairs, price each against the configurable price table for a
-    representative request shape and return the cheapest. Unpriced candidates
-    cost ``0`` and are treated as free (explicit, so a missing price is visible
-    rather than silently expensive). Ties keep input order.
+    representative request shape and return the cheapest **known-cost**
+    candidate. Unpriced models are skipped — unknown is never treated as
+    free. Ties keep input order. If every candidate is unpriced, return
+    ``None``.
     """
     if not candidates:
         return None
@@ -137,9 +138,11 @@ def cheapest_upstream(
     for candidate in candidates:
         provider = candidate.get("provider", "")
         model = candidate.get("model", "")
-        cost, _currency = price_book.compute_cost(
+        cost, _currency = price_book.known_compute_cost(
             provider, model, assumed_prompt_tokens, assumed_completion_tokens
         )
+        if cost is None:
+            continue
         if best_cost is None or cost < best_cost:
             best_cost = cost
             best = candidate

contextual_orchestrator/cost_ledger.py

@@ -35,6 +35,8 @@
 from typing import Any, Dict, List, Optional, Protocol
 import uuid
 
+from .price_honesty import optional_finite_price
+
 
 # ---------------------------------------------------------------------------
 # Attribution dimensions
@@ -114,6 +116,8 @@ class PriceEntry:
     prompt_price_per_1k: float
     completion_price_per_1k: float
     currency_code: str = "USD"
+    original_list_prompt_per_1k: float | None = None
+    original_list_completion_per_1k: float | None = None
 
     def as_dict(self) -> Dict[str, Any]:
         """Serialize the price entry for KV storage / reporting."""
@@ -123,6 +127,8 @@ def as_dict(self) -> Dict[str, Any]:
             "prompt_price_per_1k": self.prompt_price_per_1k,
             "completion_price_per_1k": self.completion_price_per_1k,
             "currency_code": self.currency_code,
+            "original_list_prompt_per_1k": self.original_list_prompt_per_1k,
+            "original_list_completion_per_1k": self.original_list_completion_per_1k,
         }
 
 
@@ -162,12 +168,20 @@ def get_price(self, provider: str, model: str) -> Optional[PriceEntry]:
             raw = self._config.get(_PRICE_CATEGORY, _price_key(provider, "*"), None)
         if raw is None:
             return None
+        if not isinstance(raw, dict):
+            return None
+        billed_keys = ("prompt_price_per_1k", "completion_price_per_1k")
+        if not any(key in raw for key in billed_keys):
+            # A stub without billed-rate keys is unpriced, not promotional-free.
+            return None
         return PriceEntry(
             provider_name=raw.get("provider_name", provider),
             model_name=raw.get("model_name", model),
             prompt_price_per_1k=float(raw.get("prompt_price_per_1k", 0.0)),
             completion_price_per_1k=float(raw.get("completion_price_per_1k", 0.0)),
             currency_code=raw.get("currency_code", self.default_currency),
+            original_list_prompt_per_1k=optional_finite_price(raw.get("original_list_prompt_per_1k")),
+            original_list_completion_per_1k=optional_finite_price(raw.get("original_list_completion_per_1k")),
         )
 
     def compute_cost(
@@ -196,6 +210,43 @@ def compute_cost(
         )
         return float(total), entry.currency_code
 
+    def known_compute_cost(
+        self,
+        provider: str,
+        model: str,
+        prompt_tokens: int,
+        completion_tokens: int,
+    ) -> tuple[float | None, str]:
+        """Return a known cost, or ``None`` when the model is unpriced.
+
+        Selection must call this — not :meth:`compute_cost`, which records
+        unpriced usage as ``0.0`` so the ledger never fails. A promotional
+        billed rate of ``0`` with ``original_list_*`` uses the list price.
+        """
+        entry = self.get_price(provider, model)
+        if entry is None:
+            return None, self.default_currency
+        billed_prompt = entry.prompt_price_per_1k
+        billed_completion = entry.completion_price_per_1k
+        if (
+            billed_prompt == 0.0
+            and billed_completion == 0.0
+            and (
+                entry.original_list_prompt_per_1k is not None
+                or entry.original_list_completion_per_1k is not None
+            )
+        ):
+            billed_prompt = entry.original_list_prompt_per_1k or 0.0
+            billed_completion = entry.original_list_completion_per_1k or 0.0
+        prompt_cost = (Decimal(prompt_tokens) / Decimal(1000)) * Decimal(str(billed_prompt))
+        completion_cost = (Decimal(completion_tokens) / Decimal(1000)) * Decimal(
+            str(billed_completion)
+        )
+        total = (prompt_cost + completion_cost).quantize(
+            Decimal("0.000001"), rounding=ROUND_HALF_UP
+        )
+        return float(total), entry.currency_code
+
 
 # ---------------------------------------------------------------------------
 # Usage records + stores
@@ -554,6 +605,66 @@ def __len__(self) -> int:
     "currency_code",
 )
 
+# Module-constant SQL selected by DB-API paramstyle. Semgrep flags f-string
+# interpolation at execute(); these statements are fixed literals and bind
+# values only through DB-API parameters.
+_USAGE_COLUMN_LIST = (
+    "usage_record_id, created_at, workflow_run_id, request_channel, route_mode, "
+    "provider_name, model_name, account_name, service_name, upstream_api, "
+    "team_name, group_name, company_name, prompt_tokens, completion_tokens, "
+    "total_tokens, cost_amount, currency_code"
+)
+SELECT_DIMENSION_SQL = {
+    "qmark": "SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = ?",
+    "pyformat": "SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = %s",
+}
+INSERT_DIMENSION_SQL = {
+    "qmark": (
+        "INSERT INTO cost_attribution_dimensions "
+        "(dimension_name, dimension_label, dimension_order) VALUES (?, ?, ?)"
+    ),
+    "pyformat": (
+        "INSERT INTO cost_attribution_dimensions "
+        "(dimension_name, dimension_label, dimension_order) VALUES (%s, %s, %s)"
+    ),
+}
+INSERT_USAGE_SQL = {
+    "qmark": (
+        "INSERT INTO llm_usage_records ("
+        + _USAGE_COLUMN_LIST
+        + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
+    ),
+    "pyformat": (
+        "INSERT INTO llm_usage_records ("
+        + _USAGE_COLUMN_LIST
+        + ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
+    ),
+}
+SELECT_USAGE_SQL = {
+    ("qmark", False, False): "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records",
+    ("qmark", True, False): (
+        "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records WHERE created_at >= ?"
+    ),
+    ("qmark", False, True): (
+        "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records WHERE created_at < ?"
+    ),
+    ("qmark", True, True): (
+        "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records "
+        "WHERE created_at >= ? AND created_at < ?"
+    ),
+    ("pyformat", False, False): "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records",
+    ("pyformat", True, False): (
+        "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records WHERE created_at >= %s"
+    ),
+    ("pyformat", False, True): (
+        "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records WHERE created_at < %s"
+    ),
+    ("pyformat", True, True): (
+        "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records "
+        "WHERE created_at >= %s AND created_at < %s"
+    ),
+}
+
 
 class SqlLedgerStore:
     """PEP-249 SQL ledger store (stdlib ``sqlite3`` or ``psycopg``).
@@ -569,9 +680,6 @@ def __init__(self, connection: Any, paramstyle: str = "qmark") -> None:
         self._create_schema()
         self._seed_dimension_catalog()
 
-    def _placeholder(self) -> str:
-        return "?" if self._paramstyle == "qmark" else "%s"
-
     def _create_schema(self) -> None:
         cur = self._conn.cursor()
         for statement in SCHEMA_SQL.strip().split(";"):
@@ -580,54 +688,35 @@ def _create_schema(self) -> None:
         self._conn.commit()
 
     def _seed_dimension_catalog(self) -> None:
-        ph = self._placeholder()
         cur = self._conn.cursor()
+        select_sql = SELECT_DIMENSION_SQL[self._paramstyle]
+        insert_sql = INSERT_DIMENSION_SQL[self._paramstyle]
         for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG):
-            # (name,) is a real, separately-bound parameter, never string-concatenated into the
-            # query; this is a raw DB-API cursor (sqlite3/psycopg), not SQLAlchemy.
-            cur.execute(  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
-                f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}",  # nosec B608 - ph is a DB-API placeholder.
-                (name,),
-            )
+            cur.execute(select_sql, (name,))
             if cur.fetchone() is None:
-                cur.execute(
-                    "INSERT INTO cost_attribution_dimensions "
-                    f"(dimension_name, dimension_label, dimension_order) VALUES ({ph}, {ph}, {ph})",  # nosec B608 - ph is a DB-API placeholder.
-                    (name, label, order),
-                )
+                cur.execute(insert_sql, (name, label, order))
         self._conn.commit()
 
     def append(self, record: UsageRecord) -> None:
         """Insert a usage record row."""
         row = record.as_dict()
-        ph = self._placeholder()
-        placeholders = ", ".join(ph for _ in _USAGE_COLUMNS)
-        columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        # values are bound via the separate params tuple below; raw DB-API cursor, not SQLAlchemy.
-        cur.execute(  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
-            f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})",  # nosec B608 - columns are fixed _USAGE_COLUMNS.
+        cur.execute(
+            INSERT_USAGE_SQL[self._paramstyle],
             tuple(row.get(column) for column in _USAGE_COLUMNS),
         )
         self._conn.commit()
 
     def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[Dict[str, Any]]:
         """Return record rows in the optional half-open window."""
-        ph = self._placeholder()
-        clauses: List[str] = []
         params: List[Any] = []
         if start is not None:
-            clauses.append(f"created_at >= {ph}")
             params.append(start)
         if end is not None:
-            clauses.append(f"created_at < {ph}")
             params.append(end)
-        where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
-        columns = ", ".join(_USAGE_COLUMNS)
+        statement = SELECT_USAGE_SQL[(self._paramstyle, start is not None, end is not None)]
         cur = self._conn.cursor()
-        # clauses only ever contain the two fixed literal fragments above, never interpolated
-        # values (those are in params); raw DB-API cursor, not SQLAlchemy.
-        cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params))  # nosec B608 - columns and clauses are fixed.  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query
+        cur.execute(statement, tuple(params))
         return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()]