feat: Kaapi Agentic Onboarding Chatbot Flow (#1300)

Co-authored-by: Ayush <80516839+Ayush8923@users.noreply.github.com>

Author: Ayush8923

backend/app/alembic/versions/088_add_chatbot_session.py

@@ -0,0 +1,315 @@
+"""add chatbot, flow version, session and message tables + CHATBOT feature flag key
+
+Revision ID: 088
+Revises: 087
+Create Date: 2026-10-20 00:00:00.000000
+"""
+
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision = "088"
+down_revision = "087"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "chatbot",
+        sa.Column(
+            "id",
+            sa.Uuid(),
+            nullable=False,
+            comment="Chatbot identifier used in API paths",
+        ),
+        sa.Column(
+            "name",
+            sqlmodel.sql.sqltypes.AutoString(length=255),
+            nullable=False,
+            comment="Display name of the chatbot shown in the UI",
+        ),
+        sa.Column(
+            "organization_id",
+            sa.Integer(),
+            nullable=False,
+            comment="Organization that owns the chatbot",
+        ),
+        sa.Column(
+            "project_id",
+            sa.Integer(),
+            nullable=False,
+            comment="Project that owns the chatbot",
+        ),
+        sa.Column(
+            "inserted_at",
+            sa.DateTime(),
+            nullable=False,
+            comment="Timestamp when the chatbot was created",
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(),
+            nullable=False,
+            comment="Timestamp when the chatbot was last updated",
+        ),
+        sa.Column(
+            "deleted_at",
+            sa.DateTime(),
+            nullable=True,
+            comment="Soft-delete timestamp; the row is kept because sessions reference it",
+        ),
+        sa.ForeignKeyConstraint(
+            ["organization_id"], ["organization.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(["project_id"], ["project.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_chatbot_organization_id"),
+        "chatbot",
+        ["organization_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_chatbot_project_id"), "chatbot", ["project_id"], unique=False
+    )
+
+    op.create_table(
+        "chatbot_flow_version",
+        sa.Column(
+            "id",
+            sa.Uuid(),
+            nullable=False,
+            comment="Flow version identifier pinned by sessions",
+        ),
+        sa.Column(
+            "chatbot_id",
+            sa.Uuid(),
+            nullable=False,
+            comment="Chatbot this flow version belongs to",
+        ),
+        sa.Column(
+            "version",
+            sa.Integer(),
+            nullable=False,
+            comment="Sequential version number per chatbot, starting at 1",
+        ),
+        sa.Column(
+            "nodes",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=False,
+            comment="React Flow nodes as saved by the UI: [{id, type (input/output/coordinator/onboarding/rag), position {x, y}, data}]; agent node data carries config_id, config_version, prompt",
+        ),
+        sa.Column(
+            "edges",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=False,
+            server_default=sa.text("'[]'::jsonb"),
+            comment="React Flow edges as saved by the UI: [{id, source, target, sourceHandle, targetHandle, data}]",
+        ),
+        sa.Column(
+            "is_live",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.text("false"),
+            comment="True for the single published version new sessions run on; live versions are immutable, the highest non-live version is the draft",
+        ),
+        sa.Column(
+            "inserted_at",
+            sa.DateTime(),
+            nullable=False,
+            comment="Timestamp when the version was created",
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(),
+            nullable=False,
+            comment="Timestamp when the draft was last saved or the version was published/unpublished",
+        ),
+        sa.ForeignKeyConstraint(["chatbot_id"], ["chatbot.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "chatbot_id",
+            "version",
+            name="uq_chatbot_flow_version_chatbot_version",
+        ),
+    )
+    op.create_index(
+        op.f("ix_chatbot_flow_version_chatbot_id"),
+        "chatbot_flow_version",
+        ["chatbot_id"],
+        unique=False,
+    )
+    op.create_index(
+        "uq_chatbot_flow_version_live_chatbot",
+        "chatbot_flow_version",
+        ["chatbot_id"],
+        unique=True,
+        postgresql_where=sa.text("is_live"),
+    )
+
+    op.create_table(
+        "chatbot_session",
+        sa.Column(
+            "id",
+            sa.Uuid(),
+            nullable=False,
+            comment="Session identifier returned to the caller",
+        ),
+        sa.Column(
+            "chatbot_id",
+            sa.Uuid(),
+            nullable=False,
+            comment="Chatbot this session runs; tenant scope comes from the chatbot",
+        ),
+        sa.Column(
+            "flow_version_id",
+            sa.Uuid(),
+            nullable=False,
+            comment="Live flow version pinned at creation, so publishing never changes a running conversation",
+        ),
+        sa.Column(
+            "contact_ref",
+            sqlmodel.sql.sqltypes.AutoString(length=500),
+            nullable=False,
+            comment="Caller-side identifier of the end user (e.g. a Glific contact id); opaque to Kaapi",
+        ),
+        sa.Column(
+            "active_node_id",
+            sqlmodel.sql.sqltypes.AutoString(length=255),
+            nullable=False,
+            comment="React Flow node id inside the pinned flow version that handles the next turn",
+        ),
+        sa.Column(
+            "status",
+            sa.Enum(
+                "active",
+                "inactive",
+                name="chatbotsessionstatusenum",
+            ),
+            nullable=False,
+            comment="Session in use: active, or inactive once superseded by a newer session for the same chatbot + contact_ref or the chatbot was deleted",
+        ),
+        sa.Column(
+            "completion_status",
+            sa.Enum(
+                "in_progress",
+                "complete",
+                "abandoned",
+                name="chatbotcompletionstatusenum",
+            ),
+            nullable=False,
+            server_default="in_progress",
+            comment="Field collection progress: in_progress, complete once every required field is gathered, or abandoned when deactivated while in_progress",
+        ),
+        sa.Column(
+            "collected_fields",
+            postgresql.JSON(astext_type=sa.Text()),
+            nullable=False,
+            comment="Flat map of field name -> value gathered from the user so far",
+        ),
+        sa.Column(
+            "inserted_at",
+            sa.DateTime(),
+            nullable=False,
+            comment="Timestamp when the session was created",
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(),
+            nullable=False,
+            comment="Timestamp when the session was last updated",
+        ),
+        sa.ForeignKeyConstraint(["chatbot_id"], ["chatbot.id"], ondelete="RESTRICT"),
+        sa.ForeignKeyConstraint(
+            ["flow_version_id"], ["chatbot_flow_version.id"], ondelete="RESTRICT"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_chatbot_session_flow_version_id"),
+        "chatbot_session",
+        ["flow_version_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_chatbot_session_contact_ref"),
+        "chatbot_session",
+        ["contact_ref"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_chatbot_session_chatbot_contact",
+        "chatbot_session",
+        ["chatbot_id", "contact_ref"],
+        unique=False,
+    )
+    op.create_index(
+        "uq_chatbot_session_active_chatbot_contact",
+        "chatbot_session",
+        ["chatbot_id", "contact_ref"],
+        unique=True,
+        postgresql_where=sa.text("status = 'active'"),
+    )
+
+    op.create_table(
+        "chatbot_session_message",
+        sa.Column("id", sa.Uuid(), nullable=False, comment="Unique identifier"),
+        sa.Column(
+            "session_id",
+            sa.Uuid(),
+            nullable=False,
+            comment="Reference to the owning chatbot session",
+        ),
+        sa.Column(
+            "role",
+            sa.Enum("system", "user", "assistant", name="chatbotmessageroleenum"),
+            nullable=False,
+            comment="Speaker: user or assistant (system is reserved; the prompt lives on the flow node)",
+        ),
+        sa.Column(
+            "content",
+            sqlmodel.sql.sqltypes.AutoString(),
+            nullable=False,
+            comment="Message text",
+        ),
+        sa.Column(
+            "node_id",
+            sqlmodel.sql.sqltypes.AutoString(length=255),
+            nullable=True,
+            comment="React Flow node id of the pinned flow that was active for this turn",
+        ),
+        sa.Column(
+            "inserted_at",
+            sa.DateTime(),
+            nullable=False,
+            comment="Timestamp when the message was stored; defines transcript order",
+        ),
+        sa.ForeignKeyConstraint(
+            ["session_id"], ["chatbot_session.id"], ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_chatbot_session_message_session_id"),
+        "chatbot_session_message",
+        ["session_id"],
+        unique=False,
+    )
+
+    # ADD VALUE cannot run inside the migration's transaction on all PG versions.
+    with op.get_context().autocommit_block():
+        op.execute("ALTER TYPE featureflagkey ADD VALUE IF NOT EXISTS 'CHATBOT'")
+
+
+def downgrade():
+    op.drop_table("chatbot_session_message")
+    op.drop_table("chatbot_session")
+    op.drop_table("chatbot_flow_version")
+    op.drop_table("chatbot")
+    sa.Enum(name="chatbotmessageroleenum").drop(op.get_bind())
+    sa.Enum(name="chatbotcompletionstatusenum").drop(op.get_bind())
+    sa.Enum(name="chatbotsessionstatusenum").drop(op.get_bind())

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

@@ -0,0 +1,3 @@
+Create a chatbot in the caller's project and return it (`201`).
+
+Only `name` is required (1–255 characters; display only). A new chatbot has no flow: its first `PUT /chatbots/{chatbot_id}/flow` goes live at once, after which sessions can start. Later saves are drafts that go live with `POST /chatbots/{chatbot_id}/flow/publish`.

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

@@ -0,0 +1,5 @@
+Start a conversation between a chatbot and an end user and return `{session_id, chatbot_id, flow_version, active_node_id}` (`201`).
+
+The session is pinned to the chatbot's live flow version and starts on its entry node (the agent node the `input` node leads to, or the flow's only agent node). Every turn runs that pinned version, so publishing a newer flow applies to new sessions only. `contact_ref` is the caller's own identifier for the end user and is stored as-is.
+
+A contact has at most one active session per chatbot. Creating a session marks the contact's previous `active` sessions of this chatbot as `inactive`, in the same transaction; those still `in_progress` become `abandoned`, while `complete` ones stay `complete`. Sessions of other chatbots or other contacts are untouched. Returns 404 if the chatbot doesn't exist in this project or was deleted, 409 if the chatbot has no published flow (publish the flow first) or another session for the same contact was created concurrently (retry).

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

@@ -0,0 +1,3 @@
+Soft-delete a chatbot; returns `{"message": "Chatbot deleted successfully"}`. It disappears from list/get and its flow can no longer be edited, published, or used for new sessions.
+
+In the same transaction every `active` session of the chatbot becomes `inactive` (those still `in_progress` become `abandoned`; `complete` ones stay `complete`), so further messages to them return 409. Stored sessions and transcripts are kept. Returns 404 if the chatbot doesn't exist in this project or was already deleted.

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

@@ -0,0 +1,3 @@
+Get one chatbot of the caller's project by `chatbot_id`.
+
+Returns 404 if the chatbot doesn't exist in this project or was deleted. The chatbot's flow is fetched separately with `GET /chatbots/{chatbot_id}/flow`.

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

@@ -0,0 +1,5 @@
+Get a version of the chatbot's React Flow graph: `{id, chatbot_id, version, is_live, nodes, edges, inserted_at, updated_at}`.
+
+With `?version=N` that version is returned. Without it, the draft is returned if one exists (the highest version when it is not live), otherwise the live version. `nodes` and `edges` come back exactly as saved, with React Flow keys (`sourceHandle`, `targetHandle`, and UI-only keys such as `measured`). Agent nodes of a published version carry the pinned `data.config_version`.
+
+Returns 404 if the chatbot doesn't exist in this project, has no flow yet, or has no such version.

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

@@ -0,0 +1,5 @@
+Get the most recently created chatbot session for `contact_ref` in the caller's project, optionally narrowed to one chatbot with `chatbot_id`.
+
+`contact_ref` is required and matched exactly. Omit `chatbot_id` to get the contact's newest session across the project's chatbots — the response's `chatbot_id` and `chatbot_name` say which one. Sessions of deleted chatbots are not returned.
+
+The response carries the pinned `flow_version`, the `active_node_id` and that node's `prompt`, the session `status` (`active`, or `inactive` once a newer session for the same contact and chatbot was created or the chatbot was deleted), its `completion_status` (`in_progress` until every field the prompt asks for is collected, then `complete`; `abandoned` if it was deactivated while still `in_progress`), and the flat `collected_fields` map (`{field_name: value}`) gathered so far. Returns 404 when no matching session exists in this project.

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

@@ -0,0 +1,3 @@
+List the caller's project's chatbots, most recently updated first. Deleted chatbots are excluded.
+
+Paginate with `skip` (default 0) and `limit` (1–100, default 100).

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

@@ -0,0 +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 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

@@ -0,0 +1,7 @@
+Save the chatbot's flow: `{nodes, edges}` in React Flow shape. Returns the saved version.
+
+The **first** save of a chatbot creates version 1 and makes it live immediately, so the chatbot can run without a separate publish. Because it goes live, it gets the same strict validation as publish (see the publish endpoint) and its missing `config_version`s are pinned to the latest; an invalid first flow returns 422 and nothing is saved.
+
+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.