feat(worker): add named Celery queues (high/low priority) with task routing by repo size

app/worker/celery_app.py

@@ -1,23 +1,60 @@
 from celery import Celery
+from kombu import Exchange, Queue
 
 from app.core.config import get_settings
 
 settings = get_settings()
 
+# Two named queues:
+#   high  — small repos (< 10 MB), expected to complete in < 30s
+#   low   — large repos (>= 10 MB), may take several minutes
+#
+# Both are durable (survives Redis restart). The worker consumes high
+# first due to queue ordering in the -Q argument (see docker-compose.yml).
+
+default_exchange = Exchange("default", type="direct")
+
+HIGH_QUEUE = "high"
+LOW_QUEUE  = "low"
+
 celery_app = Celery(
-    "ai_code_review_agent",
+    "ai_code_review",
     broker=settings.redis_url,
     backend=settings.celery_result_backend or settings.redis_url,
 )
 
 celery_app.conf.update(
+    # Queue definitions
+    task_queues=(
+        Queue(HIGH_QUEUE, default_exchange, routing_key="high", durable=True),
+        Queue(LOW_QUEUE,  default_exchange, routing_key="low",  durable=True),
+    ),
+    task_default_queue=HIGH_QUEUE,
+    task_default_exchange="default",
+    task_default_routing_key="high",
+
+    # Route analyze_repository_task to high or low based on a size hint
+    # passed as a task header (set in tasks.py before .apply_async())
+    task_routes={
+        "analyze_repository_task": {"queue": HIGH_QUEUE},  # default; overridden at call site
+    },
+
+    # Serialisation
     task_serializer="json",
-    accept_content=["json"],
     result_serializer="json",
-    task_track_started=True,
-    task_time_limit=600,
-    task_soft_time_limit=540,
-    result_expires=3600,
+    accept_content=["json"],
+
+    # Results expire after 24 h — matches the cache TTL
+    result_expires=86400,
+
+    # Eagerness (test mode)
     task_always_eager=settings.celery_task_always_eager,
-    task_store_eager_result=settings.celery_task_always_eager,
+
+    # Acknowledge only after the task completes, not on receipt.
+    # Prevents a crashed worker from silently dropping a job.
+    task_acks_late=True,
+    task_reject_on_worker_lost=True,
+
+    # Worker settings
+    worker_prefetch_multiplier=1,   # one task at a time per worker slot
 )

app/worker/tasks.py

@@ -2,23 +2,54 @@ import asyncio
 import threading
 from typing import Any, cast
 
+from app.core.config import get_settings
 from app.core.logger import get_logger
 from app.models.report import EngineeringReport
 from app.services.analysis_service import run_full_analysis
-from app.worker.celery_app import celery_app
+from app.worker.celery_app import celery_app, HIGH_QUEUE, LOW_QUEUE
 
 logger = get_logger(__name__)
 
+# Repos larger than this threshold are routed to the low-priority queue
+# so they do not starve small, fast jobs.
+LARGE_REPO_MB_THRESHOLD = 50
+
+
+def route_queue_for_repo(estimated_size_mb: float | None) -> str:
+    """Return the appropriate queue name based on estimated repo size."""
+    if estimated_size_mb is not None and estimated_size_mb >= LARGE_REPO_MB_THRESHOLD:
+        return LOW_QUEUE
+    return HIGH_QUEUE
+
+
+def submit_analysis_task(
+    github_url: str,
+    base_sha: str | None = None,
+    estimated_size_mb: float | None = None,
+) -> Any:
+    """
+    Submit an analysis task to the correct priority queue.
+
+    Call this instead of calling .delay() or .apply_async() directly so
+    that queue routing logic stays in one place.
+    """
+    queue = route_queue_for_repo(estimated_size_mb)
+    logger.info(
+        "Submitting analysis task",
+        extra={"url": github_url, "queue": queue, "size_mb": estimated_size_mb},
+    )
+    return analyze_repository_task.apply_async(
+        args=[github_url, base_sha],
+        queue=queue,
+    )
+
 
 def _run_coro_sync(coro: Any) -> Any:
     """
-    Run an awaitable to completion from synchronous code.
+    Run an awaitable to completion from synchronous Celery worker code.
 
-    Uses asyncio.run() directly when no event loop is running (the normal
-    case in a Celery prefork worker). If a loop is already running (e.g.
-    Celery's eager-execution mode invoked from within a pytest-asyncio
-    test), runs the coroutine in a separate thread with its own loop to
-    avoid "asyncio.run() cannot be called from a running event loop".
+    Uses asyncio.run() when no event loop is running (normal prefork worker).
+    Falls back to a thread when a loop is already running (eager test mode).
     """
     try:
         asyncio.get_running_loop()
@@ -38,12 +69,16 @@ def _run_coro_sync(coro: Any) -> Any:
 
 @celery_app.task(name="analyze_repository_task", bind=True, max_retries=2)
 def analyze_repository_task(
-    self: Any, github_url: str, base_sha: str | None = None
+    self: Any,
+    github_url: str,
+    base_sha: str | None = None,
 ) -> dict[str, Any]:
     """
     Celery task wrapper around the async analysis pipeline.
-    Runs the full multi-agent pipeline and returns a JSON-serializable
-    EngineeringReport.
+
+    Runs the full multi-agent pipeline and returns a JSON-serialisable
+    EngineeringReport dict. Retries up to 2 times on unexpected failure
+    with exponential backoff.
     """
     try:
         report = cast(
@@ -53,6 +88,8 @@ def analyze_repository_task(
         return cast(dict[str, Any], report.model_dump())
     except Exception as exc:
         logger.exception(
-            "Analysis task failed", extra={"url": github_url, "error": str(exc)}
+            "Analysis task failed",
+            extra={"url": github_url, "queue": self.request.delivery_info.get("routing_key"), "error": str(exc)},
         )
-        raise
+        # Exponential backoff: 60s, 120s
+        raise self.retry(exc=exc, countdown=60 * (self.request.retries + 1))

docker-compose.yml

@@ -41,7 +41,7 @@ services:
   worker:
     build: .
     entrypoint: ["celery"]
-    command: ["-A", "app.worker.celery_app", "worker", "--loglevel=info", "--concurrency=2"]
+    command: ["-A", "app.worker.celery_app", "worker", "--loglevel=info", "--concurrency=2", "-Q", "high,low"]
     environment:
       - OPENROUTER_API_KEY=${OPENROUTER_API_KEY}
       - API_KEY=${API_KEY}

docs/architecture.md

@@ -56,4 +56,20 @@ Default model: `meta-llama/llama-3.3-70b-instruct`
 Alternative models:
 - `mistralai/mistral-7b-instruct`
 - `deepseek/deepseek-chat`
-- `google/gemma-3-27b-it`
+- `google/gemma-3-27b-it`
+
+## Task queue design
+
+Jobs are routed to one of two Celery queues backed by Redis:
+
+| Queue | Threshold | Expected duration |
+|---|---|---|
+| `high` | repos < 50 MB | < 60 s |
+| `low` | repos ≥ 50 MB | 1–5 min |
+
+Workers consume `high` before `low`, so a flood of large-repo jobs cannot
+starve small fast ones. Both queues are durable — a Redis restart does not
+lose in-flight tasks because `task_acks_late=True` means a task is only
+acknowledged after it completes, not on receipt.
+
+Retry policy: up to 2 retries with 60 s / 120 s exponential backoff.