diff --git a/backend/app/api/routes/cron.py b/backend/app/api/routes/cron.py index c31288ce8..49e31bfe5 100644 --- a/backend/app/api/routes/cron.py +++ b/backend/app/api/routes/cron.py @@ -1,4 +1,5 @@ import logging +from typing import Any import sentry_sdk from fastapi import APIRouter, Depends @@ -8,6 +9,7 @@ from app.api.permissions import Permission, require_permission from app.core.config import settings from app.crud.evaluations import process_all_pending_evaluations +from app.services.health_probes import run_health_probe_tick from app.services.job_monitoring import monitor_pending_jobs logger = logging.getLogger(__name__) @@ -15,13 +17,11 @@ router = APIRouter(tags=["Cron"]) EVALUATION_CRON_MONITOR_CONFIG: MonitorConfig = { - # Expected cadence: a check-in every CRON_INTERVAL_MINUTES minutes. "schedule": { "type": "interval", "value": settings.CRON_INTERVAL_MINUTES, "unit": "minute", }, - # Timezone for the schedule (only affects crontab-style schedules). "timezone": "UTC", # Grace period (minutes) before a late check-in is marked as missed. "checkin_margin": 2, @@ -123,6 +123,28 @@ async def evaluation_cron_job( raise +@router.get( + "/cron/health-probes", + include_in_schema=False, + dependencies=[Depends(require_permission(Permission.SUPERUSER))], +) +def health_probes_cron_job() -> dict[str, Any]: + # Runs synchronously (no Celery task of its own) — the probe rides the + # existing LLM_API job pipeline via `start_job`. + logger.info("[health_probes_cron_job] Cron job invoked") + try: + result = run_health_probe_tick() + logger.info( + f"[health_probes_cron_job] Tick complete | job_id: {result.get('job_id')}, " + f"probe_index: {result.get('probe_index')}" + ) + return result + except Exception as e: + logger.error(f"[health_probes_cron_job] Tick failed: {e}", exc_info=True) + sentry_sdk.capture_exception(e) + raise + + @router.get( "/cron/pending-jobs", include_in_schema=False, diff --git a/backend/app/assets/health_probe_hello.ogg b/backend/app/assets/health_probe_hello.ogg new file mode 100644 index 000000000..933c0ef6f Binary files /dev/null and b/backend/app/assets/health_probe_hello.ogg differ diff --git a/backend/app/celery/tasks/job_execution.py b/backend/app/celery/tasks/job_execution.py index 000c6025a..fcf7b1cfe 100644 --- a/backend/app/celery/tasks/job_execution.py +++ b/backend/app/celery/tasks/job_execution.py @@ -18,7 +18,6 @@ from opentelemetry import context as otel_context from opentelemetry import trace from opentelemetry.propagate import extract - from app.celery.celery_app import celery_app from app.celery.utils import gevent_timeout from app.core.config import settings diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 8ebc37b53..15ea78a1c 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -193,6 +193,10 @@ def AWS_S3_BUCKET(self) -> str: EVAL_FAST_STALL_THRESHOLD_MINUTES: int = 15 PENDING_JOB_QUERY_TIMEOUT_MS: int = 1000 + HEALTH_PROBE_ORG_ID: int | None = None + HEALTH_PROBE_PROJECT_ID: int | None = None + HEALTH_PROBE_INTERVAL_MINUTES: int = 3 + # AI-assisted prompt improvement settings. # See docs/srd-ai-prompt-improvement.md for the full design rationale. # Platform-owned Anthropic key shared by every org/project for this feature, diff --git a/backend/app/core/telemetry.py b/backend/app/core/telemetry.py index 99d2fc959..95f64f13f 100644 --- a/backend/app/core/telemetry.py +++ b/backend/app/core/telemetry.py @@ -190,11 +190,15 @@ def setup_telemetry(service_name: str | None = None) -> None: resource = _build_resource(service_name) tracer_provider = TracerProvider(resource=resource) - # Bridge OTel spans into Sentry as Sentry transactions and spans, with full attribute and error capture. if settings.SENTRY_DSN: - from sentry_sdk.integrations.opentelemetry import SentrySpanProcessor + from opentelemetry.propagate import set_global_textmap + from sentry_sdk.integrations.opentelemetry import ( + SentryPropagator, + SentrySpanProcessor, + ) tracer_provider.add_span_processor(SentrySpanProcessor()) + set_global_textmap(SentryPropagator()) trace.set_tracer_provider(tracer_provider) diff --git a/backend/app/services/health_probes.py b/backend/app/services/health_probes.py new file mode 100644 index 000000000..be6b4a7f8 --- /dev/null +++ b/backend/app/services/health_probes.py @@ -0,0 +1,262 @@ +import base64 +import logging +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Literal +from uuid import UUID + +import redis +from sentry_sdk.crons import MonitorStatus, capture_checkin +from sentry_sdk.types import MonitorConfig +from sqlmodel import Session + +from app.core.config import settings +from app.core.db import engine +from app.crud.jobs import JobCrud +from app.models.job import JobStatus +from app.models.llm.constants import CompletionType +from app.models.llm.request import ( + AudioContent, + AudioInput, + ConfigBlob, + KaapiCompletionConfig, + LLMCallConfig, + LLMCallRequest, + QueryParams, +) +from app.services.llm.jobs import start_job + +logger = logging.getLogger(__name__) + +_PROBE_INPUT = "ping" + +# ~1sec "Hello" clip used as the STT probe's input audio. +_STT_AUDIO_PATH = ( + Path(__file__).resolve().parents[1] / "assets" / "health_probe_hello.ogg" +) +_STT_AUDIO_MIME = "audio/ogg" +_stt_audio_b64: str | None = None + +HEALTH_PROBES_MONITOR_SLUG = "health-probes-cron-job" +HEALTH_PROBES_MONITOR_CONFIG: MonitorConfig = { + "schedule": { + "type": "interval", + "value": settings.HEALTH_PROBE_INTERVAL_MINUTES, + "unit": "minute", + }, + "timezone": "UTC", + "checkin_margin": 2, + "max_runtime": 2 * settings.HEALTH_PROBE_INTERVAL_MINUTES, + "failure_issue_threshold": 2, + "recovery_threshold": 1, +} + +# Rotation state lives in Redis, not a table — no CRUD, no tenant isolation needed. +_LAST_JOB_ID_KEY = "health_probe:last_job_id" +_INDEX_KEY = "health_probe:index" + +_redis_client: redis.Redis = redis.from_url(settings.REDIS_URL, decode_responses=True) + +Modality = Literal["text", "tts", "stt"] + + +@dataclass(frozen=True) +class Probe: + provider: str + model: str + modality: Modality + # Extra Kaapi params a provider's mapper requires beyond `model` + # (Sarvam TTS needs `language`, ElevenLabs TTS needs `voice`). + params: dict[str, Any] = field(default_factory=dict) + + +_PROBES: list[Probe] = [ + # Text + Probe(provider="openai", model="gpt-4o-mini", modality="text"), + Probe(provider="google", model="gemini-2.5-flash", modality="text"), + Probe(provider="google", model="gemini-2.5-pro", modality="text"), + Probe(provider="anthropic", model="claude-sonnet-4-6", modality="text"), + # TTS + Probe(provider="google", model="gemini-2.5-flash-preview-tts", modality="tts"), + Probe(provider="google", model="gemini-3.1-flash-tts-preview", modality="tts"), + Probe(provider="google", model="gemini-2.5-pro-preview-tts", modality="tts"), + Probe( + provider="sarvamai", + model="bulbul:v3", + modality="tts", + params={"language": "en-IN"}, + ), + Probe( + provider="elevenlabs", + model="eleven_v3", + modality="tts", + params={"voice": "Sarah"}, + ), + # STT + Probe(provider="google", model="gemini-2.5-pro", modality="stt"), + Probe(provider="google", model="gemini-2.5-flash", modality="stt"), + Probe(provider="google", model="gemini-3.1-pro-preview", modality="stt"), + Probe(provider="sarvamai", model="saaras:v3", modality="stt"), + Probe(provider="elevenlabs", model="scribe_v2", modality="stt"), +] + + +def _load_stt_audio_b64() -> str: + global _stt_audio_b64 + if _stt_audio_b64 is None: + with open(_STT_AUDIO_PATH, "rb") as f: + _stt_audio_b64 = base64.b64encode(f.read()).decode("ascii") + return _stt_audio_b64 + + +def _build_probe_request(probe: Probe, index: int) -> LLMCallRequest: + completion_type = { + "text": CompletionType.TEXT, + "tts": CompletionType.TTS, + "stt": CompletionType.STT, + }[probe.modality] + + kaapi_config = KaapiCompletionConfig( + provider=probe.provider, + type=completion_type, + params={"model": probe.model, **probe.params}, + ) + + if probe.modality == "stt": + query_input: str | AudioInput = AudioInput( + content=AudioContent( + format="base64", + value=_load_stt_audio_b64(), + mime_type=_STT_AUDIO_MIME, + ) + ) + else: + query_input = _PROBE_INPUT + + # No callback_url: the probe reads its result by polling `Job.status` on + # the next tick, so callback delivery (a best-effort side channel that + # never affects Job.status) buys nothing here. + return LLMCallRequest( + query=QueryParams(input=query_input), + config=LLMCallConfig(blob=ConfigBlob(completion=kaapi_config)), + request_metadata={ + "health_probe": True, + "probe_index": index, + "provider": probe.provider, + "modality": probe.modality, + "model": probe.model, + }, + ) + + +def _check_previous_probe(project_id: int) -> JobStatus | None: + # Missing key (first run, Redis eviction) means skip the check-in, not fail the tick. + try: + last_job_id = _redis_client.get(_LAST_JOB_ID_KEY) + except redis.RedisError as e: + logger.warning(f"[_check_previous_probe] Redis GET failed | error: {e}") + return None + + if last_job_id is None: + logger.warning( + f"[_check_previous_probe] {_LAST_JOB_ID_KEY} missing — skipping check-in" + ) + return None + + with Session(engine) as session: + job = JobCrud(session=session).get( + job_id=UUID(str(last_job_id)), project_id=project_id + ) + + if job is None: + logger.warning(f"[_check_previous_probe] Job not found | job_id: {last_job_id}") + return None + + return job.status + + +def _capture_previous_result_checkin(previous_status: JobStatus | None) -> None: + if previous_status is None: + return + # SUCCESS means the probe actually ran and returned; FAILED or a + # still-PENDING/PROCESSING job both mean a real failure. + monitor_status = ( + MonitorStatus.OK + if previous_status == JobStatus.SUCCESS + else MonitorStatus.ERROR + ) + capture_checkin( + monitor_slug=HEALTH_PROBES_MONITOR_SLUG, + status=monitor_status, + monitor_config=HEALTH_PROBES_MONITOR_CONFIG, + ) + + +def _claim_next_probe_index() -> int: + # Redis INCR is atomic, so two overlapping ticks can never claim the same + # slot (which would fire the same probe twice while skipping another). + try: + count = _redis_client.incr(_INDEX_KEY) + except redis.RedisError as e: + logger.warning(f"[_claim_next_probe_index] Redis INCR failed | error: {e}") + return 0 + + try: + return (int(count) - 1) % len(_PROBES) + except (TypeError, ValueError): + logger.warning(f"[_claim_next_probe_index] Non-integer index | raw: {count!r}") + return 0 + + +def _store_last_job_id(job_id: UUID) -> None: + try: + _redis_client.set(_LAST_JOB_ID_KEY, str(job_id)) + except redis.RedisError as e: + logger.error( + f"[_store_last_job_id] Redis SET failed | job_id: {job_id}, error: {e}" + ) + + +def run_health_probe_tick() -> dict[str, Any]: + """Check the previous probe's result, then fire the next one round-robin.""" + org_id = settings.HEALTH_PROBE_ORG_ID + project_id = settings.HEALTH_PROBE_PROJECT_ID + if org_id is None or project_id is None: + logger.warning( + "[run_health_probe_tick] Health probe org/project not configured — skipping" + ) + return { + "enqueued": False, + "job_id": None, + "probe_index": None, + "previous_job_status": None, + } + + previous_status = _check_previous_probe(project_id) + _capture_previous_result_checkin(previous_status) + + index = _claim_next_probe_index() + probe = _PROBES[index] + request = _build_probe_request(probe, index) + + with Session(engine) as session: + job_id = start_job( + db=session, + request=request, + project_id=project_id, + organization_id=org_id, + ) + + _store_last_job_id(job_id) + + logger.info( + f"[run_health_probe_tick] Fired probe | provider: {probe.provider}, " + f"model: {probe.model}, modality: {probe.modality}, job_id: {job_id}, " + f"previous_job_status: {previous_status}" + ) + return { + "enqueued": True, + "job_id": job_id, + "probe_index": index, + "previous_job_status": previous_status.value if previous_status else None, + } diff --git a/backend/app/tests/api/routes/test_cron_health_probes.py b/backend/app/tests/api/routes/test_cron_health_probes.py new file mode 100644 index 000000000..8ccdc897a --- /dev/null +++ b/backend/app/tests/api/routes/test_cron_health_probes.py @@ -0,0 +1,74 @@ +from typing import Any +from unittest.mock import patch +from uuid import uuid4 + +from fastapi.testclient import TestClient + +from app.core.config import settings +from app.tests.utils.auth import TestAuthContext + + +def _fake_tick_result(**overrides: Any) -> dict[str, Any]: + result = { + "enqueued": True, + "job_id": str(uuid4()), + "probe_index": 4, + "previous_job_status": "SUCCESS", + } + result.update(overrides) + return result + + +def test_health_probes_cron_runs_tick_and_returns_result( + client: TestClient, + superuser_api_key: TestAuthContext, +) -> None: + canned = _fake_tick_result() + + with patch( + "app.api.routes.cron.run_health_probe_tick", return_value=canned + ) as tick_mock: + response = client.get( + f"{settings.API_V1_STR}/cron/health-probes", + headers={"X-API-KEY": superuser_api_key.key}, + ) + + assert response.status_code == 200 + assert response.json() == canned + tick_mock.assert_called_once_with() + + +def test_health_probes_cron_requires_superuser( + client: TestClient, + user_api_key: TestAuthContext, +) -> None: + with patch( + "app.api.routes.cron.run_health_probe_tick", + return_value=_fake_tick_result(), + ) as tick_mock: + response = client.get( + f"{settings.API_V1_STR}/cron/health-probes", + headers={"X-API-KEY": user_api_key.key}, + ) + + assert response.status_code == 403 + assert "Insufficient permissions" in response.json()["error"] + tick_mock.assert_not_called() + + +def test_health_probes_cron_requires_authentication(client: TestClient) -> None: + with patch( + "app.api.routes.cron.run_health_probe_tick", + return_value=_fake_tick_result(), + ) as tick_mock: + response = client.get(f"{settings.API_V1_STR}/cron/health-probes") + + assert response.status_code in (401, 403) + tick_mock.assert_not_called() + + +def test_health_probes_cron_not_in_openapi_schema(client: TestClient) -> None: + response = client.get(f"{settings.API_V1_STR}/openapi.json") + assert response.status_code == 200 + paths = response.json().get("paths", {}) + assert f"{settings.API_V1_STR}/cron/health-probes" not in paths diff --git a/backend/app/tests/services/test_health_probes.py b/backend/app/tests/services/test_health_probes.py new file mode 100644 index 000000000..be614747b --- /dev/null +++ b/backend/app/tests/services/test_health_probes.py @@ -0,0 +1,232 @@ +import threading +from types import TracebackType +from typing import Any +from unittest.mock import patch +from uuid import UUID, uuid4 + +import pytest +from sqlmodel import Session + +from app.core.config import settings +from app.crud.jobs import JobCrud +from app.models.job import JobStatus, JobType, JobUpdate +from app.models.llm.request import KaapiCompletionConfig +from app.services import health_probes +from app.services.health_probes import _PROBES, _build_probe_request +from app.services.llm.mappers import transform_kaapi_config_to_native +from app.tests.utils.test_data import create_test_project + + +class _NonClosingSession: + """Stands in for `Session(engine)` inside health_probes, backed by the + conftest transactional `db` session instead of closing a real one.""" + + def __init__(self, session: Session): + self._session = session + + def __enter__(self) -> Session: + return self._session + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> bool: + return False + + +class _FakeRedis: + """In-memory stand-in for `_redis_client`. `incr` is lock-protected so it + actually replicates Redis's atomicity for the concurrency test below.""" + + def __init__(self) -> None: + self.store: dict[str, str] = {} + self._lock = threading.Lock() + + def get(self, key: str) -> str | None: + return self.store.get(key) + + def set(self, key: str, value: Any) -> None: + self.store[key] = str(value) + + def incr(self, key: str) -> int: + with self._lock: + count = int(self.store.get(key, "0")) + 1 + self.store[key] = str(count) + return count + + +@pytest.fixture +def probe_env(db: Session, monkeypatch: pytest.MonkeyPatch): + project = create_test_project(db) + monkeypatch.setattr(settings, "HEALTH_PROBE_ORG_ID", project.organization_id) + monkeypatch.setattr(settings, "HEALTH_PROBE_PROJECT_ID", project.id) + monkeypatch.setattr( + health_probes, "Session", lambda _engine: _NonClosingSession(db) + ) + fake_redis = _FakeRedis() + monkeypatch.setattr(health_probes, "_redis_client", fake_redis) + return project, fake_redis + + +def _create_job(db: Session, project_id: int, status: JobStatus) -> UUID: + job = JobCrud(session=db).create(job_type=JobType.LLM_API, project_id=project_id) + JobCrud(session=db).update(job_id=job.id, job_update=JobUpdate(status=status)) + return job.id + + +class TestFiresExactlyOneProbe: + def test_tick_enqueues_exactly_one_job( + self, probe_env: tuple[Any, _FakeRedis] + ) -> None: + _, _fake_redis = probe_env + expected_job_id = uuid4() + + with patch( + "app.services.health_probes.start_job", return_value=expected_job_id + ) as start_job_mock: + result = health_probes.run_health_probe_tick() + + start_job_mock.assert_called_once() + assert result["enqueued"] is True + assert result["job_id"] == expected_job_id + + +class TestRoundRobinRotation: + def test_index_advances_and_wraps_across_full_registry( + self, probe_env: tuple[Any, _FakeRedis] + ) -> None: + seen_indexes = [] + with patch( + "app.services.health_probes.start_job", side_effect=lambda **_k: uuid4() + ): + for _ in range(len(_PROBES)): + result = health_probes.run_health_probe_tick() + seen_indexes.append(result["probe_index"]) + + assert seen_indexes == list(range(len(_PROBES))) + + wrapped = health_probes.run_health_probe_tick() + assert wrapped["probe_index"] == 0 + + def test_concurrent_claims_never_collide_or_skip_a_slot( + self, probe_env: tuple[Any, _FakeRedis] + ) -> None: + # N concurrent claims must land on N distinct, evenly spread slots. + from concurrent.futures import ThreadPoolExecutor + + rounds = 5 + total_claims = rounds * len(_PROBES) + with ThreadPoolExecutor(max_workers=16) as pool: + claimed = list( + pool.map( + lambda _: health_probes._claim_next_probe_index(), + range(total_claims), + ) + ) + + counts = {i: claimed.count(i) for i in range(len(_PROBES))} + assert all(count == rounds for count in counts.values()), counts + + +class TestMissingRedisKeysDoNotFailTick: + def test_missing_index_key_defaults_to_zero( + self, db: Session, probe_env: tuple[Any, _FakeRedis] + ) -> None: + project, fake_redis = probe_env + # last_job_id present (not a first-ever run), index key absent. + existing_job_id = _create_job(db, project.id, JobStatus.SUCCESS) + fake_redis.store[health_probes._LAST_JOB_ID_KEY] = str(existing_job_id) + + with patch( + "app.services.health_probes.start_job", return_value=uuid4() + ) as start_job_mock: + result = health_probes.run_health_probe_tick() + + start_job_mock.assert_called_once() + assert result["enqueued"] is True + assert result["probe_index"] == 0 + + def test_missing_last_job_id_key_skips_checkin_but_still_fires( + self, probe_env: tuple[Any, _FakeRedis] + ) -> None: + _, fake_redis = probe_env + fake_redis.store[health_probes._INDEX_KEY] = "2" + + with ( + patch( + "app.services.health_probes.start_job", return_value=uuid4() + ) as start_job_mock, + patch("app.services.health_probes.capture_checkin") as checkin_mock, + ): + result = health_probes.run_health_probe_tick() + + start_job_mock.assert_called_once() + checkin_mock.assert_not_called() + assert result["enqueued"] is True + assert result["probe_index"] == 2 + assert result["previous_job_status"] is None + + +class TestPreviousProbeCheckin: + def test_previous_job_failed_reports_error_checkin( + self, db: Session, probe_env: tuple[Any, _FakeRedis] + ) -> None: + project, fake_redis = probe_env + failed_job_id = _create_job(db, project.id, JobStatus.FAILED) + fake_redis.store[health_probes._LAST_JOB_ID_KEY] = str(failed_job_id) + + with ( + patch("app.services.health_probes.start_job", return_value=uuid4()), + patch("app.services.health_probes.capture_checkin") as checkin_mock, + ): + result = health_probes.run_health_probe_tick() + + checkin_mock.assert_called_once() + assert ( + checkin_mock.call_args.kwargs["status"] == health_probes.MonitorStatus.ERROR + ) + assert result["previous_job_status"] == JobStatus.FAILED.value + + def test_previous_job_success_reports_ok_checkin( + self, db: Session, probe_env: tuple[Any, _FakeRedis] + ) -> None: + project, fake_redis = probe_env + success_job_id = _create_job(db, project.id, JobStatus.SUCCESS) + fake_redis.store[health_probes._LAST_JOB_ID_KEY] = str(success_job_id) + + with ( + patch("app.services.health_probes.start_job", return_value=uuid4()), + patch("app.services.health_probes.capture_checkin") as checkin_mock, + ): + result = health_probes.run_health_probe_tick() + + checkin_mock.assert_called_once() + assert checkin_mock.call_args.kwargs["status"] == health_probes.MonitorStatus.OK + assert result["previous_job_status"] == JobStatus.SUCCESS.value + + +class TestProbeRegistryPayloadsAreValid: + # FR-8/FR-10/FR-11: every registry entry must resolve through the real + # Kaapi -> native mapper with zero warnings (catches e.g. Sarvam TTS + # missing `language`, ElevenLabs TTS missing `voice`). + + @pytest.mark.parametrize( + "probe", + _PROBES, + ids=[f"{p.provider}-{p.model}-{p.modality}" for p in _PROBES], + ) + def test_probe_config_resolves_with_no_mapper_warnings( + self, db: Session, probe: health_probes.Probe + ) -> None: + request = _build_probe_request(probe, index=0) + + kaapi_config = request.config.blob.completion # type: ignore[union-attr] + assert isinstance(kaapi_config, KaapiCompletionConfig) + + _native_config, warnings = transform_kaapi_config_to_native( + session=db, kaapi_config=kaapi_config + ) + + assert warnings == [] diff --git a/docs/wiki/INDEX.md b/docs/wiki/INDEX.md index 9bc299d34..4417c9d28 100644 --- a/docs/wiki/INDEX.md +++ b/docs/wiki/INDEX.md @@ -22,7 +22,7 @@ Deep design narrative lives in `docs/architecture/*.md`; open those only for des - [modules/responses.md](modules/responses.md) — OpenAI Responses API integration, conversations, threads, assistants. No deep-dive doc yet. - [modules/assessment.md](modules/assessment.md) — assessments and assessment runs. No deep-dive doc yet. - [modules/tenancy.md](modules/tenancy.md) — users, orgs, projects, API keys, onboarding, login. -- [modules/platform.md](modules/platform.md) — analytics, notifications, feature flags, languages, credentials, model config, cron. +- [modules/platform.md](modules/platform.md) — analytics, notifications, feature flags, languages, credentials, model config, cron (incl. round-robin health probes through the real `/llm/call` pipeline). ### Cross-cutting - [cross-cutting/auth.md](cross-cutting/auth.md) — JWT, API keys, org/project permission model. diff --git a/docs/wiki/modules/platform.md b/docs/wiki/modules/platform.md index 0f34e5d87..961394ef6 100644 --- a/docs/wiki/modules/platform.md +++ b/docs/wiki/modules/platform.md @@ -12,5 +12,5 @@ All paths relative to `backend/app/`. | Languages | `api/routes/languages.py` | `global.languages` (`models/language.py`) | `crud/language.py` | | Credentials | `api/routes/credentials.py` | `credential` (`models/credentials.py`) | `crud/credentials.py`; provider keys per org/project | | Model config | `api/routes/model_config.py` | `model_config` (`models/model_config.py`) | `crud/model_config.py` | -| Cron | `api/routes/cron.py` | — | triggers batch polling (`crud/evaluations/cron.py`) | +| Cron | `api/routes/cron.py` | — | triggers batch polling (`crud/evaluations/cron.py`); health probes (`services/health_probes.py`) fire one round-robin probe per tick through the real `/llm/call` pipeline (`services/llm/jobs.py::start_job`), rotation state in Redis (`health_probe:index`, `health_probe:last_job_id`), no dedicated table — result rides the `job` table | | Jobs | — | `job` (`models/job.py`), `batch_job` (`models/batch_job.py`) | `crud/jobs.py`, `crud/job/`, `services/job_monitoring.py` | diff --git a/features/health-probes/SRD.md b/features/health-probes/SRD.md new file mode 100644 index 000000000..e73af5646 --- /dev/null +++ b/features/health-probes/SRD.md @@ -0,0 +1,110 @@ +# Health Probes SRD + +## Introduction & Purpose + +This SRD redefines Kaapi's provider health probes so a probe run is a real, checkable test of the `POST /llm/call` path, not a call that only confirms a Celery task got enqueued. + +The current probe (`app/services/health_probes.py`) calls `provider.execute()` directly, bundles all ~13 provider/modality combinations into a single Celery task, and wraps the Sentry check-in around that same enqueue-only task. This bypasses `LLMCallConfig` resolution and the provider mappers entirely, so a malformed per-provider config (ElevenLabs' TTS param is `model_id`, not `model`; Sarvam's TTS/STT calls require a `language` param absent from the probe payload) never reaches validation and never flags. There is also no independent per-probe record: pass/fail exists only inside a single combined Sentry check-in for the whole batch. + +The feature produces, per cron tick: +- Exactly one probe fired through the real `/llm/call` job pipeline (config blob -> mapper -> provider), using the same `job` table record every production LLM call gets. +- A Sentry check-in that reflects the previous tick's actual job outcome (`SUCCESS`/`FAILED`), not just that a task was enqueued. + +**Phase 1:** round-robin one-probe-per-tick firing through `start_job`, previous-probe status check via the existing `job` table, Redis-held rotation state. +**Phase 2+:** none identified; the hardcoded probe registry is maintained by hand as providers/models change. + +Intent: the probe is a real end-to-end regression guard against config/mapper/backward-compat breaks, with provider uptime as a secondary signal, not the reverse. + +## Goals + +- A probe traverses the full `/llm/call` path (`LLMCallConfig` resolution, mapper, provider call) via the same job pipeline production traffic uses, not a direct provider client call. +- Malformed or incompatible per-provider params (e.g. a wrong param name, a missing required param) surface as a validation or job failure, because the probe now goes through the same config validation every real call goes through. +- The registry's ElevenLabs TTS and Sarvam TTS entries carry the provider-required Kaapi params (`voice` and `language` respectively) so those two probes pass, not merely fail loudly. +- Each cron tick tests one probe from the registry, round-robin, instead of all combinations at once. +- Sentry reflects the real outcome of a specific probe's job, not just that a Celery task was enqueued. +- A missing rotation-state key in Redis degrades to a logged, Sentry-visible entry and the flow continues; it never fails the cron tick. + +## Assumptions & Constraints + +- **Out of scope:** a probe results table or dashboard (the `job` table is the record of pass/fail); config storage for probes (each probe's config is hardcoded in its `LLMCallRequest` payload); alerting rules beyond the existing Sentry cron-monitor check-in pattern. +- **Cadence:** cron fires every `HEALTH_PROBE_INTERVAL_MINUTES` (~3); one probe per tick. With N probes in the registry, a given probe's effective retest cadence is `N * HEALTH_PROBE_INTERVAL_MINUTES`. +- **Rotation state, no new table:** the round-robin index and the last-fired job ID live in Redis (`REDIS_URL`, already Kaapi's Celery result backend), not Postgres. Keys: `health_probe:index` (advanced via atomic `INCR`, `mod` over registry length, so two overlapping ticks can never claim the same slot) and `health_probe:last_job_id`. +- **Registry:** the probe list stays a hardcoded Python list, same shape as the current `_PROBES` (provider, model, modality), plus an explicit per-entry `params` dict for any Kaapi param a provider's mapper requires beyond `model`. Concretely: the Sarvam TTS entry carries `language` (its mapper hard-requires it for `target_language_code`, `app/services/llm/mappers.py` `map_kaapi_to_sarvam_params`), and the ElevenLabs TTS entry carries `voice` (its mapper hard-requires it to resolve `voice_id`, `map_kaapi_to_elevenlabs_params`). Edited by hand as providers/models change (e.g. drop a model once Kaapi stops using it). +- **Reuse:** no new tables. The probe's job rides the existing `job` table (`JobType.LLM_API`) via `start_job`, the same entry point `POST /llm/call` uses. `HEALTH_PROBE_ORG_ID` / `HEALTH_PROBE_PROJECT_ID` (existing settings) scope the probe's tenant. +- **Cost:** each probe tick is a real, billed provider call against the configured health-probe org/project; this pollutes that org's usage aggregation by one call per tick. Accepted cost, the point of routing through the real pipeline is the end-to-end regression guard, not a free check. +- **No callback:** the probe's `LLMCallRequest` carries no `callback_url`. Result reading is by polling `Job.status` on the next tick, not by callback delivery, so there is nothing for a callback to add, and no SSRF-guard/domain concern to solve (see Design Decisions). + +## Detailed Design (Execution Flow) + +Each cron tick does two things in one request: resolve the previous tick's probe result, then fire the next one. + +### Cron tick: check previous, fire next + +--- + +**>> PLACE IMAGE HERE: `assets/flow-a.png`, cron tick, check previous probe result then fire the next.** +System-level sequence: the external scheduler, the Kaapi backend, Redis, Postgres, Sentry, the LLM call pipeline, and the provider. + +--- + +Resolution order: read `health_probe:last_job_id` from Redis first. If present, look up that `Job` row; `SUCCESS` means the LLM was actually invoked and returned, `FAILED` (including a provider-side error surfaced through the pipeline) or still `PENDING` past the tick means a real failure, and either drives the Sentry check-in status. If the key is missing (first run, or a Redis eviction), log it and continue without failing the tick, no check-in is possible for a probe that was never recorded. + +Rotation: claim `health_probe:index` via an atomic `INCR` (default to `0` on a missing key or a Redis error, logged, non-fatal), select `registry[index mod len(registry)]`, call `start_job` with that probe's hardcoded `LLMCallRequest`, then write the new job ID to `health_probe:last_job_id`. The probe travels through the identical pipeline a real `/llm/call` request does; a provider-side error or a config validation error both land the job in `FAILED`, which is what the next tick reports. + +## Functional Requirements (Testing) + +| ID | What (user-facing behavior) | Acceptance criteria | Status | +|----|-----------------------------|---------------------|--------| +| FR-1 | One probe fired per cron tick | A single `GET /cron/health-probes` invocation enqueues exactly one `LLMCallRequest` job, not all registry entries | Not Started | +| FR-2 | Probe uses the real LLM call pipeline | The enqueued job resolves through `LLMCallConfig` and the provider mapper (the same path `POST /llm/call` uses), not a direct provider client call | Not Started | +| FR-3 | Round-robin selection | Across consecutive ticks, the probe index advances by one (mod registry length) each time; after `len(registry)` ticks every registry entry has been fired exactly once | Not Started | +| FR-4 | Missing rotation key does not fail the tick | With `health_probe:index` absent from Redis, the tick still fires a probe (defaulting to index 0) and logs the miss | Not Started | +| FR-5 | Missing last-job key does not fail the tick | With `health_probe:last_job_id` absent from Redis, the tick skips the previous-result check (logs it), and still fires the next probe | Not Started | +| FR-6 | Real job failure reported | When the previous probe's `Job.status` is `FAILED` (including a provider-side error), the tick's Sentry check-in reports an error state, not OK | Not Started | +| FR-7 | Real job success reported | When the previous probe's `Job.status` is `SUCCESS`, the tick's Sentry check-in reports OK | Not Started | +| FR-8 | Malformed per-provider config now surfaces | A registry entry whose hardcoded config is invalid for that provider (e.g. a required or misnamed param) fails at config validation/mapper resolution and lands the job in `FAILED`, it is not silently swallowed | Not Started | +| FR-9 | Concurrent ticks never collide | Two overlapping calls to the index claim never return the same index; across N concurrent claims, every registry slot is claimed the same number of times | Not Started | +| FR-10 | Sarvam TTS probe carries the required `language` param | The Sarvam TTS registry entry's mapped native params include `target_language_code`; its job reaches the provider call instead of failing at mapper resolution with "Missing required 'language' parameter for TTS" | Not Started | +| FR-11 | ElevenLabs TTS probe carries the required `voice` param | The ElevenLabs TTS registry entry's mapped native params include `voice_id`; its job reaches the provider call instead of failing at mapper resolution with "Missing required 'voice' parameter for TTS" | Not Started | + +## Endpoints + +### `GET /cron/health-probes` (existing, behavior replaced) + +Still hidden from Swagger, still superuser-only. No longer enqueues a bespoke all-probes Celery task; instead runs the check-previous-then-fire-next tick described above and returns immediately once the next probe is enqueued. + +**Response:** + +```json +{ + "enqueued": true, + "job_id": "9c2e4d6f-1a2b-3c4d-5e6f-7a8b9c0d1e2f", + "probe_index": 4, + "previous_job_status": "SUCCESS" +} +``` + +`previous_job_status` is `null` when `health_probe:last_job_id` was missing (first tick, or the key expired). + +## Database Schema + +No new tables. The probe's per-tick result rides the existing `job` table (`models/job.py`), the same table every `/llm/call` request writes to; `JobType.LLM_API`, scoped to `HEALTH_PROBE_PROJECT_ID`. + +## Configuration + +| Setting | Type | Default | Description | +|---------|------|---------|-------------| +| HEALTH_PROBE_INTERVAL_MINUTES | int | 3 | Cron tick interval, one probe per tick | +| HEALTH_PROBE_ORG_ID | int \| None | existing | Org the probe's job is created under | +| HEALTH_PROBE_PROJECT_ID | int \| None | existing | Project the probe's job is created under | + +## Design Decisions / Known Limitations + +- **Probe fires through `start_job`, not `provider.execute()`.** This is the core fix: the old design's biggest gap was bypassing config resolution and mappers, which is exactly where the ElevenLabs/Sarvam param mismatches lived undetected. Routing through the same entry point as production traffic means a config-shape regression fails the probe the same way it would fail a real caller. +- **No probe-specific Sentry span inside a bespoke Celery task.** The previous design's Sentry instrumentation sat on the enqueue route, so it never saw the actual provider call. This redesign removes the need for that span entirely: the probe now runs inside the standard LLM job pipeline (already traced), and the check-in is driven by reading the `Job.status` a tick later, which is a stronger signal than a span around a single synchronous call. +- **Redis over a new table for rotation state.** Redis is already wired up as the Celery result backend with unused capacity; a probe-index table would duplicate what two keys already do, and rotation state has no need for query, audit, or FK/tenant isolation. +- **Atomic `INCR`, not GET-then-SET, for the rotation index.** A read-then-write pair lets two overlapping ticks read the same index, fire the same probe, and silently skip another registry entry for a cycle. `INCR` is one atomic Redis command, so concurrent callers always land on distinct, consecutive slots. +- **No callback_url, no callback endpoint.** An earlier pass added a dummy `POST /internal/health-probe-callback` on the assumption the pipeline needs a callback target and Kaapi's SSRF guard would reject `localhost`. Neither holds: `callback_url` on `LLMCallRequest` is optional, `Job.status` is set to `SUCCESS`/`FAILED` in `execute_job` unconditionally, and callback delivery is a best-effort side channel that never gates it. Since the probe already reads its result by polling `Job.status` on the next tick, the callback bought nothing, so `callback_url` is simply omitted. +- **Calling `start_job` directly, not `POST /llm/call` over HTTP.** `start_job` is exactly what the route calls after its two thin wrapper concerns (the auth dependency and callback SSRF validation, both irrelevant to a cron-triggered internal probe with no callback). Going over HTTP instead would add API-key provisioning, a self-referential network hop, and JSON (de)serialization, all to rebuild the same `LLMCallRequest` this call already builds in Python, for no additional coverage. +- **Known limitation:** if a probe's config becomes stale (a model retired, a param shape changed upstream) the registry must be edited and redeployed by hand, there is no self-healing or drift detection on the registry itself. +- **ElevenLabs and Sarvam TTS payloads are fixed, not just detected.** The prior probe passed only `model` for every provider; ElevenLabs TTS silently needs `voice` (to resolve `voice_id`) and Sarvam TTS silently needs `language` (to resolve `target_language_code`), both hard requirements in their respective mappers. Phase 1 adds these fields to the registry entries directly, so both probes are expected to pass on first tick, not merely fail loudly. The redesigned pipeline (routing through the real mapper) is what surfaces this class of gap at all; the registry fix is what closes it for these two entries specifically. diff --git a/features/health-probes/assets/flow-a.mmd b/features/health-probes/assets/flow-a.mmd new file mode 100644 index 000000000..ce38c8a81 --- /dev/null +++ b/features/health-probes/assets/flow-a.mmd @@ -0,0 +1,27 @@ +sequenceDiagram + participant Scheduler as External Scheduler + participant Cron as Kaapi backend (cron route) + participant Redis + participant DB as Postgres (job table) + participant Sentry + participant Celery as LLM call pipeline (Celery) + participant Provider as LLM Provider + + Scheduler->>Cron: GET /cron/health-probes + Cron->>Redis: GET last_job_id + alt key present + Cron->>DB: look up previous Job by id + DB-->>Cron: JobStatus (SUCCESS / FAILED / PENDING) + Cron->>Sentry: check-in reflecting previous probe's real status + else key missing + Cron->>Sentry: log entry, continue (no flow failure) + end + Cron->>Redis: INCR probe_index (atomic claim) + Cron->>Cron: pick probe = registry[index mod len(registry)] + Cron->>Celery: start_job(LLMCallRequest for probe, no callback_url) + Celery-->>Cron: job_id + Cron->>Redis: SET last_job_id + Cron-->>Scheduler: 200 OK (enqueued) + Celery->>Provider: resolved config + input (config -> mapper -> provider) + Provider-->>Celery: response or provider error + Celery->>DB: update Job status (SUCCESS / FAILED) diff --git a/features/health-probes/assets/flow-a.png b/features/health-probes/assets/flow-a.png new file mode 100644 index 000000000..a2bf8ee6a Binary files /dev/null and b/features/health-probes/assets/flow-a.png differ diff --git a/scripts/python/invoke-cron.py b/scripts/python/invoke-cron.py index 64df37b25..64b409573 100644 --- a/scripts/python/invoke-cron.py +++ b/scripts/python/invoke-cron.py @@ -18,6 +18,7 @@ ENDPOINTS = [ "/api/v1/cron/evaluations", "/api/v1/cron/pending-jobs", + "/api/v1/cron/health-probes", ] REQUEST_TIMEOUT = 30 # Timeout for requests in seconds