Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
10ed0a4
Refactor error handling across services to use domain-specific except…
vprashrex Jul 29, 2026
64cafc5
Refactor exception handling to use HTTPException for consistent statu…
vprashrex Jul 29, 2026
73576c2
chore: add HTTPException import to test files and enhance exception h…
vprashrex Jul 29, 2026
7a2c055
Merge branch 'main' into fix/834-status-code-consistency
vprashrex Aug 4, 2026
ddd22cd
Merge branch 'main' into fix/834-status-code-consistency
vprashrex Aug 20, 2026
486c909
fix(tests): update exception handling in decrypt_credentials tests to…
vprashrex Aug 20, 2026
1f4dda2
Merge remote-tracking branch 'origin/main' into fix/834-status-code-c…
vprashrex Oct 15, 2026
f055e4c
feat(exception-handling): Enhance DB error handling with specific sta…
vprashrex Oct 15, 2026
162480f
test: Enhance database error handling tests with specific status codes
vprashrex Oct 18, 2026
93c63cc
Merge branch 'main' into fix/834-status-code-consistency
vprashrex Oct 18, 2026
84950d3
Refactor error handling for database integrity violations
vprashrex Oct 21, 2026
cccf26f
Merge remote-tracking branch 'origin/main' into fix/834-status-code-c…
vprashrex Oct 21, 2026
0b6517a
refactor: Improve status code consistency and error messaging in exce…
vprashrex Oct 21, 2026
bc3dbaf
Merge branch 'main' into fix/834-status-code-consistency
vprashrex Oct 21, 2026
2ffc604
Merge remote-tracking branch 'origin/main' into fix/834-status-code-c…
vprashrex Oct 23, 2026
df82ec4
fix: standardize HTTP status codes across evaluation endpoints and im…
vprashrex Oct 23, 2026
6e8a7c9
Merge branch 'main' into fix/834-status-code-consistency
vprashrex Oct 23, 2026
a151abb
fix: improve observability by tagging Sentry tasks with tenant inform…
vprashrex Oct 24, 2026
d0bfb1f
Merge remote-tracking branch 'origin/main' into fix/834-status-code-c…
vprashrex Oct 24, 2026
8df56ca
fix: add additional GenAI content keys for improved data handling
vprashrex Oct 24, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 11 additions & 13 deletions .claude/conventions/celery.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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

Expand Down Expand Up @@ -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/<domain>/`. 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:
Expand Down
4 changes: 2 additions & 2 deletions backend/app/api/deps.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion backend/app/api/routes/collections.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand Down
15 changes: 13 additions & 2 deletions backend/app/celery/celery_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Loading
Loading