fix(discovery): bound OpenRouter free-model endpoint fan-out to one deadline

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,

contextual_orchestrator/model_discovery.py

@@ -13,19 +13,21 @@
 
 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
 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 (
@@ -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