"""API Gate. Validates configuration before downstream screens are enabled. Surfaces the precise status strings from the build plan. Resolves and validates the configured text provider — one of OpenAI, Anthropic, Groq, or Gemini — by checking provider support, API-key presence, and model presence. Configuration-level checks only: no expensive or live API calls are made here unless ``live=True`` is requested. A lightweight format-only placeholder check is applied to the Anthropic key specifically (``sk-ant-`` prefix); the other providers rely on the optional live reachability check. :func:`run_interview_capability_test` is the same machinery aimed at the interview provider. It is deliberately a separate entry point so it can be run when the user opens the interview and never at app start: the dossier route must keep working with no interview key configured. The API key value is never logged, displayed, or included in error output. """ from __future__ import annotations import inspect from dataclasses import dataclass from app import config as app_config from app.config import Settings, get_settings _SUPPORTED_PROVIDERS = ("anthropic", "openai", "groq", "gemini") # The interview leg is pinned to one provider. ``app.config`` owns the # canonical value; the literal fallback keeps this module importable against # a configuration that predates the constant. INTERVIEW_PROVIDER: str = ( str(getattr(app_config, "INTERVIEW_PROVIDER", "groq") or "").strip().lower() or "groq" ) class ApiGateError(str): """Status strings returned by the gate. Mirrors the build plan exactly.""" NO_API_ADDED = "NO API ADDED" INVALID_API_KEY = "INVALID API KEY" # The key works but may not use this model (HTTP 403): a plan, billing # or organisation-verification gate, which re-copying the key won't fix. ACCESS_DENIED = "API ACCESS DENIED" API_QUOTA_EXCEEDED = "API QUOTA EXCEEDED" API_CONNECTION_FAILED = "API CONNECTION FAILED" VISION_MODEL_NOT_AVAILABLE = "VISION MODEL NOT AVAILABLE" MODEL_NOT_CONFIGURED = "MODEL NOT CONFIGURED" # A model IS set, but the provider did not accept it (retired or # misspelt name). Kept apart from MODEL_NOT_CONFIGURED, which means no # model was set at all. MODEL_NOT_ACCEPTED = "MODEL NOT ACCEPTED" API_OK = "API OK" @dataclass(frozen=True) class CapabilityTestResult: status: str model: str | None = None provider: str | None = None @property def ok(self) -> bool: return self.status == ApiGateError.API_OK @property def error(self) -> str | None: return None if self.ok else self.status def _resolve_provider(provider: str | None, settings: Settings) -> str: raw = provider if provider is not None else settings.llm_provider return (raw or "").strip().lower() def _resolve_api_key(api_key: str | None, provider: str, settings: Settings) -> str | None: if api_key is not None: key = api_key.strip() return key or None if provider == "anthropic": return settings.anthropic_api_key if provider == "openai": return settings.openai_api_key if provider == "groq": return settings.groq_api_key if provider == "gemini": return settings.gemini_api_key return None def _resolve_model(model: str | None, provider: str, settings: Settings) -> str | None: if model is not None: m = model.strip() return m or None if provider == "anthropic": return settings.anthropic_model if provider == "openai": return settings.openai_model if provider == "groq": return settings.groq_model if provider == "gemini": return settings.gemini_model return None def _anthropic_key_format_ok(api_key: str) -> bool: """Cheap structural check — no network call. The Anthropic console issues keys prefixed with `sk-ant-`. This is a format sanity check only; it does not prove the key is live. """ key = api_key.strip() return key.startswith("sk-ant-") and len(key) >= 20 def run_capability_test( api_key: str | None = None, model: str | None = None, provider: str | None = None, settings: Settings | None = None, *, live: bool = False, route_provider: bool = False, task_name: str = "api_check", ) -> CapabilityTestResult: """Run configuration-level checks against the resolved provider. Returns a CapabilityTestResult whose `status` is one of the build-plan status strings. Never raises on bad input; never echoes the API key. When ``live=True``, after the offline checks succeed, route through :func:`app.services.llm_client.call_text_llm` with task ``api_check`` and the prompt ``"Reply with exactly: API OK"``. The live response only has to be received — content matching is tolerant — so this works as a lightweight reachability check. ``route_provider`` aims that live ping at the resolved provider instead of whichever one Settings has selected. It defaults to False so the Setup gate keeps testing exactly what it tested before. """ settings = settings if settings is not None else get_settings() resolved_provider = _resolve_provider(provider, settings) if not resolved_provider: return CapabilityTestResult(status=ApiGateError.MODEL_NOT_CONFIGURED) if resolved_provider not in _SUPPORTED_PROVIDERS: return CapabilityTestResult( status=ApiGateError.MODEL_NOT_CONFIGURED, provider=resolved_provider, ) resolved_key = _resolve_api_key(api_key, resolved_provider, settings) if not resolved_key: return CapabilityTestResult( status=ApiGateError.NO_API_ADDED, provider=resolved_provider, ) resolved_model = _resolve_model(model, resolved_provider, settings) if not resolved_model: return CapabilityTestResult( status=ApiGateError.MODEL_NOT_CONFIGURED, provider=resolved_provider, ) if resolved_provider == "anthropic" and not _anthropic_key_format_ok(resolved_key): return CapabilityTestResult( status=ApiGateError.INVALID_API_KEY, provider=resolved_provider, model=resolved_model, ) if live: return _run_live_check( api_key=resolved_key, model=resolved_model, provider=resolved_provider, settings=settings, route_provider=route_provider, task_name=task_name, ) return CapabilityTestResult( status=ApiGateError.API_OK, provider=resolved_provider, model=resolved_model, ) # --------------------------------------------------------------------------- # Live reachability check # --------------------------------------------------------------------------- _LIVE_STATUS_MAP: dict[str, str] = { # llm_client.STATUS_* → ApiGateError.* "ok": ApiGateError.API_OK, "no_api": ApiGateError.NO_API_ADDED, "invalid_key": ApiGateError.INVALID_API_KEY, "access_denied": ApiGateError.ACCESS_DENIED, "quota": ApiGateError.API_QUOTA_EXCEEDED, # A context-window overflow is a prompt-size problem, NOT a billing # quota; it must never be reported as "quota exceeded". For the tiny # api_check ping it also proves the key/endpoint work, so the gate # treats it as reachable. "context_overflow": ApiGateError.API_OK, "connection": ApiGateError.API_CONNECTION_FAILED, # The ping went out with the model that is set and the provider turned # it down (404 / model_not_found) — not the same as no model being set. "model_missing": ApiGateError.MODEL_NOT_ACCEPTED, "unsupported_provider": ApiGateError.MODEL_NOT_CONFIGURED, # parse_error and generic exception both indicate the call reached # the provider but didn't come back clean — treat as connection issue. "parse_error": ApiGateError.API_CONNECTION_FAILED, # A truncated ping reply still proves the provider is reachable and the # key works (the api_check call only needs *a* response). "truncated": ApiGateError.API_OK, "exception": ApiGateError.API_CONNECTION_FAILED, "skipped": ApiGateError.API_CONNECTION_FAILED, } def _supported_call_overrides() -> frozenset[str]: """Per-call override arguments ``call_text_llm`` accepts.""" # Import lazily to avoid a circular import at module load. from app.services import llm_client try: params = inspect.signature(llm_client.call_text_llm).parameters except (TypeError, ValueError): # defensive: never break the gate return frozenset() return frozenset({"provider", "model"} & set(params)) def _run_live_check( *, api_key: str, model: str, provider: str, settings: Settings, route_provider: bool = False, task_name: str = "api_check", ) -> CapabilityTestResult: """Lightweight real-API check via the central llm_client. With ``route_provider`` the ping is sent to ``provider`` rather than the one selected in Settings, so a check for one provider can run while a different one stays configured for the rest of the app. """ # Import lazily to avoid a circular import at module load. from app.services import llm_client extra: dict[str, str] = {} if route_provider: overrides = _supported_call_overrides() if "provider" not in overrides: # Without per-call routing the ping would test the configured # provider and the answer would be reported as if it came from # this one — a rejected key elsewhere would wrongly block this # provider. The offline checks above stand on their own. return CapabilityTestResult( status=ApiGateError.API_OK, provider=provider, model=model, ) extra["provider"] = provider if "model" in overrides: # Ping the exact model this check reports on, so a retired or # misspelt name surfaces here rather than mid-flow. extra["model"] = model result = llm_client.call_text_llm( task_name=task_name, system_prompt="", user_prompt="Reply with exactly: API OK", expected_json=False, max_tokens=16, settings=settings, **extra, ) if result.success: return CapabilityTestResult( status=ApiGateError.API_OK, provider=provider, model=model, ) mapped = _LIVE_STATUS_MAP.get(result.status, ApiGateError.API_CONNECTION_FAILED) if result.status == "model_missing" and not getattr(result, "used_api", True): # The client found no model to send, so nothing was rejected. mapped = ApiGateError.MODEL_NOT_CONFIGURED return CapabilityTestResult( status=mapped, provider=provider, model=model, ) # --------------------------------------------------------------------------- # Interview provider check # --------------------------------------------------------------------------- def _resolve_interview_model(model: str | None, settings: Settings) -> str | None: """Return the model the interview will use. The dedicated ``interview_model`` setting wins; otherwise the interview provider's own configured model is used. """ if model is not None: stripped = model.strip() return stripped or None configured = getattr(settings, "interview_model", None) if isinstance(configured, str) and configured.strip(): return configured.strip() return _resolve_model(None, INTERVIEW_PROVIDER, settings) def run_interview_capability_test( api_key: str | None = None, model: str | None = None, settings: Settings | None = None, *, live: bool = False, prefer: str | None = None, ) -> CapabilityTestResult: """Check the interview provider, for use on entry to the interview. Kept apart from :func:`run_capability_test`, which gates the whole app at Setup. This one must never run at app start: a dossier user with no interview key configured has to reach every other screen unaffected. The verdict uses the same vocabulary as the Setup gate, so the two failures that read alike to a user stay apart — an exhausted free tier comes back as ``API_QUOTA_EXCEEDED`` while a rejected key comes back as ``INVALID_API_KEY``. With ``live=False`` (the default) only the offline checks run: provider supported, key present, model set. ``api_key`` overrides the offline key-presence check only; as on the Setup check, the live ping resolves the key from Settings. """ settings = settings if settings is not None else get_settings() # Probe whatever the interview will actually run on: a Groq key if one # is set, otherwise the provider from Setup (see # ``llm_client.interview_route``). With neither, the Groq slot is # checked, which reports the missing key. from app.services import llm_client # lazy: avoids an import cycle route = llm_client.interview_route(settings, prefer=prefer) if route is None: provider = INTERVIEW_PROVIDER resolved_model = _resolve_interview_model(model, settings) else: provider, route_model = route explicit = (model or "").strip() if model is not None else "" resolved_model = ( explicit or route_model or _resolve_model(None, provider, settings) ) return run_capability_test( api_key=api_key, model=resolved_model, provider=provider, settings=settings, live=live, route_provider=True, # Logged apart from the Setup check, so a failed interview key is # never reported as a problem with the main API. task_name="interview_check", )