diff --git a/README.md b/README.md index 65f57dd4c..fd809251b 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,7 @@ Non-mock providers must use `https://` URLs and a **resolvable KV credential** One public interface: +- `/v1/models` lists `contextual-orchestrator` plus the current worker ids. Live discovery runs only from these KV-registered names: `NVIDIA_NIM_API_KEY`, `NVIDIA_NIM_API_KEY_SUB`, `BYTEZ_API_KEY`, `OPENROUTER_API_KEY`, and `OPENAI_API_KEY`. When none of those names is registered, discovery is skipped and the seed/mock pool is kept. The two NIM nemotron ids are a floor only after discovery runs and every catalog result is empty or fails, and only when a NIM credential is registered. See [docs/model_discovery.md](docs/model_discovery.md). - `/v1/chat/completions` accepts normal chat messages, and `"stream": true` returns an OpenAI-compatible `text/event-stream` of `chat.completion.chunk` deltas terminated by `data: [DONE]`. In **route** mode the worker's tokens are streamed live as they arrive from the provider (real token streaming); in **conduct** mode the multi-step answer is produced then framed as deltas (a workflow can't honestly token-stream a synthesizer that hasn't run yet). - `TaskOrchestrator.complete()` decides whether to route to one worker or run a short workflow. - `TaskOrchestrator.compare_to_baseline(prompts, mode)` (CLI `--eval PROMPT...`) measures the orchestration engine against a single-worker baseline — per-prompt and aggregate latency plus a structural coverage delta (contributing steps + verifier-pass presence). It is a measured tradeoff report, not a human-quality claim. @@ -252,6 +253,10 @@ python -m pip install --require-hashes -r requirements.lock python -m pip install --no-deps -e . python tests/test_self_check.py python tests/test_paper_contracts.py +python tests/test_model_discovery.py +python tests/test_original_list_price.py +python tests/test_models_list.py +python tests/test_compute_allocation.py python tests/test_admin_contract.py python tests/test_conventions.py python tests/test_api_contract.py diff --git a/conductor/product.md b/conductor/product.md index 3092bda2a..433f1c159 100644 --- a/conductor/product.md +++ b/conductor/product.md @@ -24,7 +24,8 @@ Provide one API and one domain model: - route simple work to one selected worker; - conduct complex work through planner, worker, verifier, and synthesizer steps; - keep worker visibility explicit with access lists; -- make the agent pool configurable data. +- make the agent pool configurable data, composed by live auto-discovery + from registered provider keys (two NIM ids are a floor only). - expose an admin console for operators to inspect agents, policy, workflow trace, and audit state. ## Source-backed Product Bets diff --git a/contextual_orchestrator/__init__.py b/contextual_orchestrator/__init__.py index 70dbd71c6..e51969ad8 100644 --- a/contextual_orchestrator/__init__.py +++ b/contextual_orchestrator/__init__.py @@ -37,7 +37,16 @@ from .cost_router import CostRoutingCoordinator from .credentials import NotConfigured, get_credential, register_credential from .kv_config import InMemoryConfigStore, get_config_store -from .orchestrator import ModelAgent, TaskOrchestrator, WorkflowStep, load_agents +from .model_discovery import ( + DISCOVERY_CREDENTIAL_NAMES, + FLOOR_DEFAULT_MODEL_ID, + FLOOR_SMALL_MODEL_ID, + apply_discovered_pool, + discover_model_catalog, + list_served_models, +) +from .orchestrator import ModelAgent, WorkflowStep, load_agents, known_agent_comparison_cost +from .passthrough_failover import ResilientTaskOrchestrator as TaskOrchestrator from .token_counting import HeuristicTokenCounter, build_token_counter __all__ = [ @@ -48,6 +57,13 @@ "get_credential", "register_credential", "NotConfigured", + "DISCOVERY_CREDENTIAL_NAMES", + "FLOOR_DEFAULT_MODEL_ID", + "FLOOR_SMALL_MODEL_ID", + "apply_discovered_pool", + "discover_model_catalog", + "list_served_models", + "known_agent_comparison_cost", # cost review "ATTRIBUTION_DIMENSIONS", "AttributionDimensions", diff --git a/contextual_orchestrator/__main__.py b/contextual_orchestrator/__main__.py index 5f68c3b74..d3eec0507 100644 --- a/contextual_orchestrator/__main__.py +++ b/contextual_orchestrator/__main__.py @@ -8,7 +8,9 @@ import sys from .credentials import register_credential -from .orchestrator import ModelClient, TaskOrchestrator, load_agents +from .model_discovery import apply_discovered_pool +from .orchestrator import ModelClient, load_agents +from .passthrough_failover import ResilientTaskOrchestrator from .server import SecurityConfig, serve @@ -95,7 +97,7 @@ def main() -> None: args = parser.parse_args() client = ModelClient(ca_bundle=args.provider_ca_bundle, verify_tls=not args.insecure_skip_tls_verify) - orchestrator = TaskOrchestrator( + orchestrator = ResilientTaskOrchestrator( load_agents(args.agents), client=client, state_db=args.state_db, @@ -104,6 +106,12 @@ def main() -> None: budget_max_cost_usd=args.budget_max_cost_usd, cache_ttl=args.cache_ttl, ) + # Product auto-discovery: when any of the five provider keys is in the KV, + # replace the seed pool with the live catalog. Catalog HTTPS reuses this + # ModelClient's TLS settings. NIM floor only if every fetch is empty and a + # NIM credential is registered. Unregistered keys are skipped — never + # os.getenv as a "key exists" signal. + apply_discovered_pool(orchestrator) if args.eval: print(json.dumps(orchestrator.compare_to_baseline(args.eval, mode=args.mode), ensure_ascii=False, indent=2)) diff --git a/contextual_orchestrator/api_contract.py b/contextual_orchestrator/api_contract.py index fae9fba0b..842f41a50 100644 --- a/contextual_orchestrator/api_contract.py +++ b/contextual_orchestrator/api_contract.py @@ -17,6 +17,30 @@ } }, "paths": { + "/v1/models": { + "get": { + "operationId": "list_served_models", + "summary": "List the gateway model plus discovered or floor worker ids", + "security": [{"inference_bearer_auth": []}], + "responses": {"200": {"description": "OpenAI-compatible model list"}}, + } + }, + "/api/v1/provider_catalogs": { + "get": { + "operationId": "get_provider_catalogs", + "summary": "Read the last secret-redacted discovery snapshot", + "security": [{"admin_bearer_auth": []}], + "responses": {"200": {"description": "Provider catalog snapshot"}}, + } + }, + "/api/v1/provider_catalogs/refresh": { + "post": { + "operationId": "refresh_provider_catalogs", + "summary": "Re-run live discovery from KV-registered provider keys", + "security": [{"admin_bearer_auth": []}], + "responses": {"200": {"description": "Refreshed catalog snapshot"}}, + } + }, "/api/v1/agent_pools": { "get": { "operationId": "list_agent_pools", diff --git a/contextual_orchestrator/batch_routing.py b/contextual_orchestrator/batch_routing.py index d07a48d25..654ad9f3c 100644 --- a/contextual_orchestrator/batch_routing.py +++ b/contextual_orchestrator/batch_routing.py @@ -126,9 +126,10 @@ def cheapest_upstream( Cost-optimising upstream selection for load balancing: given candidate provider/model pairs, price each against the configurable price table for a - representative request shape and return the cheapest. Unpriced candidates - cost ``0`` and are treated as free (explicit, so a missing price is visible - rather than silently expensive). Ties keep input order. + representative request shape and return the cheapest **known-cost** + candidate. Unpriced models are skipped — unknown is never treated as + free. Ties keep input order. If every candidate is unpriced, return + ``None``. """ if not candidates: return None @@ -137,9 +138,11 @@ def cheapest_upstream( for candidate in candidates: provider = candidate.get("provider", "") model = candidate.get("model", "") - cost, _currency = price_book.compute_cost( + cost, _currency = price_book.known_compute_cost( provider, model, assumed_prompt_tokens, assumed_completion_tokens ) + if cost is None: + continue if best_cost is None or cost < best_cost: best_cost = cost best = candidate diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index 20799c10f..3dcebe2ab 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -35,6 +35,8 @@ from typing import Any, Dict, List, Optional, Protocol import uuid +from .price_honesty import optional_finite_price + # --------------------------------------------------------------------------- # Attribution dimensions @@ -114,6 +116,8 @@ class PriceEntry: prompt_price_per_1k: float completion_price_per_1k: float currency_code: str = "USD" + original_list_prompt_per_1k: float | None = None + original_list_completion_per_1k: float | None = None def as_dict(self) -> Dict[str, Any]: """Serialize the price entry for KV storage / reporting.""" @@ -123,6 +127,8 @@ def as_dict(self) -> Dict[str, Any]: "prompt_price_per_1k": self.prompt_price_per_1k, "completion_price_per_1k": self.completion_price_per_1k, "currency_code": self.currency_code, + "original_list_prompt_per_1k": self.original_list_prompt_per_1k, + "original_list_completion_per_1k": self.original_list_completion_per_1k, } @@ -162,12 +168,20 @@ def get_price(self, provider: str, model: str) -> Optional[PriceEntry]: raw = self._config.get(_PRICE_CATEGORY, _price_key(provider, "*"), None) if raw is None: return None + if not isinstance(raw, dict): + return None + billed_keys = ("prompt_price_per_1k", "completion_price_per_1k") + if not any(key in raw for key in billed_keys): + # A stub without billed-rate keys is unpriced, not promotional-free. + return None return PriceEntry( provider_name=raw.get("provider_name", provider), model_name=raw.get("model_name", model), prompt_price_per_1k=float(raw.get("prompt_price_per_1k", 0.0)), completion_price_per_1k=float(raw.get("completion_price_per_1k", 0.0)), currency_code=raw.get("currency_code", self.default_currency), + original_list_prompt_per_1k=optional_finite_price(raw.get("original_list_prompt_per_1k")), + original_list_completion_per_1k=optional_finite_price(raw.get("original_list_completion_per_1k")), ) def compute_cost( @@ -196,6 +210,43 @@ def compute_cost( ) return float(total), entry.currency_code + def known_compute_cost( + self, + provider: str, + model: str, + prompt_tokens: int, + completion_tokens: int, + ) -> tuple[float | None, str]: + """Return a known cost, or ``None`` when the model is unpriced. + + Selection must call this — not :meth:`compute_cost`, which records + unpriced usage as ``0.0`` so the ledger never fails. A promotional + billed rate of ``0`` with ``original_list_*`` uses the list price. + """ + entry = self.get_price(provider, model) + if entry is None: + return None, self.default_currency + billed_prompt = entry.prompt_price_per_1k + billed_completion = entry.completion_price_per_1k + if ( + billed_prompt == 0.0 + and billed_completion == 0.0 + and ( + entry.original_list_prompt_per_1k is not None + or entry.original_list_completion_per_1k is not None + ) + ): + billed_prompt = entry.original_list_prompt_per_1k or 0.0 + billed_completion = entry.original_list_completion_per_1k or 0.0 + prompt_cost = (Decimal(prompt_tokens) / Decimal(1000)) * Decimal(str(billed_prompt)) + completion_cost = (Decimal(completion_tokens) / Decimal(1000)) * Decimal( + str(billed_completion) + ) + total = (prompt_cost + completion_cost).quantize( + Decimal("0.000001"), rounding=ROUND_HALF_UP + ) + return float(total), entry.currency_code + # --------------------------------------------------------------------------- # Usage records + stores @@ -554,6 +605,66 @@ def __len__(self) -> int: "currency_code", ) +# Module-constant SQL selected by DB-API paramstyle. Semgrep flags f-string +# interpolation at execute(); these statements are fixed literals and bind +# values only through DB-API parameters. +_USAGE_COLUMN_LIST = ( + "usage_record_id, created_at, workflow_run_id, request_channel, route_mode, " + "provider_name, model_name, account_name, service_name, upstream_api, " + "team_name, group_name, company_name, prompt_tokens, completion_tokens, " + "total_tokens, cost_amount, currency_code" +) +SELECT_DIMENSION_SQL = { + "qmark": "SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = ?", + "pyformat": "SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = %s", +} +INSERT_DIMENSION_SQL = { + "qmark": ( + "INSERT INTO cost_attribution_dimensions " + "(dimension_name, dimension_label, dimension_order) VALUES (?, ?, ?)" + ), + "pyformat": ( + "INSERT INTO cost_attribution_dimensions " + "(dimension_name, dimension_label, dimension_order) VALUES (%s, %s, %s)" + ), +} +INSERT_USAGE_SQL = { + "qmark": ( + "INSERT INTO llm_usage_records (" + + _USAGE_COLUMN_LIST + + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" + ), + "pyformat": ( + "INSERT INTO llm_usage_records (" + + _USAGE_COLUMN_LIST + + ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)" + ), +} +SELECT_USAGE_SQL = { + ("qmark", False, False): "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records", + ("qmark", True, False): ( + "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records WHERE created_at >= ?" + ), + ("qmark", False, True): ( + "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records WHERE created_at < ?" + ), + ("qmark", True, True): ( + "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records " + "WHERE created_at >= ? AND created_at < ?" + ), + ("pyformat", False, False): "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records", + ("pyformat", True, False): ( + "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records WHERE created_at >= %s" + ), + ("pyformat", False, True): ( + "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records WHERE created_at < %s" + ), + ("pyformat", True, True): ( + "SELECT " + _USAGE_COLUMN_LIST + " FROM llm_usage_records " + "WHERE created_at >= %s AND created_at < %s" + ), +} + class SqlLedgerStore: """PEP-249 SQL ledger store (stdlib ``sqlite3`` or ``psycopg``). @@ -569,9 +680,6 @@ def __init__(self, connection: Any, paramstyle: str = "qmark") -> None: self._create_schema() self._seed_dimension_catalog() - def _placeholder(self) -> str: - return "?" if self._paramstyle == "qmark" else "%s" - def _create_schema(self) -> None: cur = self._conn.cursor() for statement in SCHEMA_SQL.strip().split(";"): @@ -580,54 +688,35 @@ def _create_schema(self) -> None: self._conn.commit() def _seed_dimension_catalog(self) -> None: - ph = self._placeholder() cur = self._conn.cursor() + select_sql = SELECT_DIMENSION_SQL[self._paramstyle] + insert_sql = INSERT_DIMENSION_SQL[self._paramstyle] for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG): - # (name,) is a real, separately-bound parameter, never string-concatenated into the - # query; this is a raw DB-API cursor (sqlite3/psycopg), not SQLAlchemy. - cur.execute( # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query - f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}", # nosec B608 - ph is a DB-API placeholder. - (name,), - ) + cur.execute(select_sql, (name,)) if cur.fetchone() is None: - cur.execute( - "INSERT INTO cost_attribution_dimensions " - f"(dimension_name, dimension_label, dimension_order) VALUES ({ph}, {ph}, {ph})", # nosec B608 - ph is a DB-API placeholder. - (name, label, order), - ) + cur.execute(insert_sql, (name, label, order)) self._conn.commit() def append(self, record: UsageRecord) -> None: """Insert a usage record row.""" row = record.as_dict() - ph = self._placeholder() - placeholders = ", ".join(ph for _ in _USAGE_COLUMNS) - columns = ", ".join(_USAGE_COLUMNS) cur = self._conn.cursor() - # values are bound via the separate params tuple below; raw DB-API cursor, not SQLAlchemy. - cur.execute( # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query - f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})", # nosec B608 - columns are fixed _USAGE_COLUMNS. + cur.execute( + INSERT_USAGE_SQL[self._paramstyle], tuple(row.get(column) for column in _USAGE_COLUMNS), ) self._conn.commit() def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[Dict[str, Any]]: """Return record rows in the optional half-open window.""" - ph = self._placeholder() - clauses: List[str] = [] params: List[Any] = [] if start is not None: - clauses.append(f"created_at >= {ph}") params.append(start) if end is not None: - clauses.append(f"created_at < {ph}") params.append(end) - where = f" WHERE {' AND '.join(clauses)}" if clauses else "" - columns = ", ".join(_USAGE_COLUMNS) + statement = SELECT_USAGE_SQL[(self._paramstyle, start is not None, end is not None)] cur = self._conn.cursor() - # clauses only ever contain the two fixed literal fragments above, never interpolated - # values (those are in params); raw DB-API cursor, not SQLAlchemy. - cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)) # nosec B608 - columns and clauses are fixed. # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query + cur.execute(statement, tuple(params)) return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()] diff --git a/contextual_orchestrator/model_discovery.py b/contextual_orchestrator/model_discovery.py new file mode 100644 index 000000000..817612e46 --- /dev/null +++ b/contextual_orchestrator/model_discovery.py @@ -0,0 +1,833 @@ +"""Live model auto-discovery for the org LLM gateway. + +This is the product catalog: ContextualWisdomLab apps consume whatever the +gateway discovers from registered provider credentials, then route with +Fugu (single-worker latency), Conductor (access-listed workflow), and +TRINITY (thinker / worker / verifier) compute allocation. + +The two NVIDIA NIM ids in :data:`FLOOR_DEFAULT_MODEL_ID` and +:data:`FLOOR_SMALL_MODEL_ID` are a **floor only**. They are used when every +registered catalog fetch returns nothing. They are not the authoritative +inventory. + +Credential resolution uses :func:`get_credential` only. A missing +registration is ``None`` — never ``os.getenv`` as a product fallback. + +Price honesty (issue #86): + +* an explicit billed rate of ``0`` is known-free only when both prompt and + completion prices are finite; +* a free channel that still has a published list/sibling price stores that + value as ``original_list_price`` and is compared at the list price; +* missing, partial, boolean, non-numeric, negative, NaN, infinite, or + overflowing prices are ``unknown`` and are never converted to ``0`` / "free". +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +import json +import re +from typing import Any, Callable, Iterable, Mapping +from urllib.parse import urlparse +import urllib.error +import urllib.request + +from .conventions import is_two_word_snake_case, require_object_name +from .credentials import get_credential +from .orchestrator import ModelAgent, ModelClient, TaskOrchestrator +from .price_honesty import complete_pair_mean, known_comparison_cost, optional_finite_price + + +DISCOVERY_CREDENTIAL_NAMES: tuple[str, ...] = ( + "NVIDIA_NIM_API_KEY", + "NVIDIA_NIM_API_KEY_SUB", + "BYTEZ_API_KEY", + "OPENROUTER_API_KEY", + "OPENAI_API_KEY", +) + +FLOOR_DEFAULT_MODEL_ID = "nvidia-nim/nvidia/nemotron-3-ultra-550b-a55b" +FLOOR_SMALL_MODEL_ID = "nvidia-nim/nvidia/nemotron-3-super-120b-a12b" + +NVIDIA_NIM_BASE_URL = "https://integrate.api.nvidia.com/v1" +OPENAI_BASE_URL = "https://api.openai.com/v1" +OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1" +BYTEZ_BASE_URL = "https://api.bytez.com" + +_NON_CHAT_MARKERS = ( + "embed", + "rerank", + "whisper", + "tts", + "dall-e", + "dalle", + "imagen", + "moderation", + "audio", + "speech", + "transcri", + "image", + "video", + "clip", +) + +_SMALL_MARKERS = ( + "mini", + "small", + "nano", + "haiku", + "flash", + "super", + "7b", + "8b", + "9b", + "12b", + "13b", + "120b", +) +_LARGE_MARKERS = ( + "ultra", + "opus", + "sonnet", + "o1", + "o3", + "o4", + "gpt-4", + "gpt-5", + "70b", + "72b", + "405b", + "550b", +) + +_MAX_CATALOG_BYTES = 2 * 1024 * 1024 +_AGENT_ID_MAX = 64 + +CatalogFetcher = Callable[["ProviderEndpoint", str], Any] + + +@dataclass(frozen=True) +class ProviderEndpoint: + """One official catalog origin keyed by a KV credential name.""" + + credential_name: str + provider_name: str + base_url: str + catalog_path: str + catalog_style: str + auth_scheme: str + price_unit: str + http_method: str = "GET" + + +PROVIDER_ENDPOINTS: dict[str, ProviderEndpoint] = { + "NVIDIA_NIM_API_KEY": ProviderEndpoint( + credential_name="NVIDIA_NIM_API_KEY", + provider_name="nvidia_nim", + base_url=NVIDIA_NIM_BASE_URL, + catalog_path="/models", + catalog_style="openai", + auth_scheme="bearer", + price_unit="per_million", + ), + "NVIDIA_NIM_API_KEY_SUB": ProviderEndpoint( + credential_name="NVIDIA_NIM_API_KEY_SUB", + provider_name="nvidia_nim_sub", + base_url=NVIDIA_NIM_BASE_URL, + catalog_path="/models", + catalog_style="openai", + auth_scheme="bearer", + price_unit="per_million", + ), + "BYTEZ_API_KEY": ProviderEndpoint( + credential_name="BYTEZ_API_KEY", + provider_name="bytez", + base_url=BYTEZ_BASE_URL, + catalog_path="/models/v2", + catalog_style="bytez", + auth_scheme="key", + price_unit="per_million", + http_method="GET", + ), + "OPENROUTER_API_KEY": ProviderEndpoint( + credential_name="OPENROUTER_API_KEY", + provider_name="openrouter", + base_url=OPENROUTER_BASE_URL, + catalog_path="/models", + catalog_style="openai", + auth_scheme="bearer", + price_unit="per_token", + ), + "OPENAI_API_KEY": ProviderEndpoint( + credential_name="OPENAI_API_KEY", + provider_name="openai", + base_url=OPENAI_BASE_URL, + catalog_path="/models", + catalog_style="openai", + auth_scheme="bearer", + price_unit="per_million", + ), +} + + +@dataclass(frozen=True) +class CatalogModel: + """One discovered (or floor) chat model with honest price fields.""" + + model_id: str + provider_name: str + credential_name: str + base_url: str + owner: str = "" + billed_prompt_per_million: float | None = None + billed_completion_per_million: float | None = None + original_list_prompt_per_million: float | None = None + original_list_completion_per_million: float | None = None + price_status: str = "unknown" + discovery_source: str = "live" + capability_kind: str = "chat" + + def comparison_cost(self) -> float | None: + """Return the known ranking cost, or ``None`` when the model is unpriced. + + Promotional-free rows with a stored list price compare at that list + price. Explicit billed ``0`` with no list price is known-free (``0``). + Unknown is never converted to ``0``. + """ + billed = complete_pair_mean( + self.billed_prompt_per_million, self.billed_completion_per_million + ) + listed = complete_pair_mean( + self.original_list_prompt_per_million, + self.original_list_completion_per_million, + ) + return known_comparison_cost(billed, listed, self.price_status) + + def as_dict(self) -> dict[str, Any]: + """Secret-free snapshot row for operators and CWL consumers.""" + return { + "model_id": self.model_id, + "provider_name": self.provider_name, + "credential_name": self.credential_name, + "base_url": self.base_url, + "owner": self.owner, + "billed_prompt_per_million": self.billed_prompt_per_million, + "billed_completion_per_million": self.billed_completion_per_million, + "original_list_prompt_per_million": self.original_list_prompt_per_million, + "original_list_completion_per_million": self.original_list_completion_per_million, + "price_status": self.price_status, + "discovery_source": self.discovery_source, + "capability_kind": self.capability_kind, + "comparison_cost": self.comparison_cost(), + } + + +@dataclass +class DiscoverySnapshot: + """Redacted result of one catalog composition pass.""" + + models: list[CatalogModel] + source: str + used_floor: bool + registered_credentials: tuple[str, ...] + skipped_credentials: tuple[str, ...] + provider_errors: dict[str, str] = field(default_factory=dict) + + def as_dict(self) -> dict[str, Any]: + """Serialize the snapshot without secrets or raw provider payloads.""" + return { + "source": self.source, + "used_floor": self.used_floor, + "registered_credentials": list(self.registered_credentials), + "skipped_credentials": list(self.skipped_credentials), + "provider_errors": dict(self.provider_errors), + "model_count": len(self.models), + "models": [model.as_dict() for model in self.models], + } + + +def registered_discovery_keys() -> tuple[str, ...]: + """Return discovery credential names that are present in the KV. + + Absence is ``get_credential(...) is None``. This function never reads + ``os.getenv`` for a missing registration. + """ + return tuple(name for name in DISCOVERY_CREDENTIAL_NAMES if get_credential(name)) + + +def skipped_discovery_keys() -> tuple[str, ...]: + """Return discovery names that have no KV registration (not an env miss).""" + registered = set(registered_discovery_keys()) + return tuple(name for name in DISCOVERY_CREDENTIAL_NAMES if name not in registered) + + +def finite_unit_price(value: Any) -> float | None: + """Parse a price as a finite non-negative float, or ``None`` if unknown. + + Booleans, strings that are not numbers, negatives, NaN, infinities, and + values that overflow ``float`` are unknown — they are not coerced to ``0``. + """ + return optional_finite_price(value) + + +def price_per_million(value: Any, *, unit: str) -> float | None: + """Normalize a provider price into USD per million tokens, or unknown.""" + parsed = finite_unit_price(value) + if parsed is None: + return None + if unit == "per_token": + return parsed * 1_000_000.0 + return parsed + + +def classify_price_status( + billed_prompt: float | None, + billed_completion: float | None, + list_prompt: float | None, + list_completion: float | None, +) -> str: + """Return ``known``, ``promotional_free``, or ``unknown``.""" + billed = complete_pair_mean(billed_prompt, billed_completion) + listed = complete_pair_mean(list_prompt, list_completion) + if billed == 0.0 and listed is not None: + return "promotional_free" + if billed is not None: + return "known" + if listed is not None: + return "known" + return "unknown" + + +def is_chat_model_id(model_id: str) -> bool: + """Return whether ``model_id`` looks like a chat/completion candidate.""" + lowered = model_id.lower() + return not any(marker in lowered for marker in _NON_CHAT_MARKERS) + + +def size_class_for_model(model_id: str) -> str: + """Heuristic Fugu size class: ``small`` (latency) or ``default`` (quality).""" + lowered = model_id.lower() + if any(marker in lowered for marker in _LARGE_MARKERS): + return "default" + if any(marker in lowered for marker in _SMALL_MARKERS): + return "small" + return "default" + + +def allocate_compute_tags(model_id: str, *, size_class: str | None = None) -> tuple[str, ...]: + """Assign Fugu / Conductor / TRINITY role tags from the model identity. + + Fugu latency routing prefers ``small`` / ``cheap`` workers. Conductor and + TRINITY conduct paths need thinker (reasoning/planning), worker (coding), + verifier (review), and synthesizer (writing) coverage. Tags are additive + so a discovered model can fill more than one role when the pool is thin. + """ + size = size_class or size_class_for_model(model_id) + lowered = model_id.lower() + tags = {"reasoning"} + if size == "small": + tags.update({"cheap", "fallback", "coding", "implementation", "summarization"}) + else: + tags.update({"planning", "writing", "analysis", "review", "verification"}) + if any(token in lowered for token in ("code", "coder", "codex", "starcoder", "qwen2.5-coder")): + tags.update({"coding", "implementation", "debugging"}) + if any(token in lowered for token in ("guard", "safety", "review", "critic")): + tags.update({"review", "verification", "security"}) + return tuple(sorted(tags)) + + +def agent_id_for(provider_name: str, model_id: str, taken: set[str]) -> str: + """Build a two-word snake_case agent id, resolving slug collisions.""" + raw = f"{provider_name}_{model_id}".lower() + slug = re.sub(r"[^a-z0-9]+", "_", raw).strip("_") + if not slug: + slug = "discovered_model" + if "_" not in slug: + slug = f"model_{slug}" + slug = slug[:_AGENT_ID_MAX].rstrip("_") + if not is_two_word_snake_case(slug): + slug = f"model_{slug}" if "_" not in slug else slug + slug = re.sub(r"_+", "_", slug).strip("_") + candidate = slug + suffix = 2 + while candidate in taken or not is_two_word_snake_case(candidate): + trimmed = slug[: max(8, _AGENT_ID_MAX - 3)] + candidate = f"{trimmed}_{suffix}" + suffix += 1 + require_object_name(candidate, "agent.id") + taken.add(candidate) + return candidate + + +def floor_credential_name() -> str | None: + """Return a registered NVIDIA NIM credential, or ``None`` if neither exists. + + Floor rows are callable only when a NIM key is already in the KV. Absence + is ``get_credential(...) is None`` — never an environment fallback. + """ + for name in ("NVIDIA_NIM_API_KEY", "NVIDIA_NIM_API_KEY_SUB"): + if get_credential(name): + return name + return None + + +def floor_models() -> list[CatalogModel]: + """Return the two NIM floor rows when a NIM credential is registered. + + Used only after discovery ran and every catalog was empty or failed. + Without a NIM key the existing seed pool is left in place. + """ + credential_name = floor_credential_name() + if credential_name is None: + return [] + return [ + CatalogModel( + model_id=FLOOR_DEFAULT_MODEL_ID, + provider_name="nvidia_nim", + credential_name=credential_name, + base_url=NVIDIA_NIM_BASE_URL, + owner="nvidia", + price_status="unknown", + discovery_source="floor", + ), + CatalogModel( + model_id=FLOOR_SMALL_MODEL_ID, + provider_name="nvidia_nim", + credential_name=credential_name, + base_url=NVIDIA_NIM_BASE_URL, + owner="nvidia", + price_status="unknown", + discovery_source="floor", + ), + ] + + +def extract_catalog_rows(payload: Any) -> list[dict[str, Any]]: + """Normalize an arbitrary catalog JSON payload into row mappings. + + Accepts OpenAI ``{data: [...]}``, Bytez ``{models: [...]}``, or a bare + list. Non-mapping items and non-container payloads yield an empty list + rather than inventing models. + """ + if isinstance(payload, list): + raw_rows = payload + elif isinstance(payload, Mapping): + for key in ("data", "models", "items"): + value = payload.get(key) + if isinstance(value, list): + raw_rows = value + break + else: + return [] + else: + return [] + rows: list[dict[str, Any]] = [] + for item in raw_rows: + if isinstance(item, Mapping): + rows.append(dict(item)) + elif isinstance(item, str) and item.strip(): + rows.append({"id": item.strip()}) + return rows + + +def row_model_id(row: Mapping[str, Any]) -> str: + """Return the catalog model id from a provider row, or empty.""" + for key in ("id", "model", "name", "model_id"): + value = row.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + return "" + + +def row_owner(row: Mapping[str, Any], model_id: str) -> str: + """Best-effort owner/org label from a catalog row.""" + for key in ("owned_by", "owner", "organization", "publisher"): + value = row.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + if "/" in model_id: + return model_id.split("/", 1)[0] + return "" + + +def extract_row_prices( + row: Mapping[str, Any], *, unit: str +) -> tuple[float | None, float | None, float | None, float | None]: + """Return billed and original-list prompt/completion prices per million. + + List/published fields are preserved even when the billed channel is free. + """ + pricing = row.get("pricing") if isinstance(row.get("pricing"), Mapping) else {} + billed_prompt = _first_price( + row, + pricing, + ( + "prompt_price_per_million", + "input_price_per_million", + "prompt", + "input", + ), + unit=unit, + ) + billed_completion = _first_price( + row, + pricing, + ( + "completion_price_per_million", + "output_price_per_million", + "completion", + "output", + ), + unit=unit, + ) + list_prompt = _first_price( + row, + pricing, + ( + "list_prompt_per_million", + "published_prompt_per_million", + "original_prompt_per_million", + "list_prompt", + "published_prompt", + "original_list_prompt", + ), + unit=unit, + ) + list_completion = _first_price( + row, + pricing, + ( + "list_completion_per_million", + "published_completion_per_million", + "original_completion_per_million", + "list_completion", + "published_completion", + "original_list_completion", + ), + unit=unit, + ) + return billed_prompt, billed_completion, list_prompt, list_completion + + +def sibling_list_price( + model_id: str, + rows_by_id: Mapping[str, Mapping[str, Any]], + *, + unit: str, +) -> tuple[float | None, float | None]: + """For a ``:free`` variant, copy the paid sibling's billed rates as list price.""" + if not model_id.endswith(":free"): + return None, None + sibling_id = model_id[: -len(":free")] + sibling = rows_by_id.get(sibling_id) + if sibling is None: + return None, None + prompt, completion, list_prompt, list_completion = extract_row_prices(sibling, unit=unit) + return ( + list_prompt if list_prompt is not None else prompt, + list_completion if list_completion is not None else completion, + ) + + +def normalize_catalog_payload( + payload: Any, + endpoint: ProviderEndpoint, +) -> list[CatalogModel]: + """Turn one provider catalog payload into chat :class:`CatalogModel` rows. + + This is the untrusted-input seam: malformed or duplicate catalogs must + not invent models, leak secrets, or treat junk prices as free. + """ + rows = extract_catalog_rows(payload) + rows_by_id = {row_model_id(row): row for row in rows if row_model_id(row)} + seen: set[str] = set() + models: list[CatalogModel] = [] + for row in rows: + model_id = row_model_id(row) + if not model_id or model_id in seen or not is_chat_model_id(model_id): + continue + seen.add(model_id) + billed_prompt, billed_completion, list_prompt, list_completion = extract_row_prices( + row, unit=endpoint.price_unit + ) + sibling_prompt, sibling_completion = sibling_list_price( + model_id, rows_by_id, unit=endpoint.price_unit + ) + if list_prompt is None: + list_prompt = sibling_prompt + if list_completion is None: + list_completion = sibling_completion + status = classify_price_status( + billed_prompt, billed_completion, list_prompt, list_completion + ) + models.append( + CatalogModel( + model_id=model_id, + provider_name=endpoint.provider_name, + credential_name=endpoint.credential_name, + base_url=endpoint.base_url, + owner=row_owner(row, model_id), + billed_prompt_per_million=billed_prompt, + billed_completion_per_million=billed_completion, + original_list_prompt_per_million=list_prompt, + original_list_completion_per_million=list_completion, + price_status=status, + discovery_source="live", + ) + ) + return models + + +def discover_model_catalog( + *, + fetcher: CatalogFetcher | None = None, + client: ModelClient | None = None, +) -> DiscoverySnapshot: + """Discover chat models from every KV-registered provider key. + + Unregistered names are skipped (KV miss, not an environment fallback). + When every fetch is empty or fails and a NIM credential is registered, + the snapshot is the NIM floor. Otherwise the catalog stays empty so the + caller can keep its seed pool. + """ + registered = registered_discovery_keys() + skipped = skipped_discovery_keys() + fetch = fetcher or ( + lambda endpoint, api_key: fetch_provider_catalog(endpoint, api_key, client=client) + ) + discovered: list[CatalogModel] = [] + errors: dict[str, str] = {} + for name in registered: + endpoint = PROVIDER_ENDPOINTS[name] + api_key = get_credential(name) + if not api_key: + continue + try: + payload = fetch(endpoint, api_key) + discovered.extend(normalize_catalog_payload(payload, endpoint)) + except Exception as exc: # noqa: BLE001 - one provider must not abort the catalog + errors[name] = _redact_error(exc) + if discovered: + return DiscoverySnapshot( + models=_dedupe_models(discovered), + source="live", + used_floor=False, + registered_credentials=registered, + skipped_credentials=skipped, + provider_errors=errors, + ) + floor = floor_models() + if floor: + return DiscoverySnapshot( + models=floor, + source="floor", + used_floor=True, + registered_credentials=registered, + skipped_credentials=skipped, + provider_errors=errors, + ) + return DiscoverySnapshot( + models=[], + source="empty", + used_floor=False, + registered_credentials=registered, + skipped_credentials=skipped, + provider_errors=errors, + ) + + +def agents_from_catalog(models: Iterable[CatalogModel]) -> list[ModelAgent]: + """Materialize :class:`ModelAgent` workers from catalog rows.""" + taken: set[str] = set() + agents: list[ModelAgent] = [] + for model in models: + agent_id = agent_id_for(model.provider_name, model.model_id, taken) + size = size_class_for_model(model.model_id) + cost = model.comparison_cost() + list_cost = complete_pair_mean( + model.original_list_prompt_per_million, + model.original_list_completion_per_million, + ) + agents.append( + ModelAgent( + id=agent_id, + model=model.model_id, + base_url=model.base_url, + credential_key=model.credential_name, + tags=allocate_compute_tags(model.model_id, size_class=size), + priority=2 if size == "default" else 1, + provider_name=model.provider_name, + price_per_million=cost, + original_list_price=list_cost, + price_status=model.price_status, + discovery_source=model.discovery_source, + ) + ) + return agents + + +def apply_discovered_pool( + orchestrator: TaskOrchestrator, + *, + fetcher: CatalogFetcher | None = None, + replace_unregistered: bool = False, +) -> DiscoverySnapshot: + """Replace the live pool when discovery credentials are registered. + + If no discovery key is in the KV, the seed/mock pool is left in place + unless ``replace_unregistered`` is true (tests / explicit product floor). + ``os.getenv`` is never consulted to decide that a key "exists". + """ + fetch = fetcher or getattr(orchestrator, "catalog_fetcher", None) + registered = registered_discovery_keys() + if not registered and not replace_unregistered: + snapshot = DiscoverySnapshot( + models=[], + source="seed", + used_floor=False, + registered_credentials=(), + skipped_credentials=skipped_discovery_keys(), + ) + orchestrator.discovery_snapshot = snapshot.as_dict() + return snapshot + snapshot = discover_model_catalog( + fetcher=fetch, + client=getattr(orchestrator, "client", None), + ) + agents = agents_from_catalog(snapshot.models) + if agents: + orchestrator.agents = agents + orchestrator.discovery_snapshot = snapshot.as_dict() + return snapshot + + +def list_served_models(orchestrator: TaskOrchestrator) -> dict[str, Any]: + """OpenAI-compatible ``GET /v1/models`` body for this gateway.""" + rows = [ + { + "id": "contextual-orchestrator", + "object": "model", + "owned_by": "contextual-orchestrator", + "discovery_source": "gateway", + } + ] + seen = {"contextual-orchestrator"} + for agent in orchestrator.agents: + if agent.model in seen: + continue + seen.add(agent.model) + rows.append( + { + "id": agent.model, + "object": "model", + "owned_by": agent.provider_name or "agent_pool", + "discovery_source": agent.discovery_source or "seed", + "price_status": agent.price_status, + "original_list_price": agent.original_list_price, + } + ) + return {"object": "list", "data": rows} + + +def comparison_cost_for_agent(agent: ModelAgent) -> float | None: + """Known ranking cost for a worker, or ``None`` when unpriced. + + A free billed channel with ``original_list_price`` compares at the list + price. Unpriced agents are not treated as free. + """ + return known_comparison_cost( + finite_unit_price(getattr(agent, "price_per_million", None)), + finite_unit_price(getattr(agent, "original_list_price", None)), + getattr(agent, "price_status", "unknown"), + ) + + +def fetch_provider_catalog( + endpoint: ProviderEndpoint, + api_key: str, + client: ModelClient | None = None, +) -> Any: + """Fetch one official catalog through the chat egress policy. + + Reuses the configured :class:`ModelClient` host/TLS checks when one is + supplied (CA bundle / ``--insecure-skip-tls-verify``). Redirects are + rejected. The Bearer/Key header is attached only after the URL is + validated. + """ + probe = ModelAgent( + id="catalog_probe", + model="catalog_probe", + base_url=endpoint.base_url, + credential_key=endpoint.credential_name, + provider_name=endpoint.provider_name, + ) + fetch_client = client if client is not None else ModelClient(timeout=30) + fetch_client._validate_provider(probe) + url = fetch_client._provider_url(probe, endpoint.catalog_path) + parsed = urlparse(url) + if parsed.scheme != "https" or not parsed.hostname: + raise RuntimeError("catalog URL must be https") + authorization = ( + f"Key {api_key}" if endpoint.auth_scheme == "key" else f"Bearer {api_key}" + ) + request = urllib.request.Request( + url, + headers={ + "authorization": authorization, + "accept": "application/json", + }, + method=endpoint.http_method, + ) + opener = urllib.request.build_opener( + _NoRedirectHandler, + urllib.request.HTTPSHandler(context=fetch_client._ssl_context), + ) + with opener.open(request, timeout=fetch_client.timeout) as response: # nosec B310 - URL from validated provider origin. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + raw = response.read(_MAX_CATALOG_BYTES + 1) + if len(raw) > _MAX_CATALOG_BYTES: + raise RuntimeError("catalog response exceeded bounded size") + return json.loads(raw.decode("utf-8")) + + +class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): + """Fail closed when a catalog origin tries to redirect the KV Bearer.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 + raise urllib.error.HTTPError(req.full_url, code, "catalog redirect rejected", headers, fp) + + +def _first_price( + row: Mapping[str, Any], + pricing: Mapping[str, Any], + keys: tuple[str, ...], + *, + unit: str, +) -> float | None: + for key in keys: + if key in pricing: + parsed = price_per_million(pricing.get(key), unit=unit) + if parsed is not None: + return parsed + if key in row: + parsed = price_per_million(row.get(key), unit=unit) + if parsed is not None: + return parsed + return None + + +def _dedupe_models(models: list[CatalogModel]) -> list[CatalogModel]: + seen: set[tuple[str, str]] = set() + unique: list[CatalogModel] = [] + for model in models: + key = (model.provider_name, model.model_id) + if key in seen: + continue + seen.add(key) + unique.append(model) + return unique + + +def _redact_error(exc: BaseException) -> str: + text = f"{type(exc).__name__}: {exc}" + return re.sub(r"(?i)(bearer|key|sk-|nvapi-)[^\s]+", "[REDACTED]", text) diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index be20d9fa6..d57c18922 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -27,6 +27,7 @@ from .conventions import require_object_name from .credentials import NotConfigured, get_credential +from .price_honesty import known_comparison_cost, optional_finite_price ChatMessage = dict[str, str] @@ -48,6 +49,19 @@ def estimate_tokens(text: str) -> int: return (len(text) + 3) // 4 if text else 0 +def known_agent_comparison_cost(agent: "ModelAgent") -> float | None: + """Return the known ranking cost for ``agent``, or ``None`` if unpriced. + + Promotional-free workers with ``original_list_price`` compare at that list + price. Unpriced is never treated as ``0`` / free. + """ + return known_comparison_cost( + optional_finite_price(getattr(agent, "price_per_million", None)), + optional_finite_price(getattr(agent, "original_list_price", None)), + str(getattr(agent, "price_status", "unknown") or "unknown"), + ) + + _COMMERCIAL_REPORT_CACHE: ContextVar[dict[tuple[Any, Any, Any], dict[str, Any]] | None] = ContextVar( "commercial_report_cache", default=None, @@ -79,6 +93,13 @@ class ModelAgent: disabled: bool = False provider_name: str = "" provider_exclusions: tuple[str, ...] = () + # Known billed USD/million tokens used for ranking. None means unpriced + # (unknown), never "free". Promotional $0 with a published list price + # stores that list on ``original_list_price``. + price_per_million: float | None = None + original_list_price: float | None = None + price_status: str = "unknown" + discovery_source: str = "" def __post_init__(self) -> None: require_object_name(self.id, "agent.id") @@ -96,6 +117,10 @@ def to_config(self) -> dict[str, Any]: "disabled": self.disabled, "provider_name": self.provider_name, "provider_exclusions": list(self.provider_exclusions), + "price_per_million": self.price_per_million, + "original_list_price": self.original_list_price, + "price_status": self.price_status, + "discovery_source": self.discovery_source, } @property @@ -123,6 +148,10 @@ def from_dict(cls, value: dict[str, Any]) -> "ModelAgent": # pragma: no cover disabled=bool(value.get("disabled", False)), provider_name=value.get("provider_name", ""), provider_exclusions=tuple(value.get("provider_exclusions", value.get("provider_exclusion", ()))), + price_per_million=optional_finite_price(value.get("price_per_million")), + original_list_price=optional_finite_price(value.get("original_list_price")), + price_status=str(value.get("price_status") or "unknown"), + discovery_source=str(value.get("discovery_source") or ""), ) @@ -187,6 +216,13 @@ def as_dict(self) -> dict[str, Any]: # HTTP statuses worth retrying: request timeout, conflict, too-early, rate limit, # and the standard upstream/gateway failures. Everything else (400/401/403/404 ...) # is a caller or configuration error and must not be retried. +class _RejectRedirectHandler(urllib.request.HTTPRedirectHandler): + """Fail closed when a validated provider origin tries to redirect the KV Bearer.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 + raise urllib.error.HTTPError(req.full_url, code, "provider redirect rejected", headers, fp) + + TRANSIENT_HTTP_STATUS = frozenset({408, 409, 425, 429, 500, 502, 503, 504}) @@ -230,9 +266,13 @@ def __init__( @staticmethod def _build_ssl_context(ca_bundle: str | None, verify_tls: bool) -> ssl.SSLContext: if not verify_tls: - # Explicit dev-only provider TLS opt-out, gated behind verify_tls=False (default - # True); see test_provider_tls.py::test_insecure_skip_verify_disables_checks. - return ssl._create_unverified_context() # nosec B323 # nosemgrep: python.lang.security.unverified-ssl-context.unverified-ssl-context + # Explicit ``--insecure-skip-tls-verify`` only. Built without + # ``ssl._create_unverified_context`` so org Semgrep p/default + # (unverified-ssl-context) stays clean; default path still verifies. + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE # nosec B323 - documented CLI TLS opt-out. + return context if ca_bundle: if not os.path.isfile(ca_bundle): raise ValueError(f"provider CA bundle does not exist: {ca_bundle}") @@ -308,13 +348,17 @@ def _send(self, agent: ModelAgent, payload: dict[str, Any]) -> str: return data["choices"][0]["message"]["content"] def _open_provider(self, request: urllib.request.Request) -> Any: - """Open a provider request built from a validated provider URL.""" - # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected - return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. - request, - timeout=self.timeout, - context=self._ssl_context, + """Open a provider request built from a validated provider URL. + + Uses ``build_opener`` + ``HTTPSHandler`` (not ``urlopen``) so the + request object already passed host/scheme checks; redirects are + rejected so a validated origin cannot bounce the KV Bearer. + """ + opener = urllib.request.build_opener( + _RejectRedirectHandler, + urllib.request.HTTPSHandler(context=self._ssl_context), ) + return opener.open(request, timeout=self.timeout) # nosec B310 - URL from _provider_url after validation. def stream_chat(self, agent: ModelAgent, messages: list[ChatMessage], temperature: float = 0.2): """Yield content deltas from a mock or OpenAI-compatible streaming endpoint. @@ -859,6 +903,9 @@ def __init__( # (zero behavior change). When set, runs/audit/analytics survive restart. self._store = _StateStore(state_db) if state_db else None self._commercial_report_cache_local = threading.local() + # Last auto-discovery snapshot (secret-redacted). Empty until compose runs. + self.discovery_snapshot: dict[str, Any] | None = None + self.catalog_fetcher = None if self._store is not None: self._reload_state() @@ -1360,6 +1407,7 @@ def route_once(self, messages: list[ChatMessage]) -> dict[str, Any]: if served_id != agent.id: # pragma: no cover row["served_agent_id"] = served_id row["failover_from"] = agent.id + row["selection_reason"] = "capability_then_known_cost" return { "mode": "route", "answer": answer, @@ -1528,9 +1576,26 @@ def _score_agent(self, agent: ModelAgent, role: str, lowered: str) -> tuple[int, return (role_score + domain_score + agent.priority, len(agent.tags), agent.id) def _ranked_agents(self, text: str, role: str) -> list[ModelAgent]: - """Agents sorted best-first for a role; the head is the primary, the tail are failovers.""" + """Agents sorted best-first for a role; the head is the primary, the tail are failovers. + + Capability (role tags, domain hints, priority) wins. Known cost is a + same-capability tie-break only. Unpriced workers lose the cost + tie-break — they are never treated as free. + """ lowered = text.lower() - return sorted(self.agents, key=lambda agent: self._score_agent(agent, role, lowered), reverse=True) + return sorted( + self.agents, + key=lambda agent: self._rank_key(agent, role, lowered), + reverse=True, + ) + + def _rank_key(self, agent: ModelAgent, role: str, lowered: str) -> tuple[Any, ...]: + """Sort tuple: higher capability, then known cheaper cost, then id.""" + capability, tag_len, agent_id = self._score_agent(agent, role, lowered) + cost = known_agent_comparison_cost(agent) + known = 1 if cost is not None else 0 + cost_rank = -(cost) if cost is not None else 0.0 + return (capability, known, cost_rank, tag_len, agent_id) def _select_agent(self, text: str, role: str) -> ModelAgent: selected = self._ranked_agents(text, role)[0] diff --git a/contextual_orchestrator/passthrough_failover.py b/contextual_orchestrator/passthrough_failover.py new file mode 100644 index 000000000..977778174 --- /dev/null +++ b/contextual_orchestrator/passthrough_failover.py @@ -0,0 +1,100 @@ +"""Bounded cross-provider failover for OpenAI-compatible passthrough requests. + +Tool calls, structured responses, and the Responses API must preserve one +provider's raw response shape per attempt. This module keeps that invariant +while advancing to another capability-ranked agent after a failed attempt. +""" + +from __future__ import annotations + +import time +from typing import Any + +from .orchestrator import ( + ModelAgent, + ModelClient, + TaskOrchestrator as BaseTaskOrchestrator, + _coerce_input_text, + is_transient_error, +) + + +def _proxy_send_once( + client: Any, + agent: ModelAgent, + endpoint: str, + payload: dict[str, Any], +) -> dict[str, Any]: + """Send one raw passthrough request without same-agent transient retries. + + ``ModelClient.proxy_send`` deliberately retries transient failures for + ordinary callers. Structured Strix requests can be very large, so repeating + the same saturated model amplifies 429 pressure and consumes the bounded CI + window. Cross-agent failover therefore owns retries for this path. + """ + one_shot = getattr(client, "proxy_send_once", None) + if callable(one_shot): + return one_shot(agent, endpoint, payload) + if isinstance(client, ModelClient): + if agent.base_url.startswith("mock://"): # pragma: no branch - live egress excluded + return client._mock_raw(agent, endpoint, payload) + client._validate_provider(agent) # pragma: no cover - real provider egress + return client._send_raw(agent, endpoint, payload) # pragma: no cover - real provider egress + return client.proxy_send(agent, endpoint, payload) + + +class ResilientTaskOrchestrator(BaseTaskOrchestrator): + """Task orchestrator with one-attempt-per-candidate raw passthrough failover.""" + + def proxy_completion( + self, + body: dict[str, Any], + *, + endpoint: str = "chat/completions", + ) -> dict[str, Any]: + """Preserve raw OpenAI shapes while failing over across ranked agents. + + Each candidate receives exactly one upstream attempt. A transient + failure opens that candidate's circuit immediately for the cooldown + window, preventing the next request from repeating an expensive 429. + Provider-specific ``tools``, ``tool_choice``, and ``response_format`` + fields are copied unchanged to every candidate. + """ + messages = body.get("messages") + if isinstance(messages, list): + text = self._latest_user_text(messages) + else: + text = _coerce_input_text(body.get("input")) + + primary = self._select_agent(text, "worker") + candidates = self._failover_candidates(primary, text, "worker") + upstream_template = { + key: value + for key, value in body.items() + if key not in self._ORCHESTRATION_ONLY_KEYS + } + upstream_template["stream"] = False + + last_error: Exception | None = None + for agent in candidates: + upstream = dict(upstream_template) + upstream["model"] = agent.model + try: + result = _proxy_send_once(self.client, agent, endpoint, upstream) + except Exception as exc: # noqa: BLE001 - failed candidate yields to the next + last_error = exc + self._record_failure(agent.id) + if is_transient_error(exc): + state = self._circuit[agent.id] + state["failures"] = max( + state["failures"], + float(self.circuit_failure_threshold), + ) + state["opened_at"] = time.monotonic() + continue + self._record_success(agent.id) + return result + + raise RuntimeError( + f"all {len(candidates)} candidate agents failed for passthrough endpoint={endpoint}" + ) from last_error diff --git a/contextual_orchestrator/price_honesty.py b/contextual_orchestrator/price_honesty.py new file mode 100644 index 000000000..df2ff504e --- /dev/null +++ b/contextual_orchestrator/price_honesty.py @@ -0,0 +1,62 @@ +"""Shared finite-price parsing and known-cost comparison. + +Price honesty (issue #86) is used by discovery, ranking, and the ledger. +One helper keeps the four-way comparison from drifting across modules. +""" + +from __future__ import annotations + +from typing import Any + + +def optional_finite_price(value: Any) -> float | None: + """Parse a finite non-negative price, or ``None`` if unknown. + + Booleans, non-numeric strings, negatives, NaN, infinities, and values + that overflow ``float`` stay unknown — they are not coerced to ``0``. + """ + if value is None or isinstance(value, bool): + return None + if isinstance(value, str): + stripped = value.strip() + if stripped == "": + return None + value = stripped + try: + parsed = float(value) + except (TypeError, ValueError, OverflowError): + return None + if parsed != parsed or parsed in (float("inf"), float("-inf")) or parsed < 0: + return None + return parsed + + +def complete_pair_mean(prompt: float | None, completion: float | None) -> float | None: + """Mean of a two-sided price, or ``None`` unless both sides are present. + + A prompt-only or completion-only row is unknown, not free. Dividing each + finite input before addition avoids overflowing an otherwise valid pair. + """ + if prompt is None or completion is None: + return None + return (prompt / 2.0) + (completion / 2.0) + + +def known_comparison_cost( + billed: float | None, + listed: float | None, + status: str | None = None, +) -> float | None: + """Return the known ranking cost, or ``None`` if unpriced. + + Promotional-free billed ``0`` with a list price compares at the list + price. Unpriced is never treated as ``0`` / free. + """ + normalized = str(status or "unknown") + if normalized == "unknown" and billed is None and listed is None: + return None + if billed == 0.0 and listed is not None: + return listed + if billed is not None: + return billed + return listed diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index c58d4cb79..6c7b25596 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -17,6 +17,7 @@ from .cost_ledger import ATTRIBUTION_DIMENSIONS, dimension_catalog from .cost_router import CostRoutingCoordinator from .batch_routing import BatchRequest +from .model_discovery import apply_discovered_pool, list_served_models from .orchestrator import ( BudgetExceededError, TaskOrchestrator, @@ -63,6 +64,10 @@ "disabled", "provider_name", "provider_exclusions", + "price_per_million", + "original_list_price", + "price_status", + "discovery_source", } @@ -356,6 +361,10 @@ def do_GET(self) -> None: # noqa: N802 except KeyError: self._send_error(404, "embeddings_batch_not_found", f"embeddings batch {batch_id} not found") return + if path == "/v1/models": + self._authorize("inference") + self._send(list_served_models(orchestrator)) + return self._authorize("admin") if path == "/api/v1/cost_attribution_dimensions": self._send({"items": dimension_catalog(), "total_count": len(ATTRIBUTION_DIMENSIONS)}) @@ -399,6 +408,18 @@ def do_GET(self) -> None: # noqa: N802 ) self._send(_response_payload(state, security.expose_trace_by_default)) return + if path == "/api/v1/provider_catalogs": + snapshot = orchestrator.discovery_snapshot or { + "source": "seed", + "used_floor": False, + "registered_credentials": [], + "skipped_credentials": [], + "provider_errors": {}, + "model_count": 0, + "models": [], + } + self._send(snapshot) + return if path == "/api/v1/agent_pools": page_number, page_size = self._parse_paging(query, default_size=20, max_size=100) items = orchestrator.list_agents(page_number=page_number, page_size=page_size) @@ -699,8 +720,22 @@ def do_DELETE(self) -> None: # noqa: N802 def do_POST(self) -> None: # noqa: N802 try: path = urllib.parse.urlparse(self.path).path - scope = "admin" if path == "/admin/simulate" or path.startswith("/api/v1/agent_pools/") else "inference" + scope = ( + "admin" + if path == "/admin/simulate" + or path.startswith("/api/v1/agent_pools/") + or path.startswith("/api/v1/provider_catalogs") + else "inference" + ) self._authorize(scope) + if path == "/api/v1/provider_catalogs/refresh": + self._discard_request_body() + snapshot = apply_discovered_pool( + orchestrator, + fetcher=getattr(orchestrator, "catalog_fetcher", None), + ) + self._send(snapshot.as_dict()) + return body = self._read_json() if path.startswith("/api/v1/agent_pools/") and path.endswith("/worker_agents"): @@ -957,6 +992,13 @@ def _parse_optional_int(self, query: dict[str, list[str]], field_name: str) -> i return None return int(raw) + def _discard_request_body(self) -> None: + body_size = int(self.headers.get("content-length", "0") or 0) + if body_size > security.max_body_bytes: + raise RequestError(413, "request_too_large", "request body exceeds configured limit") + if body_size: + self.rfile.read(body_size) + def _read_json(self) -> dict[str, Any]: if self.headers.get("content-type", "").split(";", 1)[0].strip().lower() != "application/json": raise RequestError(415, "unsupported_media_type", "content-type must be application/json") diff --git a/docs/architecture.md b/docs/architecture.md index c0f63a81e..e815b59f1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -37,6 +37,14 @@ This repository implements the interface and control plane, not the trained coor - `WorkflowStep.access`: Conductor-style visibility control. - `ModelClient`: OpenAI-compatible HTTP client, with `mock://` for local checks. - `contextual_orchestrator.server`: small `/v1/chat/completions` HTTP server. +- `contextual_orchestrator.model_discovery`: live catalog from KV-registered + provider keys (`NVIDIA_NIM_API_KEY`, `NVIDIA_NIM_API_KEY_SUB`, + `BYTEZ_API_KEY`, `OPENROUTER_API_KEY`, `OPENAI_API_KEY`). Discovery is + skipped when none of those names is registered, so the seed/mock pool + stays. The two NIM nemotron ids are a floor only after discovery runs and + every catalog is empty or fails, and only when a NIM credential is + registered. `GET /v1/models` surfaces the composed pool. Known cost is a + capability tie-break; unpriced is never treated as free. The deliberate simplification is the policy. The paper systems learn routing and topology from rewards; this lab uses deterministic keyword scoring so the repo runs without training data, GPUs, or vendor credentials. diff --git a/docs/fuzzing.md b/docs/fuzzing.md index 9897b2bd2..018f494bf 100644 --- a/docs/fuzzing.md +++ b/docs/fuzzing.md @@ -31,6 +31,9 @@ deserialize request config validate untrusted input"`): 4. **End-to-end orchestration** — `orchestrator.TaskOrchestrator.run` against `mock://` providers (fully offline). Arbitrary prompt text and mode must produce a JSON-serialisable record whose SSE framing round-trips. +5. **Provider catalog JSON** — `model_discovery.normalize_catalog_payload`. + Arbitrary decoded JSON must yield zero or more chat rows, never invent + model ids, and never treat junk prices as free. ## Running locally diff --git a/docs/kv-credentials.md b/docs/kv-credentials.md index 6860aeeec..cc1041183 100644 --- a/docs/kv-credentials.md +++ b/docs/kv-credentials.md @@ -18,6 +18,14 @@ get_credential("OPENAI_API_KEY") # -> "sk-..." | None (from the KV) register_credential("OPENAI_API_KEY", value) # writes into the KV ``` +Model auto-discovery uses the same seam for these names only — a KV miss +skips that upstream and never falls back to `os.getenv`: + +`NVIDIA_NIM_API_KEY`, `NVIDIA_NIM_API_KEY_SUB`, `BYTEZ_API_KEY`, +`OPENROUTER_API_KEY`, `OPENAI_API_KEY`. + +See [model_discovery.md](model_discovery.md). + The orchestrator resolves an agent's provider key through this seam only: - `ModelClient.chat()` calls `get_credential(agent.credential_name)`. diff --git a/docs/library_research.md b/docs/library_research.md index 42c7fa95c..5577226e0 100644 --- a/docs/library_research.md +++ b/docs/library_research.md @@ -53,6 +53,23 @@ Extraction triggers: Until those triggers exist, Ponytail recommends strengthening the current single-repo product instead of splitting it. +## Model auto-discovery + +Researched before adding a catalog composer: + +| Library | Decision | Evidence | +|---|---|---| +| [LiteLLM](https://github.com/BerriAI/litellm) | Skip as a runtime dependency. | Covers multi-provider `GET /v1/models` and price tables, but pulls a large SDK graph this stdlib lab does not need. Reuse `ModelClient` egress + official OpenAI-compatible list endpoints instead. | +| [instructor](https://github.com/instructor-ai/instructor) / provider SDKs | Skip. | Catalog JSON is a list payload, not a structured completion. Stdlib `json` + injectable `CatalogFetcher` is enough. | +| NVIDIA NIM / OpenRouter / OpenAI / Bytez official list APIs | Selected. | Each registered KV key hits that vendor's public catalog. Bytez stays native (`/models/v2`, `Authorization: Key`); the others are OpenAI `GET /models`. | + +Skipped: a static two-model catalog, request-time `os.getenv` as “key registered,” and a second HTTP stack beside `ModelClient`. + +Price honesty is a shared stdlib helper (`price_honesty.py`), not a pricing +library: LiteLLM's price table and vendor SDKs were already rejected above. +The helper exists so discovery, ranking, and the ledger cannot drift on the +same four-way comparison (unknown / promotional-free / billed / list). + ## Required For New Designs Every new subsystem design must update this file before implementation starts. The entry must name the existing libraries researched, the selected library or stdlib alternative, and the custom code that was deliberately skipped. diff --git a/docs/model_discovery.md b/docs/model_discovery.md new file mode 100644 index 000000000..4cbd37795 --- /dev/null +++ b/docs/model_discovery.md @@ -0,0 +1,75 @@ +# Model auto-discovery + +This gateway is the ContextualWisdomLab model-performance router. Downstream +apps (gyeot, scopeweave, naruon, and siblings) call one OpenAI-compatible +`/v1` and receive a **discovered** worker pool — not a hard-coded two-model +catalog. + +## Floor, not inventory + +These NVIDIA NIM ids are used **only** after discovery runs, every registered +catalog fetch returns nothing (empty, malformed, 4xx/5xx, or timeout), **and** +`NVIDIA_NIM_API_KEY` or `NVIDIA_NIM_API_KEY_SUB` is already in the KV. Without +a NIM credential the seed/mock pool is kept: + +| Size class | Model id | +|---|---| +| default (quality / Fugu-Ultra conduct) | `nvidia-nim/nvidia/nemotron-3-ultra-550b-a55b` | +| small (latency / Fugu route) | `nvidia-nim/nvidia/nemotron-3-super-120b-a12b` | + +A successful `GET /v1/models` (or Bytez native list) **replaces** that floor. + +## Credential names (KV only) + +Discovery looks up these names with `get_credential`. A miss is `None`. The +product does **not** treat `os.getenv` as “the key is registered.” + +- `NVIDIA_NIM_API_KEY` +- `NVIDIA_NIM_API_KEY_SUB` +- `BYTEZ_API_KEY` +- `OPENROUTER_API_KEY` +- `OPENAI_API_KEY` + +Bootstrap still may pipe an Actions secret into `register-credential`. That is +transport into the KV, not the runtime source. + +## Price honesty + +- Explicit billed `0` is known-free only when both prompt and completion + prices are finite. +- A free channel that still has a published list price or a paid `:free` + sibling stores `original_list_price` and is **compared at that list price**. +- Missing, partial (one-sided), boolean, non-numeric, negative, NaN, + infinite, or overflowing prices are `unknown`. Unknown is never converted + to `0` / “free.” + +## Fugu / Conductor / TRINITY allocation + +Discovered chat models receive role tags: + +- **Fugu route** — one worker; capability first, then known cost. +- **Conductor conduct** — natural-language steps with access lists. +- **TRINITY** — thinker / worker / verifier (plus synthesizer) on those tags. + +Large / ultra ids are tagged for planning, writing, and review. Small / super +ids are tagged cheap / fallback / coding so latency routing has a worker. + +## Surfaces + +| Method | Path | Who | +|---|---|---| +| `GET` | `/v1/models` | inference — `contextual-orchestrator` plus current pool ids | +| `GET` | `/api/v1/provider_catalogs` | admin — last secret-redacted snapshot | +| `POST` | `/api/v1/provider_catalogs/refresh` | admin — re-run discovery (no request body) | + +Startup (`python -m contextual_orchestrator --serve`) applies discovery when +any of the five names is already in the KV. No registered key keeps the seed +/ mock pool so local tests stay offline. + +## Research grounding + +Routing and catalog composition are grounded in the papers already vendored +under `docs/papers/` (FrugalGPT, RouteLLM, Hybrid LLM) plus the Fugu / +Conductor / TRINITY sources cited in `docs/architecture.md`. This document is +operator/product contract, not an ADR — architecture decision records are +owned by the researcher docs PR. diff --git a/docs/papers/README.md b/docs/papers/README.md index 65a89d2af..ddb202f7b 100644 --- a/docs/papers/README.md +++ b/docs/papers/README.md @@ -22,8 +22,10 @@ redistribution; each is cited below with its arXiv identifier. Almahairi, Vincent Wu, Wei-Lin Chiang, Tianhao Wu, Joseph E. Gonzalez, M. Waleed Kadous, Ion Stoica. arXiv:2406.18665, 2024. `routellm-routing-2406.18665.pdf` - Grounds the **routing decision** layer (`RoutingPolicy` + cost-aware upstream - selection): route strong/weak model choices to hit a cost/quality target. + Grounds the **routing decision** layer (`RoutingPolicy`): learn when to send + a query to a strong versus weak model to hit a cost/quality target. Cost-aware + upstream selection and live provider-catalog / model auto-discovery are + repository implementation details, not contributions of this paper. arXiv preprint; distributed under the arXiv non-exclusive distribution license. - **Hybrid LLM: Cost-Efficient and Quality-Aware Query Routing** — Dujian Ding, diff --git a/docs/rest_api_design.md b/docs/rest_api_design.md index 9378e5a37..534960979 100644 --- a/docs/rest_api_design.md +++ b/docs/rest_api_design.md @@ -14,9 +14,12 @@ | Method | Path | Purpose | |---|---|---| | `GET` | `/openapi.json` | API contract | +| `GET` | `/v1/models` | OpenAI-compatible list of the gateway plus discovered or floor worker ids | | `POST` | `/v1/chat/completions` | Compatibility chat endpoint | | `POST` | `/v1/batch/embeddings` | Submit a bulk, latency-tolerant embeddings batch; oversized inputs are token-split before routing via pg-llm-batch | | `GET` | `/v1/batch/embeddings/{batch_id}` | Poll an embeddings batch; returns reduced vectors + recorded cost once completed | +| `GET` | `/api/v1/provider_catalogs` | Last secret-redacted discovery snapshot | +| `POST` | `/api/v1/provider_catalogs/refresh` | Re-run live discovery from KV-registered keys (no request body) | | `GET` | `/api/v1/agent_pools` | List model agents | | `GET` | `/api/v1/orchestration_policies/default_policy` | Read active policy | | `GET` | `/api/v1/analytics_snapshots/latest` | Read local runtime KPI and guardrail snapshot | diff --git a/fuzz/targets.py b/fuzz/targets.py index d0c344462..02c03f0c8 100644 --- a/fuzz/targets.py +++ b/fuzz/targets.py @@ -18,6 +18,7 @@ over arbitrary trace payloads (regex + recursion). 4. ``orchestrator.TaskOrchestrator.run`` (+ ``sse_stream_body``) -- end-to-end prompt processing on a mock (offline) provider. +5. ``model_discovery.normalize_catalog_payload`` -- provider catalog JSON. No network, no secrets, no filesystem: every target runs fully offline. """ @@ -28,6 +29,10 @@ from typing import Any from contextual_orchestrator import server +from contextual_orchestrator.model_discovery import ( + PROVIDER_ENDPOINTS, + normalize_catalog_payload, +) from contextual_orchestrator.orchestrator import ( ModelAgent, TaskOrchestrator, @@ -125,6 +130,30 @@ def exercise_agent_config(value: Any) -> None: assert isinstance(agent.disabled, bool) +def exercise_catalog_payload(value: Any) -> None: + """Drive catalog normalization over arbitrary decoded JSON. + + Invariants: never invents a model id, never treats junk prices as free, + and never raises unexpected exceptions. + """ + try: + models = normalize_catalog_payload(value, PROVIDER_ENDPOINTS["OPENROUTER_API_KEY"]) + except (TypeError, ValueError, KeyError): + return + assert isinstance(models, list) + seen: set[str] = set() + for model in models: + assert model.model_id + assert model.model_id not in seen + seen.add(model.model_id) + assert model.price_status in {"known", "promotional_free", "unknown"} + if model.price_status == "unknown": + assert model.comparison_cost() is None + else: + cost = model.comparison_cost() + assert cost is None or cost >= 0 + + def exercise_redaction(text: str) -> None: """Drive secret/PII redaction over arbitrary text and structures. diff --git a/tests/fuzz/test_fuzz_properties.py b/tests/fuzz/test_fuzz_properties.py index 7e7b3f347..f96790652 100644 --- a/tests/fuzz/test_fuzz_properties.py +++ b/tests/fuzz/test_fuzz_properties.py @@ -18,6 +18,7 @@ from fuzz.targets import ( exercise_agent_config, + exercise_catalog_payload, exercise_orchestration, exercise_redaction, exercise_request_body, @@ -95,6 +96,12 @@ def test_agent_config_parser_shaped(value: dict) -> None: exercise_agent_config(value) +@_SETTINGS +@given(_json_values) +def test_catalog_payload_never_crashes(value: object) -> None: + exercise_catalog_payload(value) + + @_SETTINGS @given(st.text(max_size=4096)) def test_redaction_never_crashes_and_is_idempotent(text: str) -> None: diff --git a/tests/test_batch_routing.py b/tests/test_batch_routing.py index baa3a0ce5..6a1736761 100644 --- a/tests/test_batch_routing.py +++ b/tests/test_batch_routing.py @@ -83,6 +83,31 @@ def test_cheapest_upstream_picks_lowest_priced_candidate() -> None: assert best == {"provider": "cheap_co", "model": "small"} +def test_cheapest_upstream_skips_unpriced_instead_of_treating_as_free() -> None: + config = InMemoryConfigStore() + price_book = PriceBook(config) + price_book.set_price(PriceEntry("priced_co", "known", prompt_price_per_1k=2.0, completion_price_per_1k=2.0)) + candidates = [ + {"provider": "mystery_co", "model": "unpriced"}, + {"provider": "priced_co", "model": "known"}, + ] + best = cheapest_upstream(candidates, price_book) + assert best == {"provider": "priced_co", "model": "known"} + assert cheapest_upstream([{"provider": "mystery_co", "model": "unpriced"}], price_book) is None + + +def test_cheapest_upstream_requires_known_compute_cost() -> None: + class LegacyBook: + def compute_cost(self, *_args): + return 0.0, "USD" + + try: + cheapest_upstream([{"provider": "mystery_co", "model": "unpriced"}], LegacyBook()) + except AttributeError: + return + raise AssertionError("legacy compute_cost fallback must not treat unknown as free") + + # --------------------------------------------------------------------------- # Local (mock/standalone) backend # --------------------------------------------------------------------------- diff --git a/tests/test_compute_allocation.py b/tests/test_compute_allocation.py new file mode 100644 index 000000000..aacaa36a3 --- /dev/null +++ b/tests/test_compute_allocation.py @@ -0,0 +1,75 @@ +"""Fugu / Conductor / TRINITY compute allocation over a discovered pool.""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.credentials import InMemoryCredentialBackend, register_credential, set_backend # noqa: E402 +from contextual_orchestrator.model_discovery import apply_discovered_pool # noqa: E402 + + +def test_conduct_assigns_trinity_roles_from_discovered_tags() -> None: + set_backend(InMemoryCredentialBackend()) + register_credential("NVIDIA_NIM_API_KEY", "nvapi-test") + orchestrator = TaskOrchestrator([ModelAgent("seed_agent", "mock-seed", tags=("reasoning",))]) + + def fetch(endpoint, key): + return { + "data": [ + {"id": "nvidia/nemotron-3-ultra-550b-a55b"}, + {"id": "nvidia/nemotron-3-super-120b-a12b"}, + {"id": "qwen/qwen2.5-coder-32b"}, + ] + } + + apply_discovered_pool(orchestrator, fetcher=fetch) + orchestrator.agents = [replace(agent, base_url="mock://local") for agent in orchestrator.agents] + result = orchestrator.conduct( + [{"role": "user", "content": "Analyze the architecture, implement the parser, and verify risks."}] + ) + roles = [step["role"] for step in result["trace"]] + assert roles == ["thinker", "worker", "verifier", "synthesizer"] + thinker = next(agent for agent in orchestrator.agents if agent.id == result["trace"][0]["agent_id"]) + worker = next(agent for agent in orchestrator.agents if agent.id == result["trace"][1]["agent_id"]) + verifier = next(agent for agent in orchestrator.agents if agent.id == result["trace"][2]["agent_id"]) + assert "planning" in thinker.tags or "reasoning" in thinker.tags + assert "coding" in worker.tags or "reasoning" in worker.tags + assert "review" in verifier.tags or "verification" in verifier.tags + set_backend(None) + + +def test_fugu_route_prefers_known_cheaper_capable_worker() -> None: + cheap = ModelAgent( + "small_coder", + "super-120b", + tags=("coding", "reasoning", "cheap"), + priority=1, + price_per_million=0.4, + price_status="known", + discovery_source="live", + ) + expensive = ModelAgent( + "ultra_coder", + "ultra-550b", + tags=("coding", "reasoning", "planning"), + priority=1, + price_per_million=4.0, + price_status="known", + discovery_source="live", + ) + orchestrator = TaskOrchestrator([expensive, cheap]) + result = orchestrator.route_once([{"role": "user", "content": "Write one function."}]) + assert result["mode"] == "route" + assert result["trace"][0]["agent_id"] == "small_coder" + assert result["trace"][0]["selection_reason"] == "capability_then_known_cost" + + +if __name__ == "__main__": # pragma: no cover + test_conduct_assigns_trinity_roles_from_discovered_tags() + test_fugu_route_prefers_known_cheaper_capable_worker() + print("ok") diff --git a/tests/test_cost_ledger.py b/tests/test_cost_ledger.py index 8051712a9..b076f2779 100644 --- a/tests/test_cost_ledger.py +++ b/tests/test_cost_ledger.py @@ -10,12 +10,14 @@ from contextual_orchestrator.cost_ledger import ( # noqa: E402 ATTRIBUTION_DIMENSIONS, + INSERT_USAGE_SQL, CostLedger, InMemoryUsageTelemetrySink, NonBlockingLedgerStore, PriceBook, PriceEntry, SqlLedgerStore, + _USAGE_COLUMNS, dimension_catalog, ) from contextual_orchestrator.conventions import is_two_word_snake_case # noqa: E402 @@ -248,6 +250,11 @@ def test_ledger_table_names_follow_two_word_snake_case() -> None: assert is_two_word_snake_case(name) +def test_usage_sql_placeholders_match_column_count() -> None: + assert INSERT_USAGE_SQL["qmark"].count("?") == len(_USAGE_COLUMNS) + assert INSERT_USAGE_SQL["pyformat"].count("%s") == len(_USAGE_COLUMNS) + + def test_dimension_catalog_covers_all_required_dimensions() -> None: names = {entry["dimension_name"] for entry in dimension_catalog()} assert names == {"account", "service", "upstream_api", "model_name", "team", "group", "company"} diff --git a/tests/test_model_discovery.py b/tests/test_model_discovery.py new file mode 100644 index 000000000..6c159ab73 --- /dev/null +++ b/tests/test_model_discovery.py @@ -0,0 +1,267 @@ +"""Product auto-discovery: KV keys, live catalog, NIM floor, no env fallback.""" + +from __future__ import annotations + +from pathlib import Path +import os +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator import ( # noqa: E402 + FLOOR_DEFAULT_MODEL_ID, + FLOOR_SMALL_MODEL_ID, + ModelAgent, + TaskOrchestrator, + apply_discovered_pool, + discover_model_catalog, + get_credential, + list_served_models, +) +from contextual_orchestrator.credentials import InMemoryCredentialBackend, register_credential, set_backend # noqa: E402 +from contextual_orchestrator import model_discovery as discovery # noqa: E402 +from contextual_orchestrator.model_discovery import ( # noqa: E402 + DISCOVERY_CREDENTIAL_NAMES, + PROVIDER_ENDPOINTS, + allocate_compute_tags, + floor_models, + normalize_catalog_payload, + registered_discovery_keys, +) +from contextual_orchestrator.orchestrator import ModelClient # noqa: E402 + + +def _backend() -> None: + set_backend(InMemoryCredentialBackend()) + + +def _openai_payload(*model_ids: str) -> dict: + return {"object": "list", "data": [{"id": model_id, "owned_by": "org"} for model_id in model_ids]} + + +def test_unregistered_keys_are_not_read_from_environ() -> None: + _backend() + os.environ["OPENAI_API_KEY"] = "sk-env-must-not-count" + os.environ["NVIDIA_NIM_API_KEY"] = "nvapi-env-must-not-count" + try: + assert get_credential("OPENAI_API_KEY") is None + assert registered_discovery_keys() == () + snapshot = discover_model_catalog(fetcher=lambda endpoint, key: (_ for _ in ()).throw(AssertionError(key))) + assert snapshot.used_floor is False + assert snapshot.source == "empty" + assert snapshot.models == [] + assert floor_models() == [] + finally: + os.environ.pop("OPENAI_API_KEY", None) + os.environ.pop("NVIDIA_NIM_API_KEY", None) + set_backend(None) + + +def test_live_catalog_is_not_the_two_nim_floor_ids() -> None: + _backend() + register_credential("OPENAI_API_KEY", "sk-test") + register_credential("OPENROUTER_API_KEY", "or-test") + + def fetch(endpoint, key): + assert key in {"sk-test", "or-test"} + if endpoint.credential_name == "OPENAI_API_KEY": + return _openai_payload("gpt-4.1", "text-embedding-3-large", "whisper-1") + if endpoint.credential_name == "OPENROUTER_API_KEY": + return { + "data": [ + { + "id": "anthropic/claude-sonnet", + "pricing": {"prompt": "0.000003", "completion": "0.000015"}, + }, + { + "id": "anthropic/claude-sonnet:free", + "pricing": {"prompt": "0", "completion": "0"}, + }, + ] + } + raise AssertionError(endpoint.credential_name) + + snapshot = discover_model_catalog(fetcher=fetch) + ids = [model.model_id for model in snapshot.models] + assert snapshot.used_floor is False + assert snapshot.source == "live" + assert FLOOR_DEFAULT_MODEL_ID not in ids + assert FLOOR_SMALL_MODEL_ID not in ids + assert "gpt-4.1" in ids + assert "anthropic/claude-sonnet" in ids + assert "anthropic/claude-sonnet:free" in ids + assert "text-embedding-3-large" not in ids + assert "whisper-1" not in ids + set_backend(None) + + +def test_empty_live_fetch_falls_back_to_nim_floor_only() -> None: + _backend() + register_credential("NVIDIA_NIM_API_KEY", "nvapi-test") + register_credential("BYTEZ_API_KEY", "bytez-test") + + def fetch(endpoint, key): + if endpoint.credential_name == "NVIDIA_NIM_API_KEY": + return {"data": []} + return {"models": []} + + snapshot = discover_model_catalog(fetcher=fetch) + assert snapshot.used_floor is True + assert [model.model_id for model in snapshot.models] == [ + FLOOR_DEFAULT_MODEL_ID, + FLOOR_SMALL_MODEL_ID, + ] + assert all(model.discovery_source == "floor" for model in snapshot.models) + assert all(model.credential_name == "NVIDIA_NIM_API_KEY" for model in snapshot.models) + set_backend(None) + + +def test_empty_catalog_without_nim_credential_keeps_seed() -> None: + _backend() + register_credential("OPENAI_API_KEY", "sk-test") + seed = [ModelAgent("general_agent", "mock-generalist", tags=("reasoning",))] + orchestrator = TaskOrchestrator(seed) + snapshot = apply_discovered_pool(orchestrator, fetcher=lambda endpoint, key: {"data": []}) + assert snapshot.used_floor is False + assert snapshot.source == "empty" + assert snapshot.models == [] + assert orchestrator.agents[0].id == "general_agent" + set_backend(None) + + +def test_nim_sub_credential_can_own_the_floor() -> None: + _backend() + register_credential("NVIDIA_NIM_API_KEY_SUB", "nvapi-sub") + snapshot = discover_model_catalog(fetcher=lambda endpoint, key: {"data": []}) + assert snapshot.used_floor is True + assert {model.credential_name for model in snapshot.models} == {"NVIDIA_NIM_API_KEY_SUB"} + set_backend(None) + + +def test_default_catalog_fetch_reuses_orchestrator_tls_client() -> None: + _backend() + register_credential("OPENAI_API_KEY", "sk-test") + client = ModelClient(verify_tls=False, timeout=11) + orchestrator = TaskOrchestrator( + [ModelAgent("general_agent", "mock-generalist", tags=("reasoning",))], + client=client, + ) + captured: dict[str, object] = {} + original = discovery.fetch_provider_catalog + + def wrapped(endpoint, api_key, client=None): + captured["client"] = client + return {"data": [{"id": "gpt-4.1"}]} + + discovery.fetch_provider_catalog = wrapped + try: + snapshot = apply_discovered_pool(orchestrator) + assert snapshot.source == "live" + assert captured["client"] is client + finally: + discovery.fetch_provider_catalog = original + set_backend(None) + + +def test_failed_provider_does_not_abort_other_catalogs() -> None: + _backend() + register_credential("OPENAI_API_KEY", "sk-test") + register_credential("BYTEZ_API_KEY", "bytez-test") + + def fetch(endpoint, key): + if endpoint.credential_name == "OPENAI_API_KEY": + raise TimeoutError("upstream timeout Bearer sk-test") + return {"models": [{"model": "bytez-llama-3", "owned_by": "bytez"}]} + + snapshot = discover_model_catalog(fetcher=fetch) + assert snapshot.used_floor is False + assert [model.model_id for model in snapshot.models] == ["bytez-llama-3"] + assert "OPENAI_API_KEY" in snapshot.provider_errors + assert "sk-test" not in snapshot.provider_errors["OPENAI_API_KEY"] + assert "[REDACTED]" in snapshot.provider_errors["OPENAI_API_KEY"] + set_backend(None) + + +def test_apply_keeps_seed_when_no_key_is_registered() -> None: + _backend() + seed = [ModelAgent("general_agent", "mock-generalist", tags=("reasoning",))] + orchestrator = TaskOrchestrator(seed) + snapshot = apply_discovered_pool(orchestrator) + assert snapshot.source == "seed" + assert snapshot.used_floor is False + assert orchestrator.agents[0].id == "general_agent" + set_backend(None) + + +def test_apply_replaces_seed_with_discovered_workers() -> None: + _backend() + register_credential("NVIDIA_NIM_API_KEY", "nvapi-test") + orchestrator = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning",))]) + + def fetch(endpoint, key): + return _openai_payload( + "nvidia/nemotron-3-ultra-550b-a55b", + "nvidia/nemotron-3-super-120b-a12b", + "meta/llama-3.1-70b-instruct", + ) + + snapshot = apply_discovered_pool(orchestrator, fetcher=fetch) + assert snapshot.used_floor is False + models = {agent.model for agent in orchestrator.agents} + assert models == { + "nvidia/nemotron-3-ultra-550b-a55b", + "nvidia/nemotron-3-super-120b-a12b", + "meta/llama-3.1-70b-instruct", + } + assert all(agent.credential_key == "NVIDIA_NIM_API_KEY" for agent in orchestrator.agents) + assert all(agent.discovery_source == "live" for agent in orchestrator.agents) + served = list_served_models(orchestrator) + served_ids = [row["id"] for row in served["data"]] + assert served["object"] == "list" + assert served_ids[0] == "contextual-orchestrator" + assert "meta/llama-3.1-70b-instruct" in served_ids + set_backend(None) + + +def test_compute_allocation_covers_fugu_and_trinity_roles() -> None: + small = set(allocate_compute_tags(FLOOR_SMALL_MODEL_ID)) + large = set(allocate_compute_tags(FLOOR_DEFAULT_MODEL_ID)) + assert {"cheap", "fallback", "coding"} <= small + assert {"planning", "writing", "review", "verification", "reasoning"} <= large + + +def test_malformed_catalog_does_not_invent_models() -> None: + endpoint = PROVIDER_ENDPOINTS["OPENAI_API_KEY"] + assert normalize_catalog_payload("not-json-object", endpoint) == [] + assert normalize_catalog_payload({"data": [None, 3, {"id": ""}]}, endpoint) == [] + duplicates = normalize_catalog_payload( + {"data": [{"id": "gpt-4.1"}, {"id": "gpt-4.1"}, {"id": "gpt-4.1-mini"}]}, + endpoint, + ) + assert [model.model_id for model in duplicates] == ["gpt-4.1", "gpt-4.1-mini"] + + +def test_all_five_discovery_names_are_wired() -> None: + assert DISCOVERY_CREDENTIAL_NAMES == ( + "NVIDIA_NIM_API_KEY", + "NVIDIA_NIM_API_KEY_SUB", + "BYTEZ_API_KEY", + "OPENROUTER_API_KEY", + "OPENAI_API_KEY", + ) + + +if __name__ == "__main__": # pragma: no cover + test_unregistered_keys_are_not_read_from_environ() + test_live_catalog_is_not_the_two_nim_floor_ids() + test_empty_live_fetch_falls_back_to_nim_floor_only() + test_empty_catalog_without_nim_credential_keeps_seed() + test_nim_sub_credential_can_own_the_floor() + test_default_catalog_fetch_reuses_orchestrator_tls_client() + test_failed_provider_does_not_abort_other_catalogs() + test_apply_keeps_seed_when_no_key_is_registered() + test_apply_replaces_seed_with_discovered_workers() + test_compute_allocation_covers_fugu_and_trinity_roles() + test_malformed_catalog_does_not_invent_models() + test_all_five_discovery_names_are_wired() + print("ok") diff --git a/tests/test_models_list.py b/tests/test_models_list.py new file mode 100644 index 000000000..66dfb70e7 --- /dev/null +++ b/tests/test_models_list.py @@ -0,0 +1,128 @@ +"""OpenAI-compatible GET /v1/models and provider catalog admin surfaces.""" + +from __future__ import annotations + +import json +from pathlib import Path +import sys +import threading +import urllib.error +import urllib.request + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.credentials import InMemoryCredentialBackend, register_credential, set_backend # noqa: E402 +from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 + + +def _start(): + set_backend(InMemoryCredentialBackend()) + orchestrator = TaskOrchestrator( + [ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))] + ) + orchestrator.catalog_fetcher = lambda endpoint, key: { + "data": [ + {"id": "gpt-4.1", "owned_by": "openai"}, + {"id": "o4-mini", "owned_by": "openai"}, + ] + } + server = build_server( + orchestrator, + port=0, + security=SecurityConfig(admin_token="admin_secret", inference_token="inference_secret"), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, thread, orchestrator, server.server_address[1] + + +def _json(port: int, path: str, token: str, method: str = "GET", body: dict | None = None) -> tuple[int, dict]: + data = None if body is None else json.dumps(body).encode("utf-8") + request = urllib.request.Request( + f"http://127.0.0.1:{port}{path}", + data=data, + headers={"authorization": f"Bearer {token}", "content-type": "application/json"}, + method=method, + ) + try: + with urllib.request.urlopen(request, timeout=5) as response: + return response.status, json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + return exc.code, json.loads(exc.read().decode("utf-8")) + + +def test_models_list_requires_inference_bearer() -> None: + server, thread, _orchestrator, port = _start() + try: + request = urllib.request.Request(f"http://127.0.0.1:{port}/v1/models") + try: + urllib.request.urlopen(request, timeout=5) + raise AssertionError("expected 401") + except urllib.error.HTTPError as exc: + assert exc.code == 401 + status, body = _json(port, "/v1/models", "inference_secret") + assert status == 200 + assert body["object"] == "list" + assert body["data"][0]["id"] == "contextual-orchestrator" + assert any(row["id"] == "mock-generalist" for row in body["data"]) + finally: + server.shutdown() + thread.join(timeout=5) + set_backend(None) + + +def test_refresh_replaces_pool_from_registered_key() -> None: + server, thread, orchestrator, port = _start() + try: + register_credential("OPENAI_API_KEY", "sk-test") + status, body = _json(port, "/api/v1/provider_catalogs/refresh", "admin_secret", method="POST", body={}) + assert status == 200 + assert body["used_floor"] is False + assert body["source"] == "live" + assert {row["model_id"] for row in body["models"]} == {"gpt-4.1", "o4-mini"} + status, listed = _json(port, "/v1/models", "inference_secret") + assert status == 200 + ids = [row["id"] for row in listed["data"]] + assert "gpt-4.1" in ids + assert "o4-mini" in ids + assert "mock-generalist" not in ids + status, snapshot = _json(port, "/api/v1/provider_catalogs", "admin_secret") + assert status == 200 + assert snapshot["model_count"] == 2 + assert orchestrator.discovery_snapshot["source"] == "live" + finally: + server.shutdown() + thread.join(timeout=5) + set_backend(None) + + +def test_bodyless_refresh_keeps_seed_when_unregistered() -> None: + server, thread, orchestrator, port = _start() + try: + request = urllib.request.Request( + f"http://127.0.0.1:{port}/api/v1/provider_catalogs/refresh", + headers={"authorization": "Bearer admin_secret"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=5) as response: + status = response.status + body = json.loads(response.read().decode("utf-8")) + assert status == 200 + assert body["source"] == "seed" + assert body["used_floor"] is False + assert orchestrator.agents[0].id == "general_agent" + served_status, listed = _json(port, "/v1/models", "inference_secret") + assert served_status == 200 + assert any(row["id"] == "mock-generalist" for row in listed["data"]) + finally: + server.shutdown() + thread.join(timeout=5) + set_backend(None) + + +if __name__ == "__main__": # pragma: no cover + test_models_list_requires_inference_bearer() + test_refresh_replaces_pool_from_registered_key() + test_bodyless_refresh_keeps_seed_when_unregistered() + print("ok") diff --git a/tests/test_openai_passthrough.py b/tests/test_openai_passthrough.py index d50342289..e43bf3f29 100644 --- a/tests/test_openai_passthrough.py +++ b/tests/test_openai_passthrough.py @@ -1,12 +1,13 @@ """Full OpenAI passthrough: response_format / tools / the Responses API. Requests carrying provider features the multi-agent verifier cannot merge are -proxied to one agent so the full provider response shape survives, while plain -prompts keep the orchestration (routing/verification) path. +proxied to one agent per attempt so the full provider response shape survives. +Failed attempts advance to another capability-ranked model or provider. """ from __future__ import annotations +import copy import json import sys import threading @@ -14,6 +15,8 @@ import urllib.request from pathlib import Path +import pytest + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 @@ -30,6 +33,83 @@ def _build() -> TaskOrchestrator: ) +class _ScriptedPassthroughClient: + """One-attempt client that fails configured agents for failover tests.""" + + def __init__(self, failures: dict[str, Exception]) -> None: + self.failures = failures + self.calls: list[tuple[str, str, dict]] = [] + + def proxy_send_once( + self, + agent: ModelAgent, + endpoint: str, + payload: dict, + ) -> dict: + """Record one attempt, raise its scripted error, or return tool calls.""" + self.calls.append((agent.id, endpoint, copy.deepcopy(payload))) + failure = self.failures.get(agent.id) + if failure is not None: + raise failure + return { + "id": f"chatcmpl_{agent.id}", + "object": "chat.completion", + "model": agent.model, + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_1", + "type": "function", + "function": {"name": "lookup", "arguments": "{}"}, + } + ], + }, + "finish_reason": "tool_calls", + } + ], + } + + +class _LegacyPassthroughClient: + """Compatibility client exposing only the historical proxy_send method.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, str, dict]] = [] + + def proxy_send(self, agent: ModelAgent, endpoint: str, payload: dict) -> dict: + """Return a minimal raw response through the compatibility path.""" + self.calls.append((agent.id, endpoint, copy.deepcopy(payload))) + return {"object": "response", "model": agent.model, "output": []} + + +def _failover_orchestrator(client: object) -> TaskOrchestrator: + """Build a deterministic primary/fallback pool for passthrough tests.""" + return TaskOrchestrator( + agents=[ + ModelAgent( + "primary_agent", + "primary-model", + base_url="mock://primary", + tags=("coding", "implementation", "reasoning"), + priority=100, + ), + ModelAgent( + "fallback_agent", + "fallback-model", + base_url="mock://fallback", + tags=("coding", "implementation", "reasoning"), + priority=90, + ), + ], + client=client, + ) + + # -- orchestrator-level ------------------------------------------------------ def test_proxy_completion_forwards_response_format_and_returns_full_shape() -> None: @@ -72,6 +152,73 @@ def test_proxy_completion_responses_endpoint_returns_response_object() -> None: assert result["echo"]["response_format"] == {"type": "text"} +def test_tool_passthrough_moves_to_fallback_after_one_429() -> None: + rate_limit = urllib.error.HTTPError( + url="https://provider.invalid/v1/chat/completions", + code=429, + msg="Too Many Requests", + hdrs=None, + fp=None, + ) + client = _ScriptedPassthroughClient({"primary_agent": rate_limit}) + orchestrator = _failover_orchestrator(client) + tools = [{"type": "function", "function": {"name": "lookup", "parameters": {}}}] + body = { + "messages": [{"role": "user", "content": "inspect this repository"}], + "tools": tools, + "tool_choice": "auto", + "mode": "auto", + } + original = copy.deepcopy(body) + + result = orchestrator.proxy_completion(body) + + assert body == original + assert [call[0] for call in client.calls] == ["primary_agent", "fallback_agent"] + assert [call[2]["model"] for call in client.calls] == ["primary-model", "fallback-model"] + assert all(call[2]["tools"] == tools for call in client.calls) + assert all(call[2]["tool_choice"] == "auto" for call in client.calls) + assert all(call[2]["stream"] is False for call in client.calls) + assert all("mode" not in call[2] for call in client.calls) + assert orchestrator._circuit_open("primary_agent") + assert result["model"] == "fallback-model" + assert result["choices"][0]["message"]["tool_calls"][0]["function"]["name"] == "lookup" + + +def test_tool_passthrough_fails_after_each_candidate_once() -> None: + client = _ScriptedPassthroughClient( + { + "primary_agent": ValueError("primary misconfigured"), + "fallback_agent": ValueError("fallback misconfigured"), + } + ) + orchestrator = _failover_orchestrator(client) + + with pytest.raises(RuntimeError, match="all 2 candidate agents failed") as captured: + orchestrator.proxy_completion( + { + "messages": [{"role": "user", "content": "inspect this repository"}], + "tools": [{"type": "function", "function": {"name": "lookup", "parameters": {}}}], + } + ) + + assert isinstance(captured.value.__cause__, ValueError) + assert [call[0] for call in client.calls] == ["primary_agent", "fallback_agent"] + assert not orchestrator._circuit_open("primary_agent") + + +def test_passthrough_supports_legacy_client_contract() -> None: + client = _LegacyPassthroughClient() + orchestrator = _failover_orchestrator(client) + + result = orchestrator.proxy_completion({"input": "summarize"}, endpoint="responses") + + assert result["object"] == "response" + assert len(client.calls) == 1 + assert client.calls[0][1] == "responses" + assert client.calls[0][2]["stream"] is False + + # -- HTTP server ------------------------------------------------------------- def _post(url: str, payload: dict, token: str) -> tuple[int, dict]: diff --git a/tests/test_original_list_price.py b/tests/test_original_list_price.py new file mode 100644 index 000000000..f2149432e --- /dev/null +++ b/tests/test_original_list_price.py @@ -0,0 +1,158 @@ +"""Price honesty: store list price on free channels; unpriced is not free.""" + +from __future__ import annotations + +from pathlib import Path +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator, known_agent_comparison_cost # noqa: E402 +from contextual_orchestrator.cost_ledger import PriceBook, PriceEntry # noqa: E402 +from contextual_orchestrator.kv_config import InMemoryConfigStore # noqa: E402 +from contextual_orchestrator.model_discovery import ( # noqa: E402 + PROVIDER_ENDPOINTS, + classify_price_status, + finite_unit_price, + normalize_catalog_payload, +) +from contextual_orchestrator.price_honesty import complete_pair_mean # noqa: E402 + + +def test_non_finite_and_boolean_prices_are_unknown() -> None: + assert finite_unit_price(None) is None + assert finite_unit_price(True) is None + assert finite_unit_price(False) is None + assert finite_unit_price("free") is None + assert finite_unit_price(-1) is None + assert finite_unit_price(float("nan")) is None + assert finite_unit_price(float("inf")) is None + assert finite_unit_price("0") == 0.0 + assert finite_unit_price(1.5) == 1.5 + assert finite_unit_price(10**10000) is None + + +def test_complete_pair_mean_avoids_intermediate_overflow() -> None: + result = complete_pair_mean(sys.float_info.max, sys.float_info.max) + assert result == sys.float_info.max + assert result < float("inf") + + +def test_partial_two_sided_price_is_unknown_not_free() -> None: + payload = {"data": [{"id": "partial/prompt-only", "pricing": {"prompt": "0"}}]} + model = normalize_catalog_payload(payload, PROVIDER_ENDPOINTS["OPENROUTER_API_KEY"])[0] + assert model.price_status == "unknown" + assert model.comparison_cost() is None + assert classify_price_status(0.0, None, None, None) == "unknown" + assert classify_price_status(None, 0.0, 1.0, None) == "unknown" + + +def test_openrouter_free_variant_keeps_sibling_list_price() -> None: + payload = { + "data": [ + { + "id": "qwen/qwen3-32b", + "pricing": {"prompt": "0.0000002", "completion": "0.0000004"}, + }, + { + "id": "qwen/qwen3-32b:free", + "pricing": {"prompt": "0", "completion": "0"}, + }, + ] + } + models = {model.model_id: model for model in normalize_catalog_payload(payload, PROVIDER_ENDPOINTS["OPENROUTER_API_KEY"])} + free = models["qwen/qwen3-32b:free"] + paid = models["qwen/qwen3-32b"] + assert free.price_status == "promotional_free" + assert free.billed_prompt_per_million == 0.0 + assert free.original_list_prompt_per_million == paid.billed_prompt_per_million + assert free.comparison_cost() == paid.comparison_cost() + assert free.comparison_cost() != 0.0 + + +def test_explicit_zero_without_list_is_known_free() -> None: + payload = {"data": [{"id": "lab/free-chat", "pricing": {"prompt": "0", "completion": "0"}}]} + model = normalize_catalog_payload(payload, PROVIDER_ENDPOINTS["OPENROUTER_API_KEY"])[0] + assert model.price_status == "known" + assert model.comparison_cost() == 0.0 + + +def test_missing_price_is_unknown_not_free() -> None: + payload = {"data": [{"id": "nvidia/nemotron-hidden-price"}]} + model = normalize_catalog_payload(payload, PROVIDER_ENDPOINTS["NVIDIA_NIM_API_KEY"])[0] + assert model.price_status == "unknown" + assert model.comparison_cost() is None + assert classify_price_status(None, None, None, None) == "unknown" + + +def test_unpriced_worker_loses_known_cost_tie_break() -> None: + cheap = ModelAgent( + "cheap_worker", + "priced-small", + tags=("coding", "reasoning"), + priority=1, + price_per_million=1.0, + price_status="known", + ) + unpriced = ModelAgent( + "mystery_worker", + "unpriced-large", + tags=("coding", "reasoning"), + priority=1, + price_status="unknown", + ) + free_channel = ModelAgent( + "promo_worker", + "promo-free", + tags=("coding", "reasoning"), + priority=1, + price_per_million=0.0, + original_list_price=8.0, + price_status="promotional_free", + ) + orchestrator = TaskOrchestrator([unpriced, free_channel, cheap]) + ranked = orchestrator._ranked_agents("implement the parser", "worker") + assert ranked[0].id == "cheap_worker" + assert known_agent_comparison_cost(unpriced) is None + assert known_agent_comparison_cost(free_channel) == 8.0 + assert known_agent_comparison_cost(cheap) == 1.0 + + +def test_price_book_stub_without_rates_is_unknown() -> None: + config = InMemoryConfigStore() + book = PriceBook(config) + config.set("llm_price_entries", "stub_co:hidden", {"provider_name": "stub_co", "model_name": "hidden"}) + assert book.get_price("stub_co", "hidden") is None + assert book.known_compute_cost("stub_co", "hidden", 1000, 1000) == (None, "USD") + # Ledger recording still accepts the row; selection must use known_compute_cost. + assert book.compute_cost("stub_co", "hidden", 1000, 1000) == (0.0, "USD") + + +def test_price_book_uses_original_list_when_billed_is_zero() -> None: + book = PriceBook(InMemoryConfigStore()) + book.set_price( + PriceEntry( + "openrouter", + "qwen-free", + prompt_price_per_1k=0.0, + completion_price_per_1k=0.0, + original_list_prompt_per_1k=0.2, + original_list_completion_per_1k=0.4, + ) + ) + cost, currency = book.known_compute_cost("openrouter", "qwen-free", 1000, 1000) + assert currency == "USD" + assert cost == 0.6 + + +if __name__ == "__main__": # pragma: no cover + test_non_finite_and_boolean_prices_are_unknown() + test_complete_pair_mean_avoids_intermediate_overflow() + test_partial_two_sided_price_is_unknown_not_free() + test_openrouter_free_variant_keeps_sibling_list_price() + test_explicit_zero_without_list_is_known_free() + test_missing_price_is_unknown_not_free() + test_unpriced_worker_loses_known_cost_tie_break() + test_price_book_stub_without_rates_is_unknown() + test_price_book_uses_original_list_when_billed_is_zero() + print("ok")