feat(chatbot): Enable coordinator node routing

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

@@ -1,5 +1,5 @@
 Publish the chatbot's draft so new sessions run on it. Returns the now-live version.
 
-The draft is validated strictly: node ids are unique; at most one `input` and one `output` node; every edge's `source` and `target` is a node in the flow; every agent node has `data.config_id` and a non-empty `data.prompt`, and its config resolves to an Anthropic text config in this project with a model set (and a valid thinking budget). The project must have Anthropic credentials configured, since agent nodes call Anthropic with the project's own key. Only `onboarding` agent nodes can run for now — `coordinator` and `rag` nodes are rejected as not supported yet. The flow needs an entry node: with no edges it must contain exactly one agent node; otherwise the `input` node must have exactly one outgoing edge, leading to an agent node. A node without `data.config_version` gets the config's latest version written into the stored node, so later config edits don't change the published flow.
+The draft is validated strictly: node ids are unique; at most one `input` and one `output` node; every edge's `source` and `target` is a node in the flow; every agent node has `data.config_id` and a non-empty `data.prompt`, and its config resolves to an Anthropic text config in this project with a model set (and a valid thinking budget). The project must have Anthropic credentials configured, since agent nodes call Anthropic with the project's own key. `onboarding` and `coordinator` agent nodes can run; `rag` nodes are rejected as not supported yet. A flow has at most one `coordinator`; it needs at least one outgoing edge, every outgoing edge must lead to an `onboarding` node, and each such edge needs a non-empty `data.intent` that is unique among that coordinator's edges (`data.description` is optional and helps the routing). The flow needs an entry node: with no edges it must contain exactly one agent node; otherwise the `input` node must have exactly one outgoing edge, leading to an agent node. A node without `data.config_version` gets the config's latest version written into the stored node, so later config edits don't change the published flow.
 
 The previous live version (if any) stops being live in the same transaction. Sessions already running stay pinned to the version they started on. Returns 404 for an unknown chatbot, 409 if there is no draft to publish (save the flow first) or on a concurrent change, and 422 listing every validation problem.

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

@@ -4,4 +4,4 @@ The **first** save of a chatbot creates version 1 and makes it live immediately,
 
 Every later save is a draft that needs a publish to go live: if the highest version is not live it is overwritten, otherwise a new version `max + 1` is created with `is_live: false`. Live versions are never modified, so editing never affects running conversations.
 
-Draft saves check only the shape, so half-built flows save: each node needs `id` (1–255 chars), `type` (`input`, `output`, `coordinator`, `onboarding`, `rag`), `position {x, y}` and an object `data`; agent nodes (`coordinator`, `onboarding`, `rag`) may carry `data.config_id` (UUID), `data.config_version` (≥ 1) and `data.prompt`. Each edge needs `id`, `source`, `target`; `sourceHandle`, `targetHandle` and `data` (e.g. `intent`, `description`) are optional. Extra keys are stored and returned as sent. Returns 404 for an unknown chatbot, 409 if the flow was changed concurrently (reload and retry), and 422 for an invalid first save.
+Draft saves check only the shape, so half-built flows save: each node needs `id` (1–255 chars), `type` (`input`, `output`, `coordinator`, `onboarding`, `rag`), `position {x, y}` and an object `data`; agent nodes (`coordinator`, `onboarding`, `rag`) may carry `data.config_id` (UUID), `data.config_version` (≥ 1) and `data.prompt`. Each edge needs `id`, `source`, `target`; `sourceHandle`, `targetHandle` and `data` (e.g. `intent`, `description`) are optional; a coordinator routes on its outgoing edges' `data.intent` / `data.description`, and the intent rules are enforced only at publish. Extra keys are stored and returned as sent. Returns 404 for an unknown chatbot, 409 if the flow was changed concurrently (reload and retry), and 422 for an invalid first save.

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

@@ -4,4 +4,6 @@ The user `message` is stored immediately (tagged with the session's active node)
 
 The turn runs the session's active node of its pinned flow version: the node's `data.prompt` is the system prompt and its `data.config_id` / `data.config_version` supply the model. Each turn re-reads the conversation against that prompt and saves every field the user has given into `collected_fields`, a flat `{field_name: value}` map with snake_case keys (values are only ever taken from what the user said; a correction overwrites the old value). `completion_status` is `complete` once every required field is collected and `in_progress` otherwise; it is computed from a structured extraction, not from the reply text. A `complete` session still accepts messages. `status` (`active`/`inactive`) is separate and says whether the session is in use: once a newer session is created for the same contact and chatbot, or the chatbot is deleted, this one is `inactive` and rejects new messages with 409. A turn that was already queued when that happened is not run; Kaapi POSTs the failure payload with error `[KAAPI] Chatbot session was superseded by a newer session; start a new session.` instead.
 
+When the active node is a `coordinator`, it classifies the message against its outgoing edges' intents. On a clear match the session moves to that `onboarding` node, which writes the reply in the same turn (one webhook). If nothing clearly matches, the coordinator replies with a clarifying question and the session stays on it; the opening greeting is always the coordinator's own. `node_id` in the webhook is the node that produced the reply. When a routed agent finishes (`completion_status: complete`), the session goes back to the coordinator, so the next message is routed afresh and `completion_status` returns to `in_progress`. In coordinator flows, treat `complete` as "this sub-flow is done" for that webhook only, not as the end of the session. `collected_fields` is one map shared by every agent in the session.
+
 On the first call of a session, omit `message` to get the bot's opening greeting; after that `message` is required. `webhook_url` must be a public HTTPS URL. Returns 404 if the session isn't in the caller's project, 409 if the session is inactive, and 422 for an invalid `webhook_url` or a missing `message` on a started session.

backend/app/crud/chatbot.py

@@ -367,18 +367,21 @@ def update_chatbot_session_progress(
     session_id: UUID,
     collected_fields: dict[str, Any],
     completion_status: ChatbotCompletionStatusEnum,
+    active_node_id: str | None = None,
 ) -> ChatbotSession | None:
+    """`active_node_id=None` leaves the session's current node unchanged."""
     chatbot_session = session.get(ChatbotSession, session_id)
     if chatbot_session is None:
         return None
 
-    chatbot_session.sqlmodel_update(
-        {
-            "collected_fields": collected_fields,
-            "completion_status": completion_status,
-            "updated_at": now(),
-        }
-    )
+    updates: dict[str, Any] = {
+        "collected_fields": collected_fields,
+        "completion_status": completion_status,
+        "updated_at": now(),
+    }
+    if active_node_id is not None:
+        updates["active_node_id"] = active_node_id
+    chatbot_session.sqlmodel_update(updates)
     session.add(chatbot_session)
     session.commit()
     session.refresh(chatbot_session)

backend/app/services/chatbot/agents/coordinator/graph.py

@@ -0,0 +1,166 @@
+"""Coordinator turn: one structured-output call that routes the user to an agent or asks to clarify.
+
+No LangGraph here — a single model call doesn't need a graph.
+"""
+
+import asyncio
+import logging
+from dataclasses import dataclass
+from typing import NamedTuple
+
+import anthropic
+from langchain_core.exceptions import OutputParserException
+from langchain_core.language_models.chat_models import BaseChatModel
+from langchain_core.messages import BaseMessage
+from pydantic import BaseModel, Field, ValidationError
+
+from app.services.chatbot.agents.coordinator.prompts import (
+    FALLBACK_CLARIFYING_QUESTION,
+    ROUTES_TEMPLATE,
+    ROUTING_INSTRUCTION,
+)
+from app.services.chatbot.exceptions import ChatbotInputError, ChatbotProviderError
+from app.services.chatbot.history import summarize_if_needed, to_langchain_messages
+from app.services.chatbot.types import ChatbotMessageDict
+from app.services.chatbot.utils import (
+    STRUCTURED_OUTPUT_METHOD,
+    ChatbotModelSettings,
+    get_chat_model,
+    to_provider_error,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class RouteDecision(BaseModel):
+    intent: str | None = Field(
+        default=None,
+        description="Exactly one intent name from the list, or null when none clearly matches.",
+    )
+    clarifying_question: str | None = Field(
+        default=None,
+        description="Question to ask the user when intent is null; otherwise null.",
+    )
+
+
+class CoordinatorRoute(NamedTuple):
+    intent: str
+    description: str | None
+    target_node_id: str
+
+
+@dataclass(frozen=True)
+class CoordinatorDecision:
+    """Exactly one of `target_node_id` / `clarifying_question` is set."""
+
+    target_node_id: str | None
+    clarifying_question: str | None
+
+
+def _build_routing_prompt(*, system_prompt: str, routes: list[CoordinatorRoute]) -> str:
+    route_lines = [
+        f"- {route.intent}" + (f": {route.description}" if route.description else "")
+        for route in routes
+    ]
+    return (
+        system_prompt
+        + ROUTING_INSTRUCTION
+        + ROUTES_TEMPLATE.format(routes="\n".join(route_lines))
+    )
+
+
+def _decide_route(
+    chat_model: BaseChatModel,
+    messages: list[BaseMessage],
+    *,
+    history_token_limit: int,
+) -> RouteDecision:
+    conversation_view = summarize_if_needed(
+        chat_model, messages, history_token_limit=history_token_limit
+    )
+    routing_model = chat_model.with_structured_output(
+        RouteDecision, method=STRUCTURED_OUTPUT_METHOD
+    )
+    # Both are ValueError subclasses; mapped here so the caller's ValueError
+    # catch doesn't mislabel them as a model-configuration problem.
+    try:
+        decision = routing_model.invoke(conversation_view)
+    except (OutputParserException, ValidationError) as e:
+        error_message = (
+            "[ANTHROPIC] Coordinator routing returned malformed structured output — "
+            f"retry the turn. If persistent, contact Kaapi. (code: {type(e).__name__})"
+        )
+        logger.warning(f"[_decide_route] {error_message}", exc_info=True)
+        raise ChatbotProviderError(error_message) from e
+
+    if not isinstance(decision, RouteDecision):
+        error_message = (
+            "[ANTHROPIC] Coordinator routing returned an unexpected output shape — "
+            "retry the turn. If persistent, contact Kaapi. (code: UnexpectedOutputType)"
+        )
+        logger.warning(f"[_decide_route] {error_message}")
+        raise ChatbotProviderError(error_message)
+    return decision
+
+
+def _match_route(
+    intent: str | None, routes: list[CoordinatorRoute]
+) -> CoordinatorRoute | None:
+    if intent is None:
+        return None
+    wanted = intent.strip().casefold()
+    return next((route for route in routes if route.intent.casefold() == wanted), None)
+
+
+async def run_coordinator_turn(
+    *,
+    system_prompt: str,
+    history: list[ChatbotMessageDict],
+    user_message: str | None,
+    routes: list[CoordinatorRoute],
+    model_settings: ChatbotModelSettings,
+) -> CoordinatorDecision:
+    """Classify the latest user message against `routes`, or return a clarifying question."""
+    if user_message is None and history:
+        raise ChatbotInputError(
+            "[KAAPI] `message` is required once `history` is non-empty; omit it only "
+            "on the first call to get the opening greeting."
+        )
+
+    messages = to_langchain_messages(
+        system_prompt=_build_routing_prompt(system_prompt=system_prompt, routes=routes),
+        history=history,
+        user_message=user_message,
+    )
+    try:
+        chat_model = get_chat_model(model_settings)
+        # ChatAnthropic.invoke is sync; keep it off the event loop.
+        decision = await asyncio.to_thread(
+            _decide_route,
+            chat_model,
+            messages,
+            history_token_limit=model_settings.history_token_limit,
+        )
+    except (anthropic.APIConnectionError, anthropic.APIStatusError, ValueError) as e:
+        raise to_provider_error(e, model=model_settings.model) from e
+
+    # The opening turn has only the synthetic kickoff message, so it must never hand off.
+    if user_message is not None:
+        route = _match_route(decision.intent, routes)
+        if route is not None:
+            logger.info(
+                f"[run_coordinator_turn] Routed | intent: {route.intent}, "
+                f"target_node_id: {route.target_node_id}"
+            )
+            return CoordinatorDecision(
+                target_node_id=route.target_node_id, clarifying_question=None
+            )
+
+    question = decision.clarifying_question
+    if question is None or not question.strip():
+        question = FALLBACK_CLARIFYING_QUESTION
+    logger.info(
+        f"[run_coordinator_turn] Clarifying | model_intent: {decision.intent}, "
+        f"opening_turn: {user_message is None}"
+    )
+    return CoordinatorDecision(target_node_id=None, clarifying_question=question)

backend/app/services/chatbot/agents/coordinator/prompts.py

@@ -0,0 +1,14 @@
+ROUTING_INSTRUCTION = (
+    "\n\nYou route the user to the right assistant. Read the conversation and decide "
+    "which of the intents below the user's latest message expresses.\n"
+    "1. intent: copy exactly one intent name from the list when the user's need clearly "
+    "matches it; otherwise null. Never invent an intent that is not listed.\n"
+    "2. clarifying_question: when intent is null, a short, friendly question in the "
+    "user's language that helps them say which of the listed options they need; "
+    "otherwise null."
+)
+ROUTES_TEMPLATE = "\n\n<intents>\n{routes}\n</intents>"
+
+FALLBACK_CLARIFYING_QUESTION = (
+    "Could you tell me a bit more about what you need help with?"
+)

backend/app/services/chatbot/agents/onboarding/graph.py

@@ -17,7 +17,6 @@
 from app.services.chatbot.agents.onboarding.prompts import (
     EXTRACTION_FLOW_PROMPT_TEMPLATE,
     EXTRACTION_INSTRUCTION,
-    EXTRACTION_METHOD,
     EXTRACTION_PRIOR_FIELDS_TEMPLATE,
     PROGRESS_COLLECTED_TEMPLATE,
     PROGRESS_COMPLETE_NOTE,
@@ -37,7 +36,12 @@
     to_langchain_messages,
 )
 from app.services.chatbot.types import ChatbotMessageDict, ChatbotTurnResult
-from app.services.chatbot.utils import ChatbotModelSettings, get_chat_model
+from app.services.chatbot.utils import (
+    STRUCTURED_OUTPUT_METHOD,
+    ChatbotModelSettings,
+    get_chat_model,
+    to_provider_error,
+)
 
 ExtractionModel = Runnable[LanguageModelInput, dict | BaseModel]
 
@@ -163,28 +167,8 @@ def chatbot_node(state: OnboardingState) -> dict[str, list[BaseMessage]]:
 
 
 def _get_extraction_model(chat_model: BaseChatModel) -> ExtractionModel:
-    return chat_model.with_structured_output(FieldExtraction, method=EXTRACTION_METHOD)
-
-
-def _to_provider_error(error: Exception, *, model: str) -> ChatbotProviderError:
-    if isinstance(error, anthropic.APITimeoutError):
-        return ChatbotProviderError(
-            "[KAAPI] Request timed out — retry smaller. If persistent, contact Kaapi. "
-            "(code: APITimeoutError)"
-        )
-    if isinstance(error, anthropic.APIConnectionError):
-        return ChatbotProviderError(
-            "[KAAPI] Network/DNS issue reaching provider — check connectivity. "
-            "If persistent, contact Kaapi. (code: APIConnectionError)"
-        )
-    if isinstance(error, anthropic.APIStatusError):
-        return ChatbotProviderError(
-            f"[ANTHROPIC] Chat model call failed: {error.message} "
-            f"(code: {error.status_code}, request_id: {error.request_id})"
-        )
-    return ChatbotProviderError(
-        f"[KAAPI] Unsupported model configuration for '{model}': {error} "
-        "— update the config's model, thinking, or effort. (code: ValueError)"
+    return chat_model.with_structured_output(
+        FieldExtraction, method=STRUCTURED_OUTPUT_METHOD
     )
 
 
@@ -223,7 +207,7 @@ async def run_chatbot_turn(
         # ChatAnthropic.invoke is sync; keep it off the event loop.
         result = await asyncio.to_thread(graph.invoke, initial_state)
     except (anthropic.APIConnectionError, anthropic.APIStatusError, ValueError) as e:
-        raise _to_provider_error(e, model=model_settings.model) from e
+        raise to_provider_error(e, model=model_settings.model) from e
 
     return ChatbotTurnResult(
         reply=extract_final_reply(result["messages"]),

backend/app/services/chatbot/agents/onboarding/prompts.py

@@ -1,5 +1,3 @@
-EXTRACTION_METHOD = "json_schema"
-
 EXTRACTION_INSTRUCTION = (
     "You extract data from a chatbot conversation. You do not talk to the user.\n"
     "The admin flow prompt below defines the goal, the fields to collect, and the rules.\n"