diff --git a/contextual_orchestrator/__main__.py b/contextual_orchestrator/__main__.py index ffd82536d..821539b6c 100644 --- a/contextual_orchestrator/__main__.py +++ b/contextual_orchestrator/__main__.py @@ -59,6 +59,44 @@ def _bootstrap_telemetry_config() -> InMemoryConfigStore: return config +_LOG_LEVEL_NAMES = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL") + + +def _resolve_log_level(name: str | None) -> int: + """Map a log-level name to its :mod:`logging` constant, defaulting to WARNING. + + Accepts case-insensitively any of :data:`_LOG_LEVEL_NAMES`; anything else + (including ``None`` or unset) resolves to ``logging.WARNING``, matching + this module's behavior before verbose logging was configurable. + """ + normalized = (name or "").strip().upper() + if normalized not in _LOG_LEVEL_NAMES: + return logging.WARNING + return getattr(logging, normalized) + + +def _configure_logging(explicit_level: str | None = None) -> None: + """Attach a stderr handler at the requested verbosity; a no-op if none was requested. + + Verbosity is never a secret channel: only provider/model identifiers, + counts, and elapsed time are ever logged (see + :mod:`contextual_orchestrator.model_discovery`) -- never an ``api_key`` + or provider payload/content -- so raising this to DEBUG is safe in any + environment, CI included. With no explicit CLI selection, this leaves the process's logging + configuration untouched (Python's stdlib WARNING default applies, same + as before verbose logging was configurable) rather than pinning an + explicit level on every ``main()`` call -- ``main()`` runs many times + in-process in this repo's own test suite, and an unconditional + ``setLevel`` here would permanently shadow a later, unrelated test's + ``caplog.at_level("DEBUG")`` for the whole process. + """ + if not explicit_level: + return + level = _resolve_log_level(explicit_level) + logging.basicConfig(level=level, format="%(asctime)s %(levelname)s %(name)s %(message)s") + logging.getLogger("contextual_orchestrator").setLevel(level) + + def _positive_int(value: str) -> int: """Parse a strictly positive integer for an argparse option.""" try: @@ -273,6 +311,12 @@ def _discover_models_command(argv: list[str]) -> None: action="store_true", help="Emit secret-free provider discovery diagnostics to stderr.", ) + parser.add_argument( + "--log-level", + choices=_LOG_LEVEL_NAMES, + default=None, + help="Verbosity for provider discovery diagnostics (default: WARNING).", + ) parser.add_argument( "--agents-db", default=None, @@ -306,8 +350,10 @@ def _discover_models_command(argv: list[str]) -> None: help="Optional reviewed CA bundle for configured-gateway discovery TLS verification.", ) args = parser.parse_args(argv) - if args.verbose: - logging.basicConfig(level=logging.DEBUG) + if args.log_level: + _configure_logging(args.log_level) + elif args.verbose: + _configure_logging("DEBUG") if args.enable_cheapest and not args.agents_db: parser.error("--enable-cheapest requires --agents-db") @@ -594,9 +640,19 @@ def main(argv: list[str] | None = None) -> None: action="store_true", help="discover source-declared chat-capable models at startup and activate them", ) + parser.add_argument( + "--log-level", + choices=_LOG_LEVEL_NAMES, + default=None, + help="Verbosity for provider-discovery and server diagnostics (default: WARNING). " + "DEBUG never logs an " + "api_key or provider payload/content.", + ) args = parser.parse_args(arguments) - if args.verbose: - logging.basicConfig(level=logging.DEBUG) + if args.log_level: + _configure_logging(args.log_level) + elif args.verbose: + _configure_logging("DEBUG") client = ModelClient( ca_bundle=args.provider_ca_bundle, diff --git a/contextual_orchestrator/model_discovery.py b/contextual_orchestrator/model_discovery.py index b95d01758..001967ab8 100644 --- a/contextual_orchestrator/model_discovery.py +++ b/contextual_orchestrator/model_discovery.py @@ -13,11 +13,13 @@ from __future__ import annotations +import _thread from decimal import Decimal -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import FIRST_COMPLETED, Future, wait import json import logging import math +import queue import re import ssl import time @@ -25,7 +27,7 @@ import urllib.request import certifi from dataclasses import dataclass, replace -from typing import TYPE_CHECKING, Any, Literal, Mapping +from typing import TYPE_CHECKING, Any, Callable, Literal, Mapping, Sequence from urllib.parse import quote, urlsplit, urlunsplit from .chat_capability import ( @@ -45,8 +47,23 @@ if TYPE_CHECKING: from .cost_ledger import PriceBook -DISCOVERY_TIMEOUT_SECONDS = 15.0 +# Never logs an ``api_key``, a provider payload, or model content -- only +# provider/model identifiers, counts, and elapsed time -- so this logger is +# safe to enable at DEBUG in any environment, including CI. Silent by default +# (module loggers have no handler until a caller configures one, e.g. via +# ``logging.basicConfig`` selected by CLI flags in ``__main__.main()``), +# matching this repo's existing per-module logger +# convention (``server.py``, ``telemetry.py``, ``video_jobs.py``). _LOGGER = logging.getLogger(__name__) + +_OPENROUTER_ENDPOINT_WORKERS = 8 +_OPENROUTER_ENDPOINT_QUEUE: queue.SimpleQueue[ + tuple[Future[Any], Callable[..., Any], tuple[Any, ...]] +] = queue.SimpleQueue() +_OPENROUTER_ENDPOINT_THREADS: tuple[int, ...] = () +_OPENROUTER_ENDPOINT_THREADS_LOCK = _thread.allocate_lock() + +DISCOVERY_TIMEOUT_SECONDS = 15.0 # One bounded retry for a provider's primary model-list fetch, reusing the same # transient-vs-terminal classification completion calls already trust # (is_transient_error). A short, fixed delay and a shortened retry timeout keep @@ -296,7 +313,14 @@ def __init__(self, provider_name: str, error_code: str) -> None: super().__init__(f"model discovery failed for provider {provider_name!r}: {error_code}") -def _fetch_json(url: str, *, api_key: str = "", auth_scheme: str = "Bearer", timeout: float) -> Any: +def _fetch_json( + url: str, + *, + api_key: str = "", + auth_scheme: str = "Bearer", + timeout: float, + deadline: float | None = None, +) -> Any: if not url.startswith("https://"): # Every caller passes one of the hardcoded PROVIDER_SOURCES chat_base_url # constants below, never external input -- but urlopen also honors @@ -307,15 +331,23 @@ def _fetch_json(url: str, *, api_key: str = "", auth_scheme: str = "Bearer", tim if api_key: headers["authorization"] = format_authorization_header(auth_scheme, api_key) request = urllib.request.Request(url, headers=headers, method="GET") + def remaining_timeout() -> float: + if deadline is None: + return timeout + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("model discovery deadline exceeded") + return min(timeout, remaining) + # Scheme is enforced to https:// immediately above; url is never attacker-controlled. try: - response = urllib.request.urlopen(request, timeout=timeout) # noqa: S310 - fixed provider inventory # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected + response = urllib.request.urlopen(request, timeout=remaining_timeout()) # noqa: S310 - fixed provider inventory # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected except urllib.error.URLError as exc: if not isinstance(exc.reason, ssl.SSLCertVerificationError): raise context = ssl.create_default_context(cafile=certifi.where()) response = urllib.request.urlopen( # noqa: S310 - fixed provider inventory # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected - request, timeout=timeout, context=context + request, timeout=remaining_timeout(), context=context ) with response: return json.loads(response.read().decode("utf-8")) @@ -882,7 +914,21 @@ def _merge_openrouter_provider_privacy( def _openrouter_free_model_endpoints( payload: Any, *, api_key: str, timeout: float ) -> dict[str, Any]: - """Fetch endpoint/provider mappings only for explicitly zero-price models.""" + """Fetch endpoint/provider mappings only for explicitly zero-price models. + + Bounded to one overall ``timeout``-second deadline for the whole free + catalog, not ``timeout`` seconds per model: with a fixed-size thread + pool, a per-request timeout alone still lets total wall time grow with + the free-model count (``ceil(len(model_ids) / max_workers)`` sequential + timeout waves), which can make this single enrichment step consume a + caller's entire startup budget on its own as the free catalog grows. + Models whose fetch has not completed by the shared deadline are + reported as unmapped (``None``) rather than waited on further; this + data is best-effort provider-privacy enrichment, not required for a + model to be discovered. In-flight stdlib HTTP calls cannot be cancelled, + so a process-wide pool of eight daemon workers caps their accumulation + without extending one-shot process shutdown. + """ rows = payload.get("data") if isinstance(payload, dict) else None model_ids = [ row["id"] @@ -892,6 +938,15 @@ def _openrouter_free_model_endpoints( and isinstance(row.get("pricing"), dict) and _pricing_is_free(row.get("pricing")) ] + if not model_ids: + return {} + _LOGGER.debug( + "openrouter_free_endpoints_started free_model_count=%d timeout=%.1f", + len(model_ids), + timeout, + ) + + deadline = time.monotonic() + max(0.0, timeout) def fetch(model_id: str) -> tuple[str, Any]: author, separator, slug = model_id.partition("/") @@ -902,12 +957,79 @@ def fetch(model_id: str) -> tuple[str, Any]: f"https://openrouter.ai/api/v1/models/{quote(author, safe='')}/{quote(slug, safe=':')}/endpoints", api_key=api_key, timeout=timeout, + deadline=deadline, ).get("data") except (AttributeError, urllib.error.URLError, TimeoutError, ValueError, OSError): return model_id, None - with ThreadPoolExecutor(max_workers=min(8, len(model_ids) or 1)) as executor: - return dict(executor.map(fetch, model_ids)) + _ensure_openrouter_endpoint_workers() + started = time.monotonic() + results: dict[str, Any] = dict.fromkeys(model_ids) + model_iter = iter(model_ids) + pending: dict[Future[Any], str] = {} + + def submit_next() -> bool: + try: + model_id = next(model_iter) + except StopIteration: + return False + future: Future[Any] = Future() + pending[future] = model_id + _OPENROUTER_ENDPOINT_QUEUE.put((future, fetch, (model_id,))) + return True + + for _ in range(min(_OPENROUTER_ENDPOINT_WORKERS, len(model_ids))): + submit_next() + completed = 0 + while pending: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + done, _ = wait(pending, timeout=remaining, return_when=FIRST_COMPLETED) + if not done: + break + for future in done: + pending.pop(future) + model_id, endpoint = future.result() + results[model_id] = endpoint + completed += 1 + submit_next() + for future in pending: + future.cancel() + _LOGGER.debug( + "openrouter_free_endpoints_finished elapsed=%.2f completed=%d deadline_exceeded=%d", + time.monotonic() - started, + completed, + len(model_ids) - completed, + ) + return results + + +def _ensure_openrouter_endpoint_workers() -> None: + """Start the fixed daemon pool used by best-effort endpoint enrichment.""" + global _OPENROUTER_ENDPOINT_THREADS + if _OPENROUTER_ENDPOINT_THREADS: + return + with _OPENROUTER_ENDPOINT_THREADS_LOCK: + if _OPENROUTER_ENDPOINT_THREADS: + return + + def worker() -> None: + while True: + future, function, args = _OPENROUTER_ENDPOINT_QUEUE.get() + if not future.set_running_or_notify_cancel(): + continue + try: + future.set_result(function(*args)) + except BaseException as exc: + future.set_exception(exc) + + # Low-level threads are daemon workers by definition and are not joined + # during interpreter shutdown; the fixed tuple is the process-wide cap. + _OPENROUTER_ENDPOINT_THREADS = tuple( + _thread.start_new_thread(worker, ()) + for _ in range(_OPENROUTER_ENDPOINT_WORKERS) + ) def _privacy_policy_urls( @@ -1197,6 +1319,7 @@ def discover_provider_models( source.provider_name, ) return [] + started = time.monotonic() _LOGGER.debug( "model discovery started account=%s", source.provider_name, @@ -1236,9 +1359,10 @@ def discover_provider_models( if last_exc is not None: error_code = _provider_discovery_error_code(last_exc) _LOGGER.debug( - "model discovery failed account=%s error_code=%s", + "model discovery failed account=%s error_code=%s elapsed=%.2f", source.provider_name, error_code, + time.monotonic() - started, ) raise ProviderDiscoveryError(source.provider_name, error_code) from None if source.models_dev_provider_id: @@ -1286,9 +1410,10 @@ def discover_provider_models( discovered = _parse_openai_compatible(payload, source) result = [replace(model, evidence_only=source.evidence_only) for model in discovered] _LOGGER.debug( - "model discovery completed account=%s model_count=%d", + "model discovery completed account=%s model_count=%d elapsed=%.2f", source.provider_name, len(result), + time.monotonic() - started, ) return result @@ -1311,6 +1436,8 @@ def discover_all_models( hand every source the identical parsed payload, instead of each source independently repeating the fetch inside :func:`discover_provider_models`. """ + started = time.monotonic() + _LOGGER.debug("discover_all_models_started source_count=%d", len(sources)) discovered: list[DiscoveredModel] = [] errors: list[ProviderDiscoveryError] = [] models_dev_metadata: Any = _NOT_FETCHED @@ -1347,6 +1474,12 @@ def discover_all_models( routed, openrouter_paid_inference_available(timeout=timeout), ) + _LOGGER.info( + "discover_all_models_finished elapsed=%.2f discovered=%d errors=%d", + time.monotonic() - started, + len(routed), + len(errors), + ) return routed, errors diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index cbad42f7f..89f998ecf 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,5 +1,49 @@ # Contextual Orchestrator: Product & Technical Gap Baseline +## 2026-08-31 OpenRouter free-discovery deadline bound (PR #939) + orchestrator/free tool-call-capability gap (issue #940) + +Investigating Devin Review's second finding on `ContextualWisdomLab/.github#1463` ("Discovery exhausts +sidecar startup budget"): `_openrouter_free_model_endpoints()` (`contextual_orchestrator/model_discovery.py`) +used a per-request `timeout` but no overall deadline, so total wall time scaled with the free-catalog +count (`ceil(len(model_ids) / 8)` sequential timeout waves via `ThreadPoolExecutor` — up to +`12 * 15s = 180s` at today's ~94-model OpenRouter free catalog, for this one enrichment step alone). +Confirmed genuinely new relative to `.github`'s old vendored pin (`git log -S +"_openrouter_free_model_endpoints"` → single commit `6376d85`), so this is a real regression risk +introduced since that pin was last bumped, not a pre-existing characteristic. `.github`'s sidecar has a +180s startup watchdog covering discovery + catalog build + preflight combined (already tracked as tight +in `ContextualWisdomLab/.github#1455`), so this one call could exhaust it alone in the worst case — on +top of ~6 other sequential discovery calls. Live evidence from `.github#1463`'s own Strix run +(`33348306414`) showed the sidecar provisioning step actually succeeding in ~5m41s, so this is hardening +against a worst-case tail risk, not a fix for an observed outage. + +**Fix**: [PR #939](https://github.com/ContextualWisdomLab/contextual-orchestrator/pull/939) bounds the +fetch to one shared `concurrent.futures.wait(futures, timeout=timeout)` deadline instead of +`executor.map()` (no total-time bound); any model not done by the deadline is reported unmapped rather +than waited on further — this data is best-effort provider-privacy enrichment, not required for +discovery. Regression test proves the bound holds regardless of catalog size. Also adds opt-in +verbose/debug discovery logging (`--log-level`, never logs +`api_key` or payload content) for exactly this kind of future timing diagnosis. + +**Separate, larger gap found while root-causing `.github#1463`'s live Strix failure** (not caused by +that PR's diff — confirmed via `git show c10a557:...` that the relevant candidate-selection code +(`CATALOG_FAMILY_CAP=8`) predates the branch): NVIDIA NIM's `meta/llama-3.2-11b-vision-instruct` +rejects any tools request that isn't restricted to one tool call at a time +(`openai.BadRequestError: ... "This model only supports single tool-calls at once!"`, +[job log](https://github.com/ContextualWisdomLab/.github/actions/runs/33348306414/job/99356649771)). +Traced the full request/retry/candidate-selection path (`server.py`, `orchestrator.py`, +`model_discovery.py`): `parallel_tool_calls` is forwarded to upstream verbatim with no gateway default; +the passthrough failover path only retries on a fixed HTTP-status allowlist that excludes plain `400`, +so one incompatible candidate hard-fails the whole request with no failover to the next pool candidate; +and `DiscoveredModel` tracks modality/price/ZDR capability but nothing about tool-call semantics, so +there's no way to exclude such models from tool-calling-eligible pools at candidate-selection time. +`docs/planning/adrs/0035-structured-provider-orchestration.md`'s existing "capability tags are positive +declarations only" stance and `tool_fallback.py`'s deliberately status-only (not message-sniffing) +retry classification both cut against a quick patch — this needs a proper design pass. Tracked as +[issue #940](https://github.com/ContextualWisdomLab/contextual-orchestrator/issues/940) with the +recommended direction (a new, evidence-based negative capability signal feeding +`is_routable_discovered_model`'s existing filter shape, not a hardcoded denylist) rather than rushed +into this PR. + ## 2026-08-30 provider-catalog-sync: no scheduled run has succeeded in 5 days over one provider; workflow check was too strict `provider-catalog-sync.yml` (run `33312773022`, job `99260685380`) failed with `credential diff --git a/tests/test_discover_models_cli.py b/tests/test_discover_models_cli.py index 3b369fac7..b35726633 100644 --- a/tests/test_discover_models_cli.py +++ b/tests/test_discover_models_cli.py @@ -69,6 +69,23 @@ def test_discover_models_provider_ca_bundle_help_is_available() -> None: assert "--provider-ca-bundle" in stdout.getvalue() +def test_discover_models_log_level_flag_wins_over_verbose() -> None: + """The subcommand accepts an explicit verbosity selection.""" + stdout = StringIO() + with ( + patch.object( + sys, + "argv", + ["contextual-orchestrator", "discover-models", "--verbose", "--log-level", "INFO"], + ), + patch.object(sys, "stdout", stdout), + patch("contextual_orchestrator.__main__._configure_logging") as configure_logging, + ): + main() + + assert configure_logging.call_args_list[-1].args == ("INFO",) + + def test_discover_models_with_no_credentials_reports_zero_and_succeeds() -> None: set_backend(InMemoryCredentialBackend()) stdout = StringIO() diff --git a/tests/test_model_discovery.py b/tests/test_model_discovery.py index c753a3d6a..e4e4f063f 100644 --- a/tests/test_model_discovery.py +++ b/tests/test_model_discovery.py @@ -3,7 +3,10 @@ from __future__ import annotations import json +import ssl +import subprocess import sys +import time import urllib.error import urllib.parse from dataclasses import replace @@ -15,6 +18,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator import model_discovery as model_discovery_module # noqa: E402 from contextual_orchestrator.orchestrator import AUTH_SCHEME_RAW_TOKEN # noqa: E402 from contextual_orchestrator.credentials import ( # noqa: E402 InMemoryCredentialBackend, @@ -34,6 +38,7 @@ _deduplicate_discovered_models, _fetch_json, _merge_configured_gateway_metadata, + _openrouter_free_model_endpoints, _merge_openrouter_provider_privacy, _merge_openrouter_zdr_metadata, _price_per_1k, @@ -670,6 +675,96 @@ def test_openrouter_skips_model_endpoint_fetches_when_provider_policies_fail() - assert [model.model_id for model in discovered] == ["free/model"] endpoint_fetch.assert_not_called() + + +def test_openrouter_free_model_endpoint_fetch_bounded_by_one_overall_deadline() -> None: + """Wall time must stay near one ``timeout`` regardless of free-catalog size. + + A fixed-size thread pool with only a per-request timeout still lets total + wall time grow with the model count (sequential timeout waves), which can + make this single enrichment step consume a caller's entire startup + budget as the free catalog grows (see + ContextualWisdomLab/.github#1463 review discussion). Every one of a large + number of models hangs past the deadline here; if the fix regresses to + per-model timeout waves, this proves it by taking far longer than the + shared deadline. + """ + model_ids = [f"vendor/free-{i}" for i in range(200)] + payload = { + "data": [ + {"id": model_id, "pricing": {"prompt": "0", "completion": "0"}} + for model_id in model_ids + ] + } + + def hang(*_args, **_kwargs): + time.sleep(5) + raise AssertionError("must not be awaited past the shared deadline") + + with patch("contextual_orchestrator.model_discovery._fetch_json", side_effect=hang): + started = time.monotonic() + result = _openrouter_free_model_endpoints(payload, api_key="sk-router", timeout=0.2) + elapsed = time.monotonic() - started + + # Unbounded-per-model waves would need ceil(200 / 8) * 5s = 125s here; + # the shared deadline keeps this well under that regardless of catalog size. + assert elapsed < 2.0 + assert result == {model_id: None for model_id in model_ids} + + +def test_openrouter_free_endpoint_deadline_does_not_hold_process_open() -> None: + """Timed-out endpoint workers must not extend one-shot CLI process lifetime.""" + script = """ +import time +from unittest.mock import patch +from contextual_orchestrator.model_discovery import _openrouter_free_model_endpoints +payload = {'data': [{'id': 'vendor/free', 'pricing': {'prompt': '0', 'completion': '0'}}]} +with patch('contextual_orchestrator.model_discovery._fetch_json', side_effect=lambda *a, **k: time.sleep(60)): + _openrouter_free_model_endpoints(payload, api_key='secret', timeout=0.05) +""" + started = time.monotonic() + subprocess.run([sys.executable, "-c", script], check=True, timeout=2) + assert time.monotonic() - started < 2 + + +def test_openrouter_free_endpoint_workers_have_a_fixed_ceiling() -> None: + """Repeated timed-out calls reuse one bounded daemon pool.""" + payload = { + "data": [ + {"id": f"vendor/free-{index}", "pricing": {"prompt": "0", "completion": "0"}} + for index in range(20) + ] + } + + with patch( + "contextual_orchestrator.model_discovery._fetch_json", + side_effect=lambda *_args, **_kwargs: time.sleep(0.2), + ): + _openrouter_free_model_endpoints(payload, api_key="secret", timeout=0.005) + worker_ids = model_discovery_module._OPENROUTER_ENDPOINT_THREADS + for _ in range(12): + _openrouter_free_model_endpoints(payload, api_key="secret", timeout=0.005) + + assert len(worker_ids) == 8 + assert model_discovery_module._OPENROUTER_ENDPOINT_THREADS == worker_ids + + +def test_fetch_json_tls_retry_uses_only_remaining_deadline() -> None: + """Certificate fallback cannot start after the caller's shared deadline.""" + certificate_error = ssl.SSLCertVerificationError(1, "untrusted") + with ( + patch( + "contextual_orchestrator.model_discovery.urllib.request.urlopen", + side_effect=urllib.error.URLError(certificate_error), + ) as urlopen, + patch("contextual_orchestrator.model_discovery.time.monotonic", side_effect=[0.0, 2.0]), + pytest.raises(TimeoutError, match="deadline exceeded"), + ): + _fetch_json("https://provider.example/models", timeout=5.0, deadline=1.0) + + assert urlopen.call_count == 1 + + def test_non_text_model_does_not_gain_structured_response_capability() -> None: """A provider parameter alone cannot make an image-only model a synthesizer.""" register_credential("OPENROUTER_API_KEY", "sk-router")