From af172362459814ee17256b7abf3175d59e7fdaa5 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 02:18:15 +0000 Subject: [PATCH 1/6] fix(discovery): bound OpenRouter free-model endpoint fan-out to one deadline _openrouter_free_model_endpoints() used a per-request timeout but no overall deadline: with a fixed 8-worker thread pool, total wall time still grew with the free-model count (ceil(n/8) sequential timeout waves). At today's ~94-model free catalog that's up to 180s on its own -- enough on its own to exhaust the contextual-orchestrator review sidecar's 180s startup watchdog in ContextualWisdomLab/.github, flagged by Devin Review on .github#1463. Bounded it to one shared timeout via concurrent.futures.wait(), reporting any model not done by the deadline as unmapped rather than waiting on further batches; 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 logging for provider discovery (contextual_orchestrator.model_discovery: per-provider and per-fetch timing/counts, never api_key or provider payload/content) via --log-level or CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL, off by default and a true no-op when unset so it can't leak state across this repo's many in-process main() calls in tests. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Kj32ABZLZ2a6TPTyvYrRkg --- contextual_orchestrator/__main__.py | 53 ++++++++++++++ contextual_orchestrator/model_discovery.py | 84 +++++++++++++++++++--- tests/test_model_discovery.py | 39 ++++++++++ 3 files changed, 168 insertions(+), 8 deletions(-) diff --git a/contextual_orchestrator/__main__.py b/contextual_orchestrator/__main__.py index f1c92fbad..70e639544 100644 --- a/contextual_orchestrator/__main__.py +++ b/contextual_orchestrator/__main__.py @@ -4,6 +4,7 @@ import argparse import json +import logging import os import sys from dataclasses import replace @@ -56,6 +57,47 @@ 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. ``explicit_level`` (e.g. a ``--log-level`` CLI + flag) wins over the ``CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL`` environment + variable. With neither set, 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. + """ + requested = explicit_level or os.environ.get("CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL") + if not requested: + return + level = _resolve_log_level(requested) + 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: @@ -440,6 +482,7 @@ def _auto_discover_runtime_agents(orchestrator: TaskOrchestrator) -> dict[str, l def main(argv: list[str] | None = None) -> None: """Parse CLI options and run bootstrap, prompt completion, or the HTTP server.""" arguments = list(sys.argv[1:] if argv is None else argv) + _configure_logging() if arguments and arguments[0] == "register-credential": _register_credential_command(arguments[1:]) return @@ -545,7 +588,17 @@ 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: " + "$CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL or WARNING). DEBUG never logs an " + "api_key or provider payload/content.", + ) args = parser.parse_args(arguments) + if args.log_level: + _configure_logging(args.log_level) client = ModelClient( ca_bundle=args.provider_ca_bundle, diff --git a/contextual_orchestrator/model_discovery.py b/contextual_orchestrator/model_discovery.py index 07bc06517..5a7faacf4 100644 --- a/contextual_orchestrator/model_discovery.py +++ b/contextual_orchestrator/model_discovery.py @@ -14,8 +14,9 @@ from __future__ import annotations from decimal import Decimal -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import ThreadPoolExecutor, wait import json +import logging import math import re import ssl @@ -39,6 +40,15 @@ if TYPE_CHECKING: from .cost_ledger import PriceBook +# 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`` gated on ``CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL`` in +# ``__main__.main()``), matching this repo's existing per-module logger +# convention (``server.py``, ``telemetry.py``, ``video_jobs.py``). +_LOGGER = logging.getLogger(__name__) + DISCOVERY_TIMEOUT_SECONDS = 15.0 # Some discovery endpoints (verified live: models.dev returns Cloudflare HTTP # 403 error 1010) reject urllib's default "Python-urllib/X.Y" user agent as a @@ -859,7 +869,19 @@ 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. + """ rows = payload.get("data") if isinstance(payload, dict) else None model_ids = [ row["id"] @@ -869,6 +891,13 @@ 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, + ) def fetch(model_id: str) -> tuple[str, Any]: author, separator, slug = model_id.partition("/") @@ -883,8 +912,22 @@ def fetch(model_id: str) -> tuple[str, Any]: 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)) + started = time.monotonic() + executor = ThreadPoolExecutor(max_workers=min(8, len(model_ids))) + try: + futures = {executor.submit(fetch, model_id): model_id for model_id in model_ids} + done, not_done = wait(futures, timeout=timeout) + results: dict[str, Any] = {futures[future]: None for future in not_done} + results.update(dict(future.result() for future in done)) + _LOGGER.debug( + "openrouter_free_endpoints_finished elapsed=%.2f completed=%d deadline_exceeded=%d", + time.monotonic() - started, + len(done), + len(not_done), + ) + return results + finally: + executor.shutdown(wait=False, cancel_futures=True) def _privacy_policy_urls( @@ -1114,6 +1157,8 @@ def discover_provider_models( url = source.list_url if source.task_filter: url = f"{url}?task={source.task_filter}" + started = time.monotonic() + _LOGGER.debug("provider_discovery_started provider=%s", source.provider_name) try: fetch = ( _fetch_configured_gateway_json @@ -1131,7 +1176,14 @@ def discover_provider_models( # OSError covers ConnectionError/reset failures that are not URLError # subclasses, so a raw provider transport failure can never escape the # discovery boundary with provider text attached. - raise ProviderDiscoveryError(source.provider_name, _provider_discovery_error_code(exc)) from None + error_code = _provider_discovery_error_code(exc) + _LOGGER.debug( + "provider_discovery_failed provider=%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: if models_dev_metadata is _NOT_FETCHED: metadata = _fetch_models_dev_metadata(timeout=timeout) @@ -1175,7 +1227,14 @@ def discover_provider_models( discovered = _parse_bytez(payload, source) else: discovered = _parse_openai_compatible(payload, source) - return [replace(model, evidence_only=source.evidence_only) for model in discovered] + result = [replace(model, evidence_only=source.evidence_only) for model in discovered] + _LOGGER.debug( + "provider_discovery_finished provider=%s model_count=%d elapsed=%.2f", + source.provider_name, + len(result), + time.monotonic() - started, + ) + return result def discover_all_models( @@ -1196,6 +1255,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 @@ -1219,10 +1280,17 @@ def discover_all_models( # The OpenRouter catalog is evidence-only; its public ZDR endpoint supplies # matching privacy evidence for discovered models from other providers. It # is never selected as an inference upstream here. - return _apply_discovered_model_evidence( + result = _apply_discovered_model_evidence( _deduplicate_discovered_models(discovered), _openrouter_zdr_model_ids(timeout=timeout), - ), errors + ) + _LOGGER.info( + "discover_all_models_finished elapsed=%.2f discovered=%d errors=%d", + time.monotonic() - started, + len(result), + len(errors), + ) + return result, errors def openrouter_paid_inference_available( diff --git a/tests/test_model_discovery.py b/tests/test_model_discovery.py index 3dc2a96df..056e70bca 100644 --- a/tests/test_model_discovery.py +++ b/tests/test_model_discovery.py @@ -4,6 +4,7 @@ import json import sys +import time import urllib.error import urllib.parse from dataclasses import replace @@ -33,6 +34,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, @@ -623,6 +625,43 @@ 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_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") From b15f9f77ebf067bdbff35d5f11bd8e41a282ee76 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 02:24:43 +0000 Subject: [PATCH 2/6] docs(gaps): record the discovery-deadline fix and a new tool-call-capability gap Documents PR #939's fix and the separate, larger orchestrator/free gap found while root-causing .github#1463's live Strix failure (tracked as issue #940) -- a single model that rejects multi-tool- call requests hard-fails the whole request with no failover, and there's no capability signal to exclude it from candidate selection. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Kj32ABZLZ2a6TPTyvYrRkg --- docs/product-technical-gap-baseline.md | 44 ++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index d8dca7395..65bbbec9e 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` / `CONTEXTUAL_ORCHESTRATOR_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 From 7bcb15e83411c1dc93b3e9edad741dc699eef54f Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 31 Aug 2026 02:53:31 +0000 Subject: [PATCH 3/6] test(discovery): fix bootstrap-selector test missed by #941's family removal #941 removed _provider_family() (nvidia_nim/nvidia_nim_sub collapsing) but didn't touch tests/test_discovery_bootstrap_selection.py, leaving test_bootstrap_selector_treats_nim_primary_and_sub_as_one_outage_domain asserting the now-removed collapsing behavior. Updated to assert the correct independent-provider behavior. Also folds this PR's logging additions to match #941's "account="-prefixed message convention. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Kj32ABZLZ2a6TPTyvYrRkg --- contextual_orchestrator/model_discovery.py | 8 ++++---- tests/test_discovery_bootstrap_selection.py | 13 ++++++++++--- 2 files changed, 14 insertions(+), 7 deletions(-) diff --git a/contextual_orchestrator/model_discovery.py b/contextual_orchestrator/model_discovery.py index f6a1f5fee..4d18e26e7 100644 --- a/contextual_orchestrator/model_discovery.py +++ b/contextual_orchestrator/model_discovery.py @@ -52,6 +52,7 @@ # ``__main__.main()``), matching this repo's existing per-module logger # convention (``server.py``, ``telemetry.py``, ``video_jobs.py``). _LOGGER = logging.getLogger(__name__) + DISCOVERY_TIMEOUT_SECONDS = 15.0 # Some discovery endpoints (verified live: models.dev returns Cloudflare HTTP # 403 error 1010) reject urllib's default "Python-urllib/X.Y" user agent as a @@ -1161,6 +1162,7 @@ def discover_provider_models( source.provider_name, ) return [] + started = time.monotonic() _LOGGER.debug( "model discovery started account=%s", source.provider_name, @@ -1168,8 +1170,6 @@ def discover_provider_models( url = source.list_url if source.task_filter: url = f"{url}?task={source.task_filter}" - started = time.monotonic() - _LOGGER.debug("provider_discovery_started provider=%s", source.provider_name) try: fetch = ( _fetch_configured_gateway_json @@ -1189,7 +1189,7 @@ def discover_provider_models( # discovery boundary with provider text attached. error_code = _provider_discovery_error_code(exc) _LOGGER.debug( - "provider_discovery_failed provider=%s error_code=%s elapsed=%.2f", + "model discovery failed account=%s error_code=%s elapsed=%.2f", source.provider_name, error_code, time.monotonic() - started, @@ -1240,7 +1240,7 @@ def discover_provider_models( discovered = _parse_openai_compatible(payload, source) result = [replace(model, evidence_only=source.evidence_only) for model in discovered] _LOGGER.debug( - "provider_discovery_finished provider=%s model_count=%d elapsed=%.2f", + "model discovery completed account=%s model_count=%d elapsed=%.2f", source.provider_name, len(result), time.monotonic() - started, diff --git a/tests/test_discovery_bootstrap_selection.py b/tests/test_discovery_bootstrap_selection.py index a6124fee3..93c862ffb 100644 --- a/tests/test_discovery_bootstrap_selection.py +++ b/tests/test_discovery_bootstrap_selection.py @@ -323,8 +323,15 @@ def test_bootstrap_selector_prefers_provider_diversity_before_duplicates() -> No assert selected == [router_cheapest, nim_model, openai_model] -def test_bootstrap_selector_treats_nim_primary_and_sub_as_one_outage_domain() -> None: - """Two NIM keys must not displace an independently hosted provider.""" +def test_bootstrap_selector_treats_nim_primary_and_sub_as_independent_providers() -> None: + """Two NIM keys are independent credential boundaries, not one outage domain. + + Every KV credential is its own independent provider-account/catalog + boundary (see ContextualWisdomLab/contextual-orchestrator#941); the + selector never infers that ``nvidia_nim`` and ``nvidia_nim_sub`` share + fate merely because of the name resemblance, so cheapest-first filling + admits both before a pricier, unrelated provider. + """ selector = getattr( model_discovery, "select_bootstrap_discovered_agents", @@ -346,7 +353,7 @@ def test_bootstrap_selector_treats_nim_primary_and_sub_as_one_outage_domain() -> 2, ) - assert selected == [nim_primary, openrouter] + assert selected == [nim_primary, nim_sub] def test_bootstrap_selector_is_deterministic_when_every_model_is_unpriced() -> None: From 2aa4b8c3232065add97871e325ef8b85880d5fe4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 01:32:35 +0900 Subject: [PATCH 4/6] fix(discovery): make endpoint deadline process-bounded --- contextual_orchestrator/__main__.py | 13 ++- contextual_orchestrator/model_discovery.py | 125 +++++++++++++++++---- tests/test_discover_models_cli.py | 18 +++ tests/test_model_discovery.py | 54 +++++++++ 4 files changed, 188 insertions(+), 22 deletions(-) diff --git a/contextual_orchestrator/__main__.py b/contextual_orchestrator/__main__.py index 12500a1dd..d773f510e 100644 --- a/contextual_orchestrator/__main__.py +++ b/contextual_orchestrator/__main__.py @@ -314,6 +314,13 @@ 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: " + "$CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL or WARNING).", + ) parser.add_argument( "--agents-db", default=None, @@ -347,8 +354,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") diff --git a/contextual_orchestrator/model_discovery.py b/contextual_orchestrator/model_discovery.py index 5b8b6b6bc..8d1e62441 100644 --- a/contextual_orchestrator/model_discovery.py +++ b/contextual_orchestrator/model_discovery.py @@ -14,18 +14,20 @@ from __future__ import annotations from decimal import Decimal -from concurrent.futures import ThreadPoolExecutor, wait +from concurrent.futures import FIRST_COMPLETED, Future, wait import json import logging import math +import queue import re import ssl +import threading import time import urllib.error 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 ( @@ -54,6 +56,13 @@ # 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[threading.Thread, ...] = () +_OPENROUTER_ENDPOINT_THREADS_LOCK = threading.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 @@ -304,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 @@ -315,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")) @@ -901,7 +925,9 @@ def _openrouter_free_model_endpoints( 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. + 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 = [ @@ -920,6 +946,8 @@ def _openrouter_free_model_endpoints( timeout, ) + deadline = time.monotonic() + max(0.0, timeout) + def fetch(model_id: str) -> tuple[str, Any]: author, separator, slug = model_id.partition("/") if not separator or not author or not slug: @@ -929,26 +957,83 @@ 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 + _ensure_openrouter_endpoint_workers() started = time.monotonic() - executor = ThreadPoolExecutor(max_workers=min(8, len(model_ids))) - try: - futures = {executor.submit(fetch, model_id): model_id for model_id in model_ids} - done, not_done = wait(futures, timeout=timeout) - results: dict[str, Any] = {futures[future]: None for future in not_done} - results.update(dict(future.result() for future in done)) - _LOGGER.debug( - "openrouter_free_endpoints_finished elapsed=%.2f completed=%d deadline_exceeded=%d", - time.monotonic() - started, - len(done), - len(not_done), + 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) + + _OPENROUTER_ENDPOINT_THREADS = tuple( + threading.Thread( + target=worker, + name=f"openrouter-endpoint-{index}", + daemon=True, + ) + for index in range(_OPENROUTER_ENDPOINT_WORKERS) ) - return results - finally: - executor.shutdown(wait=False, cancel_futures=True) + for thread in _OPENROUTER_ENDPOINT_THREADS: + thread.start() def _privacy_policy_urls( diff --git a/tests/test_discover_models_cli.py b/tests/test_discover_models_cli.py index 3b369fac7..c9c9774cd 100644 --- a/tests/test_discover_models_cli.py +++ b/tests/test_discover_models_cli.py @@ -69,6 +69,24 @@ 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_and_environment() -> None: + """The subcommand accepts explicit verbosity with documented precedence.""" + stdout = StringIO() + with ( + patch.dict(os.environ, {"CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL": "ERROR"}, clear=True), + 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 fde50dc26..d2a8593ad 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 threading import time import urllib.error import urllib.parse @@ -708,6 +711,57 @@ def hang(*_args, **_kwargs): 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), + ): + for _ in range(12): + _openrouter_free_model_endpoints(payload, api_key="secret", timeout=0.005) + + workers = [thread for thread in threading.enumerate() if thread.name.startswith("openrouter-endpoint-")] + assert len(workers) == 8 + + +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") From 8969302f88adcfda0e0d2e8554d28c3cb31a813c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 01:43:00 +0900 Subject: [PATCH 5/6] fix(discovery): keep deadline workers detached --- contextual_orchestrator/model_discovery.py | 18 +++++++----------- tests/test_model_discovery.py | 8 +++++--- 2 files changed, 12 insertions(+), 14 deletions(-) diff --git a/contextual_orchestrator/model_discovery.py b/contextual_orchestrator/model_discovery.py index e881cb6d4..8e4f4468c 100644 --- a/contextual_orchestrator/model_discovery.py +++ b/contextual_orchestrator/model_discovery.py @@ -13,6 +13,7 @@ from __future__ import annotations +import _thread from decimal import Decimal from concurrent.futures import FIRST_COMPLETED, Future, wait import json @@ -21,7 +22,6 @@ import queue import re import ssl -import threading import time import urllib.error import urllib.request @@ -60,8 +60,8 @@ _OPENROUTER_ENDPOINT_QUEUE: queue.SimpleQueue[ tuple[Future[Any], Callable[..., Any], tuple[Any, ...]] ] = queue.SimpleQueue() -_OPENROUTER_ENDPOINT_THREADS: tuple[threading.Thread, ...] = () -_OPENROUTER_ENDPOINT_THREADS_LOCK = threading.Lock() +_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 @@ -1024,16 +1024,12 @@ def worker() -> None: 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( - threading.Thread( - target=worker, - name=f"openrouter-endpoint-{index}", - daemon=True, - ) - for index in range(_OPENROUTER_ENDPOINT_WORKERS) + _thread.start_new_thread(worker, ()) + for _ in range(_OPENROUTER_ENDPOINT_WORKERS) ) - for thread in _OPENROUTER_ENDPOINT_THREADS: - thread.start() def _privacy_policy_urls( diff --git a/tests/test_model_discovery.py b/tests/test_model_discovery.py index 22ab5bbcc..e4e4f063f 100644 --- a/tests/test_model_discovery.py +++ b/tests/test_model_discovery.py @@ -6,7 +6,6 @@ import ssl import subprocess import sys -import threading import time import urllib.error import urllib.parse @@ -19,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, @@ -740,11 +740,13 @@ def test_openrouter_free_endpoint_workers_have_a_fixed_ceiling() -> None: "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) - workers = [thread for thread in threading.enumerate() if thread.name.startswith("openrouter-endpoint-")] - assert len(workers) == 8 + 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: From 2df8935656e93251061c6b071e907f7779b7ba6b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 1 Sep 2026 04:25:41 +0900 Subject: [PATCH 6/6] fix: keep discovery logging config on CLI --- contextual_orchestrator/__main__.py | 17 ++++++----------- contextual_orchestrator/model_discovery.py | 4 ++-- docs/product-technical-gap-baseline.md | 2 +- tests/test_discover_models_cli.py | 5 ++--- 4 files changed, 11 insertions(+), 17 deletions(-) diff --git a/contextual_orchestrator/__main__.py b/contextual_orchestrator/__main__.py index d773f510e..821539b6c 100644 --- a/contextual_orchestrator/__main__.py +++ b/contextual_orchestrator/__main__.py @@ -82,9 +82,7 @@ def _configure_logging(explicit_level: str | None = None) -> None: 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. ``explicit_level`` (e.g. a ``--log-level`` CLI - flag) wins over the ``CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL`` environment - variable. With neither set, this leaves the process's logging + 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 @@ -92,10 +90,9 @@ def _configure_logging(explicit_level: str | None = None) -> None: ``setLevel`` here would permanently shadow a later, unrelated test's ``caplog.at_level("DEBUG")`` for the whole process. """ - requested = explicit_level or os.environ.get("CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL") - if not requested: + if not explicit_level: return - level = _resolve_log_level(requested) + 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) @@ -318,8 +315,7 @@ def _discover_models_command(argv: list[str]) -> None: "--log-level", choices=_LOG_LEVEL_NAMES, default=None, - help="Verbosity for provider discovery diagnostics (default: " - "$CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL or WARNING).", + help="Verbosity for provider discovery diagnostics (default: WARNING).", ) parser.add_argument( "--agents-db", @@ -534,7 +530,6 @@ def _auto_discover_runtime_agents(orchestrator: TaskOrchestrator) -> dict[str, l def main(argv: list[str] | None = None) -> None: """Parse CLI options and run bootstrap, prompt completion, or the HTTP server.""" arguments = list(sys.argv[1:] if argv is None else argv) - _configure_logging() if arguments and arguments[0] == "register-credential": _register_credential_command(arguments[1:]) return @@ -649,8 +644,8 @@ def main(argv: list[str] | None = None) -> None: "--log-level", choices=_LOG_LEVEL_NAMES, default=None, - help="Verbosity for provider-discovery and server diagnostics (default: " - "$CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL or WARNING). DEBUG never logs an " + 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) diff --git a/contextual_orchestrator/model_discovery.py b/contextual_orchestrator/model_discovery.py index 8e4f4468c..001967ab8 100644 --- a/contextual_orchestrator/model_discovery.py +++ b/contextual_orchestrator/model_discovery.py @@ -51,8 +51,8 @@ # 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`` gated on ``CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL`` in -# ``__main__.main()``), matching this repo's existing per-module logger +# ``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__) diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 4af216088..89f998ecf 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -21,7 +21,7 @@ fetch to one shared `concurrent.futures.wait(futures, timeout=timeout)` deadline `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` / `CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL`, never logs +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 diff --git a/tests/test_discover_models_cli.py b/tests/test_discover_models_cli.py index c9c9774cd..b35726633 100644 --- a/tests/test_discover_models_cli.py +++ b/tests/test_discover_models_cli.py @@ -69,11 +69,10 @@ 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_and_environment() -> None: - """The subcommand accepts explicit verbosity with documented precedence.""" +def test_discover_models_log_level_flag_wins_over_verbose() -> None: + """The subcommand accepts an explicit verbosity selection.""" stdout = StringIO() with ( - patch.dict(os.environ, {"CONTEXTUAL_ORCHESTRATOR_LOG_LEVEL": "ERROR"}, clear=True), patch.object( sys, "argv",