fix(services): Improve status code consistency (#1090)

Author: vprashrex

.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/<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:

.gitignore

@@ -21,3 +21,6 @@ ENV/
 
 #
 /backend/app/logs
+
+# local assessment scratch data
+z_assessment_test/

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

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,

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