diff --git a/.github/workflows/provider-catalog-sync.yml b/.github/workflows/provider-catalog-sync.yml new file mode 100644 index 000000000..1b951628a --- /dev/null +++ b/.github/workflows/provider-catalog-sync.yml @@ -0,0 +1,105 @@ +name: Provider catalog sync + +on: + workflow_dispatch: + schedule: + - cron: "17 * * * *" + +permissions: + contents: read + +concurrency: + group: provider-catalog-sync + cancel-in-progress: false + +jobs: + sync: + name: Bootstrap durable provider KV and model catalog + if: github.ref == 'refs/heads/main' + runs-on: ubuntu-latest + environment: production + timeout-minutes: 15 + steps: + - name: Checkout protected default branch + uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # actions/checkout@v7 + with: + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # actions/setup-python@v6 + with: + python-version: "3.12" + + - name: Install hash-pinned runtime dependencies + run: python -m pip install --disable-pip-version-check --no-input --require-hashes -r requirements.lock + + - name: Register credentials and refresh normalized model catalog + shell: bash + env: + CONTEXTUAL_ORCHESTRATOR_KV_BACKEND: postgres + CONTEXTUAL_ORCHESTRATOR_KV_DSN: ${{ secrets.CONTEXTUAL_ORCHESTRATOR_KV_DSN }} + CONTEXTUAL_ORCHESTRATOR_KV_PASSPHRASE: ${{ secrets.CONTEXTUAL_ORCHESTRATOR_KV_PASSPHRASE }} + NVIDIA_NIM_API_KEY: ${{ secrets.NVIDIA_NIM_API_KEY }} + NVIDIA_NIM_API_KEY_SUB: ${{ secrets.NVIDIA_NIM_API_KEY_SUB }} + BYTEZ_API_KEY: ${{ secrets.BYTEZ_API_KEY }} + OPENROUTER_API_KEY: ${{ secrets.OPENROUTER_API_KEY }} + OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} + run: | + set -euo pipefail + test -n "${CONTEXTUAL_ORCHESTRATOR_KV_DSN}" + test -n "${CONTEXTUAL_ORCHESTRATOR_KV_PASSPHRASE}" + python -m contextual_orchestrator.provider_catalog_bootstrap --model-limit 24 > provider-bootstrap-report.json + python - <<'PY' + import json + from pathlib import Path + + report = json.loads(Path('provider-bootstrap-report.json').read_text(encoding='utf-8')) + expected = { + 'NVIDIA_NIM_API_KEY', + 'NVIDIA_NIM_API_KEY_SUB', + 'BYTEZ_API_KEY', + 'OPENROUTER_API_KEY', + 'OPENAI_API_KEY', + } + registered = set(report['registered_credentials']) + if registered != expected: + raise SystemExit(f'credential inventory mismatch: {sorted(expected - registered)}') + if report['catalog_backend'] != 'postgres': + raise SystemExit('provider catalog is not durable PostgreSQL') + if report['catalog_model_count'] < 1 or report['eligible_model_count'] < 1: + raise SystemExit('provider catalog has no compatible serving model') + if not report['selected_agent_ids']: + raise SystemExit('provider catalog produced no serving candidates') + if report['enabled_agent_ids'] or report['durable_agent_pool']: + raise SystemExit('ephemeral Actions sync must not claim agent-pool activation') + print(json.dumps({ + 'registered_credentials': sorted(registered), + 'live_discovered_model_count': report['live_discovered_model_count'], + 'catalog_model_count': report['catalog_model_count'], + 'last_known_good_model_count': report['last_known_good_model_count'], + 'selected_agent_count': len(report['selected_agent_ids']), + 'catalog_refresh_failure_count': report['catalog_refresh_failure_count'], + 'providers_with_errors': report['providers_with_errors'], + }, sort_keys=True)) + PY + python - <<'PY' + import os + from pathlib import Path + + report = Path('provider-bootstrap-report.json').read_text(encoding='utf-8') + names = ( + 'NVIDIA_NIM_API_KEY', + 'NVIDIA_NIM_API_KEY_SUB', + 'BYTEZ_API_KEY', + 'OPENROUTER_API_KEY', + 'OPENAI_API_KEY', + ) + leaked = [ + name + for name in names + if os.environ[name].rstrip('\r\n') + and os.environ[name].rstrip('\r\n') in report + ] + if leaked: + raise SystemExit(f'provider bootstrap report leaked secret values for: {leaked}') + PY diff --git a/contextual_orchestrator/__main__.py b/contextual_orchestrator/__main__.py index 622d2642c..3424629af 100644 --- a/contextual_orchestrator/__main__.py +++ b/contextual_orchestrator/__main__.py @@ -16,7 +16,7 @@ agent_id_for, discover_all_models, refresh_price_book, - select_top_n_cheapest_discovered_agents, + select_bootstrap_discovered_agents, ) from .orchestrator import ( CONTEXTUAL_ORCHESTRATOR_CONTRACT_V1, @@ -211,7 +211,7 @@ def _discover_models_command(argv: list[str]) -> None: type=_non_negative_int, default=0, metavar="N", - help="Enable the N cheapest discovered agents in --agents-db (auto-optimization bootstrap; " + help="Enable a price-honest, provider-diverse discovered agent pool in --agents-db (auto-optimization bootstrap; " "requires --agents-db; 0 disables, the default, leaving every discovered agent inert).", ) args = parser.parse_args(argv) @@ -229,7 +229,7 @@ def _discover_models_command(argv: list[str]) -> None: ) bootstrap.sync_discovered_agents([agent_from_discovered(model) for model in discovered]) if args.enable_cheapest: - for model in select_top_n_cheapest_discovered_agents(discovered, price_book, args.enable_cheapest): + for model in select_bootstrap_discovered_agents(discovered, price_book, args.enable_cheapest): agent_id = agent_id_for(model) bootstrap.patch_agent("default", agent_id, {"status": "active"}) enabled_agent_ids.append(agent_id) diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index 3866396dc..daee00c62 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -31,6 +31,7 @@ import threading from dataclasses import dataclass, field from decimal import ROUND_HALF_UP, Decimal +import math import time from typing import Any, Dict, List, Optional, Protocol import uuid @@ -162,11 +163,30 @@ 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 + try: + if ( + "prompt_price_per_1k" not in raw + or "completion_price_per_1k" not in raw + ): + return None + prompt_price = float(raw["prompt_price_per_1k"]) + completion_price = float(raw["completion_price_per_1k"]) + except (OverflowError, TypeError, ValueError): + return None + if ( + not math.isfinite(prompt_price) + or not math.isfinite(completion_price) + or prompt_price < 0 + or completion_price < 0 + ): + 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)), + prompt_price_per_1k=prompt_price, + completion_price_per_1k=completion_price, currency_code=raw.get("currency_code", self.default_currency), ) diff --git a/contextual_orchestrator/credentials.py b/contextual_orchestrator/credentials.py index a466b13e0..3daf36283 100644 --- a/contextual_orchestrator/credentials.py +++ b/contextual_orchestrator/credentials.py @@ -50,6 +50,10 @@ def set(self, name: str, value: str) -> None: """Register (or replace) the secret stored under ``name``.""" ... + def delete(self, name: str) -> None: + """Remove one credential after an unvalidated candidate promotion.""" + ... + class InMemoryCredentialBackend: """Process-local credential registry for dev and tests (no Postgres needed).""" @@ -68,6 +72,11 @@ def set(self, name: str, value: str) -> None: with self._lock: self._store[name] = value + def delete(self, name: str) -> None: + """Remove ``name`` from the in-memory credential registry if present.""" + with self._lock: + self._store.pop(name, None) + # --- Postgres pgcrypto-encrypted credential registry ------------------------ # @@ -112,6 +121,15 @@ def __init__(self, dsn: str, passphrase: str) -> None: self._passphrase = passphrase self._ensured = False + @property + def connection_dsn(self) -> str: + """Return the bootstrap DSN for a colocated metadata store. + + Callers must treat this as connection material: never include it in logs, + reports, traces, or exceptions. Provider API keys remain inaccessible. + """ + return self._dsn + @classmethod def from_env(cls) -> "PostgresCredentialBackend": """Build the backend from bootstrap transport env vars (the only allowed env use). @@ -173,6 +191,17 @@ def set(self, name: str, value: str) -> None: # pragma: no cover - requires a l ) conn.commit() + def delete(self, name: str) -> None: # pragma: no cover - requires a live Postgres + """Delete one encrypted credential after a failed candidate promotion.""" + with self._connect() as conn: + self._ensure_schema(conn) + with conn.cursor() as cur: + cur.execute( + "DELETE FROM provider_credentials WHERE credential_name = %s", + (name,), + ) + conn.commit() + _backend: CredentialBackend | None = None _backend_lock = threading.Lock() @@ -216,3 +245,8 @@ def get_credential(name: str) -> str | None: def register_credential(name: str, value: str) -> None: """Register a named secret into the KV (used by the bootstrap CLI).""" get_backend().set(name, value) + + +def delete_credential(name: str) -> None: + """Remove a named credential from the KV after an unvalidated promotion.""" + get_backend().delete(name) diff --git a/contextual_orchestrator/model_discovery.py b/contextual_orchestrator/model_discovery.py index 738111039..bb9a91073 100644 --- a/contextual_orchestrator/model_discovery.py +++ b/contextual_orchestrator/model_discovery.py @@ -1,4 +1,4 @@ -"""Provider model-list discovery for chat-agent candidates. +"""Provider model-list discovery: turns registered KV credentials into agent candidates. Queries each configured provider's model-list endpoint over its OpenAI-compatible (or provider-specific) discovery API and returns :class:`DiscoveredModel` rows that @@ -9,21 +9,16 @@ (the KV registry), and a provider with nothing registered is silently skipped so registering a subset of the five supported keys still works. Stdlib only (``urllib.request``), matching this repo's dependency-free transport convention. - -This module owns the ordinary chat-agent discovery boundary. Provider catalogs may -mix chat, embedding, reranking, transcription, moderation, image, and realtime -models under one ``/models`` endpoint. Clearly non-chat identifiers are rejected -before they can be converted to workers, selected by cost, or persisted into the -chat agent pool. """ from __future__ import annotations import json +import math import re import urllib.error import urllib.request -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import TYPE_CHECKING, Any from .chat_capability import is_general_chat_agent_model_id @@ -90,7 +85,7 @@ class ProviderModelSource: @dataclass(frozen=True) class DiscoveredModel: - """One general chat-agent eligible model found on a provider, with pricing.""" + """One general-chat model found on a provider, with reported pricing.""" provider_name: str model_id: str @@ -127,25 +122,80 @@ def _fetch_json(url: str, *, api_key: str, auth_scheme: str, timeout: float) -> return json.loads(response.read().decode("utf-8")) +def _valid_price_component(value: object) -> bool: + """Return whether one price component is finite, numeric, and non-negative.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return False + try: + numeric = float(value) + except (TypeError, ValueError, OverflowError): + return False + return math.isfinite(numeric) and numeric >= 0.0 + + def _price_per_1k(value: Any) -> float | None: - """OpenAI-compatible providers report USD price per single token; convert to per-1K.""" - if value is None: + """Convert a trustworthy per-token USD price to per-1K, else return unknown.""" + if value is None or isinstance(value, bool): return None try: - return float(value) * 1000 - except (TypeError, ValueError): + per_1k = float(value) * 1000 + except (TypeError, ValueError, OverflowError): return None + return per_1k if _valid_price_component(per_1k) else None + + +def _serving_identity(model: DiscoveredModel) -> tuple[str, str]: + """Return the durable agent identity used by discovery synchronization.""" + return (model.provider_name, model.model_id) + + +def _source_tiebreaker(model: DiscoveredModel) -> tuple[str, str, str, str]: + """Choose deterministic transport metadata for an ambiguous duplicate row.""" + return ( + model.credential_name, + model.chat_base_url, + model.auth_scheme, + model.currency_code, + ) + + +def _deduplicate_discovered_models( + discovered: list[DiscoveredModel], +) -> list[DiscoveredModel]: + """Collapse duplicate agent identities and withhold conflicting price evidence. + + Exact duplicate catalog rows become one candidate. When the same provider/model + identity is repeated with conflicting metadata or prices, one deterministic + transport record is retained but its prices become unknown. Provider row order + therefore cannot fabricate a cheaper bootstrap candidate or consume failover + capacity twice. + """ + unique: dict[tuple[str, str], DiscoveredModel] = {} + for model in discovered: + identity = _serving_identity(model) + previous = unique.get(identity) + if previous is None: + unique[identity] = model + continue + if previous == model: + continue + chosen = min((previous, model), key=_source_tiebreaker) + unique[identity] = replace( + chosen, + prompt_price_per_1k=None, + completion_price_per_1k=None, + ) + return list(unique.values()) def _parse_openai_compatible(payload: Any, source: ProviderModelSource) -> list[DiscoveredModel]: - """Parse one OpenAI-compatible catalog into general chat-agent candidates.""" rows = payload.get("data") if isinstance(payload, dict) else None discovered: list[DiscoveredModel] = [] for row in rows if isinstance(rows, list) else []: if not isinstance(row, dict): continue model_id = row.get("id") - if not is_general_chat_agent_model_id(model_id): + if type(model_id) is not str or not model_id or not is_general_chat_agent_model_id(model_id): continue pricing = row.get("pricing") if isinstance(row.get("pricing"), dict) else {} discovered.append( @@ -159,18 +209,17 @@ def _parse_openai_compatible(payload: Any, source: ProviderModelSource) -> list[ completion_price_per_1k=_price_per_1k(pricing.get("completion")), ) ) - return discovered + return _deduplicate_discovered_models(discovered) def _parse_bytez(payload: Any, source: ProviderModelSource) -> list[DiscoveredModel]: - """Parse one Bytez chat catalog without admitting ineligible identifiers.""" rows = payload.get("output") if isinstance(payload, dict) else None discovered: list[DiscoveredModel] = [] for row in rows if isinstance(rows, list) else []: if not isinstance(row, dict): continue model_id = row.get("modelId") - if not is_general_chat_agent_model_id(model_id): + if type(model_id) is not str or not model_id or not is_general_chat_agent_model_id(model_id): continue discovered.append( DiscoveredModel( @@ -183,13 +232,13 @@ def _parse_bytez(payload: Any, source: ProviderModelSource) -> list[DiscoveredMo # per-1k pricing unset is more honest than a misleading estimate. ) ) - return discovered + return _deduplicate_discovered_models(discovered) def discover_provider_models( source: ProviderModelSource, *, timeout: float = DISCOVERY_TIMEOUT_SECONDS ) -> list[DiscoveredModel]: - """Discover chat candidates, or ``[]`` when the credential is not registered.""" + """Discover one provider's models, or ``[]`` if its credential is not registered.""" api_key = get_credential(source.credential_name) if not api_key: return [] @@ -210,7 +259,7 @@ def discover_all_models( *, timeout: float = DISCOVERY_TIMEOUT_SECONDS, ) -> tuple[list[DiscoveredModel], list[ProviderDiscoveryError]]: - """Discover chat candidates across providers with registered credentials. + """Discover models across every provider with a registered credential. One provider's failure never blocks the others: errors are collected and returned alongside whatever models were successfully discovered. @@ -222,7 +271,7 @@ def discover_all_models( discovered.extend(discover_provider_models(source, timeout=timeout)) except ProviderDiscoveryError as exc: errors.append(exc) - return discovered, errors + return _deduplicate_discovered_models(discovered), errors _SLUG_RE = re.compile(r"[^a-z0-9]+") @@ -239,7 +288,7 @@ def agent_id_for(discovered: DiscoveredModel) -> str: def agent_from_discovered(discovered: DiscoveredModel, *, priority: int = 0) -> ModelAgent: - """Build a disabled general chat agent or reject an ineligible record.""" + """Build a disabled general-chat agent or reject an ineligible record.""" if not is_general_chat_agent_model_id(discovered.model_id): raise ValueError("model is not eligible for a general chat agent") return ModelAgent( @@ -255,69 +304,175 @@ def agent_from_discovered(discovered: DiscoveredModel, *, priority: int = 0) -> ) +def _currency_is_comparable(currency_code: object, default_currency: object) -> bool: + """Return whether two ISO-style currency codes can be compared directly.""" + return ( + isinstance(currency_code, str) + and isinstance(default_currency, str) + and currency_code.strip().upper() == default_currency.strip().upper() + and bool(currency_code.strip()) + ) + + def refresh_price_book(discovered: list[DiscoveredModel], price_book: "PriceBook") -> int: - """Write every discovered chat model's known pricing into the price book. + """Write complete, comparable provider pricing into the discovery price book. - Returns the number of price rows written. A model without provider-reported - pricing is skipped rather than defaulted to 0 -- an unpriced model already - costs 0 under ``PriceBook.compute_cost``'s "explicit, not silently expensive" - contract, so writing a fabricated 0 row here would just hide that signal. + Both prompt and completion prices are required for the fixed 1K+1K ranking + workload. Partial, conflicting, non-finite, negative, or cross-currency + evidence remains unknown rather than acquiring an invented zero component. """ from .cost_ledger import PriceEntry written = 0 - for model in discovered: + for model in _deduplicate_discovered_models(discovered): if not is_general_chat_agent_model_id(model.model_id): continue - if model.prompt_price_per_1k is None and model.completion_price_per_1k is None: + if not ( + _valid_price_component(model.prompt_price_per_1k) + and _valid_price_component(model.completion_price_per_1k) + and _currency_is_comparable( + model.currency_code, + price_book.default_currency, + ) + ): continue price_book.set_price( PriceEntry( provider_name=model.provider_name, model_name=model.model_id, - prompt_price_per_1k=model.prompt_price_per_1k or 0.0, - completion_price_per_1k=model.completion_price_per_1k or 0.0, - currency_code=model.currency_code, + prompt_price_per_1k=float(model.prompt_price_per_1k), + completion_price_per_1k=float(model.completion_price_per_1k), + currency_code=model.currency_code.strip().upper(), ) ) written += 1 return written +def _discovery_price_key( + model: DiscoveredModel, + price_book: "PriceBook", +) -> tuple[int, float, str, str]: + """Rank comparable trustworthy prices first, then deterministic unknowns.""" + unknown = (1, 0.0, model.provider_name, model.model_id) + try: + entry = price_book.get_price(model.provider_name, model.model_id) + except (TypeError, ValueError, OverflowError): + return unknown + if entry is None: + return unknown + if not ( + _valid_price_component(entry.prompt_price_per_1k) + and _valid_price_component(entry.completion_price_per_1k) + and _currency_is_comparable( + entry.currency_code, + price_book.default_currency, + ) + ): + return unknown + try: + cost, currency = price_book.compute_cost( + model.provider_name, + model.model_id, + 1000, + 1000, + ) + except (TypeError, ValueError, OverflowError): + return unknown + if not ( + _valid_price_component(cost) + and _currency_is_comparable(currency, price_book.default_currency) + ): + return unknown + return (0, cost, model.provider_name, model.model_id) + + +def _provider_family(provider_name: str) -> str: + """Collapse credentials that share one upstream provider outage domain.""" + if provider_name in {"nvidia_nim", "nvidia_nim_sub"}: + return "nvidia_nim" + return provider_name + + def select_cheapest_discovered_agent( discovered: list[DiscoveredModel], price_book: "PriceBook" ) -> DiscoveredModel | None: - """Pick the lowest-cost general chat-agent model per the price book. - - Uses the same representative request cost as the top-N selector. Call - :func:`refresh_price_book` first so discovered pricing is visible; an - unpriced candidate costs ``0`` - under that selector's documented contract and is treated as free, not - unknown -- so a genuinely unpriced provider (e.g. Bytez, priced by - GPU-second rather than per token) will always look cheapest here. Fine for - "auto-pick something free to try," but callers doing real cost comparison - should refresh pricing for every candidate they care about first. + """Pick the cheapest candidate with trustworthy price evidence. + + A candidate without a price row is unknown, not free. Known prices therefore + sort first; when every candidate is unpriced, provider and model identifiers + provide deterministic fallback ordering without inventing a monetary value. """ - eligible = [model for model in discovered if is_general_chat_agent_model_id(model.model_id)] + eligible = [ + model + for model in _deduplicate_discovered_models(discovered) + if is_general_chat_agent_model_id(model.model_id) + ] if not eligible: return None - return min(eligible, key=lambda model: _discovered_cost(model, price_book)) + return min(eligible, key=lambda model: _discovery_price_key(model, price_book)) def select_top_n_cheapest_discovered_agents( discovered: list[DiscoveredModel], price_book: "PriceBook", limit: int ) -> list[DiscoveredModel]: - """Return the ``limit`` cheapest general chat-agent models in ascending cost.""" + """Return up to ``limit`` unique candidates, known-priced before unknown.""" if limit <= 0: return [] - eligible = [model for model in discovered if is_general_chat_agent_model_id(model.model_id)] + eligible = [ + model + for model in _deduplicate_discovered_models(discovered) + if is_general_chat_agent_model_id(model.model_id) + ] if not eligible: return [] + return sorted( + eligible, + key=lambda model: _discovery_price_key(model, price_book), + )[:limit] + - return sorted(eligible, key=lambda model: _discovered_cost(model, price_book))[:limit] +def select_bootstrap_discovered_agents( + discovered: list[DiscoveredModel], + price_book: "PriceBook", + limit: int, +) -> list[DiscoveredModel]: + """Build a deterministic, price-honest, provider-diverse initial pool. + + Candidates retain the known-price-first ordering above, but the first pass + takes at most one model from each independent provider family. Remaining + capacity is filled in the same deterministic cost order. NVIDIA NIM primary + and sub credentials are one outage domain, so they participate in the second + pass only after independently hosted providers have had a chance to enter. + Duplicate serving identities never consume capacity twice. + """ + if limit <= 0: + return [] + eligible = [ + model + for model in _deduplicate_discovered_models(discovered) + if is_general_chat_agent_model_id(model.model_id) + ] + if not eligible: + return [] + ranked = sorted( + eligible, + key=lambda model: _discovery_price_key(model, price_book), + ) + selected: list[DiscoveredModel] = [] + deferred: list[DiscoveredModel] = [] + provider_families: set[str] = set() + + for model in ranked: + family = _provider_family(model.provider_name) + if family in provider_families: + deferred.append(model) + continue + provider_families.add(family) + selected.append(model) + if len(selected) == limit: + return selected -def _discovered_cost(model: DiscoveredModel, price_book: "PriceBook") -> float: - """Price the representative discovery request used by both selectors.""" - cost, _currency = price_book.compute_cost(model.provider_name, model.model_id, 1000, 1000) - return cost + selected.extend(deferred[: limit - len(selected)]) + return selected diff --git a/contextual_orchestrator/provider_bootstrap.py b/contextual_orchestrator/provider_bootstrap.py new file mode 100644 index 000000000..01cf7e80c --- /dev/null +++ b/contextual_orchestrator/provider_bootstrap.py @@ -0,0 +1,387 @@ +"""Durable bootstrap for the organization provider credential inventory. + +A trusted deployment process may expose the fixed provider-secret inventory to +this one-shot module. Values are validated as a complete set, written to the +configured credential KV, and then model discovery runs exclusively through the +KV-backed runtime seam. Runtime provider calls never read provider API keys from +``os.environ``. + +Bootstrap establishes a conservative serving candidate set. It does not infer +reasoning, coding, vision, or other provider capabilities from model names; +capability negotiation remains an explicit runtime/catalog responsibility. +""" + +from __future__ import annotations + +import argparse +from dataclasses import dataclass, replace +import json +import os +import re +from typing import Mapping, Sequence + +from .cost_ledger import PriceBook +from .credentials import ( + InMemoryCredentialBackend, + PostgresCredentialBackend, + get_backend, +) +from .kv_config import InMemoryConfigStore +from .model_discovery import ( + DiscoveredModel, + PROVIDER_MODEL_SOURCES, + agent_from_discovered, + agent_id_for, + discover_all_models, + refresh_price_book, +) +from .orchestrator import ModelAgent, TaskOrchestrator + + +PROVIDER_CREDENTIAL_NAMES: tuple[str, ...] = tuple( + dict.fromkeys(source.credential_name for source in PROVIDER_MODEL_SOURCES) +) +"""Fixed organization credential inventory accepted by the bootstrap boundary.""" + +_NON_CHAT_EXACT_TOKENS = frozenset( + { + "audio", + "bge", + "e5", + "embed", + "embedding", + "embeddings", + "guard", + "gte", + "image", + "images", + "moderation", + "realtime", + "rerank", + "reranker", + "safety", + "sora", + "speech", + "transcribe", + "transcription", + "tts", + "whisper", + } +) +_NON_CHAT_TOKEN_PREFIXES = ( + "embed", + "moderation", + "rerank", + "transcrib", +) +_GENERIC_SERVING_TAGS = ( + "discovered", + "chat", + "worker", + "writing", + "synthesizer", +) + + +class ProviderBootstrapError(RuntimeError): + """Raised when trusted provider bootstrap cannot establish a usable catalog.""" + + +@dataclass(frozen=True) +class ProviderBootstrapReport: + """Secret-free evidence emitted after one provider bootstrap run.""" + + registered_credentials: tuple[str, ...] + discovered_model_count: int + eligible_model_count: int + selected_agent_ids: tuple[str, ...] + enabled_agent_ids: tuple[str, ...] + durable_agent_pool: bool + providers_with_errors: tuple[str, ...] + priced_model_count: int + + def as_dict(self) -> dict[str, object]: + """Return JSON-safe evidence without credential values or provider payloads.""" + return { + "registered_credentials": list(self.registered_credentials), + "discovered_model_count": self.discovered_model_count, + "eligible_model_count": self.eligible_model_count, + "selected_agent_ids": list(self.selected_agent_ids), + "enabled_agent_ids": list(self.enabled_agent_ids), + "durable_agent_pool": self.durable_agent_pool, + "providers_with_errors": list(self.providers_with_errors), + "priced_model_count": self.priced_model_count, + } + + +def _strip_mounted_line_endings(value: str) -> str: + """Remove only CR/LF bytes commonly appended by mounted secret files.""" + return value.rstrip("\r\n") + + +def collect_provider_credentials( + environ: Mapping[str, str], *, require_all: bool = True +) -> dict[str, str]: + """Collect the fixed inventory without rewriting non-line-ending bytes.""" + values: dict[str, str] = {} + missing: list[str] = [] + for name in PROVIDER_CREDENTIAL_NAMES: + raw = environ.get(name, "") + value = _strip_mounted_line_endings(raw) if isinstance(raw, str) else "" + if value and value.strip(): + values[name] = value + else: + missing.append(name) + if require_all and missing: + raise ProviderBootstrapError( + "provider bootstrap requires the complete credential inventory: " + + ", ".join(sorted(missing)) + ) + if not values: + raise ProviderBootstrapError("provider bootstrap received no credentials") + return values + + +def register_provider_credentials_atomically( + credentials: Mapping[str, str], +) -> tuple[str, ...]: + """Register a validated credential batch with one commit where supported.""" + if not credentials: + raise ProviderBootstrapError("provider bootstrap received an empty credential batch") + unknown = sorted(set(credentials) - set(PROVIDER_CREDENTIAL_NAMES)) + if unknown: + raise ProviderBootstrapError("provider bootstrap rejected unknown credential names") + + normalized: dict[str, str] = {} + for name, value in credentials.items(): + if not isinstance(value, str): + raise ProviderBootstrapError( + f"provider bootstrap rejected an empty value for {name}" + ) + normalized_value = _strip_mounted_line_endings(value) + if not normalized_value or not normalized_value.strip(): + raise ProviderBootstrapError( + f"provider bootstrap rejected an empty value for {name}" + ) + normalized[name] = normalized_value + + backend = get_backend() + if isinstance(backend, InMemoryCredentialBackend): + with backend._lock: # noqa: SLF001 - package-internal atomic batch operation + backend._store.update(normalized) # noqa: SLF001 + elif isinstance(backend, PostgresCredentialBackend): + with backend._connect() as connection: # noqa: SLF001 - package transaction + backend._ensure_schema(connection) # noqa: SLF001 + with connection.cursor() as cursor: + for name, value in normalized.items(): + cursor.execute( + "INSERT INTO provider_credentials " + "(credential_name, encrypted_value, updated_at) " + "VALUES (%s, pgp_sym_encrypt(%s, %s), now()) " + "ON CONFLICT (credential_name) DO UPDATE SET " + "encrypted_value = EXCLUDED.encrypted_value, updated_at = now()", + (name, value, backend._passphrase), # noqa: SLF001 + ) + connection.commit() + else: + raise ProviderBootstrapError( + "provider bootstrap requires an atomic built-in credential backend" + ) + return tuple(sorted(normalized)) + + +def _model_tokens(model_id: str) -> tuple[str, ...]: + """Return conservative tokens used only to exclude obvious non-chat transports.""" + return tuple( + token for token in re.split(r"[^a-z0-9]+", model_id.casefold()) if token + ) + + +def is_chat_serving_candidate(model: DiscoveredModel) -> bool: + """Exclude catalog identifiers that clearly describe non-chat transports. + + This is a negative compatibility filter, not positive capability inference. + Models that survive receive only generic chat-serving tags until an explicit + provider/catalog capability record or measured evidence is available. + """ + tokens = _model_tokens(model.model_id) + for token in tokens: + if token in _NON_CHAT_EXACT_TOKENS: + return False + if token.startswith(_NON_CHAT_TOKEN_PREFIXES): + return False + return bool(tokens) + + +def serving_tags_for_discovered(_model: DiscoveredModel) -> tuple[str, ...]: + """Return capability-neutral tags safe for any compatible chat candidate.""" + return _GENERIC_SERVING_TAGS + + +def _known_cost_sort_key( + model: DiscoveredModel, +) -> tuple[int, float, str, str]: + """Sort known-price models before unknown-price models without inventing free cost.""" + prices = (model.prompt_price_per_1k, model.completion_price_per_1k) + known = [price for price in prices if price is not None] + if not known: + return (1, float("inf"), model.provider_name, model.model_id) + return (0, sum(known), model.provider_name, model.model_id) + + +def select_provider_diverse_models( + discovered: Sequence[DiscoveredModel], *, limit: int +) -> list[DiscoveredModel]: + """Choose a bounded compatible pool while preserving provider diversity.""" + if limit < 1: + raise ValueError("provider bootstrap model limit must be positive") + unique: dict[tuple[str, str, str], DiscoveredModel] = {} + for model in discovered: + if not is_chat_serving_candidate(model): + continue + unique[(model.provider_name, model.credential_name, model.model_id)] = model + ordered = sorted(unique.values(), key=_known_cost_sort_key) + selected: list[DiscoveredModel] = [] + seen_providers: set[str] = set() + for model in ordered: + if model.provider_name in seen_providers: + continue + selected.append(model) + seen_providers.add(model.provider_name) + if len(selected) >= limit: + return selected + selected_keys = { + (item.provider_name, item.credential_name, item.model_id) + for item in selected + } + for model in ordered: + key = (model.provider_name, model.credential_name, model.model_id) + if key in selected_keys: + continue + selected.append(model) + if len(selected) >= limit: + break + return selected + + +def _active_agent_from_discovered(model: DiscoveredModel) -> ModelAgent: + """Convert one selected chat model into an enabled capability-neutral agent.""" + return replace( + agent_from_discovered(model), + disabled=False, + tags=serving_tags_for_discovered(model), + ) + + +def _synchronize_durable_agent_pool( + agents_db: str, + selected: Sequence[DiscoveredModel], +) -> tuple[str, ...]: + """Activate exactly the selected discovered models in one durable agent pool.""" + bootstrap = TaskOrchestrator( + [ModelAgent("bootstrap_agent", "bootstrap-model")], + agents_db=agents_db, + ) + agents = [_active_agent_from_discovered(model) for model in selected] + selected_ids = {agent.id for agent in agents} + bootstrap.sync_discovered_agents(agents) + + for candidate in list(bootstrap.candidates): + if candidate.id in selected_ids: + continue + if candidate.id == "bootstrap_agent" or "discovered" in candidate.tags: + if not candidate.disabled: + bootstrap.remove_agent("default", candidate.id) + + for agent in agents: + bootstrap.patch_agent("default", agent.id, {"status": "active"}) + + enabled = tuple( + sorted(agent.id for agent in bootstrap.agents if agent.id in selected_ids) + ) + if set(enabled) != selected_ids: + raise ProviderBootstrapError( + "provider bootstrap could not activate the selected agent pool" + ) + return enabled + + +def bootstrap_provider_runtime( + *, + environ: Mapping[str, str], + require_all_credentials: bool = True, + agents_db: str | None = None, + model_limit: int = 16, +) -> ProviderBootstrapReport: + """Register trusted secrets, discover chat models, and optionally activate a pool.""" + credentials = collect_provider_credentials( + environ, require_all=require_all_credentials + ) + registered = register_provider_credentials_atomically(credentials) + discovered, errors = discover_all_models() + if not discovered: + raise ProviderBootstrapError( + "provider bootstrap discovered no usable models" + ) + + eligible = [model for model in discovered if is_chat_serving_candidate(model)] + if not eligible: + raise ProviderBootstrapError( + "provider bootstrap discovered no chat-capable models" + ) + + price_book = PriceBook(InMemoryConfigStore()) + priced_count = refresh_price_book(discovered, price_book) + selected = select_provider_diverse_models(eligible, limit=model_limit) + if not selected: + raise ProviderBootstrapError( + "provider bootstrap selected no chat-capable models" + ) + selected_ids = tuple(agent_id_for(model) for model in selected) + enabled_ids = ( + _synchronize_durable_agent_pool(agents_db, selected) + if agents_db + else () + ) + + return ProviderBootstrapReport( + registered_credentials=registered, + discovered_model_count=len(discovered), + eligible_model_count=len(eligible), + selected_agent_ids=selected_ids, + enabled_agent_ids=enabled_ids, + durable_agent_pool=bool(agents_db), + providers_with_errors=tuple( + sorted({error.provider_name for error in errors}) + ), + priced_model_count=priced_count, + ) + + +def main(argv: Sequence[str] | None = None) -> None: + """Run the one-shot provider bootstrap command used by trusted deployment jobs.""" + parser = argparse.ArgumentParser( + description="Register provider secrets and refresh the runtime model pool." + ) + parser.add_argument( + "--agents-db", + default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_AGENTS_DB") or None, + ) + parser.add_argument("--model-limit", type=int, default=16) + parser.add_argument( + "--allow-partial-credentials", + action="store_true", + help="Permit a subset of the fixed provider inventory (development only).", + ) + args = parser.parse_args(list(argv) if argv is not None else None) + report = bootstrap_provider_runtime( + environ=os.environ, + require_all_credentials=not args.allow_partial_credentials, + agents_db=args.agents_db, + model_limit=args.model_limit, + ) + print(json.dumps(report.as_dict(), ensure_ascii=False, sort_keys=True)) + + +if __name__ == "__main__": # pragma: no cover - subprocess/CLI coverage + main() diff --git a/contextual_orchestrator/provider_catalog_bootstrap.py b/contextual_orchestrator/provider_catalog_bootstrap.py new file mode 100644 index 000000000..2253caef5 --- /dev/null +++ b/contextual_orchestrator/provider_catalog_bootstrap.py @@ -0,0 +1,378 @@ +"""Trusted provider bootstrap with durable normalized model-catalog persistence. + +This command registers the complete credential inventory, performs provider- +isolated discovery, persists successful model metadata in PostgreSQL, retains +last-known-good models for failed providers, and constructs a bounded candidate +pool from the persisted catalog. +""" + +from __future__ import annotations + +import argparse +from dataclasses import dataclass +import json +import os +from typing import Callable, Mapping, Sequence + +from .cost_ledger import PriceBook +from .credentials import ( + InMemoryCredentialBackend, + PostgresCredentialBackend, + get_backend, + get_credential, +) +from .kv_config import InMemoryConfigStore +from .model_discovery import ( + DiscoveredModel, + PROVIDER_MODEL_SOURCES, + ProviderDiscoveryError, + ProviderModelSource, + agent_id_for, + discover_all_models, + refresh_price_book, +) +from .provider_bootstrap import ( + ProviderBootstrapError, + _synchronize_durable_agent_pool, + collect_provider_credentials, + is_chat_serving_candidate, + register_provider_credentials_atomically, + select_provider_diverse_models, + serving_tags_for_discovered, +) +from .provider_catalog_store import ( + InMemoryProviderCatalogStore, + PostgresProviderCatalogStore, + ProviderCatalogStore, +) + + +@dataclass(frozen=True) +class ProviderCatalogSnapshot: + """Effective persisted model snapshot after provider-isolated refresh.""" + + models: tuple[DiscoveredModel, ...] + live_model_count: int + last_known_good_model_count: int + refresh_failure_count: int + providers_with_errors: tuple[str, ...] + + +@dataclass(frozen=True) +class ProviderCatalogBootstrapReport: + """Secret-free evidence for one durable provider-catalog bootstrap. + + ``registered_credentials`` contains the credential names that remain in the + credential registry after provider-isolated rollback has completed. It is + therefore safe for a workflow to use as durable-registration evidence. + """ + + registered_credentials: tuple[str, ...] + restored_credentials: tuple[str, ...] + live_discovered_model_count: int + catalog_model_count: int + eligible_model_count: int + last_known_good_model_count: int + selected_agent_ids: tuple[str, ...] + enabled_agent_ids: tuple[str, ...] + durable_agent_pool: bool + catalog_backend: str + catalog_refresh_failure_count: int + providers_with_errors: tuple[str, ...] + priced_model_count: int + + def as_dict(self) -> dict[str, object]: + """Return the stable JSON evidence contract without secret values.""" + return { + "registered_credentials": list(self.registered_credentials), + "restored_credentials": list(self.restored_credentials), + "live_discovered_model_count": self.live_discovered_model_count, + "catalog_model_count": self.catalog_model_count, + "eligible_model_count": self.eligible_model_count, + "last_known_good_model_count": self.last_known_good_model_count, + "selected_agent_ids": list(self.selected_agent_ids), + "enabled_agent_ids": list(self.enabled_agent_ids), + "durable_agent_pool": self.durable_agent_pool, + "catalog_backend": self.catalog_backend, + "catalog_refresh_failure_count": self.catalog_refresh_failure_count, + "providers_with_errors": list(self.providers_with_errors), + "priced_model_count": self.priced_model_count, + } + + +def build_provider_catalog_store() -> ProviderCatalogStore: + """Build a catalog store colocated with the active credential backend.""" + backend = get_backend() + if isinstance(backend, PostgresCredentialBackend): + return PostgresProviderCatalogStore(backend.connection_dsn) + if isinstance(backend, InMemoryCredentialBackend): + return InMemoryProviderCatalogStore() + raise ProviderBootstrapError( + "provider catalog requires a built-in atomic credential backend" + ) + + +def _restore_provider_credentials_atomically( + previous_credentials: Mapping[str, str | None], +) -> tuple[str, ...]: + """Restore one credential snapshot in a single built-in backend transaction.""" + backend = get_backend() + ordered = tuple(sorted(previous_credentials)) + if isinstance(backend, InMemoryCredentialBackend): + with backend._lock: # noqa: SLF001 - package-internal rollback transaction + for name in ordered: + previous = previous_credentials[name] + if previous is None: + backend._store.pop(name, None) # noqa: SLF001 + else: + backend._store[name] = previous # noqa: SLF001 + return ordered + if isinstance(backend, PostgresCredentialBackend): + with backend._connect() as connection: # noqa: SLF001 - package transaction + backend._ensure_schema(connection) # noqa: SLF001 + with connection.cursor() as cursor: + for name in ordered: + previous = previous_credentials[name] + if previous is None: + cursor.execute( + "DELETE FROM provider_credentials WHERE credential_name = %s", + (name,), + ) + else: + cursor.execute( + "INSERT INTO provider_credentials " + "(credential_name, encrypted_value, updated_at) " + "VALUES (%s, pgp_sym_encrypt(%s, %s), now()) " + "ON CONFLICT (credential_name) DO UPDATE SET " + "encrypted_value = EXCLUDED.encrypted_value, updated_at = now()", + (name, previous, backend._passphrase), # noqa: SLF001 + ) + connection.commit() + return ordered + raise ProviderBootstrapError( + "provider credential rollback requires an atomic built-in backend" + ) + + +def _source_key(source: ProviderModelSource) -> tuple[str, str]: + """Return the provider-account key shared by sources and model rows.""" + return (source.provider_name, source.credential_name) + + +def _model_key(model: DiscoveredModel) -> tuple[str, str]: + """Return the provider-account key carried by one discovered model.""" + return (model.provider_name, model.credential_name) + + +def refresh_persisted_provider_catalog( + store: ProviderCatalogStore, + *, + sources: Sequence[ProviderModelSource], + registered_credentials: Sequence[str], + discovered: Sequence[DiscoveredModel], + errors: Sequence[ProviderDiscoveryError], +) -> ProviderCatalogSnapshot: + """Persist account-local refreshes and return the effective LKG snapshot.""" + registered = set(registered_credentials) + live_by_account: dict[tuple[str, str], list[DiscoveredModel]] = {} + for model in discovered: + live_by_account.setdefault(_model_key(model), []).append(model) + + failed_names = {error.provider_name for error in errors} + effective: list[DiscoveredModel] = [] + last_known_good_count = 0 + refresh_failures = 0 + providers_with_errors: set[str] = set(failed_names) + + for source in sources: + if source.credential_name not in registered: + continue + account_models = live_by_account.get(_source_key(source), []) + failed = source.provider_name in failed_names + if failed: + store.record_failure(source, error_code="provider_discovery_error") + refresh_failures += 1 + elif not account_models: + store.record_failure(source, error_code="empty_provider_catalog") + refresh_failures += 1 + providers_with_errors.add(source.provider_name) + else: + eligible_ids = { + model.model_id + for model in account_models + if is_chat_serving_candidate(model) + } + tags = { + model.model_id: serving_tags_for_discovered(model) + for model in account_models + if model.model_id in eligible_ids + } + store.record_success( + source, + account_models, + eligible_model_ids=eligible_ids, + serving_tags=tags, + ) + + persisted = store.serving_models(source) + effective.extend(persisted) + if failed or not account_models: + last_known_good_count += len(persisted) + + unique: dict[tuple[str, str, str], DiscoveredModel] = {} + for model in effective: + unique[(model.provider_name, model.credential_name, model.model_id)] = model + ordered = tuple(unique[key] for key in sorted(unique)) + return ProviderCatalogSnapshot( + models=ordered, + live_model_count=len(discovered), + last_known_good_model_count=last_known_good_count, + refresh_failure_count=refresh_failures, + providers_with_errors=tuple(sorted(providers_with_errors)), + ) + + +DiscoveryFunction = Callable[ + [tuple[ProviderModelSource, ...]], + tuple[list[DiscoveredModel], list[ProviderDiscoveryError]], +] + + +def bootstrap_provider_catalog_runtime( + *, + environ: Mapping[str, str], + require_all_credentials: bool = True, + agents_db: str | None = None, + model_limit: int = 16, + catalog_store: ProviderCatalogStore | None = None, + sources: Sequence[ProviderModelSource] = PROVIDER_MODEL_SOURCES, + discovery: DiscoveryFunction | None = None, +) -> ProviderCatalogBootstrapReport: + """Register secrets, persist catalogs, and build the effective serving pool.""" + credentials = collect_provider_credentials( + environ, + require_all=require_all_credentials, + ) + previous_credentials = { + name: get_credential(name) for name in credentials + } + registered = register_provider_credentials_atomically(credentials) + try: + store = catalog_store or build_provider_catalog_store() + source_tuple = tuple(sources) + discover = discovery or ( + lambda requested_sources: discover_all_models(requested_sources) + ) + live_models, errors = discover(source_tuple) + snapshot = refresh_persisted_provider_catalog( + store, + sources=source_tuple, + registered_credentials=registered, + discovered=live_models, + errors=errors, + ) + failed_provider_names = {error.provider_name for error in errors} + failed_credentials = { + source.credential_name + for source in source_tuple + if source.credential_name in registered + and ( + source.provider_name in failed_provider_names + or not any( + _model_key(model) == _source_key(source) + for model in live_models + ) + ) + } + restored_credentials = _restore_provider_credentials_atomically( + { + name: previous_credentials.get(name) + for name in failed_credentials + } + ) if failed_credentials else () + + usable_models = tuple( + model + for model in snapshot.models + if get_credential(model.credential_name) + ) + if not usable_models: + raise ProviderBootstrapError( + "provider bootstrap has no persisted chat-compatible model with a usable credential" + ) + + price_book = PriceBook(InMemoryConfigStore()) + priced_count = refresh_price_book(list(usable_models), price_book) + selected = select_provider_diverse_models( + usable_models, + limit=model_limit, + ) + if not selected: + raise ProviderBootstrapError( + "provider bootstrap selected no persisted chat-compatible model" + ) + selected_ids = tuple(agent_id_for(model) for model in selected) + enabled_ids = ( + _synchronize_durable_agent_pool(agents_db, selected) + if agents_db + else () + ) + durable_registered_credentials = tuple( + name for name in registered if get_credential(name) is not None + ) + + return ProviderCatalogBootstrapReport( + registered_credentials=durable_registered_credentials, + restored_credentials=tuple(restored_credentials), + live_discovered_model_count=snapshot.live_model_count, + catalog_model_count=len(snapshot.models), + eligible_model_count=len(snapshot.models), + last_known_good_model_count=snapshot.last_known_good_model_count, + selected_agent_ids=selected_ids, + enabled_agent_ids=enabled_ids, + durable_agent_pool=bool(agents_db), + catalog_backend=store.backend_name, + catalog_refresh_failure_count=snapshot.refresh_failure_count, + providers_with_errors=snapshot.providers_with_errors, + priced_model_count=priced_count, + ) + except Exception: + try: + _restore_provider_credentials_atomically(previous_credentials) + except Exception as rollback_error: + raise ProviderBootstrapError( + "provider bootstrap failed and credential rollback could not complete" + ) from rollback_error + raise + + +def main(argv: Sequence[str] | None = None) -> None: + """Run trusted durable provider bootstrap and print secret-free evidence.""" + parser = argparse.ArgumentParser( + description=( + "Register provider secrets, persist provider models, and refresh " + "the effective serving pool." + ) + ) + parser.add_argument( + "--agents-db", + default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_AGENTS_DB") or None, + ) + parser.add_argument("--model-limit", type=int, default=16) + parser.add_argument( + "--allow-partial-credentials", + action="store_true", + help="Permit a subset of the fixed provider inventory (development only).", + ) + args = parser.parse_args(list(argv) if argv is not None else None) + report = bootstrap_provider_catalog_runtime( + environ=os.environ, + require_all_credentials=not args.allow_partial_credentials, + agents_db=args.agents_db, + model_limit=args.model_limit, + ) + print(json.dumps(report.as_dict(), ensure_ascii=False, sort_keys=True)) + + +if __name__ == "__main__": # pragma: no cover - subprocess/CLI boundary + main() diff --git a/contextual_orchestrator/provider_catalog_store.py b/contextual_orchestrator/provider_catalog_store.py new file mode 100644 index 000000000..c00012ef9 --- /dev/null +++ b/contextual_orchestrator/provider_catalog_store.py @@ -0,0 +1,608 @@ +"""Normalized durable provider-model catalog persistence. + +This module owns provider-account/model metadata persistence and last-known-good +refresh behavior. It never performs network I/O and never stores credential +values. Discovery transport remains in ``model_discovery``; runtime selection +remains in the ordinary orchestrator. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +import hashlib +import math +import re +import threading +import uuid +from typing import Callable, Mapping, Protocol, Sequence + +from .model_discovery import DiscoveredModel, ProviderModelSource + + +PROVIDER_CATALOG_SCHEMA_SQL = """ +CREATE TABLE IF NOT EXISTS provider_account ( + provider_account_id text PRIMARY KEY, + provider_name text NOT NULL, + credential_name text NOT NULL, + list_url text NOT NULL, + chat_base_url text NOT NULL, + auth_scheme text NOT NULL, + discovery_style text NOT NULL, + task_filter text NOT NULL, + enabled_flag boolean NOT NULL DEFAULT true, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (provider_name, credential_name) +); + +CREATE TABLE IF NOT EXISTS provider_model ( + provider_model_id text PRIMARY KEY, + provider_account_id text NOT NULL + REFERENCES provider_account(provider_account_id) ON DELETE CASCADE, + model_name text NOT NULL, + prompt_price_per_1k numeric(20, 8), + completion_price_per_1k numeric(20, 8), + currency_code text NOT NULL, + serving_eligible_flag boolean NOT NULL DEFAULT false, + enabled_flag boolean NOT NULL DEFAULT true, + first_seen_at timestamptz NOT NULL, + last_seen_at timestamptz NOT NULL, + UNIQUE (provider_account_id, model_name) +); + +CREATE TABLE IF NOT EXISTS model_serving_tag ( + provider_model_id text NOT NULL + REFERENCES provider_model(provider_model_id) ON DELETE CASCADE, + tag_name text NOT NULL, + PRIMARY KEY (provider_model_id, tag_name) +); + +CREATE TABLE IF NOT EXISTS catalog_refresh_run ( + catalog_refresh_run_id text PRIMARY KEY, + provider_account_id text NOT NULL + REFERENCES provider_account(provider_account_id) ON DELETE CASCADE, + refresh_status text NOT NULL, + observed_model_count integer NOT NULL DEFAULT 0, + eligible_model_count integer NOT NULL DEFAULT 0, + error_code text, + started_at timestamptz NOT NULL, + finished_at timestamptz NOT NULL +); + +CREATE INDEX IF NOT EXISTS provider_model_account_idx + ON provider_model (provider_account_id, enabled_flag, serving_eligible_flag); +CREATE INDEX IF NOT EXISTS catalog_refresh_account_idx + ON catalog_refresh_run (provider_account_id, finished_at DESC); +""" +"""Third-normal-form schema for provider accounts, models, tags, and refreshes.""" + + +class ProviderCatalogError(RuntimeError): + """Raised when durable provider catalog metadata cannot be persisted or read.""" + + +@dataclass(frozen=True) +class CatalogRefreshEvidence: + """Secret-free evidence for one provider-account catalog refresh.""" + + provider_account_id: str + refresh_status: str + observed_model_count: int + eligible_model_count: int + error_code: str | None + started_at: datetime + finished_at: datetime + + +class ProviderCatalogStore(Protocol): + """Persistence boundary for provider model metadata and last-known-good rows.""" + + @property + def backend_name(self) -> str: + """Return a stable backend name for secret-free operator evidence.""" + ... + + def record_success( + self, + source: ProviderModelSource, + models: Sequence[DiscoveredModel], + *, + eligible_model_ids: set[str], + serving_tags: Mapping[str, tuple[str, ...]], + ) -> None: + """Replace one provider account's current catalog atomically.""" + ... + + def record_failure( + self, + source: ProviderModelSource, + *, + error_code: str, + ) -> None: + """Record failure without changing last-known-good enabled models.""" + ... + + def serving_models( + self, + source: ProviderModelSource, + ) -> list[DiscoveredModel]: + """Return enabled, serving-eligible last-known-good models.""" + ... + + def refresh_evidence(self) -> tuple[CatalogRefreshEvidence, ...]: + """Return refresh evidence in insertion order.""" + ... + + +_SLUG_RE = re.compile(r"[^a-z0-9]+") +_CURRENCY_RE = re.compile(r"^[A-Z]{3}$") +_ALLOWED_REFRESH_ERROR_CODES = frozenset( + {"provider_discovery_error", "empty_provider_catalog", "unknown_error"} +) + + +def provider_account_id(source: ProviderModelSource) -> str: + """Return a stable two-or-more-word snake-case provider account ID.""" + provider = _SLUG_RE.sub("_", source.provider_name.casefold()).strip("_") + credential = _SLUG_RE.sub("_", source.credential_name.casefold()).strip("_") + if not provider or not credential: + raise ProviderCatalogError("provider account identity is incomplete") + return f"{provider}_{credential}" + + +def provider_model_id(source: ProviderModelSource, model_name: str) -> str: + """Return a stable opaque ID for one account-scoped model name.""" + normalized = model_name.strip() + if not normalized: + raise ProviderCatalogError("provider model name is empty") + digest = hashlib.sha256( + f"{provider_account_id(source)}\0{normalized}".encode("utf-8") + ).hexdigest() + return f"provider_model_{digest[:32]}" + + +def _now() -> datetime: + """Return a timezone-aware UTC timestamp.""" + return datetime.now(timezone.utc) + + +def _normalize_price(value: object) -> float | None: + """Return one finite non-negative price, or ``None`` when unknown.""" + if value is None or isinstance(value, bool): + return None + try: + number = float(value) + except (TypeError, ValueError): + return None + if not math.isfinite(number) or number < 0: + return None + return number + + +def _normalize_currency(value: object) -> str: + """Return an ISO-style three-letter currency code.""" + if not isinstance(value, str): + return "USD" + normalized = value.strip().upper() + return normalized if _CURRENCY_RE.fullmatch(normalized) else "USD" + + +def _normalize_error_code(value: object) -> str: + """Return one approved secret-free provider refresh failure code.""" + if not isinstance(value, str): + return "unknown_error" + normalized = value.strip().casefold() + return normalized if normalized in _ALLOWED_REFRESH_ERROR_CODES else "unknown_error" + + +def _normalize_tags(tags: Sequence[str]) -> tuple[str, ...]: + """Return deterministic, valid, duplicate-free serving tags.""" + normalized: list[str] = [] + for raw in tags: + if not isinstance(raw, str): + continue + tag = raw.strip().casefold() + if not tag or not re.fullmatch(r"[a-z][a-z0-9_]*", tag): + continue + if tag not in normalized: + normalized.append(tag) + return tuple(normalized) + + +def normalize_discovered_model( + source: ProviderModelSource, + model: DiscoveredModel, +) -> DiscoveredModel: + """Normalize one discovered row and enforce its provider-account identity.""" + name = model.model_id.strip() if isinstance(model.model_id, str) else "" + if not name: + raise ProviderCatalogError("provider model name is empty") + if ( + model.provider_name != source.provider_name + or model.credential_name != source.credential_name + ): + raise ProviderCatalogError("provider model belongs to a different account") + return DiscoveredModel( + provider_name=source.provider_name, + model_id=name, + credential_name=source.credential_name, + chat_base_url=source.chat_base_url, + auth_scheme=source.auth_scheme, + prompt_price_per_1k=_normalize_price(model.prompt_price_per_1k), + completion_price_per_1k=_normalize_price( + model.completion_price_per_1k + ), + currency_code=_normalize_currency(model.currency_code), + ) + + +def _deduplicate_models( + source: ProviderModelSource, + models: Sequence[DiscoveredModel], +) -> dict[str, DiscoveredModel]: + """Normalize and deterministically deduplicate account-scoped models.""" + result: dict[str, DiscoveredModel] = {} + for model in models: + normalized = normalize_discovered_model(source, model) + result[normalized.model_id] = normalized + return result + + +class InMemoryProviderCatalogStore: + """Thread-safe deterministic provider catalog for tests and standalone use.""" + + def __init__(self) -> None: + self._accounts: dict[str, ProviderModelSource] = {} + self._models: dict[str, dict[str, DiscoveredModel]] = {} + self._eligible: dict[str, set[str]] = {} + self._tags: dict[tuple[str, str], tuple[str, ...]] = {} + self._refreshes: list[CatalogRefreshEvidence] = [] + self._lock = threading.RLock() + + @property + def backend_name(self) -> str: + """Return the stable in-memory backend name.""" + return "memory" + + def record_success( + self, + source: ProviderModelSource, + models: Sequence[DiscoveredModel], + *, + eligible_model_ids: set[str], + serving_tags: Mapping[str, tuple[str, ...]], + ) -> None: + """Replace one in-memory account catalog.""" + normalized = _deduplicate_models(source, models) + if not normalized: + raise ProviderCatalogError("successful provider refresh cannot be empty") + account_id = provider_account_id(source) + started_at = _now() + eligible = set(normalized).intersection(eligible_model_ids) + with self._lock: + self._accounts[account_id] = source + self._models[account_id] = normalized + self._eligible[account_id] = eligible + for key in [key for key in self._tags if key[0] == account_id]: + del self._tags[key] + for model_name in eligible: + self._tags[(account_id, model_name)] = _normalize_tags( + serving_tags.get(model_name, ()) + ) + self._refreshes.append( + CatalogRefreshEvidence( + account_id, + "succeeded", + len(normalized), + len(eligible), + None, + started_at, + _now(), + ) + ) + + def record_failure( + self, + source: ProviderModelSource, + *, + error_code: str, + ) -> None: + """Record a stable failure without mutating last-known-good models.""" + account_id = provider_account_id(source) + started_at = _now() + stable_code = _normalize_error_code(error_code) + with self._lock: + self._accounts[account_id] = source + self._refreshes.append( + CatalogRefreshEvidence( + account_id, + "failed", + 0, + 0, + stable_code, + started_at, + _now(), + ) + ) + + def serving_models( + self, + source: ProviderModelSource, + ) -> list[DiscoveredModel]: + """Return deterministic serving models for one account.""" + account_id = provider_account_id(source) + with self._lock: + models = self._models.get(account_id, {}) + eligible = self._eligible.get(account_id, set()) + return [models[name] for name in sorted(eligible) if name in models] + + def serving_tags( + self, + source: ProviderModelSource, + model_name: str, + ) -> tuple[str, ...]: + """Return persisted generic serving tags for one model.""" + with self._lock: + return self._tags.get((provider_account_id(source), model_name), ()) + + def refresh_evidence(self) -> tuple[CatalogRefreshEvidence, ...]: + """Return immutable refresh evidence in insertion order.""" + with self._lock: + return tuple(self._refreshes) + + +class PostgresProviderCatalogStore: + """PostgreSQL provider catalog sharing the credential registry database.""" + + def __init__( + self, + dsn: str, + *, + connection_factory: Callable[[], object] | None = None, + ) -> None: + if not isinstance(dsn, str) or not dsn.strip(): + raise ProviderCatalogError("provider catalog requires a PostgreSQL DSN") + self._dsn = dsn + self._connection_factory = connection_factory + self._schema_ready = False + self._schema_lock = threading.Lock() + self._evidence: list[CatalogRefreshEvidence] = [] + + @property + def backend_name(self) -> str: + """Return the stable PostgreSQL backend name.""" + return "postgres" + + def _connect(self): + """Open one catalog connection through the injected or psycopg factory.""" + if self._connection_factory is not None: + return self._connection_factory() + try: + import psycopg + except ImportError as exc: # pragma: no cover - packaging boundary + raise ProviderCatalogError( + "provider catalog requires contextual-orchestrator[db]" + ) from exc + return psycopg.connect(self._dsn) # pragma: no cover - live database + + def _ensure_schema(self, connection: object) -> None: + """Create normalized catalog objects once per store instance.""" + if self._schema_ready: + return + with self._schema_lock: + if self._schema_ready: + return + with connection.cursor() as cursor: + cursor.execute(PROVIDER_CATALOG_SCHEMA_SQL) + connection.commit() + self._schema_ready = True + + @staticmethod + def _upsert_account(cursor: object, source: ProviderModelSource) -> str: + """Upsert one provider account without credential values.""" + account_id = provider_account_id(source) + cursor.execute( + "INSERT INTO provider_account (" + "provider_account_id, provider_name, credential_name, list_url, " + "chat_base_url, auth_scheme, discovery_style, task_filter, " + "enabled_flag, created_at, updated_at" + ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, true, now(), now()) " + "ON CONFLICT (provider_account_id) DO UPDATE SET " + "provider_name = EXCLUDED.provider_name, " + "credential_name = EXCLUDED.credential_name, " + "list_url = EXCLUDED.list_url, " + "chat_base_url = EXCLUDED.chat_base_url, " + "auth_scheme = EXCLUDED.auth_scheme, " + "discovery_style = EXCLUDED.discovery_style, " + "task_filter = EXCLUDED.task_filter, " + "enabled_flag = true, updated_at = now()", + ( + account_id, + source.provider_name, + source.credential_name, + source.list_url, + source.chat_base_url, + source.auth_scheme, + source.style, + source.task_filter, + ), + ) + return account_id + + def record_success( + self, + source: ProviderModelSource, + models: Sequence[DiscoveredModel], + *, + eligible_model_ids: set[str], + serving_tags: Mapping[str, tuple[str, ...]], + ) -> None: + """Replace one PostgreSQL account catalog in a single transaction.""" + normalized = _deduplicate_models(source, models) + if not normalized: + raise ProviderCatalogError("successful provider refresh cannot be empty") + started_at = _now() + eligible = set(normalized).intersection(eligible_model_ids) + with self._connect() as connection: + self._ensure_schema(connection) + with connection.cursor() as cursor: + account_id = self._upsert_account(cursor, source) + cursor.execute( + "UPDATE provider_model SET enabled_flag = false " + "WHERE provider_account_id = %s", + (account_id,), + ) + for model_name, model in normalized.items(): + model_row_id = provider_model_id(source, model_name) + cursor.execute( + "INSERT INTO provider_model (" + "provider_model_id, provider_account_id, model_name, " + "prompt_price_per_1k, completion_price_per_1k, currency_code, " + "serving_eligible_flag, enabled_flag, first_seen_at, " + "last_seen_at" + ") VALUES (%s, %s, %s, %s, %s, %s, %s, " + "true, %s, %s) " + "ON CONFLICT (provider_model_id) DO UPDATE SET " + "model_name = EXCLUDED.model_name, " + "prompt_price_per_1k = EXCLUDED.prompt_price_per_1k, " + "completion_price_per_1k = EXCLUDED.completion_price_per_1k, " + "currency_code = EXCLUDED.currency_code, " + "serving_eligible_flag = EXCLUDED.serving_eligible_flag, " + "enabled_flag = true, last_seen_at = EXCLUDED.last_seen_at", + ( + model_row_id, + account_id, + model_name, + model.prompt_price_per_1k, + model.completion_price_per_1k, + model.currency_code, + model_name in eligible, + started_at, + started_at, + ), + ) + cursor.execute( + "DELETE FROM model_serving_tag WHERE provider_model_id = %s", + (model_row_id,), + ) + if model_name in eligible: + for tag in _normalize_tags(serving_tags.get(model_name, ())): + cursor.execute( + "INSERT INTO model_serving_tag " + "(provider_model_id, tag_name) " + "VALUES (%s, %s) ON CONFLICT DO NOTHING", + (model_row_id, tag), + ) + finished_at = _now() + cursor.execute( + "INSERT INTO catalog_refresh_run (" + "catalog_refresh_run_id, provider_account_id, refresh_status, " + "observed_model_count, eligible_model_count, error_code, " + "started_at, finished_at" + ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s)", + ( + f"catalog_refresh_{uuid.uuid4().hex}", + account_id, + "succeeded", + len(normalized), + len(eligible), + None, + started_at, + finished_at, + ), + ) + connection.commit() + self._evidence.append( + CatalogRefreshEvidence( + provider_account_id(source), + "succeeded", + len(normalized), + len(eligible), + None, + started_at, + finished_at, + ) + ) + + def record_failure( + self, + source: ProviderModelSource, + *, + error_code: str, + ) -> None: + """Record a PostgreSQL failure without disabling prior models.""" + started_at = _now() + stable_code = _normalize_error_code(error_code) + with self._connect() as connection: + self._ensure_schema(connection) + with connection.cursor() as cursor: + account_id = self._upsert_account(cursor, source) + finished_at = _now() + cursor.execute( + "INSERT INTO catalog_refresh_run (" + "catalog_refresh_run_id, provider_account_id, refresh_status, " + "observed_model_count, eligible_model_count, error_code, " + "started_at, finished_at" + ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s)", + ( + f"catalog_refresh_{uuid.uuid4().hex}", + account_id, + "failed", + 0, + 0, + stable_code, + started_at, + finished_at, + ), + ) + connection.commit() + self._evidence.append( + CatalogRefreshEvidence( + provider_account_id(source), + "failed", + 0, + 0, + stable_code, + started_at, + finished_at, + ) + ) + + def serving_models( + self, + source: ProviderModelSource, + ) -> list[DiscoveredModel]: + """Read enabled last-known-good serving models for one account.""" + account_id = provider_account_id(source) + with self._connect() as connection: + self._ensure_schema(connection) + with connection.cursor() as cursor: + cursor.execute( + "SELECT pm.model_name, pa.chat_base_url, pa.auth_scheme, " + "pm.prompt_price_per_1k, pm.completion_price_per_1k, " + "pm.currency_code FROM provider_model AS pm " + "JOIN provider_account AS pa ON pa.provider_account_id = pm.provider_account_id " + "WHERE pm.provider_account_id = %s " + "AND pm.enabled_flag = true AND pm.serving_eligible_flag = true " + "ORDER BY pm.model_name", + (account_id,), + ) + rows = cursor.fetchall() + return [ + DiscoveredModel( + provider_name=source.provider_name, + model_id=row[0], + credential_name=source.credential_name, + chat_base_url=row[1], + auth_scheme=row[2], + prompt_price_per_1k=_normalize_price(row[3]), + completion_price_per_1k=_normalize_price(row[4]), + currency_code=_normalize_currency(row[5]), + ) + for row in rows + ] + + def refresh_evidence(self) -> tuple[CatalogRefreshEvidence, ...]: + """Return evidence emitted by this store instance.""" + return tuple(self._evidence) diff --git a/docs/database_design.sql b/docs/database_design.sql index fb5892479..56c37beac 100644 --- a/docs/database_design.sql +++ b/docs/database_design.sql @@ -69,6 +69,64 @@ create table audit_event ( created_at timestamptz not null default now() ); +create table provider_credentials ( + credential_name text primary key, + encrypted_value bytea not null, + updated_at timestamptz not null default now() +); + +create table provider_account ( + provider_account_id text primary key, + provider_name text not null, + -- Credential rollback deliberately stays independent of catalog rows so a + -- failed candidate promotion cannot delete last-known-good model metadata. + -- The runtime catalog DDL uses the same application-managed relationship. + credential_name text not null, + list_url text not null, + chat_base_url text not null, + auth_scheme text not null, + discovery_style text not null, + task_filter text not null, + enabled_flag boolean not null default true, + created_at timestamptz not null default now(), + updated_at timestamptz not null default now(), + unique (provider_name, credential_name) +); + +create table provider_model ( + provider_model_id text primary key, + provider_account_id text not null + references provider_account(provider_account_id) on delete cascade, + model_name text not null, + prompt_price_per_1k numeric(20, 8), + completion_price_per_1k numeric(20, 8), + currency_code text not null, + serving_eligible_flag boolean not null default false, + enabled_flag boolean not null default true, + first_seen_at timestamptz not null, + last_seen_at timestamptz not null, + unique (provider_account_id, model_name) +); + +create table model_serving_tag ( + provider_model_id text not null + references provider_model(provider_model_id) on delete cascade, + tag_name text not null, + primary key (provider_model_id, tag_name) +); + +create table catalog_refresh_run ( + catalog_refresh_run_id text primary key, + provider_account_id text not null + references provider_account(provider_account_id) on delete cascade, + refresh_status text not null, + observed_model_count integer not null default 0, + eligible_model_count integer not null default 0, + error_code text, + started_at timestamptz not null, + finished_at timestamptz not null +); + create index workflow_run_retention_idx on workflow_run (retention_expires_at) where deleted_at is null; @@ -81,6 +139,12 @@ create index audit_event_retention_idx on audit_event (retention_expires_at) where deleted_at is null; +create index provider_model_account_idx + on provider_model (provider_account_id, enabled_flag, serving_eligible_flag); + +create index catalog_refresh_account_idx + on catalog_refresh_run (provider_account_id, finished_at desc); + create view workflow_run_safe_view as select workflow_run_id, diff --git a/docs/doctoring/current-main-provider-bootstrap.md b/docs/doctoring/current-main-provider-bootstrap.md new file mode 100644 index 000000000..383868be6 --- /dev/null +++ b/docs/doctoring/current-main-provider-bootstrap.md @@ -0,0 +1,119 @@ +# Current-main provider bootstrap + +## Decision + +The durable catalog decision is recorded in +[`ADR 0015`](../planning/adrs/0015-durable-provider-catalog.md), including the +third-normal-form dependency boundary and the last-known-good refresh contract. + +Contextual Orchestrator treats the five organization provider credentials as one +trusted bootstrap inventory: + +- `NVIDIA_NIM_API_KEY` +- `NVIDIA_NIM_API_KEY_SUB` +- `BYTEZ_API_KEY` +- `OPENROUTER_API_KEY` +- `OPENAI_API_KEY` + +GitHub Actions secrets are transport into a one-shot bootstrap process, not the +runtime credential source. Production bootstrap requires the PostgreSQL credential +backend so values are stored encrypted at rest through the existing pgcrypto +registry. Runtime model discovery resolves credential names through +`get_credential()` only. + +## Failure contract + +Production bootstrap fails closed when any fixed credential is missing, when the +configured credential backend is not atomic, or when no usable provider model can +be discovered. A provider-local discovery exception does not erase models returned +by other providers; the report contains only stable provider names and counts, never +raw exception strings or credential values. + +`registered_credentials` is the post-rollback durable inventory, not merely the +set of candidate names received from Actions. If a first-ever candidate key is +reverted after a failed provider refresh, that name is omitted from the report +and the hourly workflow fails its complete-inventory gate. Existing keys that +are restored remain listed, so a transient provider outage can preserve +last-known-good serving without falsely claiming that a missing key is durable. + +A successful generic `/models` response is not itself evidence that every row can +serve Chat Completions. OpenAI-compatible registries may mix chat models with +embeddings, rerankers, speech, image generation, moderation, safety, or realtime +transports. The bootstrap therefore applies a conservative negative compatibility +filter before selection and reports both: + +- `discovered_model_count`: every syntactically valid catalog row; and +- `eligible_model_count`: rows that are not clearly a non-chat transport. + +If no compatible row remains, bootstrap fails closed instead of activating the +cheapest incompatible model. Surviving rows receive only generic serving tags: +`discovered`, `chat`, `worker`, `writing`, and `synthesizer`. The bootstrap never +infers reasoning, verification, coding, vision, or provider-native effort support +from a model name. Those capabilities require explicit provider/catalog evidence or +measured evaluation and are negotiated by the ordinary runtime policy. + +The bootstrap pool is provider-diverse before it is cost-ordered. Missing price is +`unknown`, not zero. This avoids treating a provider such as Bytez, whose public +catalog may use a non-token billing unit, as a fabricated free route. + +Candidate selection and durable serving activation are separate claims: + +- `selected_agent_ids` records the bounded chat candidates produced by discovery + and selection; +- `enabled_agent_ids` is populated only when an explicit durable `--agents-db` + is supplied and the selected agents are confirmed active in that pool; and +- `durable_agent_pool` states whether the activation claim is backed by a + persistent agent-pool database. + +When a durable pool is refreshed, the bootstrap tombstones its synthetic seed and +previously discovered agents that are absent from the current bounded selection. +Operator-managed agents are preserved. This prevents retired, withdrawn, or newly +classified non-chat provider models from continuing to receive traffic after a +later discovery run. + +## Operational workflow + +`.github/workflows/provider-catalog-sync.yml` runs hourly on protected `main` and may +also be dispatched manually. It is intentionally absent from pull-request secret +execution. The production environment must provide: + +- the five provider secrets above; +- `CONTEXTUAL_ORCHESTRATOR_KV_DSN`; and +- `CONTEXTUAL_ORCHESTRATOR_KV_PASSPHRASE`. + +The GitHub-hosted workflow has an ephemeral filesystem. It therefore registers the +five credentials in the durable PostgreSQL KV and verifies discovery, +`eligible_model_count`, and `selected_agent_ids`; it does not claim durable +agent-pool activation. A long-running service may either use the ordinary KV-backed +startup discovery path or invoke this bootstrap with a persistent `--agents-db` +under its own deployment boundary. + +The workflow verifies that all five credential names were registered, at least one +model was discovered, at least one chat-compatible model survived classification, a +bounded serving candidate set was produced, and no exact provider secret appears in +the emitted report. + +## Research and standards grounding + +The automatic pool remains a routing input rather than an unsupported claim that a +single cheapest model is universally best. Quality/performance selection remains in +the orchestrator's paper-grounded routing and orchestration layer; this bootstrap +only establishes a compatible candidate set and failure isolation. + +National Institute of Standards and Technology. (2020). *Security and privacy +controls for information systems and organizations* (NIST Special Publication +800-53 Rev. 5). https://doi.org/10.6028/NIST.SP.800-53r5 + +National Institute of Standards and Technology. (2024). *Artificial intelligence +risk management framework: Generative artificial intelligence profile* (NIST AI +600-1). https://doi.org/10.6028/NIST.AI.600-1 + +Tang, Y., et al. (2026). *Sakana Fugu technical report*. Sakana AI. + +Xu, J., Sun, Q., Schwendeman, P., Nielsen, S., Cetin, E., & Tang, Y. (2025). +*Trinity: An evolved LLM coordinator* (arXiv:2512.04695). +https://doi.org/10.48550/arXiv.2512.04695 + +Nielsen, S., Cetin, E., Schwendeman, P., Sun, Q., Xu, J., & Tang, Y. (2025). +*Learning to orchestrate agents in natural language with the Conductor* +(arXiv:2512.04388). https://doi.org/10.48550/arXiv.2512.04388 diff --git a/docs/doctoring/provider-diverse-discovery-routing.md b/docs/doctoring/provider-diverse-discovery-routing.md new file mode 100644 index 000000000..9eb03e0ef --- /dev/null +++ b/docs/doctoring/provider-diverse-discovery-routing.md @@ -0,0 +1,48 @@ +--- +title: "Provider-diverse discovery and cost-honest failover routing" +status: "implemented" +date: "2026-08-21" +scope: "PR #770" +--- + +# Provider-diverse discovery and cost-honest failover routing + +## Decision + +PR #770 makes model discovery fail closed for invalid or unpriced catalog +rows, retains candidates whose prices are unknown only as an explicit +fallback, and selects a provider-diverse bootstrap pool before ordinary +chat routing. The selector is deterministic eligibility and cost accounting; +it is not a learned answer-quality judge and does not claim to reproduce the +learning systems in the cited work. + +## Research-to-code mapping + +| Implementation boundary | Evidence-informed reason | Acceptance evidence | +| --- | --- | --- | +| Reject malformed, negative, or non-finite price rows | A cost-aware router must not treat missing or invalid evidence as zero cost. | Discovery and persisted-price tests reject the row before selection. | +| Keep unknown-price candidates only as an explicit fallback | Cost optimization must remain honest when price evidence is incomplete. | Selection tests never rank an unknown price above a valid priced candidate. | +| Prefer distinct providers in the bootstrap pool | A gateway needs an upstream failover set rather than several aliases for one provider. | Provider-diversity tests assert the configured pool spans available providers. | +| Leave quality judgment to evaluation/review policy | Routing signals and answer-quality judgment have different failure modes. | Existing model-judge and fail-closed routing tests remain the quality boundary. | + +The routing papers and OA PDFs are already committed in the prerequisite +stack base under `docs/papers/` (`routellm-routing-2406.18665.pdf`, +`hybrid-llm-query-routing-2404.14618.pdf`, and +`frugalgpt-cost-2305.05176.pdf`). This doctoring record makes their relevance +to the exact discovery selector explicit instead of treating inherited files +as incidental documentation. + +## APA 7 references + +Chen, L., Zaharia, M., & Zou, J. (2023). *FrugalGPT: How to use large +language models while reducing cost and improving performance*. arXiv. +https://arxiv.org/abs/2305.05176 + +Ding, D., Mallick, A., Wang, C., Sim, R., Mukherjee, S., Rühle, V., +Lakshmanan, L. V. S., & Awadallah, A. H. (2024). *Hybrid LLM: +Cost-efficient and quality-aware query routing*. International Conference on +Learning Representations. https://arxiv.org/abs/2404.14618 + +Ong, I., Almahairi, A., Wu, V., Chiang, W.-L., Wu, T., Gonzalez, J. E., +Kadous, M. W., & Stoica, I. (2024). *RouteLLM: Learning to route LLMs with +preference data*. arXiv. https://arxiv.org/abs/2406.18665 diff --git a/docs/planning/adrs/0015-durable-provider-catalog.md b/docs/planning/adrs/0015-durable-provider-catalog.md new file mode 100644 index 000000000..b4daf10ce --- /dev/null +++ b/docs/planning/adrs/0015-durable-provider-catalog.md @@ -0,0 +1,93 @@ +--- +id: "0015" +title: "Durable provider catalog and last-known-good composition" +status: proposed +proposed_date: "2026-08-20" +accepted_date: null +deciders: + - "repository maintainer" +consulted: + - "NIST SP 800-53 Rev. 5" + - "NIST AI 600-1" +informed: + - "LineageWeave" + - "fast-mlsirm" + - "contributors" +affected_components: + - "contextual_orchestrator/provider_bootstrap.py" + - "contextual_orchestrator/provider_catalog_bootstrap.py" + - "contextual_orchestrator/provider_catalog_store.py" + - ".github/workflows/provider-catalog-sync.yml" +supersedes: null +superseded-by: null +related: + - path: "docs/planning/adrs/0012-gateway-only-provider-contract.md" + relation: depends-on + - path: "docs/planning/adrs/0014-gateway-owned-model-selection.md" + relation: extends +effort: L +--- + +# Durable provider catalog and last-known-good composition + +## Context + +Registering the five organization provider secrets in PostgreSQL is necessary +but not sufficient. A production process must also retain discovered provider +accounts and models so a transient catalog outage does not erase the serving +pool, and operators must distinguish live discovery from last-known-good +metadata. GitHub-hosted scheduled jobs have ephemeral filesystems, so SQLite +cannot be the authority for this catalog. + +## Decision + +Use a third-normal-form PostgreSQL catalog colocated with the encrypted +credential registry. The authority contains four two-or-more-word +`snake_case` objects: + +- `provider_account`: provider endpoint and credential name, never the value; +- `provider_model`: account-scoped model identity, known prices, compatibility + and lifecycle state; endpoint and authentication fields are joined from its + owning account; +- `model_serving_tag`: generic serving tags as a separate many-to-many relation; +- `catalog_refresh_run`: provider-local success/failure evidence. + +A successful non-empty provider refresh atomically replaces that provider +account's enabled current set. A failed or empty/malformed refresh records only +an allowlisted stable error code and preserves the account's last-known-good +models. Successful discovery of an authoritative non-chat-only catalog may +withdraw earlier chat rows. + +Model names are used only for a conservative negative compatibility filter that +excludes obvious embedding, reranking, speech, image, moderation, safety, and +realtime transports. They are never used to infer reasoning, verification, +coding, vision, or provider-native effort capabilities. Those require explicit +catalog or measured evidence under the gateway-owned policy. + +## Consequences + +- Credentials and model metadata are durable but remain separated. +- NVIDIA primary and secondary keys are independent provider accounts. +- One provider outage does not erase other providers or its own last-known-good set. +- Unknown price remains unknown rather than becoming fabricated zero cost. +- The protected hourly workflow can persist catalog metadata without claiming + that its ephemeral runner has activated a durable agent-pool database. +- Long-running deployments may separately synchronize selected catalog rows into + a persistent agent pool. + +## Verification + +The merge gate covers normalized DDL, secret-column absence, provider-account +isolation, parameterized PostgreSQL statements, last-known-good retention, +withdrawal after authoritative success, non-chat filtering, secret-free +evidence, and end-to-end recovery when one provider fails. + +## References + +National Institute of Standards and Technology. (2020). *Security and privacy +controls for information systems and organizations* (NIST Special Publication +800-53 Rev. 5). https://doi.org/10.6028/NIST.SP.800-53r5 + +National Institute of Standards and Technology. (2024). *Artificial +intelligence risk management framework: Generative artificial intelligence +profile* (NIST AI 600-1). https://doi.org/10.6028/NIST.AI.600-1 diff --git a/docs/provider_catalog_database.sql b/docs/provider_catalog_database.sql new file mode 100644 index 000000000..0f9fd2538 --- /dev/null +++ b/docs/provider_catalog_database.sql @@ -0,0 +1,53 @@ +CREATE TABLE IF NOT EXISTS provider_account ( + provider_account_id text PRIMARY KEY, + provider_name text NOT NULL, + credential_name text NOT NULL, + list_url text NOT NULL, + chat_base_url text NOT NULL, + auth_scheme text NOT NULL, + discovery_style text NOT NULL, + task_filter text NOT NULL, + enabled_flag boolean NOT NULL DEFAULT true, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (provider_name, credential_name) +); + +CREATE TABLE IF NOT EXISTS provider_model ( + provider_model_id text PRIMARY KEY, + provider_account_id text NOT NULL + REFERENCES provider_account(provider_account_id) ON DELETE CASCADE, + model_name text NOT NULL, + prompt_price_per_1k numeric(20, 8), + completion_price_per_1k numeric(20, 8), + currency_code text NOT NULL, + serving_eligible_flag boolean NOT NULL DEFAULT false, + enabled_flag boolean NOT NULL DEFAULT true, + first_seen_at timestamptz NOT NULL, + last_seen_at timestamptz NOT NULL, + UNIQUE (provider_account_id, model_name) +); + +CREATE TABLE IF NOT EXISTS model_serving_tag ( + provider_model_id text NOT NULL + REFERENCES provider_model(provider_model_id) ON DELETE CASCADE, + tag_name text NOT NULL, + PRIMARY KEY (provider_model_id, tag_name) +); + +CREATE TABLE IF NOT EXISTS catalog_refresh_run ( + catalog_refresh_run_id text PRIMARY KEY, + provider_account_id text NOT NULL + REFERENCES provider_account(provider_account_id) ON DELETE CASCADE, + refresh_status text NOT NULL, + observed_model_count integer NOT NULL DEFAULT 0, + eligible_model_count integer NOT NULL DEFAULT 0, + error_code text, + started_at timestamptz NOT NULL, + finished_at timestamptz NOT NULL +); + +CREATE INDEX IF NOT EXISTS provider_model_account_idx + ON provider_model (provider_account_id, enabled_flag, serving_eligible_flag); +CREATE INDEX IF NOT EXISTS catalog_refresh_account_idx + ON catalog_refresh_run (provider_account_id, finished_at DESC); diff --git a/tests/test_discover_models_cli.py b/tests/test_discover_models_cli.py index d00825ffd..444557f75 100644 --- a/tests/test_discover_models_cli.py +++ b/tests/test_discover_models_cli.py @@ -164,3 +164,49 @@ def urlopen(request, timeout=None): by_id = {agent.id: agent for agent in reloaded.candidates} assert by_id["openrouter_cheap_model"].disabled is False assert by_id["openai_pricey_model"].disabled is True + + +def test_enable_cheapest_bootstraps_independent_provider_families(tmp_path) -> None: + """CLI bootstrap must use the provider-diverse selector, not only the cheapest vendor.""" + from contextual_orchestrator import TaskOrchestrator + from contextual_orchestrator.orchestrator import ModelAgent + + set_backend(InMemoryCredentialBackend()) + register_credential("OPENAI_API_KEY", "sk-openai") + register_credential("OPENROUTER_API_KEY", "sk-router") + register_credential("NVIDIA_NIM_API_KEY", "nv-primary") + db_path = str(tmp_path / "pool.db") + stdout = StringIO() + + def urlopen(request, timeout=None): + host = urllib.parse.urlsplit(request.full_url).hostname + payloads = { + "api.openai.com": {"data": [{"id": "openai-model", "pricing": {"prompt": "0.001", "completion": "0.001"}}]}, + "openrouter.ai": {"data": [{"id": "router-model", "pricing": {"prompt": "0.000001", "completion": "0.000001"}}]}, + "integrate.api.nvidia.com": {"data": [{"id": "nim-model", "pricing": {"prompt": "0.000002", "completion": "0.000002"}}]}, + } + return _Response(payloads.get(host, {"data": []})) + + try: + with ( + patch.object( + sys, + "argv", + ["contextual-orchestrator", "discover-models", "--agents-db", db_path, "--enable-cheapest", "3"], + ), + patch.object(sys, "stdout", stdout), + patch("contextual_orchestrator.model_discovery.urllib.request.urlopen", side_effect=urlopen), + ): + main() + finally: + set_backend(None) + + report = json.loads(stdout.getvalue()) + assert report["enabled_agent_ids"] == [ + "openrouter_router_model", + "nvidia_nim_nim_model", + "openai_openai_model", + ] + reloaded = TaskOrchestrator([ModelAgent("seed_agent", "seed-model")], agents_db=db_path) + enabled = {agent.id for agent in reloaded.candidates if not agent.disabled} + assert enabled - {"seed_agent"} == set(report["enabled_agent_ids"]) diff --git a/tests/test_discovery_bootstrap_selection.py b/tests/test_discovery_bootstrap_selection.py new file mode 100644 index 000000000..e12eeb0b6 --- /dev/null +++ b/tests/test_discovery_bootstrap_selection.py @@ -0,0 +1,358 @@ +"""Regression coverage for honest, provider-diverse discovery bootstrap.""" + +from __future__ import annotations + +from contextual_orchestrator.cost_ledger import PriceBook, PriceEntry +from contextual_orchestrator.kv_config import InMemoryConfigStore +from contextual_orchestrator import model_discovery +from contextual_orchestrator.model_discovery import ( + DiscoveredModel, + refresh_price_book, + select_cheapest_discovered_agent, + select_top_n_cheapest_discovered_agents, +) + + +def _model(provider_name: str, model_id: str) -> DiscoveredModel: + """Build one deterministic OpenAI-compatible discovery fixture.""" + credential_name = f"{provider_name.upper()}_API_KEY" + return DiscoveredModel( + provider_name=provider_name, + model_id=model_id, + credential_name=credential_name, + chat_base_url=f"https://{provider_name}.example/v1", + auth_scheme="Bearer", + ) + + +def _priced_model( + provider_name: str, + model_id: str, + *, + prompt_price_per_1k: float | None, + completion_price_per_1k: float | None, + currency_code: str = "USD", +) -> DiscoveredModel: + """Build one discovery row carrying provider-reported price evidence.""" + base = _model(provider_name, model_id) + return DiscoveredModel( + provider_name=base.provider_name, + model_id=base.model_id, + credential_name=base.credential_name, + chat_base_url=base.chat_base_url, + auth_scheme=base.auth_scheme, + prompt_price_per_1k=prompt_price_per_1k, + completion_price_per_1k=completion_price_per_1k, + currency_code=currency_code, + ) + + +def _set_price( + price_book: PriceBook, + model: DiscoveredModel, + price_per_1k: float, + *, + currency_code: str = "USD", +) -> None: + """Record one known symmetric prompt/completion price.""" + price_book.set_price( + PriceEntry( + model.provider_name, + model.model_id, + price_per_1k, + price_per_1k, + currency_code, + ) + ) + + +def test_unpriced_discovered_model_is_unknown_not_free() -> None: + """Missing price evidence must not outrank a model with a known price.""" + price_book = PriceBook(InMemoryConfigStore()) + priced = _model("openrouter", "priced-model") + unpriced = _model("bytez", "unpriced-model") + _set_price(price_book, priced, 0.01) + + assert select_cheapest_discovered_agent([unpriced, priced], price_book) is priced + assert select_top_n_cheapest_discovered_agents( + [unpriced, priced], price_book, 2 + ) == [priced, unpriced] + + +def test_partial_provider_price_is_unknown_instead_of_fabricating_a_free_component() -> None: + """A missing prompt or completion price cannot become an invented zero.""" + price_book = PriceBook(InMemoryConfigStore()) + partial = _priced_model( + "partial_vendor", + "partial-model", + prompt_price_per_1k=0.001, + completion_price_per_1k=None, + ) + complete = _priced_model( + "openrouter", + "complete-model", + prompt_price_per_1k=1.0, + completion_price_per_1k=1.0, + ) + + assert refresh_price_book([partial, complete], price_book) == 1 + assert price_book.get_price(partial.provider_name, partial.model_id) is None + assert select_cheapest_discovered_agent([partial, complete], price_book) is complete + + +def test_persisted_price_row_missing_one_component_remains_unknown() -> None: + """KV corruption must not silently manufacture a zero-priced component.""" + store = InMemoryConfigStore() + store.set( + "llm_price_entries", + "partial_vendor:partial-model", + { + "provider_name": "partial_vendor", + "model_name": "partial-model", + "prompt_price_per_1k": 0.001, + "currency_code": "USD", + }, + ) + price_book = PriceBook(store) + + assert price_book.get_price("partial_vendor", "partial-model") is None + + +def test_invalid_catalog_prices_are_unknown_not_trusted_cost_evidence() -> None: + """Reject negative, non-finite, and boolean provider price values.""" + assert model_discovery._price_per_1k("-0.000001") is None + assert model_discovery._price_per_1k("nan") is None + assert model_discovery._price_per_1k("inf") is None + assert model_discovery._price_per_1k(True) is None + assert model_discovery._price_per_1k("0") == 0.0 + + +def test_huge_price_values_remain_unknown_without_crashing_discovery_or_ranking() -> None: + """Unbounded JSON or KV integers must not terminate bootstrap selection.""" + huge_price = 10**10000 + assert model_discovery._price_per_1k(huge_price) is None + + price_book = PriceBook(InMemoryConfigStore()) + huge = _model("huge_vendor", "huge-model") + valid = _model("openrouter", "valid-model") + _set_price(price_book, huge, huge_price) + _set_price(price_book, valid, 1.0) + + assert select_cheapest_discovered_agent([huge, valid], price_book) is valid + + +def test_malformed_price_book_row_is_unknown_instead_of_crashing_selection() -> None: + """A corrupt persisted price row must not take down the serving bootstrap.""" + store = InMemoryConfigStore() + store.set( + "llm_price_entries", + "broken_vendor:broken-model", + { + "provider_name": "broken_vendor", + "model_name": "broken-model", + "prompt_price_per_1k": "not-a-number", + "completion_price_per_1k": 0.001, + "currency_code": "USD", + }, + ) + price_book = PriceBook(store) + broken = _model("broken_vendor", "broken-model") + valid = _model("openrouter", "valid-model") + _set_price(price_book, valid, 1.0) + + assert select_cheapest_discovered_agent([broken, valid], price_book) is valid + + +def test_refresh_counts_only_complete_prices_in_the_comparison_currency() -> None: + """Cross-currency evidence is unknown until an explicit conversion exists.""" + price_book = PriceBook(InMemoryConfigStore(), default_currency="USD") + usd = _priced_model( + "openrouter", + "usd-model", + prompt_price_per_1k=1.0, + completion_price_per_1k=1.0, + currency_code="USD", + ) + eur = _priced_model( + "eur_vendor", + "eur-model", + prompt_price_per_1k=0.001, + completion_price_per_1k=0.001, + currency_code="EUR", + ) + + assert refresh_price_book([eur, usd], price_book) == 1 + assert price_book.get_price("eur_vendor", "eur-model") is None + assert price_book.get_price("openrouter", "usd-model") is not None + + +def test_invalid_or_cross_currency_price_rows_do_not_outrank_comparable_usd_cost() -> None: + """Only finite non-negative prices in the configured currency are comparable.""" + price_book = PriceBook(InMemoryConfigStore(), default_currency="USD") + valid = _model("openrouter", "valid-model") + negative = _model("negative_vendor", "negative-model") + non_finite = _model("nan_vendor", "nan-model") + foreign = _model("eur_vendor", "eur-model") + + _set_price(price_book, valid, 1.0) + _set_price(price_book, negative, -100.0) + _set_price(price_book, non_finite, float("nan")) + _set_price(price_book, foreign, 0.000001, currency_code="EUR") + + assert select_cheapest_discovered_agent( + [negative, non_finite, foreign, valid], + price_book, + ) is valid + + +def test_duplicate_serving_identity_cannot_consume_bootstrap_capacity() -> None: + """A repeated provider/model row must not masquerade as failover diversity.""" + selector = getattr( + model_discovery, + "select_bootstrap_discovered_agents", + None, + ) + assert callable(selector), "missing provider-diverse bootstrap selector" + + price_book = PriceBook(InMemoryConfigStore()) + duplicate_first = _model("openrouter", "same-model") + duplicate_second = _model("openrouter", "same-model") + independent = _model("openai", "independent-model") + _set_price(price_book, duplicate_first, 0.01) + _set_price(price_book, independent, 0.02) + + selected = selector( + [duplicate_second, independent, duplicate_first], + price_book, + 3, + ) + top_n = select_top_n_cheapest_discovered_agents( + [duplicate_second, independent, duplicate_first], + price_book, + 3, + ) + + assert [ + (model.provider_name, model.model_id) + for model in selected + ] == [ + ("openrouter", "same-model"), + ("openai", "independent-model"), + ] + assert [ + (model.provider_name, model.model_id) + for model in top_n + ] == [ + ("openrouter", "same-model"), + ("openai", "independent-model"), + ] + + +def test_conflicting_duplicate_prices_are_withheld_as_ambiguous() -> None: + """Do not let provider row order decide the trusted price for one agent id.""" + price_book = PriceBook(InMemoryConfigStore()) + cheap_claim = _priced_model( + "openrouter", + "duplicate-model", + prompt_price_per_1k=0.000001, + completion_price_per_1k=0.000001, + ) + expensive_claim = _priced_model( + "openrouter", + "duplicate-model", + prompt_price_per_1k=100.0, + completion_price_per_1k=100.0, + ) + complete = _priced_model( + "openai", + "complete-model", + prompt_price_per_1k=1.0, + completion_price_per_1k=1.0, + ) + + assert refresh_price_book( + [cheap_claim, expensive_claim, complete], + price_book, + ) == 1 + assert price_book.get_price("openrouter", "duplicate-model") is None + assert select_cheapest_discovered_agent( + [cheap_claim, expensive_claim, complete], + price_book, + ) is complete + + +def test_bootstrap_selector_prefers_provider_diversity_before_duplicates() -> None: + """The initial failover pool must span providers before repeating one.""" + selector = getattr( + model_discovery, + "select_bootstrap_discovered_agents", + None, + ) + assert callable(selector), "missing provider-diverse bootstrap selector" + + price_book = PriceBook(InMemoryConfigStore()) + router_cheapest = _model("openrouter", "router-cheapest") + router_second = _model("openrouter", "router-second") + nim_model = _model("nvidia_nim", "nim-model") + openai_model = _model("openai", "openai-model") + _set_price(price_book, router_cheapest, 0.01) + _set_price(price_book, router_second, 0.02) + _set_price(price_book, nim_model, 0.5) + _set_price(price_book, openai_model, 1.0) + + selected = selector( + [router_second, openai_model, nim_model, router_cheapest], + price_book, + 3, + ) + + assert selected == [router_cheapest, nim_model, openai_model] + + +def test_bootstrap_selector_treats_nim_primary_and_sub_as_one_outage_domain() -> None: + """Two NIM keys must not displace an independently hosted provider.""" + selector = getattr( + model_discovery, + "select_bootstrap_discovered_agents", + None, + ) + assert callable(selector), "missing provider-diverse bootstrap selector" + + price_book = PriceBook(InMemoryConfigStore()) + nim_primary = _model("nvidia_nim", "primary-model") + nim_sub = _model("nvidia_nim_sub", "sub-model") + openrouter = _model("openrouter", "router-model") + _set_price(price_book, nim_primary, 0.01) + _set_price(price_book, nim_sub, 0.02) + _set_price(price_book, openrouter, 0.5) + + selected = selector( + [nim_sub, openrouter, nim_primary], + price_book, + 2, + ) + + assert selected == [nim_primary, openrouter] + + +def test_bootstrap_selector_is_deterministic_when_every_model_is_unpriced() -> None: + """All-unpriced discovery remains usable but never order-dependent.""" + selector = getattr( + model_discovery, + "select_bootstrap_discovered_agents", + None, + ) + assert callable(selector), "missing provider-diverse bootstrap selector" + + price_book = PriceBook(InMemoryConfigStore()) + router_z = _model("openrouter", "z-model") + router_a = _model("openrouter", "a-model") + nim_b = _model("nvidia_nim", "b-model") + + selected = selector( + [router_z, nim_b, router_a], + price_book, + 3, + ) + + assert selected == [nim_b, router_a, router_z] diff --git a/tests/test_provider_bootstrap.py b/tests/test_provider_bootstrap.py new file mode 100644 index 000000000..dad0580f1 --- /dev/null +++ b/tests/test_provider_bootstrap.py @@ -0,0 +1,330 @@ +"""Contracts for durable all-provider bootstrap and provider-diverse model activation.""" + +from __future__ import annotations + +from dataclasses import replace +import json +import os + +import pytest + +from contextual_orchestrator import ModelAgent, TaskOrchestrator +from contextual_orchestrator.credentials import ( + InMemoryCredentialBackend, + get_credential, + set_backend, +) +from contextual_orchestrator.model_discovery import ( + DiscoveredModel, + agent_from_discovered, +) +from contextual_orchestrator import provider_bootstrap + + +@pytest.fixture(autouse=True) +def isolated_credential_backend(): + """Give each test a fresh process-local credential registry.""" + set_backend(InMemoryCredentialBackend()) + yield + set_backend(None) + + +def _complete_environment() -> dict[str, str]: + """Return one complete mounted-secret fixture with trailing newlines.""" + return { + name: f"secret-for-{name.lower()}\n" + for name in provider_bootstrap.PROVIDER_CREDENTIAL_NAMES + } + + +def _model( + provider: str, + credential: str, + model_id: str, + prompt: float | None, +) -> DiscoveredModel: + """Build a deterministic provider-catalog row for bootstrap tests.""" + return DiscoveredModel( + provider_name=provider, + model_id=model_id, + credential_name=credential, + chat_base_url=f"https://{provider}.example/v1", + auth_scheme="Bearer", + prompt_price_per_1k=prompt, + completion_price_per_1k=prompt, + ) + + +def test_fixed_inventory_matches_all_five_organization_secrets(): + """The bootstrap inventory must not silently lose an organization provider key.""" + assert set(provider_bootstrap.PROVIDER_CREDENTIAL_NAMES) == { + "NVIDIA_NIM_API_KEY", + "NVIDIA_NIM_API_KEY_SUB", + "BYTEZ_API_KEY", + "OPENROUTER_API_KEY", + "OPENAI_API_KEY", + } + + +def test_collect_requires_complete_inventory_without_leaking_values(): + """Production bootstrap fails before writes when one trusted secret is absent.""" + environment = _complete_environment() + removed = environment.pop("BYTEZ_API_KEY") + with pytest.raises(provider_bootstrap.ProviderBootstrapError) as raised: + provider_bootstrap.collect_provider_credentials(environment) + assert "BYTEZ_API_KEY" in str(raised.value) + assert removed.strip() not in str(raised.value) + assert all( + get_credential(name) is None + for name in provider_bootstrap.PROVIDER_CREDENTIAL_NAMES + ) + + +def test_atomic_memory_registration_strips_mounted_secret_newlines(): + """A complete inventory becomes visible together and mounted newlines are removed.""" + credentials = provider_bootstrap.collect_provider_credentials( + _complete_environment() + ) + registered = provider_bootstrap.register_provider_credentials_atomically( + credentials + ) + assert registered == tuple( + sorted(provider_bootstrap.PROVIDER_CREDENTIAL_NAMES) + ) + for name in provider_bootstrap.PROVIDER_CREDENTIAL_NAMES: + value = get_credential(name) + assert value == f"secret-for-{name.lower()}" + assert "\n" not in value + + +def test_unknown_credential_name_is_rejected_before_any_write(): + """The fixed bootstrap boundary cannot be expanded by untrusted names.""" + with pytest.raises(provider_bootstrap.ProviderBootstrapError): + provider_bootstrap.register_provider_credentials_atomically( + {"EVIL_PROVIDER_KEY": "secret"} + ) + assert get_credential("EVIL_PROVIDER_KEY") is None + + +def test_diverse_selection_prefers_known_cost_without_treating_unknown_as_free(): + """Unknown-cost candidates stay usable but cannot win as fabricated zero cost.""" + models = [ + _model("openai", "OPENAI_API_KEY", "gpt-expensive", 4.0), + _model("openai", "OPENAI_API_KEY", "gpt-cheap", 1.0), + _model("openrouter", "OPENROUTER_API_KEY", "mistral-router", 2.0), + _model("bytez", "BYTEZ_API_KEY", "llama-unknown", None), + ] + selected = provider_bootstrap.select_provider_diverse_models(models, limit=3) + assert [(item.provider_name, item.model_id) for item in selected] == [ + ("openai", "gpt-cheap"), + ("openrouter", "mistral-router"), + ("bytez", "llama-unknown"), + ] + + +def test_non_chat_catalog_rows_are_never_selected_for_chat_service(): + """Embeddings, rerankers, speech, image, moderation, and realtime rows stay inert.""" + models = [ + _model("openai", "OPENAI_API_KEY", "text-embedding-3-small", 0.1), + _model("openai", "OPENAI_API_KEY", "whisper-1", 0.1), + _model("openai", "OPENAI_API_KEY", "gpt-image-1", 0.1), + _model("openai", "OPENAI_API_KEY", "omni-moderation-latest", 0.1), + _model( + "nvidia_nim", + "NVIDIA_NIM_API_KEY", + "nv-rerankqa-mistral-4b-v3", + 0.1, + ), + _model( + "openrouter", + "OPENROUTER_API_KEY", + "openai/gpt-4.1-mini", + 2.0, + ), + ] + selected = provider_bootstrap.select_provider_diverse_models(models, limit=10) + assert [(item.provider_name, item.model_id) for item in selected] == [ + ("openrouter", "openai/gpt-4.1-mini") + ] + + +def test_serving_tags_do_not_infer_capabilities_from_model_names(): + """Reasoning, coding, and vision-looking names receive only generic tags.""" + model = _model( + "openrouter", + "OPENROUTER_API_KEY", + "qwen/qwen-vl-coder-reasoning", + 1.0, + ) + assert provider_bootstrap.serving_tags_for_discovered(model) == ( + "discovered", + "chat", + "worker", + "writing", + "synthesizer", + ) + + +def test_bootstrap_registers_then_discovers_without_environment_runtime_reads( + monkeypatch, +): + """Discovery sees KV-backed credentials after one-shot environment bootstrap.""" + environment = _complete_environment() + observed: dict[str, str | None] = {} + + def fake_discover_all_models(): + """Observe the KV from the mocked provider-discovery boundary.""" + for name in provider_bootstrap.PROVIDER_CREDENTIAL_NAMES: + observed[name] = get_credential(name) + return ( + [_model("openai", "OPENAI_API_KEY", "gpt-test", 1.0)], + [], + ) + + monkeypatch.setattr( + provider_bootstrap, + "discover_all_models", + fake_discover_all_models, + ) + report = provider_bootstrap.bootstrap_provider_runtime( + environ=environment, + model_limit=1, + ) + + assert report.discovered_model_count == 1 + assert report.eligible_model_count == 1 + assert report.selected_agent_ids == ("openai_gpt_test",) + assert report.enabled_agent_ids == () + assert report.durable_agent_pool is False + assert all( + observed[name] == environment[name].strip() + for name in observed + ) + + +def test_bootstrap_fails_closed_when_no_model_is_discovered(monkeypatch): + """Credential writes without a usable catalog are not reported service-ready.""" + monkeypatch.setattr( + provider_bootstrap, + "discover_all_models", + lambda: ([], []), + ) + with pytest.raises( + provider_bootstrap.ProviderBootstrapError, + match="no usable models", + ): + provider_bootstrap.bootstrap_provider_runtime( + environ=_complete_environment() + ) + + +def test_bootstrap_fails_closed_when_catalog_has_only_non_chat_models(monkeypatch): + """A successful catalog response is not ready without a chat candidate.""" + monkeypatch.setattr( + provider_bootstrap, + "discover_all_models", + lambda: ( + [ + _model( + "openai", + "OPENAI_API_KEY", + "text-embedding-3-small", + 0.1, + ) + ], + [], + ), + ) + with pytest.raises( + provider_bootstrap.ProviderBootstrapError, + match="no chat-capable models", + ): + provider_bootstrap.bootstrap_provider_runtime( + environ=_complete_environment() + ) + + +def test_durable_pool_withdraws_bootstrap_and_stale_discovered_agents( + monkeypatch, + tmp_path, +): + """A refresh leaves exactly the current selected discovered models active.""" + agents_db = str(tmp_path / "agents.db") + old_model = _model( + "openai", + "OPENAI_API_KEY", + "gpt-retired-model", + 1.0, + ) + old_agent = replace(agent_from_discovered(old_model), disabled=False) + seeded = TaskOrchestrator( + [ModelAgent("manual_agent", "manual-model")], + agents_db=agents_db, + ) + seeded.sync_discovered_agents([old_agent]) + + new_model = _model( + "openrouter", + "OPENROUTER_API_KEY", + "qwen-current-coder", + 2.0, + ) + monkeypatch.setattr( + provider_bootstrap, + "discover_all_models", + lambda: ([new_model], []), + ) + report = provider_bootstrap.bootstrap_provider_runtime( + environ=_complete_environment(), + agents_db=agents_db, + model_limit=1, + ) + + assert report.discovered_model_count == 1 + assert report.eligible_model_count == 1 + assert report.selected_agent_ids == ("openrouter_qwen_current_coder",) + assert report.enabled_agent_ids == ("openrouter_qwen_current_coder",) + assert report.durable_agent_pool is True + + restarted = TaskOrchestrator( + [ModelAgent("bootstrap_agent", "bootstrap-model")], + agents_db=agents_db, + ) + assert {agent.id for agent in restarted.agents} == { + "openrouter_qwen_current_coder" + } + assert restarted.agents[0].tags == ( + "discovered", + "chat", + "worker", + "writing", + "synthesizer", + ) + assert all( + agent.id not in {"bootstrap_agent", "openai_gpt_retired_model"} + for agent in restarted.agents + ) + + +def test_cli_report_never_contains_secret_values(monkeypatch, capsys): + """Operator evidence names credentials and agents but never prints secrets.""" + environment = _complete_environment() + monkeypatch.setattr(os, "environ", environment) + monkeypatch.setattr( + provider_bootstrap, + "discover_all_models", + lambda: ( + [_model("openai", "OPENAI_API_KEY", "gpt-test", 1.0)], + [], + ), + ) + provider_bootstrap.main(["--model-limit", "1"]) + output = capsys.readouterr().out + report = json.loads(output) + assert "OPENAI_API_KEY" in output + assert report["eligible_model_count"] == 1 + assert report["selected_agent_ids"] == ["openai_gpt_test"] + assert report["enabled_agent_ids"] == [] + assert report["durable_agent_pool"] is False + for value in environment.values(): + assert value.strip() not in output diff --git a/tests/test_provider_bootstrap_secret_normalization.py b/tests/test_provider_bootstrap_secret_normalization.py new file mode 100644 index 000000000..5a1877f5c --- /dev/null +++ b/tests/test_provider_bootstrap_secret_normalization.py @@ -0,0 +1,67 @@ +"""Regression coverage for mounted provider-secret normalization.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from contextual_orchestrator.credentials import ( + InMemoryCredentialBackend, + get_credential, + set_backend, +) +from contextual_orchestrator.provider_bootstrap import ( + PROVIDER_CREDENTIAL_NAMES, + collect_provider_credentials, + register_provider_credentials_atomically, +) + + +@pytest.fixture(autouse=True) +def isolated_credential_backend(): + """Give each test a fresh process-local credential registry.""" + set_backend(InMemoryCredentialBackend()) + try: + yield + finally: + set_backend(None) + + +def _complete_environment() -> dict[str, str]: + """Return a complete mounted-secret inventory.""" + return {name: f"secret-for-{name.lower()}\n" for name in PROVIDER_CREDENTIAL_NAMES} + + +def test_collection_removes_only_mounted_line_endings() -> None: + """Do not silently rewrite other credential bytes while removing CR/LF mounts.""" + environment = _complete_environment() + environment["OPENAI_API_KEY"] = " edge-sensitive-secret \r\n" + + collected = collect_provider_credentials(environment) + + assert collected["OPENAI_API_KEY"] == " edge-sensitive-secret " + assert collected["BYTEZ_API_KEY"] == "secret-for-bytez_api_key" + + +def test_atomic_registration_preserves_normalized_secret_bytes() -> None: + """The atomic backend write must not perform a second broad whitespace trim.""" + credentials = { + name: f"secret-for-{name.lower()}" + for name in PROVIDER_CREDENTIAL_NAMES + } + credentials["OPENROUTER_API_KEY"] = " edge-sensitive-router-secret " + + register_provider_credentials_atomically(credentials) + + assert get_credential("OPENROUTER_API_KEY") == " edge-sensitive-router-secret " + + +def test_catalog_sync_leak_guard_matches_secret_normalization() -> None: + """The workflow checks the exact credential bytes that bootstrap handles.""" + workflow = Path(".github/workflows/provider-catalog-sync.yml").read_text( + encoding="utf-8" + ) + + assert "os.environ[name].rstrip('\\r\\n')" in workflow + assert "os.environ[name] and os.environ[name] in report" not in workflow diff --git a/tests/test_provider_catalog_bootstrap.py b/tests/test_provider_catalog_bootstrap.py new file mode 100644 index 000000000..e4b7aa013 --- /dev/null +++ b/tests/test_provider_catalog_bootstrap.py @@ -0,0 +1,171 @@ +"""End-to-end durable provider catalog bootstrap contracts.""" + +from __future__ import annotations + +import pytest + +from contextual_orchestrator.credentials import ( + InMemoryCredentialBackend, + get_credential, + register_credential, + set_backend, +) +from contextual_orchestrator.model_discovery import ( + DiscoveredModel, + ProviderDiscoveryError, + ProviderModelSource, +) +from contextual_orchestrator.provider_bootstrap import PROVIDER_CREDENTIAL_NAMES +from contextual_orchestrator.provider_catalog_bootstrap import ( + bootstrap_provider_catalog_runtime, +) +from contextual_orchestrator.provider_catalog_store import ( + InMemoryProviderCatalogStore, +) + + +def _environment() -> dict[str, str]: + return { + name: f"value-for-{name.casefold()}" + for name in PROVIDER_CREDENTIAL_NAMES + } + + +def _source(provider: str, credential: str) -> ProviderModelSource: + return ProviderModelSource( + provider_name=provider, + credential_name=credential, + list_url=f"https://{provider}.example/v1/models", + chat_base_url=f"https://{provider}.example/v1", + ) + + +def _model(source: ProviderModelSource, model_id: str) -> DiscoveredModel: + return DiscoveredModel( + provider_name=source.provider_name, + model_id=model_id, + credential_name=source.credential_name, + chat_base_url=source.chat_base_url, + auth_scheme=source.auth_scheme, + prompt_price_per_1k=1.0, + completion_price_per_1k=2.0, + ) + + +def test_failed_provider_uses_persisted_last_known_good_model() -> None: + """A later provider outage keeps its last successful compatible model.""" + set_backend(InMemoryCredentialBackend()) + try: + openai = _source("openai", "OPENAI_API_KEY") + openrouter = _source("openrouter", "OPENROUTER_API_KEY") + store = InMemoryProviderCatalogStore() + + first = bootstrap_provider_catalog_runtime( + environ=_environment(), + catalog_store=store, + sources=(openai, openrouter), + discovery=lambda _sources: ( + [_model(openai, "gpt-live"), _model(openrouter, "router-live")], + [], + ), + model_limit=4, + ) + assert first.catalog_model_count == 2 + assert first.last_known_good_model_count == 0 + + second = bootstrap_provider_catalog_runtime( + environ=_environment(), + catalog_store=store, + sources=(openai, openrouter), + discovery=lambda _sources: ( + [_model(openrouter, "router-new")], + [ProviderDiscoveryError("openai", "secret-bearing detail")], + ), + model_limit=4, + ) + assert second.live_discovered_model_count == 1 + assert second.catalog_model_count == 2 + assert second.last_known_good_model_count == 1 + assert second.catalog_refresh_failure_count == 1 + assert second.providers_with_errors == ("openai",) + assert set(second.selected_agent_ids) == { + "openai_gpt_live", + "openrouter_router_new", + } + assert "secret-bearing detail" not in str(second.as_dict()) + finally: + set_backend(None) + + +def test_empty_catalog_preserves_lkg_but_nonchat_success_withdraws_it() -> None: + """Empty refresh is failure; authoritative non-chat success is withdrawal.""" + set_backend(InMemoryCredentialBackend()) + try: + openai = _source("openai", "OPENAI_API_KEY") + store = InMemoryProviderCatalogStore() + bootstrap_provider_catalog_runtime( + environ=_environment(), + catalog_store=store, + sources=(openai,), + discovery=lambda _sources: ([_model(openai, "gpt-live")], []), + model_limit=1, + ) + + empty = bootstrap_provider_catalog_runtime( + environ=_environment(), + catalog_store=store, + sources=(openai,), + discovery=lambda _sources: ([], []), + model_limit=1, + ) + assert empty.last_known_good_model_count == 1 + assert empty.catalog_model_count == 1 + + try: + bootstrap_provider_catalog_runtime( + environ=_environment(), + catalog_store=store, + sources=(openai,), + discovery=lambda _sources: ( + [_model(openai, "text-embedding-3-small")], + [], + ), + model_limit=1, + ) + except RuntimeError as error: + assert "no persisted chat-compatible model" in str(error) + else: + raise AssertionError("non-chat-only authoritative catalog must fail") + finally: + set_backend(None) + + +def test_unexpected_discovery_failure_restores_entire_credential_inventory() -> None: + """An unclassified bootstrap failure must not leave unvalidated secrets promoted.""" + set_backend(InMemoryCredentialBackend()) + try: + previous = { + name: f"previous-value-for-{name.casefold()}" + for name in PROVIDER_CREDENTIAL_NAMES + } + for name, value in previous.items(): + register_credential(name, value) + + def fail_discovery(_sources): + raise RuntimeError("unexpected discovery parser failure") + + with pytest.raises(RuntimeError, match="unexpected discovery parser failure"): + bootstrap_provider_catalog_runtime( + environ=_environment(), + catalog_store=InMemoryProviderCatalogStore(), + sources=(_source("openai", "OPENAI_API_KEY"),), + discovery=fail_discovery, + model_limit=1, + ) + + assert { + name: get_credential(name) + for name in PROVIDER_CREDENTIAL_NAMES + } == previous + finally: + set_backend(None) diff --git a/tests/test_provider_catalog_credential_promotion.py b/tests/test_provider_catalog_credential_promotion.py new file mode 100644 index 000000000..fc843f981 --- /dev/null +++ b/tests/test_provider_catalog_credential_promotion.py @@ -0,0 +1,192 @@ +"""Regression coverage for provider credential promotion around catalog refresh.""" + +from __future__ import annotations + +import pytest + +from contextual_orchestrator.credentials import ( + InMemoryCredentialBackend, + get_credential, + register_credential, + set_backend, +) +from contextual_orchestrator.model_discovery import ( + DiscoveredModel, + ProviderDiscoveryError, + ProviderModelSource, +) +from contextual_orchestrator.provider_bootstrap import ProviderBootstrapError +from contextual_orchestrator.provider_catalog_bootstrap import ( + bootstrap_provider_catalog_runtime, +) +from contextual_orchestrator.provider_catalog_store import ( + InMemoryProviderCatalogStore, +) + + +@pytest.fixture(autouse=True) +def isolated_credential_backend(): + """Give every promotion test a fresh credential registry.""" + set_backend(InMemoryCredentialBackend()) + try: + yield + finally: + set_backend(None) + + +def _source() -> ProviderModelSource: + return ProviderModelSource( + provider_name="openai", + credential_name="OPENAI_API_KEY", + list_url="https://api.openai.example/v1/models", + chat_base_url="https://api.openai.example/v1", + ) + + +def _model(source: ProviderModelSource, model_id: str) -> DiscoveredModel: + return DiscoveredModel( + provider_name=source.provider_name, + model_id=model_id, + credential_name=source.credential_name, + chat_base_url=source.chat_base_url, + auth_scheme=source.auth_scheme, + prompt_price_per_1k=1.0, + completion_price_per_1k=2.0, + ) + + +def _seed_last_known_good( + store: InMemoryProviderCatalogStore, + source: ProviderModelSource, +) -> None: + model = _model(source, "gpt-last-known-good") + store.record_success( + source, + [model], + eligible_model_ids={model.model_id}, + serving_tags={model.model_id: ("discovered", "chat", "worker")}, + ) + + +def test_failed_refresh_restores_previous_credential_before_using_lkg() -> None: + """An invalid candidate key must not replace the key paired with LKG models.""" + source = _source() + store = InMemoryProviderCatalogStore() + _seed_last_known_good(store, source) + register_credential(source.credential_name, "old-working-secret") + + def failing_discovery(_sources): + assert get_credential(source.credential_name) == "new-invalid-secret" + return [], [ProviderDiscoveryError(source.provider_name, "unauthorized")] + + report = bootstrap_provider_catalog_runtime( + environ={source.credential_name: "new-invalid-secret"}, + require_all_credentials=False, + catalog_store=store, + sources=(source,), + discovery=failing_discovery, + model_limit=1, + ) + + assert get_credential(source.credential_name) == "old-working-secret" + assert report.selected_agent_ids == ("openai_gpt_last_known_good",) + assert report.restored_credentials == (source.credential_name,) + + +def test_empty_refresh_restores_previous_credential_before_using_lkg() -> None: + """An empty candidate-key catalog is failure, not credential promotion.""" + source = _source() + store = InMemoryProviderCatalogStore() + _seed_last_known_good(store, source) + register_credential(source.credential_name, "old-working-secret") + + report = bootstrap_provider_catalog_runtime( + environ={source.credential_name: "new-empty-catalog-secret"}, + require_all_credentials=False, + catalog_store=store, + sources=(source,), + discovery=lambda _sources: ([], []), + model_limit=1, + ) + + assert get_credential(source.credential_name) == "old-working-secret" + assert report.selected_agent_ids == ("openai_gpt_last_known_good",) + assert report.restored_credentials == (source.credential_name,) + + +def test_failed_first_promotion_cannot_activate_lkg_without_a_prior_credential() -> None: + """Persisted models are unusable when the candidate key failed and no old key exists.""" + source = _source() + store = InMemoryProviderCatalogStore() + _seed_last_known_good(store, source) + + with pytest.raises( + ProviderBootstrapError, + match="no persisted chat-compatible model with a usable credential", + ): + bootstrap_provider_catalog_runtime( + environ={source.credential_name: "first-invalid-secret"}, + require_all_credentials=False, + catalog_store=store, + sources=(source,), + discovery=lambda _sources: ( + [], + [ProviderDiscoveryError(source.provider_name, "unauthorized")], + ), + model_limit=1, + ) + + assert get_credential(source.credential_name) is None + + +def test_report_excludes_first_promotion_credential_removed_by_rollback() -> None: + """Durable-registration evidence cannot claim a deleted first candidate key.""" + openai = _source() + openrouter = ProviderModelSource( + provider_name="openrouter", + credential_name="OPENROUTER_API_KEY", + list_url="https://openrouter.example/v1/models", + chat_base_url="https://openrouter.example/v1", + ) + live = _model(openrouter, "router-live") + + report = bootstrap_provider_catalog_runtime( + environ={ + openai.credential_name: "first-invalid-secret", + openrouter.credential_name: "working-router-secret", + }, + require_all_credentials=False, + catalog_store=InMemoryProviderCatalogStore(), + sources=(openai, openrouter), + discovery=lambda _sources: ( + [live], + [ProviderDiscoveryError(openai.provider_name, "temporary discovery failure")], + ), + model_limit=1, + ) + + assert report.restored_credentials == (openai.credential_name,) + assert report.registered_credentials == (openrouter.credential_name,) + assert get_credential(openai.credential_name) is None + assert get_credential(openrouter.credential_name) == "working-router-secret" + + +def test_successful_refresh_promotes_the_candidate_credential() -> None: + """A validated non-empty catalog commits the new provider credential.""" + source = _source() + store = InMemoryProviderCatalogStore() + register_credential(source.credential_name, "old-working-secret") + live = _model(source, "gpt-new-live") + + report = bootstrap_provider_catalog_runtime( + environ={source.credential_name: "new-working-secret"}, + require_all_credentials=False, + catalog_store=store, + sources=(source,), + discovery=lambda _sources: ([live], []), + model_limit=1, + ) + + assert get_credential(source.credential_name) == "new-working-secret" + assert report.selected_agent_ids == ("openai_gpt_new_live",) + assert report.restored_credentials == () diff --git a/tests/test_provider_catalog_store.py b/tests/test_provider_catalog_store.py new file mode 100644 index 000000000..ef59711f2 --- /dev/null +++ b/tests/test_provider_catalog_store.py @@ -0,0 +1,249 @@ +"""Provider catalog persistence and last-known-good contracts.""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest + +from contextual_orchestrator.model_discovery import ( + DiscoveredModel, + ProviderModelSource, +) +from contextual_orchestrator.provider_catalog_store import ( + InMemoryProviderCatalogStore, + PostgresProviderCatalogStore, + PROVIDER_CATALOG_SCHEMA_SQL, + ProviderCatalogError, + normalize_discovered_model, + provider_account_id, +) + + +def _source( + provider: str = "nvidia_nim", + credential: str = "NVIDIA_NIM_API_KEY", +) -> ProviderModelSource: + return ProviderModelSource( + provider_name=provider, + credential_name=credential, + list_url=f"https://{provider}.example/v1/models", + chat_base_url=f"https://{provider}.example/v1", + ) + + +def _model( + source: ProviderModelSource, + model_id: str, + prompt_price: object = 1.0, +) -> DiscoveredModel: + return DiscoveredModel( + provider_name=source.provider_name, + model_id=model_id, + credential_name=source.credential_name, + chat_base_url=source.chat_base_url, + auth_scheme=source.auth_scheme, + prompt_price_per_1k=prompt_price, + completion_price_per_1k=prompt_price, + currency_code="usd", + ) + + +def test_schema_is_normalized_and_contains_no_secret_value_column() -> None: + """Catalog DDL keeps accounts, models, tags, and refresh evidence separate.""" + for table in ( + "provider_account", + "provider_model", + "model_serving_tag", + "catalog_refresh_run", + ): + assert f"CREATE TABLE IF NOT EXISTS {table}" in PROVIDER_CATALOG_SCHEMA_SQL + lowered = PROVIDER_CATALOG_SCHEMA_SQL.casefold() + assert "api_key" not in lowered + assert "secret_value" not in lowered + assert "encrypted_value" not in lowered + model_table = lowered.split( + "create table if not exists provider_model (", 1 + )[1].split(");", 1)[0] + assert "chat_base_url" not in model_table + assert "auth_scheme" not in model_table + + +def test_primary_and_secondary_nim_accounts_have_distinct_ids() -> None: + """Two NIM credentials remain independent quota and failure domains.""" + primary = _source(credential="NVIDIA_NIM_API_KEY") + secondary = _source( + provider="nvidia_nim_sub", + credential="NVIDIA_NIM_API_KEY_SUB", + ) + assert provider_account_id(primary) != provider_account_id(secondary) + + +def test_model_normalization_rejects_cross_account_rows_and_bad_prices() -> None: + """Catalog normalization is account-bound and never stores non-finite prices.""" + source = _source() + wrong = _model( + _source(provider="openai", credential="OPENAI_API_KEY"), + "gpt-test", + ) + with pytest.raises(ProviderCatalogError, match="different account"): + normalize_discovered_model(source, wrong) + + normalized = normalize_discovered_model( + source, + _model(source, " model-a ", float("nan")), + ) + assert normalized.model_id == "model-a" + assert normalized.prompt_price_per_1k is None + assert normalized.completion_price_per_1k is None + assert normalized.currency_code == "USD" + + +def test_success_replaces_current_rows_and_failure_keeps_last_known_good() -> None: + """A failed refresh cannot erase the last successful serving model set.""" + store = InMemoryProviderCatalogStore() + source = _source() + store.record_success( + source, + [_model(source, "model-a"), _model(source, "model-b")], + eligible_model_ids={"model-a"}, + serving_tags={"model-a": ("discovered", "chat", "chat")}, + ) + assert [model.model_id for model in store.serving_models(source)] == [ + "model-a" + ] + assert store.serving_tags(source, "model-a") == ("discovered", "chat") + + store.record_failure(source, error_code="provider_timeout: secret-token") + assert [model.model_id for model in store.serving_models(source)] == [ + "model-a" + ] + assert store.refresh_evidence()[-1].error_code == "unknown_error" + + store.record_success( + source, + [_model(source, "model-c")], + eligible_model_ids={"model-c"}, + serving_tags={"model-c": ("discovered", "chat")}, + ) + assert [model.model_id for model in store.serving_models(source)] == [ + "model-c" + ] + assert [item.refresh_status for item in store.refresh_evidence()] == [ + "succeeded", + "failed", + "succeeded", + ] + + +class _FakeCursor: + """Minimal DB-API cursor recording parameterized catalog statements.""" + + def __init__(self, rows=None) -> None: + self.calls: list[tuple[str, object]] = [] + self.rows = list(rows or []) + + def __enter__(self): + return self + + def __exit__(self, *_args) -> None: + return None + + def execute(self, statement: str, params=None) -> None: + self.calls.append((statement, params)) + + def fetchall(self): + return list(self.rows) + + +class _FakeConnection: + """Minimal transaction object exercising the PostgreSQL adapter.""" + + def __init__(self, rows=None) -> None: + self.cursor_object = _FakeCursor(rows) + self.commits = 0 + + def __enter__(self): + return self + + def __exit__(self, *_args) -> None: + return None + + def cursor(self): + return self.cursor_object + + def commit(self) -> None: + self.commits += 1 + + +def test_postgres_success_is_parameterized_and_failure_does_not_disable_lkg() -> None: + """PostgreSQL success replaces rows; failure records evidence only.""" + source = _source() + connections: list[_FakeConnection] = [] + + def factory(): + connection = _FakeConnection() + connections.append(connection) + return connection + + store = PostgresProviderCatalogStore( + "postgresql://catalog.example/db", + connection_factory=factory, + ) + store.record_success( + source, + [_model(source, "model-a")], + eligible_model_ids={"model-a"}, + serving_tags={"model-a": ("discovered", "chat")}, + ) + success_sql = "\n".join( + statement for statement, _params in connections[-1].cursor_object.calls + ) + assert "UPDATE provider_model SET enabled_flag = false" in success_sql + assert "INSERT INTO model_serving_tag" in success_sql + assert connections[-1].commits >= 1 + + store.record_failure(source, error_code="provider_timeout: secret-token") + failure_sql = "\n".join( + statement for statement, _params in connections[-1].cursor_object.calls + ) + assert "UPDATE provider_model SET enabled_flag = false" not in failure_sql + assert "INSERT INTO catalog_refresh_run" in failure_sql + assert store.refresh_evidence()[-1].error_code == "unknown_error" + + +def test_postgres_serving_models_reconstructs_account_scoped_rows() -> None: + """Read-side rows become normalized DiscoveredModel records.""" + source = _source(provider="openrouter", credential="OPENROUTER_API_KEY") + connection = _FakeConnection( + [ + ( + "model-b", + source.chat_base_url, + "Bearer", + Decimal("0.25"), + Decimal("0.50"), + "usd", + ) + ] + ) + store = PostgresProviderCatalogStore( + "postgresql://catalog.example/db", + connection_factory=lambda: connection, + ) + assert store.serving_models(source) == [ + DiscoveredModel( + provider_name="openrouter", + model_id="model-b", + credential_name="OPENROUTER_API_KEY", + chat_base_url=source.chat_base_url, + auth_scheme="Bearer", + prompt_price_per_1k=0.25, + completion_price_per_1k=0.5, + currency_code="USD", + ) + ] + query, params = connection.cursor_object.calls[-1] + assert "JOIN provider_account AS pa" in query + assert "serving_eligible_flag = true" in query + assert params == (provider_account_id(source),)