diff --git a/.claude/conventions/celery.md b/.claude/conventions/celery.md index f649a550c..61b17b49c 100644 --- a/.claude/conventions/celery.md +++ b/.claude/conventions/celery.md @@ -8,7 +8,7 @@ worker pool so a fast-eval burst can't occupy `default`'s worker slots and delay LLM jobs; priority only reorders queued-not-yet-claimed messages, it can't preempt a task a worker already started). Within each queue, tasks are ordered by a per-task `priority` (higher drains first, FIFO within a band). Read `app/celery/tasks/job_execution.py` before writing — it -shows the full pattern (decorator + timeout + OTel propagation + delegation to a service). +shows the full pattern (decorator + timeout + delegation to a service). ## Canonical decorator stack @@ -19,20 +19,18 @@ def run_my_job(self, project_id: int, job_id: str, trace_id: str, **kwargs): from app.services.my_domain.jobs import do_the_work # late import to avoid cycles _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: do_the_work( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return do_the_work( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) ``` -`_set_trace`, `_run_with_otel_parent`, and `gevent_timeout` already exist in this module / -`app/celery/utils.py` — reuse them, don't reinvent. +`_set_trace` and `gevent_timeout` already exist in this module / `app/celery/utils.py` — reuse +them, don't reinvent. Enqueue through a `start_*` helper in `app/celery/utils.py`, which calls +plain `task.apply_async(kwargs={...})`. ## Queue and priority choice — be explicit @@ -61,7 +59,7 @@ inherits its priority unless you override it. - **`bind=True`** so you have `self` (the task instance) for retries, IDs, etc. - **Always pass an explicit `queue` (`"default"` or `"evaluations"`) and `priority`.** Only fast-eval chunk/aggregate tasks use `"evaluations"`; everything else uses `"default"`. - **Pass `trace_id` explicitly** as a parameter and call `_set_trace(trace_id)` first thing. This wires `asgi_correlation_id` so logs from inside the task match the originating request. -- **Wrap the work in `_run_with_otel_parent(self, lambda: ...)`** so OpenTelemetry parent context propagates from the enqueueing process. +- **Don't propagate trace context by hand.** OTel's `CeleryInstrumentor` already parents the worker's `run/...` span to the enqueue span. Injecting headers on enqueue or attaching an extracted context in the task reparents the task's spans to the API request, and Sentry splits them into separate transactions. - **Delegate to a service.** The task body should be a thin shim over `app/services//`. No DB queries, no external HTTP, no business logic inside the task itself. - **Late-import the service inside the function body** (as the canonical pattern does). Celery workers boot faster and you avoid model-import cycles. - **Idempotency.** Celery will redeliver. Either: diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index a84c3c33d..a35dc37e1 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -50,9 +50,9 @@ def _set_tenant_span_attributes(auth_context: AuthContext) -> None: if span.is_recording(): span.set_attribute("user.id", str(auth_context.user.id)) if auth_context.organization: - span.set_attribute("tenant.org_id", auth_context.organization.id) + span.set_attribute("tenant.org_id", str(auth_context.organization.id)) if auth_context.project: - span.set_attribute("tenant.project_id", auth_context.project.id) + span.set_attribute("tenant.project_id", str(auth_context.project.id)) org_id = auth_context.organization.id if auth_context.organization else None project_id = auth_context.project.id if auth_context.project else None diff --git a/backend/app/api/routes/collections.py b/backend/app/api/routes/collections.py index 828db79e5..6190a76ac 100644 --- a/backend/app/api/routes/collections.py +++ b/backend/app/api/routes/collections.py @@ -6,6 +6,7 @@ from fastapi import Path as FastPath from app.api.deps import SessionDep, AuthContextDep +from app.crud.collection.collection import CollectionNameConflictError from app.api.permissions import Permission, require_permission from app.core.telemetry import log_context from app.core.rate_monitor import monitor_rate @@ -14,7 +15,6 @@ CollectionJobCrud, DocumentCollectionCrud, ) -from app.crud.collection.collection import CollectionNameConflictError from app.core.cloud import get_cloud_storage from app.models import ( CollectionJobStatus, diff --git a/backend/app/celery/celery_app.py b/backend/app/celery/celery_app.py index b73de912c..c75a9a984 100644 --- a/backend/app/celery/celery_app.py +++ b/backend/app/celery/celery_app.py @@ -46,9 +46,20 @@ def _initialize_worker_observability() -> None: ) @task_prerun.connect(weak=False) - def _tag_sentry_task(task: "celery.Task", **_: object) -> None: # type: ignore[name-defined] - """Tag the task's Sentry scope with its name so Issues can be filtered by task.""" + def _tag_sentry_task( + task: "celery.Task", # type: ignore[name-defined] + kwargs: dict[str, object] | None = None, + **_: object, + ) -> None: + """Tag the task's Sentry scope with its name and tenant so Issues can be filtered by both.""" sentry_sdk.set_tag("celery.task_name", task.name) + task_kwargs = kwargs or {} + for tag, key in ( + ("tenant.project_id", "project_id"), + ("tenant.org_id", "organization_id"), + ): + if task_kwargs.get(key) is not None: + sentry_sdk.set_tag(tag, str(task_kwargs[key])) _sentry_initialized = True diff --git a/backend/app/celery/tasks/job_execution.py b/backend/app/celery/tasks/job_execution.py index e8cf6bacb..82feca314 100644 --- a/backend/app/celery/tasks/job_execution.py +++ b/backend/app/celery/tasks/job_execution.py @@ -22,15 +22,11 @@ """ import logging -from collections.abc import Callable -from typing import TYPE_CHECKING, TypeVar +from typing import TYPE_CHECKING from asgi_correlation_id import correlation_id from celery import Task, current_task from gevent import Timeout -from opentelemetry import context as otel_context -from opentelemetry import trace -from opentelemetry.propagate import extract from app.celery.celery_app import EVALUATIONS_QUEUE, celery_app from app.celery.utils import gevent_timeout @@ -44,8 +40,6 @@ logger = logging.getLogger(__name__) -T = TypeVar("T") - # Sentinel correlation id used when no trace id is propagated from the # enqueueing request. Matches the codebase-wide "N/A" default (see # app/core/logger.py and app/celery/utils.py). @@ -59,60 +53,18 @@ def _set_trace(trace_id: str) -> None: logger.info(f"[_set_trace] Set correlation ID: {trace_id}") -def _extract_parent_context(task_instance: Task) -> otel_context.Context: - """Extract OTel parent context from Celery headers if available.""" - headers = getattr(task_instance.request, "headers", None) or {} - carrier: dict[str, str] = {} - - if isinstance(headers, dict): - for key, value in headers.items(): - if isinstance(value, str): - carrier[str(key)] = value - - nested = headers.get("otel", {}) - if isinstance(nested, dict): - for key, value in nested.items(): - if isinstance(value, str): - carrier[str(key)] = value - - return extract(carrier) - - -def _run_with_otel_parent( - task_instance: Task, fn: Callable[[], T] -) -> T: # noqa: UP047 (black doesn't support PEP 695 generics yet) - """Attach the extracted parent context and execute `fn` under it. - - Needed because otel's Celery instrumentation misses propagation headers - under `task.request.headers`, leaving `run/...` spans unparented. - """ - parent_ctx = _extract_parent_context(task_instance) - parent_span_ctx = trace.get_current_span(parent_ctx).get_span_context() - if not (parent_span_ctx and parent_span_ctx.is_valid): - return fn() - - token = otel_context.attach(parent_ctx) - try: - return fn() - finally: - otel_context.detach(token) - - @celery_app.task(bind=True, queue="default", priority=9) @gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_llm_job") def run_llm_job(self, project_id: int, job_id: str, trace_id: str, **kwargs): from app.services.llm.jobs import execute_job _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_job( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_job( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -122,15 +74,12 @@ def run_llm_chain_job(self, project_id: int, job_id: str, trace_id: str, **kwarg from app.services.llm.jobs import execute_chain_job _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_chain_job( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_chain_job( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -140,15 +89,12 @@ def run_response_job(self, project_id: int, job_id: str, trace_id: str, **kwargs from app.services.response.jobs import execute_job _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_job( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_job( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -158,15 +104,12 @@ def run_guardrails_job(self, project_id: int, job_id: str, trace_id: str, **kwar from app.services.guardrails.jobs import execute_job _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_job( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_job( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -176,15 +119,12 @@ def run_doctransform_job(self, project_id: int, job_id: str, trace_id: str, **kw from app.services.doctransform.job import execute_job _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_job( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_job( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -196,15 +136,12 @@ def run_collection_setup_job( from app.services.collections.create_collection import execute_setup_job _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_setup_job( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_setup_job( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -216,15 +153,12 @@ def run_collection_batch_job( from app.services.collections.create_collection import execute_batch_job _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_batch_job( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_batch_job( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -236,15 +170,12 @@ def run_delete_collection_job( from app.services.collections.delete_collection import execute_job _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_job( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_job( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -256,15 +187,12 @@ def run_evaluation_batch_submission( from app.services.evaluations.batch_job import execute_evaluation_batch_submission _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_evaluation_batch_submission( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_evaluation_batch_submission( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -276,14 +204,11 @@ def run_prompt_improvement(self, project_id: int, job_id: str, trace_id: str, ** from app.services.evaluations.prompt_improvement import execute_prompt_improvement _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_prompt_improvement( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - **kwargs, - ), + return execute_prompt_improvement( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + **kwargs, ) @@ -295,15 +220,12 @@ def run_stt_batch_submission( from app.services.stt_evaluations.batch_job import execute_batch_submission _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_batch_submission( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_batch_submission( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -315,15 +237,12 @@ def run_stt_metric_computation( from app.services.stt_evaluations.metric_job import execute_metric_computation _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_metric_computation( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_metric_computation( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -335,15 +254,12 @@ def run_tts_batch_submission( from app.services.tts_evaluations.batch_job import execute_batch_submission _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_batch_submission( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_batch_submission( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -360,13 +276,10 @@ def run_assessment_pipeline( from app.services.assessment.tasks import execute_assessment_pipeline _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_assessment_pipeline( - run_id=run_id, - organization_id=organization_id, - project_id=project_id, - ), + return execute_assessment_pipeline( + run_id=run_id, + organization_id=organization_id, + project_id=project_id, ) @@ -398,13 +311,10 @@ def run_assessment_api_batch( ) _set_trace(trace_id) - result = _run_with_otel_parent( - self, - lambda: run_batch_stage( - execution_id=execution_id, - organization_id=organization_id, - project_id=project_id, - ), + result = run_batch_stage( + execution_id=execution_id, + organization_id=organization_id, + project_id=project_id, ) if result and result.get("requeue"): self.apply_async( @@ -430,15 +340,12 @@ def run_tts_result_processing( ) _set_trace(trace_id) - return _run_with_otel_parent( - self, - lambda: execute_tts_result_processing( - project_id=project_id, - job_id=job_id, - task_id=current_task.request.id, - task_instance=self, - **kwargs, - ), + return execute_tts_result_processing( + project_id=project_id, + job_id=job_id, + task_id=current_task.request.id, + task_instance=self, + **kwargs, ) @@ -465,11 +372,8 @@ def run_evaluation_fast_chunk( ) with log_context(module=LogModuleEnum.FASTEVALS): - return _run_with_otel_parent( - self, - lambda: execute_fast_evaluation_chunk( - eval_run_id=eval_run_id, chunk_index=chunk_index - ), + return execute_fast_evaluation_chunk( + eval_run_id=eval_run_id, chunk_index=chunk_index ) @@ -493,10 +397,7 @@ def run_evaluation_fast_aggregate( ) with log_context(module=LogModuleEnum.FASTEVALS): - return _run_with_otel_parent( - self, - lambda: execute_fast_evaluation_aggregate(eval_run_id=eval_run_id), - ) + return execute_fast_evaluation_aggregate(eval_run_id=eval_run_id) # Priority 6 (fast-eval tier): same band as run_prompt_improvement — one graph @@ -533,15 +434,12 @@ def run_evaluation_iteration_graph_step( f"iteration_run_id={iteration_run_id} | resume={resume} | " f"task_id={current_task.request.id}" ) - return _run_with_otel_parent( - self, - lambda: execute_evaluation_iteration_graph_step( - iteration_run_id=iteration_run_id, - resume=resume, - organization_id=organization_id, - project_id=project_id, - **kwargs, - ), + return execute_evaluation_iteration_graph_step( + iteration_run_id=iteration_run_id, + resume=resume, + organization_id=organization_id, + project_id=project_id, + **kwargs, ) diff --git a/backend/app/celery/utils.py b/backend/app/celery/utils.py index 3bd205d22..160f3cb91 100644 --- a/backend/app/celery/utils.py +++ b/backend/app/celery/utils.py @@ -9,7 +9,6 @@ from celery.result import AsyncResult from gevent import Timeout -from opentelemetry.propagate import inject from app.celery.celery_app import celery_app @@ -18,22 +17,17 @@ F = TypeVar("F", bound=Callable[..., Any]) -def _enqueue_with_trace_context(task, **kwargs) -> str: - """Publish Celery task with explicit trace context headers.""" - otel_headers: dict[str, str] = {} - inject(otel_headers) - celery_headers = dict(otel_headers) - celery_headers["otel"] = otel_headers - async_result = task.apply_async(kwargs=kwargs, headers=celery_headers) - return async_result.id - - def start_llm_job(project_id: int, job_id: str, trace_id: str = "N/A", **kwargs) -> str: from app.celery.tasks.job_execution import run_llm_job - task_id = _enqueue_with_trace_context( - run_llm_job, project_id=project_id, job_id=job_id, trace_id=trace_id, **kwargs - ) + task_id = run_llm_job.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info(f"[start_llm_job] Started job {job_id} with Celery task {task_id}") return task_id @@ -43,13 +37,14 @@ def start_llm_chain_job( ) -> str: from app.celery.tasks.job_execution import run_llm_chain_job - task_id = _enqueue_with_trace_context( - run_llm_chain_job, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_llm_chain_job.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_llm_chain_job] Started job {job_id} with Celery task {task_id}" ) @@ -61,13 +56,14 @@ def start_guardrails_job( ) -> str: from app.celery.tasks.job_execution import run_guardrails_job - task_id = _enqueue_with_trace_context( - run_guardrails_job, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_guardrails_job.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_guardrails_job] Started job {job_id} with Celery task {task_id}" ) @@ -79,13 +75,14 @@ def start_response_job( ) -> str: from app.celery.tasks.job_execution import run_response_job - task_id = _enqueue_with_trace_context( - run_response_job, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_response_job.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info(f"[start_response_job] Started job {job_id} with Celery task {task_id}") return task_id @@ -95,13 +92,14 @@ def start_doctransform_job( ) -> str: from app.celery.tasks.job_execution import run_doctransform_job - task_id = _enqueue_with_trace_context( - run_doctransform_job, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_doctransform_job.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_doctransform_job] Started job {job_id} with Celery task {task_id}" ) @@ -113,13 +111,14 @@ def start_collection_setup_job( ) -> str: from app.celery.tasks.job_execution import run_collection_setup_job - task_id = _enqueue_with_trace_context( - run_collection_setup_job, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_collection_setup_job.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_collection_setup_job] Started job {job_id} with Celery task {task_id}" ) @@ -131,13 +130,14 @@ def start_collection_batch_job( ) -> str: from app.celery.tasks.job_execution import run_collection_batch_job - task_id = _enqueue_with_trace_context( - run_collection_batch_job, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_collection_batch_job.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_collection_batch_job] Started job {job_id} with Celery task {task_id}" ) @@ -149,13 +149,14 @@ def start_delete_collection_job( ) -> str: from app.celery.tasks.job_execution import run_delete_collection_job - task_id = _enqueue_with_trace_context( - run_delete_collection_job, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_delete_collection_job.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_delete_collection_job] Started job {job_id} with Celery task {task_id}" ) @@ -167,13 +168,14 @@ def start_evaluation_batch_submission( ) -> str: from app.celery.tasks.job_execution import run_evaluation_batch_submission - task_id = _enqueue_with_trace_context( - run_evaluation_batch_submission, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_evaluation_batch_submission.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_evaluation_batch_submission] Started job {job_id} with Celery task {task_id}" ) @@ -185,13 +187,14 @@ def start_prompt_improvement( ) -> str: from app.celery.tasks.job_execution import run_prompt_improvement - task_id = _enqueue_with_trace_context( - run_prompt_improvement, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_prompt_improvement.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_prompt_improvement] Started job {job_id} with Celery task {task_id}" ) @@ -203,13 +206,14 @@ def start_stt_batch_submission( ) -> str: from app.celery.tasks.job_execution import run_stt_batch_submission - task_id = _enqueue_with_trace_context( - run_stt_batch_submission, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_stt_batch_submission.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_stt_batch_submission] Started job {job_id} with Celery task {task_id}" ) @@ -221,13 +225,14 @@ def start_stt_metric_computation( ) -> str: from app.celery.tasks.job_execution import run_stt_metric_computation - task_id = _enqueue_with_trace_context( - run_stt_metric_computation, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_stt_metric_computation.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_stt_metric_computation] Started job {job_id} with Celery task {task_id}" ) @@ -239,13 +244,14 @@ def start_tts_batch_submission( ) -> str: from app.celery.tasks.job_execution import run_tts_batch_submission - task_id = _enqueue_with_trace_context( - run_tts_batch_submission, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_tts_batch_submission.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_tts_batch_submission] Started job {job_id} with Celery task {task_id}" ) @@ -257,13 +263,14 @@ def start_tts_result_processing( ) -> str: from app.celery.tasks.job_execution import run_tts_result_processing - task_id = _enqueue_with_trace_context( - run_tts_result_processing, - project_id=project_id, - job_id=job_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_tts_result_processing.apply_async( + kwargs={ + "project_id": project_id, + "job_id": job_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_tts_result_processing] Started job {job_id} with Celery task {task_id}" ) @@ -276,12 +283,13 @@ def start_fast_evaluation_chunk( """Enqueue one parallel responses-chunk task for a fast EvaluationRun.""" from app.celery.tasks.job_execution import run_evaluation_fast_chunk - task_id = _enqueue_with_trace_context( - run_evaluation_fast_chunk, - eval_run_id=eval_run_id, - chunk_index=chunk_index, - trace_id=trace_id, - ) + task_id = run_evaluation_fast_chunk.apply_async( + kwargs={ + "eval_run_id": eval_run_id, + "chunk_index": chunk_index, + "trace_id": trace_id, + } + ).id logger.info( f"[start_fast_evaluation_chunk] Enqueued fast eval chunk | " f"eval_run_id={eval_run_id} | chunk_index={chunk_index} | task_id={task_id}" @@ -293,11 +301,9 @@ def start_fast_evaluation_aggregate(eval_run_id: int, trace_id: str = "N/A") -> """Enqueue the fan-in aggregate task once a fast run's chunks are all done.""" from app.celery.tasks.job_execution import run_evaluation_fast_aggregate - task_id = _enqueue_with_trace_context( - run_evaluation_fast_aggregate, - eval_run_id=eval_run_id, - trace_id=trace_id, - ) + task_id = run_evaluation_fast_aggregate.apply_async( + kwargs={"eval_run_id": eval_run_id, "trace_id": trace_id} + ).id logger.info( f"[start_fast_evaluation_aggregate] Enqueued fast eval aggregate | " f"eval_run_id={eval_run_id} | task_id={task_id}" @@ -321,15 +327,16 @@ def start_evaluation_iteration_round( """ from app.celery.tasks.job_execution import run_evaluation_iteration_graph_step - task_id = _enqueue_with_trace_context( - run_evaluation_iteration_graph_step, - iteration_run_id=iteration_run_id, - resume=resume, - organization_id=organization_id, - project_id=project_id, - trace_id=trace_id, - **kwargs, - ) + task_id = run_evaluation_iteration_graph_step.apply_async( + kwargs={ + "iteration_run_id": iteration_run_id, + "resume": resume, + "organization_id": organization_id, + "project_id": project_id, + "trace_id": trace_id, + **kwargs, + } + ).id logger.info( f"[start_evaluation_iteration_round] Enqueued | " f"iteration_run_id={iteration_run_id} | resume={resume} | task_id={task_id}" diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 4640731b1..1557b32c8 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -162,8 +162,6 @@ def AWS_S3_BUCKET(self) -> str: SENTRY_TRACES_SAMPLE_RATE: float = Field(default=1.0, ge=0.0, le=1.0) SENTRY_SEND_DEFAULT_PII: bool = False SENTRY_ERROR_SAMPLE_RATE: float = Field(default=1.0, ge=0.0, le=1.0) - SENTRY_PROFILE_SESSION_SAMPLE_RATE: float = Field(default=1.0, ge=0.0, le=1.0) - SENTRY_PROFILE_LIFECYCLE: Literal["manual", "trace"] = "trace" # Threshold Request Rate per minute THRESHOLD_LLM_CALL_RATE: int = 15 diff --git a/backend/app/core/exception_handlers.py b/backend/app/core/exception_handlers.py index c0d2653a5..71a507f56 100644 --- a/backend/app/core/exception_handlers.py +++ b/backend/app/core/exception_handlers.py @@ -1,17 +1,31 @@ +import logging import re from collections import defaultdict import sentry_sdk -from fastapi import FastAPI, Request, HTTPException +from asgi_correlation_id import correlation_id +from fastapi import FastAPI, HTTPException, Request # noqa: F401 (re-exported) from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse +from starlette.exceptions import HTTPException as StarletteHTTPException +from sqlalchemy.exc import DataError, OperationalError, SQLAlchemyError from starlette.status import ( + HTTP_409_CONFLICT, HTTP_422_UNPROCESSABLE_CONTENT, HTTP_500_INTERNAL_SERVER_ERROR, + HTTP_503_SERVICE_UNAVAILABLE, ) from app.utils import APIResponse +logger = logging.getLogger(__name__) + +GENERIC_ERROR_DETAIL = "An unexpected error occurred." +DB_ERROR_DETAIL = "The request could not be completed." + +UNIQUE_VIOLATION_SQLSTATE = "23505" +FOREIGN_KEY_VIOLATION_SQLSTATE = "23503" + _BRANCH_PATTERN = re.compile(r"^[A-Z]|[\[\]()]") @@ -85,6 +99,50 @@ def _sanitize_validation_errors(errors: list[dict]) -> list[dict]: return errors +def _db_error_constraint(exc: SQLAlchemyError) -> tuple[str | None, str | None]: + """Postgres SQLSTATE and violated constraint name from the driver error, when present.""" + orig = getattr(exc, "orig", None) + diag = getattr(orig, "diag", None) + return getattr(orig, "sqlstate", None), getattr(diag, "constraint_name", None) + + +def _db_error_response(exc: SQLAlchemyError) -> tuple[int, str]: + """Map a DB exception to a status and a client-safe hint; SQL and parameters stay server-side.""" + sqlstate, _ = _db_error_constraint(exc) + if sqlstate == UNIQUE_VIOLATION_SQLSTATE: + return HTTP_409_CONFLICT, "Record already exists." + if sqlstate == FOREIGN_KEY_VIOLATION_SQLSTATE: + return HTTP_422_UNPROCESSABLE_CONTENT, "Referenced record not found." + if isinstance(exc, DataError): + return ( + HTTP_422_UNPROCESSABLE_CONTENT, + "A value in the request is invalid or too long.", + ) + if isinstance(exc, OperationalError): + return ( + HTTP_503_SERVICE_UNAVAILABLE, + "Service temporarily unavailable. Please retry.", + ) + return HTTP_500_INTERNAL_SERVER_ERROR, DB_ERROR_DETAIL + + +def _tag_error_scope(request: Request, status_code: int) -> None: + """Tag the Sentry scope with route, method and status before the error is logged. + + The middleware tags the route only after the response, too late for an event raised here. + """ + if not sentry_sdk.get_client().is_active(): + return + scope = sentry_sdk.get_current_scope() + route = getattr(request.scope.get("route"), "path", None) + if route: + scope.set_tag("http.route", route) + scope.set_transaction_name(f"{request.method} {route}") + scope.set_tag("http.method", request.method) + scope.set_tag("http.status_code", str(status_code)) + scope.set_tag("error.class", "server" if status_code >= 500 else "client") + + def register_exception_handlers(app: FastAPI) -> None: @app.exception_handler(RequestValidationError) async def validation_error_handler( @@ -96,24 +154,57 @@ async def validation_error_handler( content=APIResponse.failure_response(errors).model_dump(), ) - @app.exception_handler(HTTPException) + @app.exception_handler(StarletteHTTPException) async def http_exception_handler( - request: Request, exc: HTTPException + request: Request, exc: StarletteHTTPException ) -> JSONResponse: detail = exc.detail if isinstance(detail, list): detail = _sanitize_validation_errors(detail) + + _tag_error_scope(request, exc.status_code) + if exc.status_code >= 500: + logger.error( + f"[http_exception] {request.method} {request.url.path} " + f"-> {exc.status_code}", + exc_info=exc, + ) return JSONResponse( status_code=exc.status_code, content=APIResponse.failure_response(detail).model_dump(), + headers=exc.headers, + ) + + @app.exception_handler(SQLAlchemyError) + async def db_error_handler(request: Request, exc: SQLAlchemyError) -> JSONResponse: + status_code, hint = _db_error_response(exc) + sqlstate, constraint = _db_error_constraint(exc) + _tag_error_scope(request, status_code) + if constraint and sentry_sdk.get_client().is_active(): + sentry_sdk.set_tag("db.constraint", constraint) + log = logger.error if status_code >= 500 else logger.warning + log( + f"[db_error_handler] Unhandled DB error | path: {request.url.path}, " + f"type: {type(exc).__name__}, status: {status_code}, sqlstate: {sqlstate}, " + f"constraint: {constraint}, trace_id: {correlation_id.get() or 'N/A'}", + exc_info=exc, + ) + return JSONResponse( + status_code=status_code, + content=APIResponse.failure_response(hint).model_dump(), ) @app.exception_handler(Exception) async def generic_error_handler(request: Request, exc: Exception) -> JSONResponse: - sentry_sdk.capture_exception(exc) + # Fixed detail: str(exc) here leaked provider bodies and DB text to callers. + # logger.error alone reports it; adding capture_exception makes a second event. + _tag_error_scope(request, HTTP_500_INTERNAL_SERVER_ERROR) + logger.error( + f"[generic_error_handler] Unhandled exception | " + f"path: {request.url.path}, trace_id: {correlation_id.get() or 'N/A'}", + exc_info=exc, + ) return JSONResponse( status_code=HTTP_500_INTERNAL_SERVER_ERROR, - content=APIResponse.failure_response( - str(exc) or "An unexpected error occurred." - ).model_dump(), + content=APIResponse.failure_response(GENERIC_ERROR_DETAIL).model_dump(), ) diff --git a/backend/app/core/security.py b/backend/app/core/security.py index e6d068dfd..643fd79d1 100644 --- a/backend/app/core/security.py +++ b/backend/app/core/security.py @@ -10,6 +10,7 @@ import base64 import json import logging +from fastapi import HTTPException import os import secrets from datetime import UTC, datetime, timedelta @@ -227,8 +228,12 @@ def encrypt_credentials(credentials: dict[str, Any]) -> str: return get_fernet().encrypt(credentials_str.encode()).decode() except Exception as e: # Log the real cause (may carry AWS ARNs); never surface it to callers. - logger.error(f"[encrypt_credentials] Encryption failed | error: {e}") - raise ValueError("Failed to encrypt credentials") + logger.error( + f"[encrypt_credentials] Encryption failed | error: {e}", exc_info=True + ) + raise HTTPException( + status_code=502, detail="Failed to encrypt credentials. Retry shortly." + ) def encrypt_fernet(credentials: dict[str, Any]) -> str: @@ -270,8 +275,12 @@ def decrypt_credentials(encrypted_credentials: str) -> dict[str, Any]: return json.loads(decrypted_str) except Exception as e: # Log the real cause (may carry AWS ARNs); never surface it to callers. - logger.error(f"[decrypt_credentials] Decryption failed | error: {e}") - raise ValueError("Failed to decrypt credentials") + logger.error( + f"[decrypt_credentials] Decryption failed | error: {e}", exc_info=True + ) + raise HTTPException( + status_code=502, detail="Failed to decrypt credentials. Retry shortly." + ) class APIKeyManager: diff --git a/backend/app/core/telemetry/llm.py b/backend/app/core/telemetry/llm.py index 32423f847..0ea7ee37b 100644 --- a/backend/app/core/telemetry/llm.py +++ b/backend/app/core/telemetry/llm.py @@ -127,12 +127,15 @@ def set_gen_ai_response_attributes( """Set OTel GenAI response attributes (usage, model) on `span`.""" usage = response.usage if usage: + reasoning_tokens = usage.reasoning_tokens or 0 + output_tokens = usage.output_tokens + if reasoning_tokens and usage.input_tokens + output_tokens < usage.total_tokens: + output_tokens += reasoning_tokens span.set_attribute("gen_ai.usage.input_tokens", usage.input_tokens) - span.set_attribute("gen_ai.usage.output_tokens", usage.output_tokens) + span.set_attribute("gen_ai.usage.output_tokens", output_tokens) span.set_attribute("gen_ai.usage.total_tokens", usage.total_tokens) - reasoning_tokens = getattr(usage, "reasoning_tokens", None) - if reasoning_tokens is not None: - span.set_attribute("gen_ai.usage.output_tokens.reasoning", reasoning_tokens) + if reasoning_tokens: + span.set_attribute("gen_ai.usage.reasoning.output_tokens", reasoning_tokens) if response.response and response.response.model: span.set_attribute("gen_ai.response.model", response.response.model) diff --git a/backend/app/core/telemetry/sentry/filters.py b/backend/app/core/telemetry/sentry/filters.py index 00f37a69f..65d2b3fdc 100644 --- a/backend/app/core/telemetry/sentry/filters.py +++ b/backend/app/core/telemetry/sentry/filters.py @@ -16,6 +16,8 @@ _GENAI_CONTENT_KEYS: frozenset[str] = frozenset( { "gen_ai.request.messages", + "gen_ai.input.messages", + "gen_ai.output.messages", "gen_ai.response.text", "gen_ai.response.tool_calls", "gen_ai.choice", diff --git a/backend/app/core/telemetry/sentry/init.py b/backend/app/core/telemetry/sentry/init.py index c33a631ba..138e08cde 100644 --- a/backend/app/core/telemetry/sentry/init.py +++ b/backend/app/core/telemetry/sentry/init.py @@ -1,4 +1,4 @@ -"""sentry_sdk.init() with shared sampling, profiling, PII and before_send filters.""" +"""sentry_sdk.init() with shared sampling, PII and before_send filters.""" import logging @@ -32,8 +32,6 @@ def init_sentry( instrumenter="otel", traces_sample_rate=settings.SENTRY_TRACES_SAMPLE_RATE, sample_rate=settings.SENTRY_ERROR_SAMPLE_RATE, - profile_session_sample_rate=settings.SENTRY_PROFILE_SESSION_SAMPLE_RATE, - profile_lifecycle=settings.SENTRY_PROFILE_LIFECYCLE, send_default_pii=settings.SENTRY_SEND_DEFAULT_PII, enable_logs=True, include_local_variables=False, diff --git a/backend/app/core/telemetry/tracing.py b/backend/app/core/telemetry/tracing.py index 7bd9a2982..ae54488ef 100644 --- a/backend/app/core/telemetry/tracing.py +++ b/backend/app/core/telemetry/tracing.py @@ -150,13 +150,6 @@ def setup_telemetry(service_name: str | None = None) -> None: except Exception: logger.exception("[setup_telemetry] Failed to instrument Celery") - try: - from opentelemetry.instrumentation.redis import RedisInstrumentor - - RedisInstrumentor().instrument() - except Exception: - logger.exception("[setup_telemetry] Failed to instrument Redis") - try: from opentelemetry.instrumentation.botocore import BotocoreInstrumentor diff --git a/backend/app/crud/collection/collection.py b/backend/app/crud/collection/collection.py index 06cd9b6e9..589e72ac2 100644 --- a/backend/app/crud/collection/collection.py +++ b/backend/app/crud/collection/collection.py @@ -118,7 +118,7 @@ def read_one_if_delete(self, collection_id: UUID) -> Collection: "[CollectionCrud.read_one_if_delete] Collection already deleted | " f"{{'project_id': '{self.project_id}', 'collection_id': '{collection_id}'}}" ) - raise HTTPException(status_code=400, detail="Collection already deleted") + raise HTTPException(status_code=409, detail="Collection already deleted") return collection diff --git a/backend/app/crud/config/config.py b/backend/app/crud/config/config.py index 741bdfc64..175b1ec2e 100644 --- a/backend/app/crud/config/config.py +++ b/backend/app/crud/config/config.py @@ -68,17 +68,10 @@ def create_or_raise( return config, version - except Exception as e: + except Exception: + # Each error type has a central handler (DB, HTTPException, generic). self.session.rollback() - logger.error( - f"[ConfigCrud.create] Failed to create configuration | " - f"{{'name': '{config_create.name}', 'project_id': {self.project_id}, 'error': '{str(e)}'}}", - exc_info=True, - ) - raise HTTPException( - status_code=500, - detail="Unexpected error occurred: failed to create config", - ) + raise def read_one(self, config_id: UUID) -> Config | None: statement = select(Config).where( diff --git a/backend/app/crud/config/version.py b/backend/app/crud/config/version.py index 313a4e27c..d262b4e6e 100644 --- a/backend/app/crud/config/version.py +++ b/backend/app/crud/config/version.py @@ -98,23 +98,16 @@ def create_or_raise(self, version_create: ConfigVersionUpdate) -> ConfigVersion: self.session.refresh(version) logger.info( - f"[ConfigVersionCrud.create_from_partial] Version created successfully | " + f"[ConfigVersionCrud.create_or_raise] Version created successfully | " f"{{'config_id': '{self.config_id}', 'version_id': '{version.id}'}}" ) return version - except Exception as e: + except Exception: + # Each error type has a central handler (DB, HTTPException, generic). self.session.rollback() - logger.error( - f"[ConfigVersionCrud.create_from_partial] Failed to create version | " - f"{{'config_id': '{self.config_id}', 'error': '{str(e)}'}}", - exc_info=True, - ) - raise HTTPException( - status_code=500, - detail="Unexpected error occurred: failed to create version", - ) + raise def _get_latest_version(self) -> ConfigVersion | None: """Get the latest version for the config.""" diff --git a/backend/app/crud/credentials.py b/backend/app/crud/credentials.py index 6916546dd..0881472c0 100644 --- a/backend/app/crud/credentials.py +++ b/backend/app/crud/credentials.py @@ -2,10 +2,9 @@ from typing import Any from sqlalchemy import delete -from sqlalchemy.exc import IntegrityError from sqlmodel import Session, select -from app.core.exception_handlers import HTTPException +from fastapi import HTTPException from app.core.providers import parse_provider_credentials, validate_provider from app.core.security import decrypt_credentials, encrypt_credentials from app.core.util import now @@ -39,29 +38,10 @@ def set_creds_for_org( credential=encrypted_credentials, ) credential.inserted_at = now() - try: - session.add(credential) - session.commit() - session.refresh(credential) - created_credentials.append(credential) - except IntegrityError as e: - session.rollback() - logger.error( - f"[set_creds_for_org] Integrity error while adding credentials | organization_id {organization_id}, project_id {project_id}, provider {provider}: {str(e)}", - exc_info=True, - ) - # Check if it's a duplicate constraint violation - if ( - "uq_credential_org_project_provider" in str(e) - or "unique constraint" in str(e).lower() - ): - raise HTTPException( - status_code=400, - detail=f"Credentials for provider '{provider}' already exist for this organization and project combination", - ) - raise ValueError( - f"Error while adding credentials for provider {provider}: {str(e)}" - ) + session.add(credential) + session.commit() + session.refresh(credential) + created_credentials.append(credential) logger.info( f"[set_creds_for_org] Successfully created credentials | organization_id {organization_id}, project_id {project_id}" ) @@ -179,7 +159,9 @@ def update_creds_for_org( ) -> list[Credential]: """Updates credentials for a specific provider of an organization.""" if not creds_in.provider or not creds_in.credential: - raise ValueError("Provider and credential must be provided") + raise HTTPException( + status_code=400, detail="Provider and credential must be provided" + ) provider = creds_in.provider.value credential_data = creds_in.credential_payload() @@ -272,12 +254,11 @@ def remove_provider_credential( rows_deleted = result.rowcount if rows_deleted == 0: session.rollback() - logger.error( - f"[remove_provider_credential] Failed to delete credential | organization_id {org_id}, provider {provider}, project_id {project_id}" + logger.warning( + f"[remove_provider_credential] Credential not found | organization_id {org_id}, provider {provider}, project_id {project_id}" ) raise HTTPException( - status_code=500, - detail="Failed to delete provider credential", + status_code=404, detail=f"No credentials found for provider '{provider}'." ) session.commit() logger.info( @@ -310,6 +291,16 @@ def remove_creds_for_org(*, session: Session, org_id: int, project_id: int) -> N rows_deleted = result.rowcount + if rows_deleted == 0: + session.rollback() + logger.warning( + f"[remove_creds_for_org] No credentials to delete | organization_id {org_id}, project_id {project_id}" + ) + raise HTTPException( + status_code=404, + detail="No credentials found for this organization and project.", + ) + if rows_deleted < expected_count: logger.error( f"[remove_creds_for_org] Failed to delete all credentials | organization_id {org_id}, project_id {project_id}, expected {expected_count}, deleted {rows_deleted}" diff --git a/backend/app/crud/document/doc_transformation_job.py b/backend/app/crud/document/doc_transformation_job.py index 27b1c387a..65285499e 100644 --- a/backend/app/crud/document/doc_transformation_job.py +++ b/backend/app/crud/document/doc_transformation_job.py @@ -1,16 +1,16 @@ import logging from uuid import UUID +from fastapi import HTTPException from sqlmodel import Session, and_, col, select +from app.core.util import now from app.models import ( DocTransformationJob, DocTransformJobCreate, DocTransformJobUpdate, ) from app.models.document import Document -from app.core.util import now -from app.core.exception_handlers import HTTPException logger = logging.getLogger(__name__) diff --git a/backend/app/crud/document/document.py b/backend/app/crud/document/document.py index 7790624b5..3325b4b03 100644 --- a/backend/app/crud/document/document.py +++ b/backend/app/crud/document/document.py @@ -1,11 +1,11 @@ import logging from uuid import UUID +from fastapi import HTTPException from sqlmodel import Session, and_, col, select -from app.models import Document from app.core.util import now -from app.core.exception_handlers import HTTPException +from app.models import Document logger = logging.getLogger(__name__) @@ -110,10 +110,7 @@ def update(self, document: Document) -> Document: if not document.project_id: document.project_id = self.project_id elif document.project_id != self.project_id: - error = "Invalid document ownership: project={} attempter={}".format( - self.project_id, - document.project_id, - ) + error = f"Invalid document ownership: project={self.project_id} attempter={document.project_id}" try: raise PermissionError(error) except PermissionError as err: diff --git a/backend/app/services/collections/providers/gemini.py b/backend/app/services/collections/providers/gemini.py index 43540566a..69690f561 100644 --- a/backend/app/services/collections/providers/gemini.py +++ b/backend/app/services/collections/providers/gemini.py @@ -156,6 +156,7 @@ def upload_files( doc.id, uploaded.name, str(delete_err), + exc_info=True, ) if doc.file_id: doc.file_id.pop(GOOGLE_AISTUDIO_PROVIDER, None) diff --git a/backend/app/services/documents/helpers.py b/backend/app/services/documents/helpers.py index 6b18f1af4..959b73522 100644 --- a/backend/app/services/documents/helpers.py +++ b/backend/app/services/documents/helpers.py @@ -100,12 +100,13 @@ def pre_transform_validation( - resolve actual transformer (or None if no target_format) Returns: (source_format, actual_transformer_or_none) - Raises: HTTPException(400) on client errors. + Raises: HTTPException(422) on an unreadable source format, 400 on an + unsupported transformation or transformer. """ try: source_format = get_file_format(src_filename) except ValueError as e: - raise HTTPException(status_code=400, detail=str(e)) + raise HTTPException(status_code=422, detail=str(e)) actual_transformer: Optional[str] = None if target_format: diff --git a/backend/app/services/guardrails/jobs.py b/backend/app/services/guardrails/jobs.py index 95f97c923..1ddf26c11 100644 --- a/backend/app/services/guardrails/jobs.py +++ b/backend/app/services/guardrails/jobs.py @@ -1,9 +1,9 @@ import logging +from fastapi import HTTPException from typing import Any from uuid import UUID, uuid4 from asgi_correlation_id import correlation_id -from fastapi import HTTPException from opentelemetry import trace from sqlmodel import Session @@ -85,8 +85,8 @@ def start_job( job_update=JobUpdate(status=JobStatus.FAILED, error_message=str(e)), ) raise HTTPException( - status_code=500, - detail="Internal server error while scheduling guardrails job", + status_code=503, + detail="Could not queue the guardrails job. Retry shortly.", ) logger.info( diff --git a/backend/app/services/llm/jobs.py b/backend/app/services/llm/jobs.py index 00b8690ed..09c46ce49 100644 --- a/backend/app/services/llm/jobs.py +++ b/backend/app/services/llm/jobs.py @@ -190,7 +190,7 @@ def start_job( job_update = JobUpdate(status=JobStatus.FAILED, error_message=str(e)) job_crud.update(job_id=job.id, job_update=job_update) raise HTTPException( - status_code=500, detail="Internal server error while executing LLM call" + status_code=503, detail="Could not queue the LLM call. Retry shortly." ) _set_traceability_attributes(span, task_id=str(task_id)) @@ -251,8 +251,8 @@ def start_chain_job( job_update = JobUpdate(status=JobStatus.FAILED, error_message=str(e)) job_crud.update(job_id=job.id, job_update=job_update) raise HTTPException( - status_code=500, - detail="Internal server error while executing LLM chain job", + status_code=503, + detail="Could not queue the LLM chain job. Retry shortly.", ) _set_traceability_attributes(span, task_id=str(task_id)) @@ -1433,7 +1433,8 @@ def execute_llm_call( except (Timeout, SoftTimeLimitExceeded): raise except HTTPException as e: - logger.error( + log = logger.error if e.status_code >= 500 else logger.warning + log( f"[execute_llm_call] HTTP error: {e.detail} | job_id={job_id}", exc_info=True, ) diff --git a/backend/app/tests/api/routes/documents/test_route_document_upload.py b/backend/app/tests/api/routes/documents/test_route_document_upload.py index 909249a42..2eca7ac5c 100644 --- a/backend/app/tests/api/routes/documents/test_route_document_upload.py +++ b/backend/app/tests/api/routes/documents/test_route_document_upload.py @@ -320,7 +320,7 @@ def test_upload_with_unsupported_file_extension( try: response = uploader.put(route, unsupported_file, target_format="markdown") - assert response.status_code == 400 + assert response.status_code == 422 error_data = response.json() assert "Unsupported file extension: .xyz" in error_data["error"] finally: diff --git a/backend/app/tests/api/routes/test_creds.py b/backend/app/tests/api/routes/test_creds.py index 8fd132f40..a79081eb0 100644 --- a/backend/app/tests/api/routes/test_creds.py +++ b/backend/app/tests/api/routes/test_creds.py @@ -237,7 +237,7 @@ def test_duplicate_credential_creation_fails( client: TestClient, user_api_key: TestAuthContext, ) -> None: - """Test that creating duplicate credentials fails with 400.""" + """Test that creating duplicate credentials fails with 409.""" api_key = "sk-" + generate_random_string(10) duplicate_credential = { "organization_id": user_api_key.organization_id, @@ -257,7 +257,7 @@ def test_duplicate_credential_creation_fails( headers={"X-API-KEY": user_api_key.key}, ) - assert response.status_code == 400 + assert response.status_code == 409 assert "already exist" in response.json()["error"] diff --git a/backend/app/tests/core/telemetry/test_telemetry_llm.py b/backend/app/tests/core/telemetry/test_telemetry_llm.py index 16aa496fa..20ee05d87 100644 --- a/backend/app/tests/core/telemetry/test_telemetry_llm.py +++ b/backend/app/tests/core/telemetry/test_telemetry_llm.py @@ -101,7 +101,7 @@ def test_response_attributes_with_usage_and_model(self) -> None: telemetry.set_gen_ai_response_attributes(span, response=response) set_calls = {c.args[0]: c.args[1] for c in span.set_attribute.call_args_list} assert set_calls["gen_ai.usage.total_tokens"] == 3 - assert set_calls["gen_ai.usage.output_tokens.reasoning"] == 4 + assert set_calls["gen_ai.usage.reasoning.output_tokens"] == 4 assert set_calls["gen_ai.response.model"] == "gpt-4o-2024" def test_response_attributes_without_usage(self) -> None: diff --git a/backend/app/tests/core/telemetry/test_telemetry_tracing.py b/backend/app/tests/core/telemetry/test_telemetry_tracing.py index 3fcf0e7d5..811f0e7fd 100644 --- a/backend/app/tests/core/telemetry/test_telemetry_tracing.py +++ b/backend/app/tests/core/telemetry/test_telemetry_tracing.py @@ -12,33 +12,7 @@ class TestSetupTelemetryInstrumentors: - def test_enabled_calls_instrument_on_both(self) -> None: - redis, botocore = MagicMock(), MagicMock() - with ( - patch.object(telemetry.settings, "SENTRY_DSN", None), - patch.object(telemetry, "TracerProvider", MagicMock()), - patch.object(telemetry.trace, "set_tracer_provider", MagicMock()), - patch.object(telemetry, "LoggingInstrumentor", MagicMock()), - patch.object(telemetry, "HTTPXClientInstrumentor", MagicMock()), - patch.object(telemetry, "RequestsInstrumentor", MagicMock()), - patch( - "opentelemetry.instrumentation.celery.CeleryInstrumentor", MagicMock() - ), - patch("opentelemetry.instrumentation.redis.RedisInstrumentor", redis), - patch( - "opentelemetry.instrumentation.botocore.BotocoreInstrumentor", botocore - ), - ): - telemetry.setup_telemetry() - - redis.return_value.instrument.assert_called_once() - botocore.return_value.instrument.assert_called_once() - - def test_redis_failure_does_not_propagate_and_botocore_still_instruments( - self, - ) -> None: - redis = MagicMock() - redis.return_value.instrument.side_effect = RuntimeError("boom") + def test_enabled_instruments_botocore(self) -> None: botocore = MagicMock() with ( patch.object(telemetry.settings, "SENTRY_DSN", None), @@ -50,7 +24,6 @@ def test_redis_failure_does_not_propagate_and_botocore_still_instruments( patch( "opentelemetry.instrumentation.celery.CeleryInstrumentor", MagicMock() ), - patch("opentelemetry.instrumentation.redis.RedisInstrumentor", redis), patch( "opentelemetry.instrumentation.botocore.BotocoreInstrumentor", botocore ), @@ -265,7 +238,6 @@ def test_registers_processor_and_propagator(self) -> None: patch( "opentelemetry.instrumentation.celery.CeleryInstrumentor", MagicMock() ), - patch("opentelemetry.instrumentation.redis.RedisInstrumentor", MagicMock()), patch( "opentelemetry.instrumentation.botocore.BotocoreInstrumentor", MagicMock(), diff --git a/backend/app/tests/core/test_config.py b/backend/app/tests/core/test_config.py index 4be4e932d..bd279d6f5 100644 --- a/backend/app/tests/core/test_config.py +++ b/backend/app/tests/core/test_config.py @@ -1,5 +1,4 @@ import pytest -from pydantic import TypeAdapter, ValidationError from app.core.config import Settings @@ -11,37 +10,9 @@ class TestSentryConfigDefaults: ("SENTRY_TRACES_SAMPLE_RATE", 1.0), ("SENTRY_SEND_DEFAULT_PII", False), ("SENTRY_ERROR_SAMPLE_RATE", 1.0), - ("SENTRY_PROFILE_SESSION_SAMPLE_RATE", 1.0), - ("SENTRY_PROFILE_LIFECYCLE", "trace"), ], ) def test_declared_default( self, field: str, expected: float | str | bool | None ) -> None: assert Settings.model_fields[field].default == expected - - -class TestSentryProfilingValidation: - @staticmethod - def _adapter(field: str) -> TypeAdapter[object]: - return TypeAdapter(Settings.model_fields[field].rebuild_annotation()) - - @pytest.mark.parametrize("value", [0.0, 0.25, 1.0]) - def test_session_sample_rate_accepts_unit_interval(self, value: float) -> None: - assert ( - self._adapter("SENTRY_PROFILE_SESSION_SAMPLE_RATE").validate_python(value) - == value - ) - - @pytest.mark.parametrize("value", [-0.1, 1.5]) - def test_session_sample_rate_rejects_out_of_range(self, value: float) -> None: - with pytest.raises(ValidationError): - self._adapter("SENTRY_PROFILE_SESSION_SAMPLE_RATE").validate_python(value) - - @pytest.mark.parametrize("value", ["manual", "trace"]) - def test_lifecycle_accepts_supported_modes(self, value: str) -> None: - assert self._adapter("SENTRY_PROFILE_LIFECYCLE").validate_python(value) == value - - def test_lifecycle_rejects_unknown_mode(self) -> None: - with pytest.raises(ValidationError): - self._adapter("SENTRY_PROFILE_LIFECYCLE").validate_python("always") diff --git a/backend/app/tests/core/test_exception_handlers.py b/backend/app/tests/core/test_exception_handlers.py index e7ee22a4f..dfca4f09d 100644 --- a/backend/app/tests/core/test_exception_handlers.py +++ b/backend/app/tests/core/test_exception_handlers.py @@ -1,16 +1,30 @@ +import logging +from collections.abc import Iterator from unittest.mock import patch import pytest +import sentry_sdk from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient from pydantic import BaseModel +from sentry_sdk.integrations.dedupe import DedupeIntegration +from sentry_sdk.integrations.logging import LoggingIntegration +from sentry_sdk.types import Event, Hint +from sqlalchemy.exc import ( + DataError, + OperationalError, + ProgrammingError, +) from app.core.config import settings from app.core.exception_handlers import ( + DB_ERROR_DETAIL, + GENERIC_ERROR_DETAIL, _sanitize_validation_errors, register_exception_handlers, ) from app.tests.utils.auth import TestAuthContext +from app.tests.utils.utils import make_integrity_error class TestSanitizeValidationErrors: @@ -138,16 +152,14 @@ def validate(body: Body) -> dict[str, bool]: return TestClient(app, raise_server_exceptions=False) - def test_generic_handler_captures_unhandled_exception( + def test_generic_handler_reports_unhandled_exception( self, client: TestClient ) -> None: - with patch( - "app.core.exception_handlers.sentry_sdk.capture_exception" - ) as capture: + with patch("app.core.exception_handlers.logger.error") as log_error: response = client.get("/boom") assert response.status_code == 500 - capture.assert_called_once() + log_error.assert_called_once() def test_http_exception_handler_does_not_capture(self, client: TestClient) -> None: with patch( @@ -203,3 +215,144 @@ def test_union_noise_filtered( for error in response.json()["errors"]: assert "openai-native" not in error["message"] assert "NativeCompletionConfig" not in error["field"] + + +def _app_raising(exc: Exception) -> TestClient: + app = FastAPI() + register_exception_handlers(app) + + @app.get("/boom") + def boom() -> None: + raise exc + + return TestClient(app, raise_server_exceptions=False) + + +class TestGenericErrorHandler: + def test_does_not_leak_exception_text(self) -> None: + response = _app_raising(ValueError("secret arn:aws:kms:key/abc")).get("/boom") + + assert response.status_code == 500 + assert response.json()["error"] == GENERIC_ERROR_DETAIL + assert "arn:aws:kms" not in response.text + + +class TestDbErrorHandler: + @pytest.mark.parametrize( + ("exc", "status_code"), + [ + (make_integrity_error("23505"), 409), + (make_integrity_error("23503"), 422), + (make_integrity_error("23502"), 500), + (DataError("INSERT", {}, Exception("too long")), 422), + (OperationalError("SELECT", {}, Exception("conn refused")), 503), + (ProgrammingError("SELECT", {}, Exception("bad sql")), 500), + ], + ) + def test_maps_status_without_leaking_sql( + self, exc: Exception, status_code: int + ) -> None: + response = _app_raising(exc).get("/boom") + + assert response.status_code == status_code + assert "INSERT" not in response.text + assert "9876543210" not in response.text + + def test_unmapped_db_error_uses_default_detail(self) -> None: + response = _app_raising(ProgrammingError("SELECT", {}, Exception("x"))).get( + "/boom" + ) + + assert response.json()["error"] == DB_ERROR_DETAIL + + def test_constraint_name_stays_out_of_client_message(self) -> None: + response = _app_raising( + make_integrity_error("23505", "uq_credential_org_project_provider") + ).get("/boom") + + assert response.status_code == 409 + assert response.json()["error"] == "Record already exists." + assert "uq_credential" not in response.text + + def test_client_side_db_error_logs_warning(self) -> None: + with ( + patch("app.core.exception_handlers.logger.warning") as log_warning, + patch("app.core.exception_handlers.logger.error") as log_error, + ): + _app_raising(make_integrity_error("23505")).get("/boom") + + log_warning.assert_called_once() + log_error.assert_not_called() + + +class TestHttpExceptionHandler: + def test_server_error_is_logged(self) -> None: + with patch("app.core.exception_handlers.logger.error") as log_error: + response = _app_raising( + HTTPException(status_code=502, detail="upstream failed") + ).get("/boom") + + assert response.status_code == 502 + log_error.assert_called_once() + + +@pytest.fixture +def sentry_events() -> Iterator[list[Event]]: + """Real Sentry client whose before_send records events instead of sending them.""" + events: list[Event] = [] + + def _record(event: Event, _hint: Hint) -> None: + events.append(event) + return None + + sentry_sdk.init( + dsn="https://public@sentry.invalid/1", + before_send=_record, + default_integrations=False, + auto_enabling_integrations=False, + integrations=[ + LoggingIntegration(event_level=logging.ERROR), + DedupeIntegration(), + ], + ) + yield events + sentry_sdk.get_global_scope().set_client(None) + + +class TestSentryEvents: + def test_server_http_exception_event_carries_status_tags( + self, sentry_events: list[Event] + ) -> None: + _app_raising(HTTPException(status_code=502, detail="upstream")).get("/boom") + + assert len(sentry_events) == 1 + tags = sentry_events[0]["tags"] + assert tags["http.status_code"] == "502" + assert tags["error.class"] == "server" + assert tags["http.route"] == "/boom" + assert sentry_events[0]["transaction"] == "GET /boom" + + def test_generic_handler_reports_exactly_one_event( + self, sentry_events: list[Event] + ) -> None: + _app_raising(ValueError("boom")).get("/boom") + + assert len(sentry_events) == 1 + + def test_db_server_error_event_carries_db_tags( + self, sentry_events: list[Event] + ) -> None: + _app_raising(make_integrity_error("23502", "llm_call_input_not_null")).get( + "/boom" + ) + + assert len(sentry_events) == 1 + tags = sentry_events[0]["tags"] + assert tags["http.status_code"] == "500" + assert tags["db.constraint"] == "llm_call_input_not_null" + + def test_db_conflict_creates_no_event(self, sentry_events: list[Event]) -> None: + response = _app_raising(make_integrity_error("23505")).get("/boom") + + assert response.status_code == 409 + assert sentry_events == [] diff --git a/backend/app/tests/core/test_middleware.py b/backend/app/tests/core/test_middleware.py index 6d902300f..2a56ecaf6 100644 --- a/backend/app/tests/core/test_middleware.py +++ b/backend/app/tests/core/test_middleware.py @@ -272,6 +272,9 @@ class ActiveClient: def is_active(self) -> bool: return True + def get_integration(self, _: object) -> None: + return None + monkeypatch.setattr(sentry_sdk, "get_client", lambda: ActiveClient()) monkeypatch.setattr( sentry_sdk.metrics, @@ -307,6 +310,9 @@ class InactiveClient: def is_active(self) -> bool: return False + def get_integration(self, _: object) -> None: + return None + monkeypatch.setattr(sentry_sdk, "get_client", lambda: InactiveClient()) monkeypatch.setattr( sentry_sdk.metrics, diff --git a/backend/app/tests/core/test_security.py b/backend/app/tests/core/test_security.py index eb7074ece..f45f2a6ae 100644 --- a/backend/app/tests/core/test_security.py +++ b/backend/app/tests/core/test_security.py @@ -1,10 +1,12 @@ import base64 import json from datetime import timedelta +from unittest.mock import MagicMock import boto3 import jwt import pytest +from fastapi import HTTPException from moto import mock_aws from sqlmodel import Session @@ -82,6 +84,33 @@ def test_dual_read_fernet_row_with_kms_active(self, monkeypatch, kms_key): monkeypatch.setattr(settings, "ENVIRONMENT", "staging") assert decrypt_credentials(fernet_encrypted) == creds + def test_kms_encrypt_failure_raises_502_without_leaking_cause( + self, monkeypatch, kms_key + ): + broken = MagicMock() + broken.generate_data_key.side_effect = Exception( + "arn:aws:kms:ap-south-1:secret" + ) + monkeypatch.setattr(security, "_kms_client", broken) + + with pytest.raises(HTTPException) as exc: + encrypt_credentials({"api_key": "sk-1"}) + + assert exc.value.status_code == 502 + assert "arn:aws:kms" not in exc.value.detail + + def test_kms_decrypt_failure_raises_502(self, monkeypatch, kms_key): + encrypted = encrypt_credentials({"api_key": "sk-1"}) + broken = MagicMock() + broken.decrypt.side_effect = Exception("arn:aws:kms:ap-south-1:secret") + monkeypatch.setattr(security, "_kms_client", broken) + + with pytest.raises(HTTPException) as exc: + decrypt_credentials(encrypted) + + assert exc.value.status_code == 502 + assert "arn:aws:kms" not in exc.value.detail + def test_kms_v2_envelope_roundtrip(self, kms_key): creds = {"openai": {"api_key": "sk-envelope-123"}} @@ -133,12 +162,14 @@ def test_v2_tampered_ciphertext_raises(self, kms_key): f"{base64.b64encode(bytes(ct)).decode()}" ) - with pytest.raises(ValueError, match="Failed to decrypt credentials"): + with pytest.raises(HTTPException) as exc: decrypt_credentials(tampered) + assert exc.value.status_code == 502 def test_v2_wrong_segment_count_raises(self, kms_key): - with pytest.raises(ValueError, match="Failed to decrypt credentials"): + with pytest.raises(HTTPException) as exc: decrypt_credentials(f"{KMS_ENVELOPE_PREFIX}onlyoneseg") + assert exc.value.status_code == 502 def test_encrypt_fernet_forces_fernet_even_with_kms_active(self, kms_key): creds = {"api_key": "sk-force-fernet"} diff --git a/backend/app/tests/crud/collections/collection/test_crud_collection_read_one_if_delete.py b/backend/app/tests/crud/collections/collection/test_crud_collection_read_one_if_delete.py new file mode 100644 index 000000000..1b044d51b --- /dev/null +++ b/backend/app/tests/crud/collections/collection/test_crud_collection_read_one_if_delete.py @@ -0,0 +1,57 @@ +"""Deleting an already-deleted collection is a conflict, not bad input.""" + +import uuid + +import pytest +from sqlmodel import Session + +from fastapi import HTTPException + +from app.core.util import now +from app.crud import CollectionCrud +from app.models import Collection, ProviderType +from app.tests.utils.utils import get_project + + +def _make_collection(db: Session, project_id: int, deleted: bool) -> Collection: + collection = Collection( + project_id=project_id, + knowledge_base_id="vs_test", + knowledge_base_provider="openai vector store", + provider=ProviderType.openai, + deleted_at=now() if deleted else None, + ) + db.add(collection) + db.commit() + db.refresh(collection) + return collection + + +def test_already_deleted_raises_409(db: Session) -> None: + project = get_project(db) + collection = _make_collection(db, project.id, deleted=True) + + with pytest.raises(HTTPException) as exc: + CollectionCrud(db, project.id).read_one_if_delete(collection.id) + + assert exc.value.status_code == 409 + assert "already deleted" in exc.value.detail + + +def test_active_collection_is_returned(db: Session) -> None: + project = get_project(db) + collection = _make_collection(db, project.id, deleted=False) + + assert ( + CollectionCrud(db, project.id).read_one_if_delete(collection.id).id + == collection.id + ) + + +def test_missing_collection_still_raises_404(db: Session) -> None: + project = get_project(db) + + with pytest.raises(HTTPException) as exc: + CollectionCrud(db, project.id).read_one_if_delete(uuid.uuid4()) + + assert exc.value.status_code == 404 diff --git a/backend/app/tests/crud/config/test_config.py b/backend/app/tests/crud/config/test_config.py index e3c939ba6..6c7bdac6f 100644 --- a/backend/app/tests/crud/config/test_config.py +++ b/backend/app/tests/crud/config/test_config.py @@ -2,6 +2,7 @@ import pytest from fastapi import HTTPException +from sqlalchemy.exc import IntegrityError from sqlmodel import Session from app.crud.config import ConfigCrud @@ -14,7 +15,7 @@ from app.models.config.config import ConfigTag from app.models.llm.request import NativeCompletionConfig from app.tests.utils.test_data import create_test_config, create_test_project -from app.tests.utils.utils import random_lower_string +from app.tests.utils.utils import make_integrity_error, random_lower_string @pytest.fixture @@ -569,3 +570,68 @@ def test_config_tag_is_immutable(db: Session) -> None: with pytest.raises(ValueError, match="tag is immutable"): config.tag = ConfigTag.ASSESSMENT + + +def test_create_config_concurrent_duplicate_propagates( + db: Session, example_config_blob: ConfigBlob, monkeypatch +) -> None: + """A commit-time unique violation reaches the central DB handler (409).""" + project = create_test_project(db) + config_crud = ConfigCrud(session=db, project_id=project.id) + + def _fail_commit() -> None: + raise make_integrity_error("23505") + + monkeypatch.setattr(db, "commit", _fail_commit) + + with pytest.raises(IntegrityError): + config_crud.create_or_raise( + ConfigCreate( + name=f"test-config-{random_lower_string()}", + config_blob=example_config_blob, + ) + ) + + +def test_create_config_http_exception_propagates( + db: Session, example_config_blob: ConfigBlob, monkeypatch +) -> None: + """A typed HTTPException inside the write keeps its status, not a 500.""" + project = create_test_project(db) + config_crud = ConfigCrud(session=db, project_id=project.id) + + def _fail_add(*_: object, **__: object) -> None: + raise HTTPException(status_code=503, detail="unavailable") + + monkeypatch.setattr(db, "add", _fail_add) + + with pytest.raises(HTTPException) as exc: + config_crud.create_or_raise( + ConfigCreate( + name=f"test-config-{random_lower_string()}", + config_blob=example_config_blob, + ) + ) + + assert exc.value.status_code == 503 + + +def test_create_config_foreign_key_violation_is_not_reported_as_duplicate( + db: Session, example_config_blob: ConfigBlob, monkeypatch +) -> None: + """Only a unique violation is a name conflict; other IntegrityErrors propagate.""" + project = create_test_project(db) + config_crud = ConfigCrud(session=db, project_id=project.id) + + def _fail_commit() -> None: + raise make_integrity_error("23503") + + monkeypatch.setattr(db, "commit", _fail_commit) + + with pytest.raises(IntegrityError): + config_crud.create_or_raise( + ConfigCreate( + name=f"test-config-{random_lower_string()}", + config_blob=example_config_blob, + ) + ) diff --git a/backend/app/tests/crud/config/test_version.py b/backend/app/tests/crud/config/test_version.py index ed122b721..a76431773 100644 --- a/backend/app/tests/crud/config/test_version.py +++ b/backend/app/tests/crud/config/test_version.py @@ -3,6 +3,7 @@ import pytest from fastapi import HTTPException +from sqlalchemy.exc import IntegrityError from sqlmodel import Session from app.crud.config import ConfigVersionCrud @@ -14,6 +15,7 @@ create_test_project, create_test_version, ) +from app.tests.utils.utils import make_integrity_error @pytest.fixture @@ -650,3 +652,45 @@ def test_does_not_mutate_shared_params_dict(self) -> None: # prior_blob must NOT have been mutated through the shared reference assert prior_blob["completion"]["params"]["temperature"] == 0.7 assert shared_params["temperature"] == 0.7 + + +def test_create_version_concurrent_duplicate_propagates( + db: Session, example_config_blob: ConfigBlob, monkeypatch +) -> None: + """A commit-time unique violation reaches the central DB handler (409).""" + config = create_test_config(db) + version_crud = ConfigVersionCrud( + session=db, project_id=config.project_id, config_id=config.id + ) + + def _fail_commit() -> None: + raise make_integrity_error("23505") + + monkeypatch.setattr(db, "commit", _fail_commit) + + with pytest.raises(IntegrityError): + version_crud.create_or_raise( + ConfigVersionUpdate(config_blob=example_config_blob.model_dump()) + ) + + +def test_create_version_http_exception_propagates( + db: Session, example_config_blob: ConfigBlob, monkeypatch +) -> None: + """A typed HTTPException inside the write keeps its status, not a 500.""" + config = create_test_config(db) + version_crud = ConfigVersionCrud( + session=db, project_id=config.project_id, config_id=config.id + ) + monkeypatch.setattr( + version_crud, + "_get_next_version", + MagicMock(side_effect=HTTPException(status_code=503, detail="unavailable")), + ) + + with pytest.raises(HTTPException) as exc: + version_crud.create_or_raise( + ConfigVersionUpdate(config_blob=example_config_blob.model_dump()) + ) + + assert exc.value.status_code == 503 diff --git a/backend/app/tests/crud/test_credentials.py b/backend/app/tests/crud/test_credentials.py index d61d29c2b..4473a20d4 100644 --- a/backend/app/tests/crud/test_credentials.py +++ b/backend/app/tests/crud/test_credentials.py @@ -1,8 +1,10 @@ import json +from unittest.mock import MagicMock import pytest from fastapi import HTTPException from pydantic import ValidationError +from sqlalchemy.exc import IntegrityError from sqlmodel import Session from app.crud import ( @@ -435,3 +437,124 @@ def test_langfuse_credential_validation(db: Session) -> None: ) assert len(created_credentials) == 1 assert created_credentials[0].provider == "langfuse" + + +def test_remove_provider_credential_missing_raises_404(db: Session) -> None: + """Deleting a credential that was never stored is a 404, not a 500.""" + project = create_test_project(db) + + with pytest.raises(HTTPException) as exc: + remove_provider_credential( + session=db, + org_id=project.organization_id, + provider="openai", + project_id=project.id, + ) + + assert exc.value.status_code == 404 + + +def test_remove_provider_credential_race_deletes_nothing_raises_404( + db: Session, monkeypatch +) -> None: + """Row vanishes between the existence check and the delete: 404, not 500.""" + _, project = create_test_credential(db) + monkeypatch.setattr( + "app.crud.credentials.get_provider_credential", lambda **k: MagicMock() + ) + monkeypatch.setattr(db, "exec", lambda *a, **k: MagicMock(rowcount=0)) + + with pytest.raises(HTTPException) as exc: + remove_provider_credential( + session=db, + org_id=project.organization_id, + provider="openai", + project_id=project.id, + ) + + assert exc.value.status_code == 404 + + +def test_remove_creds_for_org_when_none_exist_raises_404(db: Session) -> None: + project = create_test_project(db) + + with pytest.raises(HTTPException) as exc: + remove_creds_for_org( + session=db, + org_id=project.organization_id, + project_id=project.id, + ) + + assert exc.value.status_code == 404 + + +def test_duplicate_provider_propagates_unique_violation(db: Session) -> None: + """The central DB handler turns this into a 409; crud lets it propagate.""" + _, project = create_test_credential(db) + + creds_create = CredsCreate( + is_active=True, + credential={"openai": {"api_key": "another-key"}}, + ) + + with pytest.raises(IntegrityError) as exc: + set_creds_for_org( + session=db, + creds_add=creds_create, + organization_id=project.organization_id, + project_id=project.id, + ) + + assert exc.value.orig.diag.constraint_name == "uq_credential_org_project_provider" + + +def test_integrity_error_propagates_to_db_handler(db: Session, monkeypatch) -> None: + """Classifying DB errors is the central handler's job, not crud's.""" + project = create_test_project(db) + + def _fail_commit() -> None: + raise IntegrityError("INSERT", {}, Exception("foreign key violation")) + + monkeypatch.setattr(db, "commit", _fail_commit) + + with pytest.raises(IntegrityError): + set_creds_for_org( + session=db, + creds_add=CredsCreate( + is_active=True, credential={"openai": {"api_key": "sk-test"}} + ), + organization_id=project.organization_id, + project_id=project.id, + ) + + +def test_update_creds_without_provider_raises_400(db: Session) -> None: + project = create_test_project(db) + + with pytest.raises(HTTPException) as exc: + update_creds_for_org( + session=db, + org_id=project.organization_id, + project_id=project.id, + creds_in=CredsUpdate.model_construct(provider=None, credential=None), + ) + + assert exc.value.status_code == 400 + + +def test_remove_creds_for_org_race_deletes_nothing_raises_404( + db: Session, monkeypatch +) -> None: + """Rows vanish between the existence check and the delete: 404, not 500.""" + _, project = create_test_credential(db) + monkeypatch.setattr( + "app.crud.credentials.get_creds_by_org", lambda **k: [MagicMock()] + ) + monkeypatch.setattr(db, "exec", lambda *a, **k: MagicMock(rowcount=0)) + + with pytest.raises(HTTPException) as exc: + remove_creds_for_org( + session=db, org_id=project.organization_id, project_id=project.id + ) + + assert exc.value.status_code == 404 diff --git a/backend/app/tests/crud/test_onboarding.py b/backend/app/tests/crud/test_onboarding.py index 13f652bfa..84ac96ae2 100644 --- a/backend/app/tests/crud/test_onboarding.py +++ b/backend/app/tests/crud/test_onboarding.py @@ -1,7 +1,9 @@ import pytest from fastapi import HTTPException +from sqlalchemy.exc import IntegrityError from sqlmodel import Session, select + from app.crud.onboarding import onboard_project from app.crud import ( get_organization_by_name, @@ -19,7 +21,11 @@ APIKey, Credential, ) -from app.tests.utils.utils import random_lower_string, random_email +from app.tests.utils.utils import ( + make_integrity_error, + random_email, + random_lower_string, +) from app.tests.utils.test_data import create_test_organization, create_test_project from app.tests.utils.user import create_random_user @@ -289,3 +295,51 @@ def test_onboard_project_response_data_integrity(db: Session) -> None: assert project.name == response.project_name assert project.organization_id == response.organization_id assert user.email == response.user_email + + +def test_onboard_kms_failure_raises_502(db: Session, monkeypatch) -> None: + """A KMS outage while encrypting credentials is upstream, not our bug.""" + monkeypatch.setattr( + "app.crud.onboarding.encrypt_credentials", + lambda _: (_ for _ in ()).throw( + HTTPException(status_code=502, detail="kms down") + ), + ) + + with pytest.raises(HTTPException) as exc: + onboard_project( + session=db, + onboard_in=OnboardingRequest( + organization_name=random_lower_string(), + project_name=random_lower_string(), + email=random_email(), + user_name=random_lower_string(), + password=random_lower_string(), + credentials=[{"openai": {"api_key": "sk-test"}}], + ), + ) + + assert exc.value.status_code == 502 + + +def test_onboard_concurrent_commit_conflict_propagates( + db: Session, monkeypatch +) -> None: + """A commit-time unique violation reaches the central DB handler (409).""" + + def _raise_integrity_error() -> None: + raise make_integrity_error("23505") + + monkeypatch.setattr(db, "commit", _raise_integrity_error) + + with pytest.raises(IntegrityError): + onboard_project( + session=db, + onboard_in=OnboardingRequest( + organization_name=random_lower_string(), + project_name=random_lower_string(), + email=random_email(), + user_name=random_lower_string(), + password=random_lower_string(), + ), + ) diff --git a/backend/app/tests/services/guardrails/test_jobs.py b/backend/app/tests/services/guardrails/test_jobs.py index 84bca8290..a09da6c2b 100644 --- a/backend/app/tests/services/guardrails/test_jobs.py +++ b/backend/app/tests/services/guardrails/test_jobs.py @@ -4,6 +4,7 @@ import pytest from fastapi import HTTPException + from sqlmodel import Session, select from app.models import Job, JobStatus, JobType @@ -143,7 +144,7 @@ def test_start_job_happy_path_creates_row_and_enqueues( assert kwargs["job_id"] == str(job.id) -def test_start_job_celery_failure_marks_failed_and_raises_500( +def test_start_job_celery_failure_marks_failed_and_raises_503( db: Session, user_api_key: TestAuthContext ) -> None: with patch("app.services.guardrails.jobs.start_guardrails_job") as mock_enqueue: @@ -156,7 +157,7 @@ def test_start_job_celery_failure_marks_failed_and_raises_500( organization_id=user_api_key.organization_id, ) - assert exc.value.status_code == 500 + assert exc.value.status_code == 503 rows = db.exec(select(Job).where(Job.project_id == user_api_key.project_id)).all() failed = [r for r in rows if r.status == JobStatus.FAILED] diff --git a/backend/app/tests/services/llm/test_jobs.py b/backend/app/tests/services/llm/test_jobs.py index 0d577a44f..5365271e3 100644 --- a/backend/app/tests/services/llm/test_jobs.py +++ b/backend/app/tests/services/llm/test_jobs.py @@ -1,12 +1,13 @@ import httpx import pytest +from fastapi import HTTPException from unittest.mock import patch, MagicMock from uuid import UUID, uuid4 from celery.exceptions import SoftTimeLimitExceeded from gevent import Timeout -from fastapi import HTTPException + from sqlmodel import Session, select from app.crud import JobCrud @@ -104,10 +105,8 @@ def test_start_job_celery_scheduling_fails( with pytest.raises(HTTPException) as exc_info: start_job(db, llm_call_request, project.id, project.organization_id) - assert exc_info.value.status_code == 500 - assert "Internal server error while executing LLM call" in str( - exc_info.value.detail - ) + assert exc_info.value.status_code == 503 + assert "Could not queue the LLM call" in str(exc_info.value.detail) def test_start_job_exception_during_job_creation( self, db: Session, llm_call_request: LLMCallRequest @@ -2503,10 +2502,8 @@ def test_start_chain_job_celery_failure(self, db: Session, chain_request): with pytest.raises(HTTPException) as exc_info: start_chain_job(db, chain_request, project.id, project.organization_id) - assert exc_info.value.status_code == 500 - assert "Internal server error while executing LLM chain job" in str( - exc_info.value.detail - ) + assert exc_info.value.status_code == 503 + assert "Could not queue the LLM chain job" in str(exc_info.value.detail) class TestExecuteChainJob: diff --git a/backend/app/tests/test_utils.py b/backend/app/tests/test_utils.py index 976cad1e7..255d9e794 100644 --- a/backend/app/tests/test_utils.py +++ b/backend/app/tests/test_utils.py @@ -435,7 +435,7 @@ def test_failure_logs_status_and_payload_without_user_data( mock_session.post.return_value = MagicMock(status_code=502) mock_session_cls.return_value.__enter__.return_value = mock_session - with caplog.at_level("ERROR", logger="app.utils"): + with caplog.at_level("WARNING", logger="app.utils"): assert not send_callback("https://api.example.com/hook", self._payload) line = next( diff --git a/backend/app/tests/utils/utils.py b/backend/app/tests/utils/utils.py index 8bf7b1971..fd3a73c2b 100644 --- a/backend/app/tests/utils/utils.py +++ b/backend/app/tests/utils/utils.py @@ -1,9 +1,11 @@ import random import string +from types import SimpleNamespace from uuid import UUID from typing import Type, TypeVar from fastapi.testclient import TestClient +from sqlalchemy.exc import IntegrityError from sqlmodel import Session, select from app.core.config import settings @@ -14,6 +16,16 @@ T = TypeVar("T") +def make_integrity_error( + sqlstate: str | None, constraint: str | None = None +) -> IntegrityError: + """IntegrityError whose driver error carries psycopg's sqlstate and diag, as Postgres raises it.""" + orig = Exception("integrity violation") + orig.sqlstate = sqlstate # type: ignore[attr-defined] + orig.diag = SimpleNamespace(constraint_name=constraint) # type: ignore[attr-defined] + return IntegrityError("INSERT", {"input": "9876543210"}, orig) + + def random_lower_string() -> str: return "".join(random.choices(string.ascii_lowercase, k=32)) diff --git a/backend/app/utils.py b/backend/app/utils.py index 1d215fc9f..c8692ec96 100644 --- a/backend/app/utils.py +++ b/backend/app/utils.py @@ -608,7 +608,7 @@ def send_callback( except requests.RequestException as e: status = e.response.status_code if e.response is not None else None - logger.error( + logger.warning( f"[send_callback] Callback failed | url={callback_url} | status={status} " f"| duration_ms={(time.perf_counter() - started_at) * 1000:.2f} " f"| bytes={len(raw_body)} | payload={log_payload} | error={e}", diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 9e584035b..f4088d114 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -30,7 +30,6 @@ dependencies = [ "opentelemetry-instrumentation-requests>=0.51b0", "opentelemetry-instrumentation-logging>=0.51b0", "opentelemetry-instrumentation-sqlalchemy>=0.51b0", - "opentelemetry-instrumentation-redis>=0.51b0", "opentelemetry-instrumentation-botocore>=0.51b0", "pyjwt>=2.15.0,<3.0.0", "boto3>=1.37.20", diff --git a/backend/uv.lock b/backend/uv.lock index bacfc69d1..a9fe6e378 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -329,7 +329,6 @@ dependencies = [ { name = "opentelemetry-instrumentation-fastapi" }, { name = "opentelemetry-instrumentation-httpx" }, { name = "opentelemetry-instrumentation-logging" }, - { name = "opentelemetry-instrumentation-redis" }, { name = "opentelemetry-instrumentation-requests" }, { name = "opentelemetry-instrumentation-sqlalchemy" }, { name = "opentelemetry-sdk" }, @@ -404,7 +403,6 @@ requires-dist = [ { name = "opentelemetry-instrumentation-fastapi", specifier = ">=0.51b0" }, { name = "opentelemetry-instrumentation-httpx", specifier = ">=0.51b0" }, { name = "opentelemetry-instrumentation-logging", specifier = ">=0.51b0" }, - { name = "opentelemetry-instrumentation-redis", specifier = ">=0.51b0" }, { name = "opentelemetry-instrumentation-requests", specifier = ">=0.51b0" }, { name = "opentelemetry-instrumentation-sqlalchemy", specifier = ">=0.51b0" }, { name = "opentelemetry-sdk", specifier = ">=1.30.0" }, @@ -2807,21 +2805,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e2/8e/7577914681d77b180f8d6dcbac435be8e4ca6add6315da2d01ac4289eaa3/opentelemetry_instrumentation_logging-0.65b0-py3-none-any.whl", hash = "sha256:68365b31755c844f1e85f07dcd217839ff92f2d278a214bdf02d4dc806f9d915", size = 15727, upload-time = "2026-07-16T15:25:18.774Z" }, ] -[[package]] -name = "opentelemetry-instrumentation-redis" -version = "0.65b0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "opentelemetry-api" }, - { name = "opentelemetry-instrumentation" }, - { name = "opentelemetry-semantic-conventions" }, - { name = "wrapt" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/21/e5/b369897a9ff902eb028f1f999c9f9a4602f0b30fe1e97298d2c15fe5b8b0/opentelemetry_instrumentation_redis-0.65b0.tar.gz", hash = "sha256:f16409f189092984ff922f26939e7be79509365f5c9d202308c13fddf1368147", size = 17218, upload-time = "2026-07-16T15:26:17.243Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/9d/2a/5b874bf8824e7cb995b22e1e61e1b5bc42810023d81612c44edbfd2c48a3/opentelemetry_instrumentation_redis-0.65b0-py3-none-any.whl", hash = "sha256:7d135f61db9d72416e1f382012fbc13de6f5e726491bebd0344a9c42a60e6769", size = 14658, upload-time = "2026-07-16T15:25:29.146Z" }, -] - [[package]] name = "opentelemetry-instrumentation-requests" version = "0.65b0" diff --git a/docs/architecture/kaapi-llm-call-ARCHITECTURE.md b/docs/architecture/kaapi-llm-call-ARCHITECTURE.md index 92ce5e065..5034124ab 100644 --- a/docs/architecture/kaapi-llm-call-ARCHITECTURE.md +++ b/docs/architecture/kaapi-llm-call-ARCHITECTURE.md @@ -89,7 +89,7 @@ backend/app/ │ ├── celery/ │ ├── tasks/job_execution.py run_llm_job (high_priority, priority=9, soft timeout) -│ └── utils.py start_llm_job() — enqueue with OTel trace headers +│ └── utils.py start_llm_job() — enqueue; OTel's Celery instrumentation carries the trace > **Worker concurrency model:** Celery currently runs with the **prefork** pool. > Gevent is being evaluated and will replace prefork once validated in production. @@ -518,7 +518,7 @@ erDiagram ## 9. Observability -Three layers, all keyed off the same `trace_id` (the request `correlation_id`, propagated to Celery via injected OTel headers): +Three layers, all keyed off the same `trace_id` (the request `correlation_id`, passed to Celery as the task's `trace_id` argument; the OTel trace itself is carried by the Celery instrumentation): ```mermaid flowchart TD diff --git a/docs/architecture/kaapi-observability-ARCHITECTURE.md b/docs/architecture/kaapi-observability-ARCHITECTURE.md index 25eb4460a..2849bd7e3 100644 --- a/docs/architecture/kaapi-observability-ARCHITECTURE.md +++ b/docs/architecture/kaapi-observability-ARCHITECTURE.md @@ -2,7 +2,7 @@ ## Overview -Kaapi collects **traces, metrics and logs** from both processes, the FastAPI web app and the Celery worker, and sends all of them to **one Sentry project**. Errors, cron check-ins and profiles go to the same place, so one request can be followed from the HTTP call to the Celery task, the provider call and the callback. +Kaapi collects **traces, metrics and logs** from both processes, the FastAPI web app and the Celery worker, and sends all of them to **one Sentry project**. Errors and cron check-ins go to the same place, so one request can be followed from the HTTP call to the Celery task, the provider call and the callback. - **OpenTelemetry creates spans, Sentry stores them** (`instrumenter="otel"`). Sentry's own span integrations are off, so nothing is recorded twice. - **One package, `core/telemetry/`**, holds all of it. API and worker share one `init_sentry()`. @@ -30,7 +30,7 @@ flowchart LR FLT["sentry/filters.py\nscrub · drop"] end - Sentry[("Sentry project\nIssues · Traces · Logs · Metrics\nProfiles · Crons · Uptime")] + Sentry[("Sentry project\nIssues · Traces · Logs · Metrics\nCrons · Uptime")] LF[("Langfuse\nmessage bodies only")] CW[("CloudWatch\nstdout text")] @@ -69,14 +69,14 @@ core/telemetry/ - **One switch:** `OTEL_ENABLED` is checked only where instrumentation starts (`main.py`, `celery_app.py`, `core/db.py`). The telemetry functions themselves have no flags. - **No double recording:** Sentry's FastAPI, Starlette, SQLAlchemy, httpx (and, on the API, Celery) integrations are disabled because OTel already covers them. The worker keeps Sentry's Celery integration for task-failure Issues. -- **Sampling and profiling** come from settings, each validated to 0–1: `SENTRY_TRACES_SAMPLE_RATE`, `SENTRY_ERROR_SAMPLE_RATE`, `SENTRY_PROFILE_SESSION_SAMPLE_RATE` (continuous profiling, tied to traces). +- **Sampling** comes from settings, each validated to 0–1: `SENTRY_TRACES_SAMPLE_RATE`, `SENTRY_ERROR_SAMPLE_RATE`. Profiling is off: Sentry rate-limits profile chunks once the quota runs out, and the SDK then backs off and drops transactions too. - **Environments:** production and staging report to separate Sentry projects (`kaapi-production`, `kaapi-staging`). --- ## 1. Tracing -**What is traced:** HTTP requests, DB queries, Celery tasks, Redis, S3/KMS (botocore), outbound HTTP, and every LLM call. These feed Sentry's ready-made views: **AI Insights** (LLM calls), **Queries** (DB), **Queues** (Celery), **Caches** (Redis) and **Outbound API Requests**. +**What is traced:** HTTP requests, DB queries, Celery tasks, S3/KMS (botocore), outbound HTTP, and every LLM call. These feed Sentry's ready-made views: **AI Insights** (LLM calls), **Queries** (DB), **Queues** (Celery) and **Outbound API Requests**. Redis is not instrumented: its spans were only Celery result-backend bookkeeping and rate-limit counters. The broker (RabbitMQ) shows only as Celery's `apply_async` span, which links the API request to the worker trace. **Waterfall of one `/llm/call`**, as it appears in Sentry: @@ -112,7 +112,7 @@ flowchart LR F --> Sentry[("Sentry Traces")] ``` -- **One trace from API to worker:** the Sentry Celery integration adds trace headers on enqueue, and the task re-attaches the parent. Self re-enqueuing poll loops start a fresh trace so one trace does not span hours. +- **One trace from API to worker:** OTel's Celery instrumentation carries the trace on enqueue and parents the worker's `run/...` span to `apply_async`; task spans nest under `run/...`. Trace context is never injected or attached by hand: doing so reparented task spans to the API request and split them into separate Sentry transactions. Self re-enqueuing poll loops start a fresh trace so one trace does not span hours. - **Root attributes are kept:** Sentry's processor copies attributes only for child spans; ours copies root-span attributes too. - **Provider HTTP spans are suppressed** inside LLM calls, since the `chat ` span already covers the call. - **Spans are flushed** after every Celery task and in LLM job `finally` blocks, so a recycled worker does not lose them. diff --git a/docs/wiki/cross-cutting/exceptions.md b/docs/wiki/cross-cutting/exceptions.md index 57281a881..1c8104f85 100644 --- a/docs/wiki/cross-cutting/exceptions.md +++ b/docs/wiki/cross-cutting/exceptions.md @@ -3,9 +3,25 @@ All paths relative to `backend/app/`. ## Global handlers -- `core/exception_handlers.py` — registered on the FastAPI app; maps exception types to HTTP responses. +- `core/exception_handlers.py` — registered on the FastAPI app; maps exception types to HTTP responses. The `Exception` catch-all returns a fixed message, never `str(exc)`, so provider bodies and DB text don't reach callers. Unhandled `SQLAlchemyError` gets its own handler: a short generic message with 409 (unique violation, SQLSTATE 23505), 422 (foreign key 23503, or `DataError`), 503 (`OperationalError`) or 500 (any other); 4xx logs at warning, 5xx at error; SQL and parameters only go to logs. +- DB errors are classified only in `db_error_handler`: crud/services let `SQLAlchemyError` propagate (rollback, then `raise`), never catch `IntegrityError` to build a 409 or match error text. Clients get a short generic message per status; the SQLSTATE and constraint name go only to the log and the `db.constraint` Sentry tag. Sentry events carry `http.status_code`, `error.class` and `db.constraint` tags. Exception: eval `run_name` keeps its own 409 (`run_name_already_exists` is a client contract). - `core/middleware.py` — request-level middleware. +## Status-code selection +Raise `HTTPException` directly with the right status code from routes, crud, and service. Never raise a bare `ValueError` for a caller-facing failure: nothing catches it, so it becomes a 500 with a generic body. + +| Condition | Code | +|---|---| +| Resource absent (ours, or one we reference upstream) | 404 | +| State conflict (already exists, already deleted) | 409 | +| Payload unparseable or wrong shape | 422 | +| Valid shape, unacceptable value | 400 | +| Upstream errored or unreachable (Langfuse, OpenAI, Gemini, KMS, object storage) | 502 | +| Our async infra could not accept the work (Celery broker) | 503 | +| Our bug (DB write failed, unexpected exception) | 500 | + +- A broad `except Exception` around a DB write must re-raise `HTTPException` first, otherwise a typed 4xx/5xx gets flattened to 500. + ## Provider/SDK error convention - `.claude/conventions/error-handling.md` — the standardized pattern for wrapping provider/SDK exceptions at service/crud call sites. Follow it for any new OpenAI/Gemini/Anthropic call site. diff --git a/docs/wiki/cross-cutting/observability.md b/docs/wiki/cross-cutting/observability.md index c2915a731..43b1dd95a 100644 --- a/docs/wiki/cross-cutting/observability.md +++ b/docs/wiki/cross-cutting/observability.md @@ -14,9 +14,10 @@ OTel-first, Sentry as sole in-process sink (`instrumenter="otel"`). Init at `mai - `core/telemetry/sentry/filters.py` — `before_send_transaction_filter` (drops probe transactions, scrubs genai content; DB spans are kept so Sentry Queries has data) and `before_send_error_filter` (drops probe events, always drops the request body, scrubs headers/cookies/query when `SENTRY_SEND_DEFAULT_PII` off, redacts LLM task kwargs). - GenAI content never reaches Sentry, in either PII mode. Four locks in `core/telemetry/sentry/filters.py`, wired by `init_sentry()` in `core/telemetry/sentry/init.py` for both API and worker: `scrub_genai_content` strips `_GENAI_CONTENT_KEYS` (prompt/completion span+log attributes) from every event, `before_send_log_filter` does the same for logs, `genai_privacy_integrations()` pins Sentry's auto-enabled AI integrations to `include_prompts=False`, and `include_local_variables=False` / `max_request_body_size="never"` keep prompts out of stack frames and request bodies. Langfuse (isolated tracer provider, `core/langfuse/langfuse.py`) stays the only sink for message bodies. - Corollary for log lines on LLM paths: log lengths/ids/types, never prompt or response text — `enable_logs=True` ships every INFO record to Sentry. -- Sampling/profiling/PII: `SENTRY_TRACES_SAMPLE_RATE`, `SENTRY_PROFILE_SESSION_SAMPLE_RATE`, `SENTRY_PROFILE_LIFECYCLE`, `SENTRY_SEND_DEFAULT_PII`, `SENTRY_ERROR_SAMPLE_RATE` (config.py; defaults preserve current behavior). -- Trace propagation: `CeleryIntegration(propagate_traces=True)` links API to worker as one trace; poll-loop re-enqueues pass `SENTRY_NO_PROPAGATE_HEADERS` (`celery/tasks/job_execution.py`) to start a fresh trace. -- Error capture: generic handler in `core/exception_handlers.py` calls `capture_exception`. +- Sampling/PII: `SENTRY_TRACES_SAMPLE_RATE`, `SENTRY_SEND_DEFAULT_PII`, `SENTRY_ERROR_SAMPLE_RATE` (config.py; defaults preserve current behavior). +- Trace propagation: OTel's `CeleryInstrumentor` parents the worker's `run/...` span to `apply_async`; never inject/extract trace context by hand around tasks (it reparents task spans to the API request and Sentry splits them into separate transactions). `CeleryIntegration(propagate_traces=True)` links API to worker as one trace; poll-loop re-enqueues pass `SENTRY_NO_PROPAGATE_HEADERS` (`celery/tasks/job_execution.py`) to start a fresh trace. +- Error capture: handlers in `core/exception_handlers.py` report via `logger.error(exc_info=exc)` (5xx only). Don't add `capture_exception` next to it: on sentry-sdk 2.54 Dedupe does not merge the two, so each error becomes two events. +- Celery tasks: `task_prerun` in `celery/celery_app.py` tags each task's Sentry scope with `celery.task_name` and, from the task kwargs, `tenant.project_id` / `tenant.org_id` (Sentry isolates the scope per task, so tags don't leak between tasks). - Tenant impact (both in `core/telemetry/context.py`, called from `api/deps.py`): `set_request_log_context` puts org/project in the log context + Sentry tags; `bind_sentry_user` binds user/org/project via `sentry_sdk.set_user` so issues report users/orgs affected. The user binding is scope-only — ids in the log context would stamp per-user cardinality on every INFO record `enable_logs` ships. Per-request identity is the `correlation_id` tag (`core/middleware.py`), not the user binding. - Crons: `@sentry_sdk.monitor` on `/cron/*` endpoints (`api/routes/cron.py`), e.g. `llm-call-retention-cron-job` (daily 09:00 IST). - Runbook (alerts/dashboards/debugging): `features/sentry-utilization/SENTRY-RUNBOOK.md`. @@ -28,5 +29,5 @@ OTel-first, Sentry as sole in-process sink (`instrumenter="otel"`). Init at `mai - Sentry Logs: `module_name: [knowledgebase] app.api.routes.collection_job.get_collection_job`, `logger.name: app.api.routes.collection_job`, `code.line.number: 58`; filter with `module_name:[knowledgebase]*`. - CloudWatch (stdout format puts `module_name` where the logger name was): `2026-10-11 10:16:57,137 - [kaapi-backend] - [a1b2c3] - INFO - [knowledgebase] app.api.routes.collection_job.get_collection_job - [get_collection_job] Fetched job | job_id=8f2c`; Logs Insights `filter @message like "[knowledgebase]"`. - `OTEL_ENABLED` is checked only at the entry points (`main.py`, `celery/celery_app.py`, `core/db.py`), which decide whether to call `setup_telemetry` / `instrument_app` / `instrument_db_engine`; the telemetry functions themselves are unconditional. `LogContextFilter` is installed by `configure_logging`, so log context works with OTel off. -- OTel auto-instrumentation (in `setup_telemetry`): FastAPI, SQLAlchemy, httpx, requests, logging, Celery (Queues insight), Redis (Caches insight), botocore (S3/KMS spans). +- OTel auto-instrumentation (in `setup_telemetry`): FastAPI, SQLAlchemy, httpx, requests, logging, Celery (Queues insight), botocore (S3/KMS spans). Redis is deliberately not instrumented (only Celery result bookkeeping and rate-limit counters); keep Celery's `apply_async` span, it links API and worker traces. - `core/rate_monitor.py` — provider rate tracking