feat: Kaapi Agentic Chatbot Flow

.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 |
+| `9` | `evaluations` | Fast evaluation chunk/aggregate (top band of this queue; new eval tasks pick a lower band) |
+| `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

.github/workflows/deploy-staging-ecs.yml

@@ -33,16 +33,16 @@ jobs:
       - name: Build and Push Docker Image
         env:
           REGISTRY: ${{ steps.login-ecr.outputs.registry }}
-          REPOSITORY: ${{ vars.AWS_RESOURCE_PREFIX }}-staging-repo
+          REPOSITORY: ${{ vars.AWS_RESOURCE_PREFIX }}-backend-repo
         run: |
           docker build --build-arg GIT_SHA=${{ github.sha }} -t $REGISTRY/$REPOSITORY:latest ./backend
           docker push $REGISTRY/$REPOSITORY:latest
 
       - name: Run database migrations
         env:
-          CLUSTER: ${{ vars.AWS_RESOURCE_PREFIX }}-staging-cluster
-          SERVICE: ${{ vars.AWS_RESOURCE_PREFIX }}-staging-service
-          TASK_DEF: ${{ vars.AWS_RESOURCE_PREFIX }}-staging-task
+          CLUSTER: ${{ vars.AWS_RESOURCE_PREFIX }}-backend-cluster
+          SERVICE: ${{ vars.AWS_RESOURCE_PREFIX }}-backend-celery-service
+          TASK_DEF: ${{ vars.AWS_RESOURCE_PREFIX }}-backend-celery
         run: |
           CONTAINER_NAME=$(aws ecs describe-task-definition --task-definition $TASK_DEF --query 'taskDefinition.containerDefinitions[0].name' --output text)
 

backend/app/alembic/versions/086_seed_anthropic_text_model_configs.py

@@ -0,0 +1,45 @@
+"""seed anthropic text model_config rows
+
+Revision ID: 086
+Revises: 085
+Create Date: 2026-10-09 00:00:00.000000
+"""
+
+from alembic import op
+
+revision = "086"
+down_revision = "085"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.execute(
+        """
+        INSERT INTO global.model_config
+            (provider, model_name, completion_type, config, input_modalities,
+             output_modalities, pricing, is_active, inserted_at, updated_at)
+        VALUES
+            ('anthropic', 'claude-sonnet-4-6', '{text}',
+                '{"effort": {"type": "enum", "default": "high", "options": ["low", "medium", "high", "max"], "description": "How long the model spends reasoning. Higher = better but slower."}}',
+                '{TEXT,IMAGE,FILES}', '{TEXT}',
+                '{"response": {"input_token_cost": 3, "output_token_cost": 15}, "batch": {"input_token_cost": 1.5, "output_token_cost": 7.5}}',
+                true, NOW(), NOW()),
+            ('anthropic', 'claude-opus-5-5', '{text}',
+                '{"effort": {"type": "enum", "default": "high", "options": ["low", "medium", "high", "xhigh", "max"], "description": "How long the model spends reasoning. Higher = better but slower."}}',
+                '{TEXT,IMAGE,FILES}', '{TEXT}',
+                NULL,
+                true, NOW(), NOW())
+        ON CONFLICT (provider, model_name) DO NOTHING
+        """
+    )
+
+
+def downgrade():
+    op.execute(
+        """
+        DELETE FROM global.model_config
+        WHERE provider = 'anthropic'
+          AND model_name IN ('claude-sonnet-4-6', 'claude-opus-5-5')
+        """
+    )

backend/app/alembic/versions/087_seed_anthropic_text_model_configs.py

@@ -0,0 +1,45 @@
+"""seed anthropic text model_config rows
+
+Revision ID: 087
+Revises: 086
+Create Date: 2026-10-09 00:00:00.000000
+"""
+
+from alembic import op
+
+revision = "087"
+down_revision = "086"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.execute(
+        """
+        INSERT INTO global.model_config
+            (provider, model_name, completion_type, config, input_modalities,
+             output_modalities, pricing, is_active, inserted_at, updated_at)
+        VALUES
+            ('anthropic', 'claude-sonnet-4-6', '{text}',
+                '{"effort": {"type": "enum", "default": "high", "options": ["low", "medium", "high", "max"], "description": "How long the model spends reasoning. Higher = better but slower."}}',
+                '{TEXT,IMAGE,FILES}', '{TEXT}',
+                '{"response": {"input_token_cost": 3, "output_token_cost": 15}, "batch": {"input_token_cost": 1.5, "output_token_cost": 7.5}}',
+                true, NOW(), NOW()),
+            ('anthropic', 'claude-opus-5-5', '{text}',
+                '{"effort": {"type": "enum", "default": "high", "options": ["low", "medium", "high", "xhigh", "max"], "description": "How long the model spends reasoning. Higher = better but slower."}}',
+                '{TEXT,IMAGE,FILES}', '{TEXT}',
+                NULL,
+                true, NOW(), NOW())
+        ON CONFLICT (provider, model_name) DO NOTHING
+        """
+    )
+
+
+def downgrade():
+    op.execute(
+        """
+        DELETE FROM global.model_config
+        WHERE provider = 'anthropic'
+          AND model_name IN ('claude-sonnet-4-6', 'claude-opus-5-5')
+        """
+    )

backend/app/api/docs/chatbot/message.md

@@ -0,0 +1,7 @@
+Run one turn of a prompt-driven chatbot conversation and return the full updated transcript.
+
+The chat model comes from a saved Kaapi config: `config_id` must be a config in the caller's project whose completion is `provider: "anthropic"`, `type: "text"` with `params.model` set. `config_version` picks a version (latest if omitted). The config's `params.effort` sets reasoning effort (provider default if absent) and `params.thinking` overrides the default adaptive thinking. `max_tokens` (default 1024) and `history_token_limit` (default 4000; older turns are summarized above it) are optional per-request overrides.
+
+There is no server-side session yet: send the admin-authored `system_prompt` on every call, and send the previous response's `messages` back as `history`. The returned `messages` contain the system prompt followed by every user/assistant turn so far, ending with the new assistant reply. Any `system` entries in `history` are ignored in favour of `system_prompt`.
+
+On the very first call, send an empty `history` and omit `message` to get the bot's opening greeting. Once `history` is non-empty, `message` is required. Returns 404 if the config or version isn't found in the project, 422 for a missing `message` or a config that isn't a usable Anthropic text config, and 502 for provider failures.

backend/app/api/docs/openapi_config.py

@@ -8,9 +8,6 @@
 from app.api.docs.guardrails_schemas import GUARDRAILS_COMPONENT_SCHEMAS
 from app.api.openapi_docs import inject_component_schemas, register_component_schemas
 
-# Schemas for routes that document a contract they do not validate (the
-# guardrails proxy). Registered at import so they are in `components` before
-# any `$ref` to them is resolved.
 register_component_schemas(GUARDRAILS_COMPONENT_SCHEMAS)
 
 # Tag metadata for organizing endpoints in documentation
@@ -35,6 +32,10 @@
         "name": "LLM",
         "description": "Large Language Model inference and interaction endpoints",
     },
+    {
+        "name": "Chatbot",
+        "description": "Prompt-driven conversational chatbot turns",
+    },
     {
         "name": "Guardrails",
         "description": (
@@ -104,6 +105,7 @@
             "Collections",
             "Config Management",
             "LLM",
+            "Chatbot",
             "Guardrails",
             "Evaluation",
             "Fine Tuning",

backend/app/api/main.py

@@ -5,6 +5,7 @@
     api_keys,
     assistants,
     auth,
+    chatbot,
     collection_job,
     collections,
     config,
@@ -47,6 +48,7 @@
 api_router.include_router(assessment_api_routes.router)
 api_router.include_router(assistants.router)
 api_router.include_router(auth.router)
+api_router.include_router(chatbot.router)
 api_router.include_router(collection_job.router)
 api_router.include_router(collections.router)
 api_router.include_router(config.router)

backend/app/api/routes/chatbot/__init__.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+
+from app.api.routes.chatbot.message import router as message_router
+
+router = APIRouter(prefix="/chatbot", tags=["Chatbot"])
+router.include_router(message_router)

backend/app/api/routes/chatbot/message.py

@@ -0,0 +1,52 @@
+from fastapi import APIRouter, Depends, HTTPException
+
+from app.api.deps import AuthContextDep, SessionDep
+from app.api.permissions import Permission, require_permission
+from app.models import ChatbotMessage, ChatbotTurnRequest, ChatbotTurnResponse
+from app.services.chatbot.config import resolve_chatbot_model_settings
+from app.services.chatbot.engine import run_chatbot_turn
+from app.services.chatbot.exceptions import ChatbotInputError, ChatbotProviderError
+from app.utils import APIResponse, load_description
+
+router = APIRouter()
+
+
+@router.post(
+    "/message",
+    description=load_description("chatbot/message.md"),
+    response_model=APIResponse[ChatbotTurnResponse],
+    dependencies=[Depends(require_permission(Permission.REQUIRE_PROJECT))],
+)
+async def send_chatbot_message(
+    request: ChatbotTurnRequest,
+    session: SessionDep,
+    auth_context: AuthContextDep,
+) -> APIResponse[ChatbotTurnResponse]:
+    history: list[dict[str, str]] = []
+    for entry in request.history:
+        history.append(entry.model_dump(mode="json"))
+
+    try:
+        model_settings = resolve_chatbot_model_settings(
+            session=session,
+            project_id=auth_context.project_.id,
+            config_id=request.config_id,
+            config_version=request.config_version,
+            max_tokens=request.max_tokens,
+            history_token_limit=request.history_token_limit,
+        )
+        transcript = await run_chatbot_turn(
+            system_prompt=request.system_prompt,
+            history=history,
+            user_message=request.message,
+            model_settings=model_settings,
+        )
+    except ChatbotInputError as e:
+        raise HTTPException(status_code=422, detail=str(e))
+    except ChatbotProviderError as e:
+        raise HTTPException(status_code=502, detail=str(e))
+
+    messages: list[ChatbotMessage] = []
+    for entry in transcript:
+        messages.append(ChatbotMessage.model_validate(entry))
+    return APIResponse.success_response(ChatbotTurnResponse(messages=messages))

backend/app/celery/celery_app.py

@@ -159,20 +159,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,15 +1,24 @@
-"""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,
-       run_prompt_improvement, run_evaluation_iteration_graph_step)
+    9  LLM call + LLM chain (run_llm_job, run_llm_chain_job, run_response_job,
+       run_guardrails_job)
+    6  Prompt improvement + iteration graph (run_prompt_improvement,
+       run_evaluation_iteration_graph_step)
     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 9, top band of this queue; later eval tasks pick lower
+bands): 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
@@ -23,7 +32,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
 from app.core.telemetry import LogModuleEnum, log_context
@@ -259,7 +268,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")
@@ -433,7 +442,7 @@ def run_tts_result_processing(
     )
 
 
-@celery_app.task(bind=True, queue="default", priority=6)
+@celery_app.task(bind=True, queue=EVALUATIONS_QUEUE, priority=9)
 @gevent_timeout(settings.CELERY_TASK_SOFT_TIME_LIMIT, "run_evaluation_fast_chunk")
 def run_evaluation_fast_chunk(
     self: Task,
@@ -464,7 +473,7 @@ def run_evaluation_fast_chunk(
         )
 
 
-@celery_app.task(bind=True, queue="default", priority=6)
+@celery_app.task(bind=True, queue=EVALUATIONS_QUEUE, priority=9)
 @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