Blame for app/services/api_gate.py

"""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",
    )