feat(evaluation): Introduce dedicated evaluations queue

.claude/conventions/celery.md

@@ -1,10 +1,14 @@
 # Celery task conventions (`app/celery/tasks/`)
 
 Authoritative conventions for Celery tasks in kaapi-backend. Tasks live in `app/celery/tasks/`.
-Celery uses RabbitMQ as broker. There is a **single `default` queue** declared with
-`x-max-priority=10`; 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).
+Celery uses RabbitMQ as broker. There are **two queues**, each declared with `x-max-priority=10`:
+`default` (everything except fast-eval burst tasks) and `evaluations` (only
+`run_evaluation_fast_chunk` / `run_evaluation_fast_aggregate` — physically isolated on their own
+worker pool so a fast-eval burst can't occupy `default`'s worker slots and delay higher-priority
+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).
 
 ## Canonical decorator stack
 
@@ -30,17 +34,23 @@ def run_my_job(self, project_id: int, job_id: str, trace_id: str, **kwargs):
 `_set_trace`, `_run_with_otel_parent`, and `gevent_timeout` already exist in this module /
 `app/celery/utils.py` — reuse them, don't reinvent.
 
-## Priority choice — be explicit
+## Queue and priority choice — be explicit
 
-All tasks share the one `default` queue; set `priority` to place the task in the right band. Match
+Only `run_evaluation_fast_chunk` and `run_evaluation_fast_aggregate` use `queue="evaluations"`
+(fast-eval burst fan-out — many chunk tasks fired at once). Everything else uses
+`queue="default"`. Don't move other eval-adjacent tasks (e.g. `run_evaluation_batch_submission`,
+`send_eval_completion_notification`) onto `evaluations` — they're one-shot/fire-and-forget, no
+burstiness to isolate.
+
+Within whichever queue the task belongs to, set `priority` to place it in the right band. Match
 the bands already in use (see the `job_execution.py` module docstring):
 
-| Priority | When |
-|---|---|
-| `9` | User-blocking, interactive — LLM call / chain / response jobs |
-| `6` | Fast evaluation |
-| `2` | Default — doctransform, collections, STT/TTS evaluation, assessment |
-| `1` | Notifications and other fire-and-forget background work |
+| Priority | Queue | When |
+|---|---|---|
+| `9` | `default` | User-blocking, interactive — LLM call / chain / response jobs |
+| `6` | `evaluations` | Fast evaluation chunk/aggregate |
+| `2` | `default` | Default — doctransform, collections, STT/TTS evaluation, assessment |
+| `1` | `default` | Notifications and other fire-and-forget background work |
 
 Pick the band that matches the task's user-facing urgency; document the choice in a comment if it's
 not obvious. `task_inherit_parent_priority=True` is set, so a task enqueued from another task
@@ -49,7 +59,7 @@ inherits its priority unless you override it.
 ## Hard rules
 
 - **`bind=True`** so you have `self` (the task instance) for retries, IDs, etc.
-- **Always pass `queue="default"` and an explicit `priority`** — there is only one queue; the priority is what matters.
+- **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.
 - **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.

.env.example

@@ -71,6 +71,9 @@ REDIS_PASSWORD=
 # Celery Configuration
 # Leave CELERY_WORKER_CONCURRENCY empty to auto-detect CPU cores, or set to specific number (e.g., number_of_cores * 2)
 CELERY_WORKER_CONCURRENCY=
+# Dedicated pool for the "evaluations" queue (fast-eval chunk/aggregate tasks),
+# isolated from CELERY_WORKER_CONCURRENCY so an eval burst can't starve LLM-call tasks.
+CELERY_EVAL_WORKER_CONCURRENCY=2
 CELERY_WORKER_MAX_TASKS_PER_CHILD=1000
 CELERY_WORKER_MAX_MEMORY_PER_CHILD=200000
 CELERY_TASK_SOFT_TIME_LIMIT=300

backend/app/celery/celery_app.py

@@ -167,20 +167,31 @@ def initialize_worker(**_) -> None:
 )
 
 # Define exchanges and queues with priority
+DEFAULT_QUEUE = "default"
+EVALUATIONS_QUEUE = "evaluations"
+
 default_exchange = Exchange("default", type="direct")
 
 # Celery configuration using environment variables
 celery_app.conf.update(
     # Queue configuration with priority support
     task_queues=(
         Queue(
-            "default",
+            DEFAULT_QUEUE,
+            exchange=default_exchange,
+            routing_key=DEFAULT_QUEUE,
+            queue_arguments={"x-max-priority": 10},
+        ),
+        # Routing key must differ per queue on the shared direct exchange, or
+        # messages fan out to both.
+        Queue(
+            EVALUATIONS_QUEUE,
             exchange=default_exchange,
-            routing_key="default",
+            routing_key=EVALUATIONS_QUEUE,
             queue_arguments={"x-max-priority": 10},
         ),
     ),
-    task_default_queue="default",
+    task_default_queue=DEFAULT_QUEUE,
     # Enable priority support
     task_inherit_parent_priority=True,
     worker_prefetch_multiplier=settings.CELERY_WORKER_PREFETCH_MULTIPLIER,

backend/app/celery/tasks/job_execution.py

@@ -1,14 +1,22 @@
-"""Celery task definitions for the single priority `default` queue.
+"""Celery task definitions, split across two priority queues.
 
-All tasks share one queue (`default`, declared with `x-max-priority=10`) and are
-ordered by the per-task `priority`:
+`default` (every task below except the two fast-eval ones), ordered by the
+per-task `priority`:
 
-    9  LLM call + LLM chain (run_llm_job, run_llm_chain_job, run_response_job)
-    6  Fast evaluation (run_evaluation_fast_chunk, run_evaluation_fast_aggregate)
+    9  LLM call + LLM chain (run_llm_job, run_llm_chain_job, run_response_job,
+       run_guardrails_job)
+    6  Prompt improvement (run_prompt_improvement)
     2  Everything else (doctransform, collections, STT/TTS evaluation, assessment)
     1  Notifications (send_eval_completion_notification)
 
-Higher priority drains first; within the same priority, delivery is FIFO.
+`evaluations` (priority 6): run_evaluation_fast_chunk and
+run_evaluation_fast_aggregate only. A fast eval fans out into many chunk tasks
+at once; priority only reorders messages still waiting in a queue, so on a
+shared queue that burst occupies every worker slot and delays priority-9 LLM
+jobs. A separate queue with its own worker pool makes the isolation physical.
+
+Both queues are declared with `x-max-priority=10`. Higher priority drains first;
+within the same priority, delivery is FIFO.
 """
 
 import logging
@@ -19,7 +27,7 @@
 from opentelemetry import trace
 from opentelemetry.propagate import extract
 
-from app.celery.celery_app import celery_app
+from app.celery.celery_app import EVALUATIONS_QUEUE, celery_app
 from app.celery.utils import gevent_timeout
 from app.core.config import settings
 
@@ -246,7 +254,7 @@ def run_evaluation_batch_submission(
     )
 
 
-# Priority 6 (fast-eval tier): user-blocking interactive prompt iteration in the
+# Priority 6: user-blocking interactive prompt iteration in the
 # evaluation domain, above default batch work but below core LLM call/chain jobs.
 @celery_app.task(bind=True, queue="default", priority=6)
 @gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_prompt_improvement")
@@ -413,7 +421,7 @@ def run_tts_result_processing(
     )
 
 
-@celery_app.task(bind=True, queue="default", priority=6)
+@celery_app.task(bind=True, queue=EVALUATIONS_QUEUE, priority=6)
 @gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_evaluation_fast_chunk")
 def run_evaluation_fast_chunk(
     self: Task,
@@ -443,7 +451,7 @@ def run_evaluation_fast_chunk(
     )
 
 
-@celery_app.task(bind=True, queue="default", priority=6)
+@celery_app.task(bind=True, queue=EVALUATIONS_QUEUE, priority=6)
 @gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_evaluation_fast_aggregate")
 def run_evaluation_fast_aggregate(
     self: Task, eval_run_id: int, trace_id: str = DEFAULT_TRACE_ID

docker-compose.dev.yml

@@ -57,4 +57,20 @@ services:
         required: true
       - path: .env.secrets
         required: false
-    command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info"]
+    command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "default"]
+
+  celery_worker_eval:
+    image: "${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG:-latest}"
+    container_name: celery-worker-eval
+    restart: always
+    build:
+      context: ./backend
+    depends_on:
+      backend:
+        condition: service_healthy
+    env_file:
+      - path: .env
+        required: true
+      - path: .env.secrets
+        required: false
+    command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "evaluations", "--concurrency=${CELERY_EVAL_WORKER_CONCURRENCY:-2}"]

docker-compose.staging.yml

@@ -66,11 +66,34 @@ services:
         required: true
       - path: .env.secrets
         required: false
-    command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info"]
+    command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "default"]
     logging:
       driver: awslogs
       options:
         awslogs-region: ap-south-1
         awslogs-group: /ec2/kaapi-staging
         awslogs-stream: celery-worker
         awslogs-create-group: "true"
+
+  celery_worker_eval:
+    image: "${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG:-latest}"
+    container_name: celery-worker-eval
+    restart: always
+    build:
+      context: ./backend
+    depends_on:
+      backend:
+        condition: service_healthy
+    env_file:
+      - path: .env
+        required: true
+      - path: .env.secrets
+        required: false
+    command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "evaluations", "--concurrency=${CELERY_EVAL_WORKER_CONCURRENCY:-2}"]
+    logging:
+      driver: awslogs
+      options:
+        awslogs-region: ap-south-1
+        awslogs-group: /ec2/kaapi-staging
+        awslogs-stream: celery-worker-eval
+        awslogs-create-group: "true"

docker-compose.yml

@@ -146,7 +146,23 @@ services:
     environment:
       REDIS_HOST: redis
       RABBITMQ_HOST: rabbitmq
-    command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info"]
+    command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "default"]
+
+  celery_worker_eval:
+    image: "${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG:-latest}"
+    container_name: celery-worker-eval
+    restart: always
+    build:
+      context: ./backend
+    depends_on:
+      backend:
+        condition: service_healthy
+    env_file:
+      - .env
+    environment:
+      REDIS_HOST: redis
+      RABBITMQ_HOST: rabbitmq
+    command: ["uv", "run", "celery", "-A", "app.celery.celery_app", "worker", "--loglevel=info", "-Q", "evaluations", "--concurrency=${CELERY_EVAL_WORKER_CONCURRENCY:-2}"]
 
   celery_flower:
     image: "${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG:-latest}"

docs/wiki/services.md

@@ -10,7 +10,7 @@ FastAPI (backend/app/main.py, api/routes/*)
    │
    ├── Postgres (SQLModel, core/db.py, alembic migrations)
    ├── Celery workers ── RabbitMQ (broker) + Redis (results)
-   │      └── app/celery/tasks/job_execution.py, priority queues
+   │      └── app/celery/tasks/job_execution.py, 2 queues (default, evaluations) + priority bands
    ├── Object storage        core/cloud/storage.py
    ├── Langfuse              core/langfuse/langfuse.py (traces, scores)
    ├── Sentry                core/sentry_filters.py
@@ -25,7 +25,7 @@ FastAPI (backend/app/main.py, api/routes/*)
 | Process | Entry | Notes |
 |---|---|---|
 | API server | `fastapi run app/main.py` | routes in `app/api/routes/` |
-| Celery worker | `app/celery/` config | async jobs, priority queues |
+| Celery worker | `app/celery/` config | `default` queue (most tasks) + dedicated `evaluations` queue/pool (fast-eval chunk/aggregate, isolated so eval bursts can't delay LLM jobs), priority bands within each |
 | Celery beat / cron | `app/api/routes/cron.py` + `crud/evaluations/cron.py` | batch polling for eval/assessment runs |
 
 ## Environment