Health Probes: For llm_call endpoints

backend/app/api/routes/cron.py

@@ -7,6 +7,7 @@
 from app.api.deps import SessionDep
 from app.api.permissions import Permission, require_permission
 from app.core.config import settings
+from app.celery.tasks.job_execution import run_health_probes
 from app.crud.evaluations import process_all_pending_evaluations
 from app.services.job_monitoring import monitor_pending_jobs
 
@@ -15,13 +16,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,
@@ -33,6 +32,20 @@
     "recovery_threshold": 1,
 }
 
+HEALTH_PROBES_CRON_MONITOR_CONFIG: MonitorConfig = {
+    "schedule": {
+        "type": "interval",
+        # not required eventbridge is aleady 5 mins. So not required.
+        "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,
+}
+
 PENDING_JOBS_CRON_MONITOR_CONFIG: MonitorConfig = {
     "schedule": {
         "type": "interval",
@@ -123,6 +136,21 @@ async def evaluation_cron_job(
         raise
 
 
+@router.get(
+    "/cron/health-probes",
+    include_in_schema=False,
+    dependencies=[Depends(require_permission(Permission.SUPERUSER))],
+)
+@sentry_sdk.monitor(
+    monitor_slug="health-probes-cron-job",
+    monitor_config=HEALTH_PROBES_CRON_MONITOR_CONFIG,
+)
+def health_probes_cron_job() -> dict:
+    logger.info("[health_probes_cron_job] Cron job invoked — enqueueing task")
+    async_result = run_health_probes.delay()
+    return {"enqueued": True, "task_id": async_result.id}
+
+
 @router.get(
     "/cron/pending-jobs",
     include_in_schema=False,

backend/app/celery/tasks/job_execution.py

@@ -425,6 +425,23 @@ def run_evaluation_fast_aggregate(
     )
 
 
+@celery_app.task(bind=True, queue="default", priority=2)
+@gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_health_probes")
+def run_health_probes(self, trace_id: str = DEFAULT_TRACE_ID) -> dict:
+    from sqlmodel import Session
+
+    from app.core.db import engine
+    from app.services.health_probes import run_probes
+
+    _set_trace(trace_id)
+
+    def _do() -> dict:
+        with Session(engine) as session:
+            return run_probes(session=session)
+
+    return _run_with_otel_parent(self, _do)
+
+
 @celery_app.task(bind=True, queue="default", priority=1)
 @gevent_timeout(
     settings.CELERY_TASK_SOFT_TIME_LIMIT, "send_eval_completion_notification"

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 = 5
+
     # 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,

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)
 

backend/app/tests/api/routes/test_cron_health_probes.py

@@ -0,0 +1,65 @@
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from fastapi.testclient import TestClient
+
+from app.core.config import settings
+from app.tests.utils.auth import TestAuthContext
+
+
+def test_health_probes_cron_enqueues_and_returns_task_id(
+    client: TestClient,
+    superuser_api_key: TestAuthContext,
+) -> None:
+    fake_async_result = SimpleNamespace(id="test-task-id")
+
+    with patch(
+        "app.api.routes.cron.run_health_probes.delay",
+        return_value=fake_async_result,
+    ) as delay_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() == {"enqueued": True, "task_id": "test-task-id"}
+    delay_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_probes.delay",
+        return_value=SimpleNamespace(id="should-not-run"),
+    ) as delay_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"]
+    delay_mock.assert_not_called()
+
+
+def test_health_probes_cron_requires_authentication(
+    client: TestClient,
+) -> None:
+    with patch(
+        "app.api.routes.cron.run_health_probes.delay",
+        return_value=SimpleNamespace(id="should-not-run"),
+    ) as delay_mock:
+        response = client.get(f"{settings.API_V1_STR}/cron/health-probes")
+
+    assert response.status_code in (401, 403)
+    delay_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

backend/app/tests/celery/test_run_health_probes.py

@@ -0,0 +1,86 @@
+from unittest.mock import patch
+
+from sqlmodel import Session
+
+from app.celery.tasks import job_execution
+
+
+class _NonClosingSession:
+    def __init__(self, session: Session):
+        self._session = session
+
+    def __enter__(self) -> Session:
+        return self._session
+
+    def __exit__(self, exc_type, exc, tb) -> bool:
+        return False
+
+
+def test_run_health_probes_task_returns_run_probes_result(db: Session):
+    canned = {
+        "elapsed_ms": 42,
+        "total": 3,
+        "ok": 3,
+        "failed": 0,
+        "results": [
+            {
+                "endpoint": "llm/call",
+                "provider": "openai",
+                "modality": "text",
+                "model": "gpt-4o-1-mini",
+                "ok": True,
+                "latency_ms": 10,
+                "error": None,
+            }
+        ],
+    }
+
+    captured: dict = {}
+
+    def _fake_run_probes(*, session: Session) -> dict:
+        captured["session"] = session
+        return canned
+
+    with (
+        patch(
+            "app.services.health_probes.run_probes",
+            side_effect=_fake_run_probes,
+        ),
+        patch(
+            "app.core.db.engine",
+            new=object(),  # engine is only fed to Session() which we intercept below
+        ),
+        patch(
+            "sqlmodel.Session",
+            side_effect=lambda _engine: _NonClosingSession(db),
+        ),
+    ):
+        result = job_execution.run_health_probes.apply(args=[]).get()
+
+    assert result == canned
+    assert captured["session"] is db
+
+
+def test_run_health_probes_task_skipped_when_settings_unset(db: Session):
+    skipped = {
+        "skipped": True,
+        "reason": "health_probe_org_or_project_not_set",
+    }
+
+    with (
+        patch(
+            "app.services.health_probes.run_probes",
+            return_value=skipped,
+        ),
+        patch(
+            "app.core.db.engine",
+            new=object(),
+        ),
+        patch(
+            "sqlmodel.Session",
+            side_effect=lambda _engine: _NonClosingSession(db),
+        ),
+    ):
+        result = job_execution.run_health_probes.apply(args=[]).get()
+
+    assert result == skipped