agent_search_gateway.config

  1"""TOML and environment configuration resolution."""
  2
  3import copy
  4import math
  5import tomllib
  6from collections.abc import Mapping
  7from dataclasses import dataclass, field
  8from pathlib import Path
  9from types import MappingProxyType
 10from urllib.parse import urlsplit
 11
 12from .errors import ConfigFailure, ErrorCode
 13from .models import LLMInvocation, RetryPolicy
 14from .observability import SecretValue
 15from .providers.academic.registry import (
 16    AcademicProviderRegistration,
 17    AcademicProviderRegistry,
 18    OAResolverRegistration,
 19    OAResolverRegistry,
 20    Requirement,
 21)
 22from .providers.registry import ProviderRegistry, WebProviderRegistration
 23
 24_WEB_SHARED_KEYS = frozenset({"enable_search", "enable_fetch", "api_key_env", "max_concurrency"})
 25_ACADEMIC_SHARED_KEYS = frozenset(
 26    {"enabled", "max_concurrency", "api_key_env", "contact_email_env"}
 27)
 28_ACADEMIC_RESERVED_OPTIONS = frozenset({"executor", "api_key", "contact_email", "name"})
 29
 30
 31@dataclass(frozen=True, slots=True)
 32class ResolvedWebProviderConfig:
 33    name: str
 34    enable_search: bool
 35    enable_fetch: bool
 36    max_concurrency: int
 37    api_key_env: str | None
 38    secret: SecretValue | None
 39    options: Mapping[str, object]
 40
 41
 42@dataclass(frozen=True, slots=True)
 43class ResolvedWebProviderGroup:
 44    default_max_concurrency: int
 45    providers: tuple[ResolvedWebProviderConfig, ...]
 46
 47
 48def _config_error(message: str) -> ConfigFailure:
 49    return ConfigFailure(ErrorCode.CONFIG_ERROR, message)
 50
 51
 52def _require_mapping(value: object, label: str) -> Mapping[str, object]:
 53    if not isinstance(value, dict):
 54        raise _config_error(f"{label} must be a table")
 55    return value
 56
 57
 58def _positive_int(value: object, label: str) -> int:
 59    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
 60        raise _config_error(f"{label} must be a positive integer")
 61    return value
 62
 63
 64def _optional_bool(table: Mapping[str, object], key: str) -> bool:
 65    value = table.get(key, False)
 66    if not isinstance(value, bool):
 67        raise _config_error(f"{key} must be a boolean")
 68    return value
 69
 70
 71def _validate_options(
 72    registration: WebProviderRegistration,
 73    options: Mapping[str, object],
 74) -> None:
 75    unknown = set(options) - registration.allowed_config_keys
 76    if unknown:
 77        names = ", ".join(sorted(unknown))
 78        raise _config_error(f"unknown config key(s) for {registration.name}: {names}")
 79
 80
 81def _resolve_one_web_provider(
 82    name: str,
 83    table: Mapping[str, object],
 84    default_max_concurrency: int,
 85    registry: ProviderRegistry,
 86    environ: Mapping[str, str],
 87) -> ResolvedWebProviderConfig:
 88    enable_search = _optional_bool(table, "enable_search")
 89    enable_fetch = _optional_bool(table, "enable_fetch")
 90    options = {key: value for key, value in table.items() if key not in _WEB_SHARED_KEYS}
 91    registration = registry.get(name)
 92
 93    if registration is not None:
 94        _validate_options(registration, options)
 95
 96    max_concurrency = _positive_int(
 97        table.get("max_concurrency", default_max_concurrency),
 98        f"web provider {name} max_concurrency",
 99    )
100
101    if not enable_search and not enable_fetch:
102        return ResolvedWebProviderConfig(
103            name=name,
104            enable_search=False,
105            enable_fetch=False,
106            max_concurrency=max_concurrency,
107            api_key_env=None,
108            secret=None,
109            options=MappingProxyType(dict(options)),
110        )
111
112    if registration is None:
113        raise _config_error(f"unknown enabled web provider: {name}")
114    if enable_search and not registration.capabilities.search:
115        raise _config_error(f"web provider {name} does not support search")
116    if enable_fetch and not registration.capabilities.fetch:
117        raise _config_error(f"web provider {name} does not support fetch")
118
119    if not registration.requires_api_key:
120        if "api_key_env" in table:
121            raise _config_error(f"web provider {name} does not accept api_key_env")
122        return ResolvedWebProviderConfig(
123            name=name,
124            enable_search=enable_search,
125            enable_fetch=enable_fetch,
126            max_concurrency=max_concurrency,
127            api_key_env=None,
128            secret=None,
129            options=MappingProxyType(dict(options)),
130        )
131
132    api_key_env = table.get("api_key_env")
133    if not isinstance(api_key_env, str) or not api_key_env.strip():
134        raise _config_error(f"web provider {name} requires api_key_env")
135    secret_text = environ.get(api_key_env)
136    if not secret_text:
137        raise _config_error(f"environment variable {api_key_env} is required")
138
139    return ResolvedWebProviderConfig(
140        name=name,
141        enable_search=enable_search,
142        enable_fetch=enable_fetch,
143        max_concurrency=max_concurrency,
144        api_key_env=api_key_env,
145        secret=SecretValue(secret_text),
146        options=MappingProxyType(dict(options)),
147    )
148
149
150def resolve_web_provider_config(
151    data: Mapping[str, object],
152    registry: ProviderRegistry,
153    environ: Mapping[str, str],
154) -> ResolvedWebProviderGroup:
155    web_table = _require_mapping(data.get("web_providers", {}), "web_providers")
156    default_max_concurrency = _positive_int(
157        web_table.get("default_max_concurrency", 3),
158        "web_providers.default_max_concurrency",
159    )
160
161    providers: list[ResolvedWebProviderConfig] = []
162    for name, value in web_table.items():
163        if name == "default_max_concurrency":
164            continue
165        provider_table = _require_mapping(value, f"web_providers.{name}")
166        providers.append(
167            _resolve_one_web_provider(
168                name,
169                provider_table,
170                default_max_concurrency,
171                registry,
172                environ,
173            )
174        )
175    return ResolvedWebProviderGroup(default_max_concurrency, tuple(providers))
176
177
178@dataclass(frozen=True, slots=True)
179class ResolvedAcademicProviderConfig:
180    name: str
181    enabled: bool
182    max_concurrency: int
183    api_key_env: str | None
184    api_key: SecretValue | None
185    contact_email_env: str | None
186    contact_email: SecretValue | None
187    options: Mapping[str, object]
188
189
190@dataclass(frozen=True, slots=True)
191class ResolvedAcademicProviderGroup:
192    default_max_concurrency: int = 3
193    providers: tuple[ResolvedAcademicProviderConfig, ...] = ()
194
195
196@dataclass(frozen=True, slots=True)
197class ResolvedOAResolverConfig:
198    name: str
199    api_key_env: str | None
200    api_key: SecretValue | None
201    contact_email_env: str | None
202    contact_email: SecretValue | None
203    options: Mapping[str, object]
204
205
206def _enabled(table: Mapping[str, object], label: str) -> bool:
207    value = table.get("enabled", False)
208    if not isinstance(value, bool):
209        raise _config_error(f"{label}.enabled must be a boolean")
210    return value
211
212
213def _validate_http_url(value: object, label: str) -> None:
214    if not isinstance(value, str) or not value.strip():
215        raise _config_error(f"{label} must be a non-empty HTTP(S) URL")
216    try:
217        parsed = urlsplit(value)
218    except ValueError as exc:
219        raise _config_error(f"{label} must be a non-empty HTTP(S) URL") from exc
220    if parsed.scheme not in {"http", "https"} or parsed.hostname is None:
221        raise _config_error(f"{label} must be a non-empty HTTP(S) URL")
222
223
224def _validate_academic_options(
225    name: str,
226    registration: AcademicProviderRegistration | OAResolverRegistration,
227    options: Mapping[str, object],
228) -> None:
229    reserved = set(options) & _ACADEMIC_RESERVED_OPTIONS
230    if reserved:
231        names = ", ".join(sorted(reserved))
232        raise _config_error(f"reserved config key(s) for {name}: {names}")
233    unknown = set(options) - registration.allowed_config_keys
234    if unknown:
235        names = ", ".join(sorted(unknown))
236        raise _config_error(f"unknown config key(s) for {name}: {names}")
237    if "api_url" in options:
238        _validate_http_url(options["api_url"], f"{name}.api_url")
239
240
241def _resolve_environment_value(
242    table: Mapping[str, object],
243    *,
244    key: str,
245    requirement: Requirement,
246    label: str,
247    environ: Mapping[str, str],
248) -> tuple[str | None, SecretValue | None]:
249    if requirement == "none":
250        if key in table:
251            raise _config_error(f"{label} does not accept {key}")
252        return None, None
253    if key not in table:
254        if requirement == "required":
255            raise _config_error(f"{label} requires {key}")
256        return None, None
257    env_name = table[key]
258    if not isinstance(env_name, str) or not env_name.strip():
259        raise _config_error(f"{label}.{key} must be a non-empty string")
260    resolved_value = environ.get(env_name)
261    if not resolved_value:
262        raise _config_error(f"environment variable {env_name} is required")
263    return env_name, SecretValue(resolved_value)
264
265
266def resolve_academic_provider_config(
267    data: Mapping[str, object],
268    registry: AcademicProviderRegistry,
269    environ: Mapping[str, str],
270) -> ResolvedAcademicProviderGroup:
271    group = _require_mapping(data.get("academic_providers", {}), "academic_providers")
272    default_limit = _positive_int(
273        group.get("default_max_concurrency", 3),
274        "academic_providers.default_max_concurrency",
275    )
276    providers: list[ResolvedAcademicProviderConfig] = []
277    for name, raw_value in group.items():
278        if name == "default_max_concurrency":
279            continue
280        table = _require_mapping(raw_value, f"academic_providers.{name}")
281        registration = registry.get(name)
282        if registration is None:
283            raise _config_error(f"unknown academic provider: {name}")
284        enabled = _enabled(table, f"academic_providers.{name}")
285        options = {key: value for key, value in table.items() if key not in _ACADEMIC_SHARED_KEYS}
286        _validate_academic_options(name, registration, options)
287        limit = _positive_int(
288            table.get("max_concurrency", default_limit),
289            f"academic provider {name} max_concurrency",
290        )
291        if not enabled:
292            for key, requirement in (
293                ("api_key_env", registration.authentication),
294                ("contact_email_env", registration.contact),
295            ):
296                if requirement == "none" and key in table:
297                    raise _config_error(f"academic provider {name} does not accept {key}")
298                if key in table and (
299                    not isinstance(table[key], str) or not str(table[key]).strip()
300                ):
301                    raise _config_error(
302                        f"academic provider {name}.{key} must be a non-empty string"
303                    )
304            providers.append(
305                ResolvedAcademicProviderConfig(
306                    name,
307                    False,
308                    limit,
309                    None,
310                    None,
311                    None,
312                    None,
313                    MappingProxyType(dict(options)),
314                )
315            )
316            continue
317        auth_env, auth_value = _resolve_environment_value(
318            table,
319            key="api_key_env",
320            requirement=registration.authentication,
321            label=f"academic provider {name}",
322            environ=environ,
323        )
324        contact_env, contact_value = _resolve_environment_value(
325            table,
326            key="contact_email_env",
327            requirement=registration.contact,
328            label=f"academic provider {name}",
329            environ=environ,
330        )
331        providers.append(
332            ResolvedAcademicProviderConfig(
333                name,
334                enabled,
335                limit,
336                auth_env,
337                auth_value,
338                contact_env,
339                contact_value,
340                MappingProxyType(dict(options)),
341            )
342        )
343    return ResolvedAcademicProviderGroup(default_limit, tuple(providers))
344
345
346def resolve_oa_resolver_config(
347    data: Mapping[str, object],
348    registry: OAResolverRegistry,
349    environ: Mapping[str, str],
350) -> ResolvedOAResolverConfig | None:
351    group = _require_mapping(data.get("oa_resolvers", {}), "oa_resolvers")
352    resolved: ResolvedOAResolverConfig | None = None
353    for name, raw_value in group.items():
354        table = _require_mapping(raw_value, f"oa_resolvers.{name}")
355        registration = registry.get(name)
356        if registration is None:
357            raise _config_error(f"unknown OA resolver: {name}")
358        enabled = _enabled(table, f"oa_resolvers.{name}")
359        options = {key: value for key, value in table.items() if key not in _ACADEMIC_SHARED_KEYS}
360        _validate_academic_options(name, registration, options)
361        if not enabled:
362            continue
363        if resolved is not None:
364            raise _config_error("only one OA resolver may be enabled")
365        auth_env, auth_value = _resolve_environment_value(
366            table,
367            key="api_key_env",
368            requirement=registration.authentication,
369            label=f"OA resolver {name}",
370            environ=environ,
371        )
372        contact_env, contact_value = _resolve_environment_value(
373            table,
374            key="contact_email_env",
375            requirement=registration.contact,
376            label=f"OA resolver {name}",
377            environ=environ,
378        )
379        resolved = ResolvedOAResolverConfig(
380            name,
381            auth_env,
382            auth_value,
383            contact_env,
384            contact_value,
385            MappingProxyType(dict(options)),
386        )
387    return resolved
388
389
390@dataclass(frozen=True, slots=True)
391class LLMProviderConfig:
392    name: str
393    protocol: str
394    api_endpoint: str
395    api_url: str
396    api_key_env: str
397    secret: SecretValue
398    max_concurrency: int
399
400
401@dataclass(frozen=True, slots=True)
402class ResolvedLLMConfig:
403    default_max_concurrency: int
404    providers: tuple[LLMProviderConfig, ...]
405    search_invocations: tuple[LLMInvocation, ...]
406    judge: LLMInvocation
407    safety: LLMInvocation
408    content_clean: LLMInvocation
409    focus_summary: LLMInvocation
410
411
412@dataclass(frozen=True, slots=True)
413class ResolvedConfig:
414    web: ResolvedWebProviderGroup
415    llm: ResolvedLLMConfig
416    retry: RetryPolicy
417    academic: ResolvedAcademicProviderGroup = field(default_factory=ResolvedAcademicProviderGroup)
418    oa_resolver: ResolvedOAResolverConfig | None = None
419
420
421def _required_string(table: Mapping[str, object], key: str, label: str) -> str:
422    value = table.get(key)
423    if not isinstance(value, str) or not value.strip():
424        raise _config_error(f"{label}.{key} must be a non-empty string")
425    return value
426
427
428def _first_defined(key: str, *tables: Mapping[str, object]) -> object | None:
429    for table in tables:
430        if key in table:
431            return table[key]
432    return None
433
434
435def _resolve_extra_body(*tables: Mapping[str, object]) -> dict[str, object]:
436    value = _first_defined("extra_body", *tables)
437    if value is None:
438        return {}
439    if not isinstance(value, dict):
440        raise _config_error("LLM extra_body must be a table")
441    return copy.deepcopy(value)
442
443
444def _resolve_invocation(
445    label: str,
446    scopes: tuple[Mapping[str, object], ...],
447    *,
448    require_explicit_provider: bool = False,
449) -> LLMInvocation:
450    provider_value = (
451        scopes[0].get("provider")
452        if require_explicit_provider
453        else _first_defined("provider", *scopes)
454    )
455    model_value = _first_defined("model", *scopes)
456    if not isinstance(provider_value, str) or not provider_value.strip():
457        raise _config_error(f"{label} requires provider")
458    if not isinstance(model_value, str) or not model_value.strip():
459        raise _config_error(f"{label} requires model")
460    return LLMInvocation(
461        provider=provider_value,
462        model=model_value,
463        extra_body=_resolve_extra_body(*scopes),
464    )
465
466
467def _search_invocations(
468    data: Mapping[str, object],
469    global_table: Mapping[str, object],
470) -> tuple[LLMInvocation, ...]:
471    search_table = _require_mapping(data.get("search_llm", {}), "search_llm")
472    raw_entries = search_table.get("providers", [])
473    if not isinstance(raw_entries, list):
474        raise _config_error("search_llm.providers must be an array of tables")
475
476    resolved: list[LLMInvocation] = []
477    for index, raw_entry in enumerate(raw_entries):
478        entry = _require_mapping(raw_entry, f"search_llm.providers[{index}]")
479        resolved.append(
480            _resolve_invocation(
481                f"search_llm.providers[{index}]",
482                (entry, global_table),
483                require_explicit_provider=True,
484            )
485        )
486    return tuple(resolved)
487
488
489def _fetch_invocations(
490    data: Mapping[str, object],
491    global_table: Mapping[str, object],
492) -> tuple[LLMInvocation, LLMInvocation, LLMInvocation, LLMInvocation]:
493    fetch_table = _require_mapping(data.get("fetch_llm", {}), "fetch_llm")
494    names = ("judge", "safety", "content_clean", "focus_summary")
495    invocations: list[LLMInvocation] = []
496    for name in names:
497        stage_table = _require_mapping(fetch_table.get(name, {}), f"fetch_llm.{name}")
498        invocations.append(
499            _resolve_invocation(
500                f"fetch_llm.{name}",
501                (stage_table, fetch_table, global_table),
502            )
503        )
504    return invocations[0], invocations[1], invocations[2], invocations[3]
505
506
507def _resolve_referenced_llm_providers(
508    llm_table: Mapping[str, object],
509    referenced: set[str],
510    default_max_concurrency: int,
511    environ: Mapping[str, str],
512) -> tuple[LLMProviderConfig, ...]:
513    providers: list[LLMProviderConfig] = []
514    for name, raw_value in llm_table.items():
515        if name == "default_max_concurrency" or name not in referenced:
516            continue
517        table = _require_mapping(raw_value, f"llm_providers.{name}")
518        protocol = _required_string(table, "protocol", f"llm_providers.{name}")
519        endpoint = _required_string(table, "api_endpoint", f"llm_providers.{name}")
520        if protocol != "openai" or endpoint != "chat_completions":
521            raise _config_error(f"unsupported LLM provider protocol/endpoint: {name}")
522        api_url = _required_string(table, "api_url", f"llm_providers.{name}")
523        api_key_env = _required_string(table, "api_key_env", f"llm_providers.{name}")
524        secret_text = environ.get(api_key_env)
525        if not secret_text:
526            raise _config_error(f"environment variable {api_key_env} is required")
527        max_concurrency = _positive_int(
528            table.get("max_concurrency", default_max_concurrency),
529            f"llm provider {name} max_concurrency",
530        )
531        providers.append(
532            LLMProviderConfig(
533                name=name,
534                protocol=protocol,
535                api_endpoint=endpoint,
536                api_url=api_url,
537                api_key_env=api_key_env,
538                secret=SecretValue(secret_text),
539                max_concurrency=max_concurrency,
540            )
541        )
542
543    configured = {provider.name for provider in providers}
544    missing = referenced - configured
545    if missing:
546        raise _config_error(f"unknown referenced LLM provider(s): {', '.join(sorted(missing))}")
547    return tuple(providers)
548
549
550def resolve_llm_config(
551    data: Mapping[str, object],
552    environ: Mapping[str, str],
553) -> ResolvedLLMConfig:
554    llm_table = _require_mapping(data.get("llm_providers", {}), "llm_providers")
555    default_max_concurrency = _positive_int(
556        llm_table.get("default_max_concurrency", 2),
557        "llm_providers.default_max_concurrency",
558    )
559    global_table = _require_mapping(data.get("global_default_llm", {}), "global_default_llm")
560    search_invocations = _search_invocations(data, global_table)
561    judge, safety, content_clean, focus_summary = _fetch_invocations(data, global_table)
562    all_invocations = (*search_invocations, judge, safety, content_clean, focus_summary)
563    providers = _resolve_referenced_llm_providers(
564        llm_table,
565        {invocation.provider for invocation in all_invocations},
566        default_max_concurrency,
567        environ,
568    )
569    return ResolvedLLMConfig(
570        default_max_concurrency=default_max_concurrency,
571        providers=providers,
572        search_invocations=search_invocations,
573        judge=judge,
574        safety=safety,
575        content_clean=content_clean,
576        focus_summary=focus_summary,
577    )
578
579
580def _positive_float(value: object, label: str) -> float:
581    if (
582        isinstance(value, bool)
583        or not isinstance(value, int | float)
584        or not math.isfinite(value)
585        or value <= 0
586    ):
587        raise _config_error(f"{label} must be a positive number")
588    return float(value)
589
590
591def resolve_retry_policy(data: Mapping[str, object]) -> RetryPolicy:
592    retry_table = _require_mapping(data.get("retry", {}), "retry")
593    return RetryPolicy(
594        max_attempts=_positive_int(retry_table.get("max_attempts", 3), "retry.max_attempts"),
595        base_delay_seconds=_positive_float(
596            retry_table.get("base_delay_seconds", 0.25),
597            "retry.base_delay_seconds",
598        ),
599        max_delay_seconds=_positive_float(
600            retry_table.get("max_delay_seconds", 2.0),
601            "retry.max_delay_seconds",
602        ),
603        request_timeout_seconds=_positive_float(
604            retry_table.get("request_timeout_seconds", 30.0),
605            "retry.request_timeout_seconds",
606        ),
607    )
608
609
610def resolve_config(
611    data: Mapping[str, object],
612    registry: ProviderRegistry,
613    environ: Mapping[str, str],
614    *,
615    academic_registry: AcademicProviderRegistry | None = None,
616    oa_resolver_registry: OAResolverRegistry | None = None,
617) -> ResolvedConfig:
618    academic = (
619        resolve_academic_provider_config(data, academic_registry, environ)
620        if academic_registry is not None
621        else ResolvedAcademicProviderGroup()
622    )
623    oa_resolver = (
624        resolve_oa_resolver_config(data, oa_resolver_registry, environ)
625        if oa_resolver_registry is not None
626        else None
627    )
628    return ResolvedConfig(
629        web=resolve_web_provider_config(data, registry, environ),
630        llm=resolve_llm_config(data, environ),
631        retry=resolve_retry_policy(data),
632        academic=academic,
633        oa_resolver=oa_resolver,
634    )
635
636
637def load_toml(path: Path) -> dict[str, object]:
638    try:
639        with path.open("rb") as handle:
640            return tomllib.load(handle)
641    except (OSError, tomllib.TOMLDecodeError) as exc:
642        raise _config_error(f"failed to load config: {exc}") from exc