UpworkAutomation / app /services /api_gate.py
ananyajoshi's picture
Fix audited pipeline, session isolation and proposal grounding issues
357ee36
Raw History Blame Contribute Delete
14 kB
"""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",
)