AI Engine

TLDR — The AI module (service/api/src/modules/ai/) hosts two subsystems under one namespace: AI App Studio (current, mounted at /ai/*) — build and run AI apps as flow graphs over knowledge bases, LLM integrations and MCP tool servers; and a legacy assistant-config + RAG ingest surface (/assistant/*) that survives for per-collection ingest mapping and backfill into KAOS. Everything is gated by the app-module ai (view / manage). The old grounded /assistant/chat orchestrator (gate → retrieve → synthesize → enforceGrounding) has been retired — chat now runs through the flow executor.

Overview

The module registers under app-module ai and exposes a single Fastify router (routes.ts) that aggregates six controllers:

Controller	Prefix	Purpose
`apps.controller`	`/ai/apps`	App CRUD + publish/clone/monitoring, flow graph, logs, annotations, runtime chat
`knowledge.controller`	`/ai/knowledge`	Knowledge-base CRUD + documents + retrieval-test query
`integrations.controller`	`/ai/integrations`, `/ai/models`	LLM/model-provider CRUD + connection test + model discovery
`mcpServers.controller`	`/ai/mcp-servers`	MCP tool-server CRUD + connection test + tool listing
`sources.controller`	`/ai/sources`	Collection/field picker for collection-type KBs
`ingest.controller`	`/assistant/*`	Legacy assistant-config CRUD + RAG ingest mapping/backfill

Two data-model families coexist:

ai_* collections (uuid PK, registered via ai-schema.ts as a setup-wizard ModuleSchema) — the App Studio store. These are real collections (visible in the schema, RBAC-capable) but most are marked hidden in their meta.
odp_ai_* system tables (int PK, migrations 066/067) — the legacy assistant-config store, in SYSTEM_TABLES (hidden from /items).

The module is named ai (schema group ai), not kaos — KAOS is one knowledge-base kind, not the contract. The retrieval/LLM ports live in ports.ts; KB kinds dispatch through providers/kb/.

AI App Studio

An AI app (ai_apps row) has a type that decides how it runs:

`type`	Runtime shape
`chatbot`	Synthetic graph: `start → [knowledge-retrieval] → llm → answer` (retrieval node added only when the app config lists `knowledge_ids`).
`text-generation`	Synthetic graph: `start → llm → answer`.
`chatflow`	Author-defined graph loaded from `ai_flow_nodes` / `ai_flow_edges`.
`agent`	Author-defined graph (same loader).
`workflow`	Author-defined graph (same loader).

runChat (runtime/chat.ts) dispatches on type: chatflow|agent|workflow load the persisted graph; chatbot|text-generation build a synthetic graph from app.config. Both then run the same executor (runtime/executor.ts).

Flow graph (nodes + edges)

Graphs are stored normalized (ai_flow_nodes + ai_flow_edges) and edited through GET/PUT /ai/apps/:id/graph, which converts to/from { nodes, edges }. A node has node_key, type, label, position, config; an edge has source_node, target_node, source_handle (branch label), condition, order. The executor walks the graph trigger-driven and branch-aware: a node runs once all its incoming edges' sources have resolved and at least one incoming edge is active (branch taken); nodes whose incoming branches are all dead are marked skipped.

Node types (`runtime/registry.ts`)

A handler is { type, execute(node, ctx) }; registering a node adds a type, the executor never changes.

Node type	Role
`start`	Emits `{ query }`.
`question-classifier`	LLM classifies the query into exactly one configured class (+ optional entity extraction); sets the branch `handle`. Loose-JSON parse + normalized label match + `default_class` fallback.
`knowledge-retrieval`	Searches one or more KBs (`knowledge_ids`) via the KB registry. Text chunks → `chunks` lane + citations; structured rows → `items` lane (no citation). Supports `top_k`, `score_threshold`, and `expand: 'full-doc'` (refetch each doc's full text).
`llm`	Builds context from upstream `chunks`/`items` + history, calls the LLM. Streams tokens when it is the terminal answer LLM; runs a guarded MCP tool loop when the app has MCP servers attached and the provider supports tool-calling.
`template`	Renders a template string.
`variable-aggregator`	Pass-through join point (`{}`).
`if-else`	Evaluates a condition → `handle: 'true'
`http-request`	SSRF-guarded outbound HTTP (`safeFetch`).
`code`	Runs a JS snippet via `Function()` — disabled by default (`AI_ENABLE_CODE_NODE=1`; not a sandbox).
`tool`	Invokes one built-in tool from the allow-list registry (`runtime/tools.ts`: `http_get`, `math`, `current_datetime`).
`mcp-agent`	LLM acts as a data-fetch planner over an MCP server's tools; raw tool results flow into the `items` lane un-summarized.
`answer`	Terminal node — emits the final answer text (or falls back to the most recent LLM output); maps `config.meta` refs into response `meta`.

Knowledge bases (`ai_knowledge_bases`)

A KB is a pluggable grounding source dispatched by kind through KbProviderManager (providers/kb/manager.ts), modeled on the storage-driver pattern. Two kinds ship:

kaos (legacy type: 'rag') — wraps the KAOS RAG backend (providers/kb/kaos.provider.ts + kaos-client.ts). Capabilities: read + write + cites + ingestConfig. source_config carries base_url / api_key / space (empty → KAOS_* env). ODP forwards only small chunk knobs (chunk_config.size/overlap); KAOS owns the RAG strategy.
collection — live ODP-collection rows via ItemsService (collection.provider.ts). Capabilities: read only, cites: false, no ingest. RBAC applies through the requester's accountability.

Capability ∩ policy: each kind has intrinsic capabilities (the ceiling); each KB row carries allow_read / allow_write flags that can only narrow them (never widen). Enforced centrally in KbProviderManager (canRead/canWrite, search degrades silently, ingest/delete fail loudly). GET /ai/kb-kinds returns the catalog so admin pickers can filter.

Integrations (`ai_integrations`) — LLM/model providers only

An integration is one OpenAI-compatible provider (provider, config.base_url, config.api_key, config.models[], capabilities, enabled). The api_key is masked on read and preserved on a blank PATCH. Endpoints test the connection (GET {base_url}/models), discover the full model catalogue, and derive the allow-listed model list. An app binds a model to its provider via config.integration_id (disambiguates two rows of the same provider type).

MCP servers (`ai_mcp_servers`) — tool servers the app consumes

An MCP server row (label, url, transport = streamable-http|sse, headers, allowed_tools, timeout_ms) is a remote tool server an app attaches (via config.mcp_server_ids) so its llm / mcp-agent nodes can function-call. headers values are secrets — AES-256-GCM encrypted at rest (via SECRET), masked on read, decrypted only for the outbound request. See Runtime for the client contract.

Data model

`ai_*` collections (App Studio — uuid PK, group `ai`)

Declared in collections/ai-schema.ts (ModuleSchema, version: 4). All carry created_at/updated_at audit; several also carry created_by/updated_by.

Collection	Key fields
`ai_apps`	`name`, `description`, `type`, `icon`, `tags`, `status` (`draft`\|`published`), `config` (json), `starred`, `is_template`, `category`, `version`, `published_at`
`ai_flow_nodes`	`app_id`, `node_key`, `type`, `label`, `position` (json), `config` (json)
`ai_flow_edges`	`app_id`, `source_node`, `target_node`, `source_handle`, `condition`, `label`, `order`
`ai_app_logs`	`app_id`, `conversation_id`, `query`, `answer`, `status` (`running`\|`success`\|`partial`\|`error`), `reason`, `error`, `trace` (json), `steps` (json), `mcp_calls` (json), `citations` (json), `retrieved_chunks`, `tokens`, `latency_ms`, `user_identifier`, `started_at`, `finished_at`
`ai_conversations`	`app_id`, `title`, `source`, `external_ref`, `summary`, `message_count`
`ai_messages`	`conversation_id`, `role` (`user`\|`assistant`\|`system`\|`summary`), `content`, `meta` (json)
`ai_annotations`	`app_id`, `question`, `answer`, `hit_count`
`ai_knowledge_bases`	`name`, `description`, `type` (`rag`\|`collection`), `source_config` (json), `embedding_model`, `chunk_config` (json), `retrieval_config` (json), `classification_default`, `scopes` (json), `allow_read`, `allow_write`, `doc_count`, `chunk_count`
`ai_documents`	`knowledge_id`, `source_type`, `source_config` (json), `filename`, `content` (raw text), `content_hash` (sha256 → dedup), `status`, `chunk_count`, `word_count`, `classification`, `error_message`
`ai_integrations`	`provider`, `label`, `config` (json), `capabilities` (json), `enabled`, `install_count`, `description`
`ai_mcp_servers`	`label`, `url`, `transport`, `headers` (json, secret), `enabled`, `allowed_tools` (json), `timeout_ms`, `description`

Relations: nodes/edges/logs/conversations/annotations → ai_apps; edges' source_node/target_node → ai_flow_nodes; ai_messages → ai_conversations; ai_documents → ai_knowledge_bases.

The schema seed provides a minimal 9router integration (required) plus demo KBs and app templates (sample, loaded from plans/ai-engine/app-templates/*.json by the seed script).

`odp_ai_*` system tables (legacy assistant-config — int PK)

Created by migrations 066-ai-assistant-config and 067-ai-ingest-maps; listed in SYSTEM_TABLES (hidden from /items). Secrets are stored as env-ref names (api_key_ref = the name of an env var, never the raw key).

Table	Purpose	Notable fields
`odp_ai_llm_providers`	OpenAI-compatible LLM endpoints	`name`, `provider_type`, `base_url`, `api_key_ref`, `model`, `is_default`
`odp_ai_knowledge_backends`	RAG backends	`name`, `backend_type`, `base_url`, `api_key_ref`, `config`, `is_default`
`odp_ai_prompts`	Domain context + system prompt	`name`, `domain_context`, `system_prompt`, `refusal_message`, `variables`
`odp_ai_agents`	Binds prompt + provider + backend	`name`, `llm_provider_id`, `knowledge_backend_id`, `prompt_id`, `temperature`, `enabled`
`odp_ai_ingest_maps`	Per-collection ingest field mapping	`collection` (unique), `enabled`, `version_field`, `title_field`, `field_map`, `template`

Endpoints

Every route is guarded by app-module ai: reads use action view, writes use manage (controllers/shared.ts → gate(request, action) → validateAppAccess(acc, 'ai', action, …)). Data operations run ItemsService without accountability (already gated at the app-module layer), mirroring the core admin routes. CRUD for App-Studio collections is emitted by the generic registerCrud helper.

`/ai/apps`

Method + path	Action	Notes
CRUD `/ai/apps`, `/ai/apps/:id`	view/manage	`name` required.
`POST /ai/apps/:id/publish`	manage	Publish the app.
`POST /ai/apps/:id/clone`	manage	Duplicate the app.
`GET /ai/apps/:id/monitoring`	view	Aggregated run metrics.
`GET/PUT /ai/apps/:id/graph`	view/manage	Read/write the normalized flow graph as `{ nodes, edges }`.
`GET /ai/apps/:id/logs`	view	`ai_app_logs` scoped to the app.
`GET/POST /ai/apps/:id/annotations`, `DELETE /ai/annotations/:id`	view/manage	Q/A annotations.
`POST /ai/apps/:id/chat`	view	Runtime chat. One handler; `stream` flag (`body.stream:true` or `?stream=1`) switches between a single JSON `FlowResult` and an SSE stream of `ChatEvent` frames.

`/ai/knowledge`

Method + path	Action	Notes
CRUD `/ai/knowledge`, `/ai/knowledge/:id`	view/manage	`source_config.api_key` masked on read; blank PATCH keeps the secret.
`GET /ai/kb-kinds`	view	Registered kinds + intrinsic capabilities.
`GET /ai/knowledge/stats`	view	Per-KB document counts by status.
`GET /ai/knowledge/:id/documents`	view	Paginated docs + `filter_count` + status `stats`.
`POST /ai/knowledge/:id/documents`	manage	Upload a document (write-gated by capability ∩ policy); size-guarded; dedup by `(kb, filename)` + content hash; enqueues background ingest.
`DELETE /ai/documents/:id`	manage	Purges the doc's units from KAOS (best-effort) then deletes the row.
`POST /ai/knowledge/:id/reindex`	manage	Re-queues this KB's `error` documents.
`POST /ai/knowledge/:id/query`	view	Retrieval test — runs a query through the KB provider registry.

`/ai/integrations` and `/ai/models`

Method + path	Action	Notes
CRUD `/ai/integrations`, `/ai/integrations/:id`	view/manage	`config.api_key` masked; blank PATCH keeps it. `provider` required.
`POST /ai/integrations/:id/test`	manage	Probe `GET {base_url}/models`; never throws (`ok:false` + reason).
`GET /ai/integrations/:id/available-models`	manage	Full provider model catalogue for the allow-list picker.
`GET /ai/models?capability=`	view	Models derived from enabled integrations (each carries `integration_id`).

`/ai/mcp-servers`

Method + path	Action	Notes
CRUD `/ai/mcp-servers`, `/ai/mcp-servers/:id`	view/manage	`headers` encrypted on write, masked on read; blank PATCH keeps secrets. `label` required.
`POST /ai/mcp-servers/test`	manage	Test-before-save with ad-hoc form values.
`POST /ai/mcp-servers/:id/test`	manage	Test a saved server → `{ ok, tools[] }`.
`GET /ai/mcp-servers/:id/tools`	view	`allowed_tools`-filtered tool list for the attach picker.

`/ai/sources`

Method + path	Action	Notes
`GET /ai/sources/collections`	view	Non-system collections for collection-KB config.
`GET /ai/sources/fields?collection=`	view	Fields of a collection.

`/assistant/*` (legacy assistant-config + RAG ingest)

All gated ai / manage. CRUD (GET list, POST, GET/PATCH/DELETE /:id) over the odp_ai_* tables:

Resource	Path
LLM providers	`/assistant/providers`
Knowledge backends	`/assistant/knowledge-backends`
Prompts	`/assistant/prompts`
Agents	`/assistant/agents`
Ingest maps	`/assistant/ingest-maps`

Plus the ingest tooling (controllers/ingest.controller.ts):

GET /assistant/ingest/collections — non-system collections for the mapping dropdown.
GET /assistant/ingest/fields?collection= — a collection's fields with parsed special (relations visible).
POST /assistant/ingest/backfill — bulk-load a collection into KAOS: paginate rows (read as the requesting ai.manage user, so RBAC applies) → map each via the ingest map → write through the KB registry (an ephemeral KAOS KB from the resolved agent-config). Returns { ingested, skipped, units }.

The retired /assistant/chat, /assistant/chat/stream and /assistant/health endpoints (the gate → retrieve → synthesize → enforceGrounding orchestrator) no longer exist. Chat is served only by POST /ai/apps/:id/chat.

Runtime

The executor (runtime/executor.ts) walks the graph, keeping a variable pool (vars[node_key] = output), a citation list and a meta bag.

Provider resolution (runtime/providers.ts): resolveLLM prefers the app's bound integration_id (base_url/key), else the first enabled integration, else ASSISTANT_LLM_* env, via createOpenAICompatibleLLM. resolveDefaultModel resolves app.config.model → integration's first llm model → env. loadKb loads a KB row by uuid.
Query rewrite: when a conversation history exists, the executor rewrites a follow-up into a standalone question (query_resolved) before classify/retrieval/mcp-agent read it. No history → no LLM call → byte-for-byte identical to the stateless path.
KB retrieval: dispatched through KbProviderManager.search, which enforces capability ∩ policy (a read-blocked KB yields [] and the flow degrades rather than erroring).
MCP client (runtime/mcp-client.ts): every connection is short-lived (connect → op → close), tool listings cached per (id, updated_at). The outbound URL is vetted through the core SSRF guard (assertUrlAllowed) before connecting; allowed_tools filters both listing and invocation; the guarded tool loop is bounded to MAX_TOOL_ITERATIONS = 5 (then forces a final tool-less answer).
Result: FlowResult { status: 'answered' | 'partial' | 'error', answer, reason, citations, meta, trace }. The SSE layer (runtime/chat-stream.ts) maps this to the wire AnswerStatus (answered → answered, partial → degraded, else declined) and encodes ChatEvent frames (conversation → status → step* → answer deltas → citations → done).

Programmatic trigger (other extensions / AI workflows)

Besides the HTTP route, another extension — or an AI workflow that needs to call an app as a sub-step — can trigger an AI app by ID in-process, without self-calling /ai/apps/:id/chat. The core AI module publishes a façade on the cross-extension service registry (see Extension Guardrails §6), under the ref id ai.engine. It runs the exact same flow as the HTTP route (conversation memory + ai_app_logs + streaming), so an app behaves identically whether a browser or an extension started it.

// endpoint / hook that runs AFTER 'extensions.register' (never at a hook's top level — registry not sealed yet)
export default defineEndpoint((router, ctx) => {
  router.post('/run-ai', async (req, reply) => {
    // consume by string id — no need to import the ref (keeps you on @odp/api/types only)
    const ai = ctx.registry.tryConsume('ai.engine');
    if (!ai) return reply.status(503).send({ error: 'AI engine not available' });

    const result = await ai.run('<APP_ID>', {
      query: 'câu hỏi của user',
      variables: { orderId: 123, locale: 'vi' }, // structured params → the app variable pool ({{app.variables.*}})
    }, {
      persist: false,           // throwaway run — don't touch channel history / log timeline (default true)
      source: 'my-ext',         // channel tag on the conversation + logs (default 'extension')
      externalRef: 'wf-inst-9', // idempotent conversation mapping (e.g. a workflow instance id)
      accountability: req.accountability, // scope the flow's own data access (default null = system)
      onStep: (ev) => {},       // live node/tool progress (optional)
      onAnswerToken: (d) => {}, // live answer tokens (optional)
    });
    return reply.send({ data: result }); // FlowResult
  });
});

run(appId, input, options?) → Promise<FlowResult> — resolves (or creates) the conversation, runs the graph, resolves after the flow completes. Throws if appId does not exist.
input: { query, conversation_id?, variables? }. variables is the params channel for workflow/agent apps — merged into the variable pool so nodes read {{ app.variables.* }}.
options (all optional, defaults mirror the HTTP route): persist, source, externalRef, accountability, onAnswerToken, onStep.
Contract lives at modules/ai/engine-ref.ts (aiEngineRef, AiEngine, AiEngineInput, AiEngineOptions); provided at boot in src/index.ts before the registry is sealed. Consume by the typed aiEngineRef or the plain 'ai.engine' id.

Ingest (RAG) — two paths

Both paths write only through the KB registry (KbProviders), never a raw backend client, so capability ∩ policy is always enforced.

Studio document upload (ingest-queue.ts) — POST /ai/knowledge/:id/documents writes an ai_documents row (status: 'queued') and enqueues a BullMQ / Redis job (Redis is mandatory — no memory fallback). The single-concurrency worker reads the row, ingests to the KB (KAOS), and marks completed / error. Transient failures (KAOS 5xx/429/network/timeout) retry with backoff; permanent 4xx fail fast. A row deleted during ingest has its orphan units purged.
Event-driven collection sync (ingest-listener.ts) — when ASSISTANT_INGEST_COLLECTIONS lists a collection, items.create / items.update / items.delete actions map the row (via its odp_ai_ingest_maps entry, or a generic scalar dump) and upsert/delete it into KAOS. The base row is read WITH the triggering user's accountability (RBAC applies to what is ingested), and a failure is logged and never rethrown — an ingest error must not break the underlying CRUD write.

An extension can intervene on the built doc through the ai-engine.ingest.remap filter (fires in both the backfill and listener paths): return the (possibly modified) IngestDoc to ingest it, or null/undefined to skip the row.

Invariants & gotchas

Actor split. Control-plane /ai/* CRUD runs ItemsService without accountability — access is already gated at the app-module layer (ai view/manage), mirroring core admin routes. Ingest (both paths) runs with accountability so collection/field RBAC decides what reaches the KB.
Raw Knex only off the actor path. Direct knex(...) use is limited to runtime/config reads (graph load, integration/KB/MCP resolution, aggregate stats with fixed status literals) and the legacy odp_ai_* config services — never a per-user data mutation.
Secret masking everywhere. Integration config.api_key, KB source_config.api_key, and MCP headers are masked on read and preserved on a blank PATCH; MCP headers are additionally AES-256-GCM encrypted at rest and only decrypted for the outbound request. Legacy odp_ai_* rows store secrets as env-ref names (api_key_ref), never raw.
code node disabled by default — requires AI_ENABLE_CODE_NODE=1 in a trusted deployment (Function() eval, not a sandbox).
AI_MOCK_RETRIEVAL is dev-only — synthetic chunks let a RAG flow complete without real KAOS data; hard-disabled in production (a fabricated "grounded" answer must never ship). Mocked responses carry meta.mock = true and mock:// citations.
Redis is mandatory for ingest — with REDIS_ENABLED off, documents queue but do not ingest.

Conversation memory & logging status

These recently-landed pieces are present and wired at HEAD, but are the most actively evolving part of the module — treat them as maturing:

Streaming chat (runtime/chat-stream.ts + the SSE branch of POST /ai/apps/:id/chat). Real per-token streaming happens for the terminal answer LLM node (via onAnswerToken); other cases buffer-then-chunk the completed answer. A NOTE in chat-stream.ts still frames full per-token streaming as a forward-compatible target.
Multi-turn conversation persistence (services/conversation.service.ts) — channel-agnostic memory over ai_conversations + ai_messages: reuse-or-create a conversation, load a token-budgeted history (AI_HISTORY_TOKEN_BUDGET), append each turn, and roll older turns into a summary once history crosses ~50% of the budget (folded rows are kept, marked meta.folded). All I/O goes through ItemsService, never raw Knex.
Lifecycle logging (services/log.service.ts) — ai_app_logs uses a create-at-running → throttled step-timeline updates → finalize-in-finally lifecycle, so even a mid-flight crash leaves a durable trace instead of no log.

Environment

Var	Default	Purpose
`ASSISTANT_LLM_PROVIDER`	`openai-compatible`	Fallback LLM provider type.
`ASSISTANT_LLM_BASE_URL`	`http://localhost:20128/v1`	Fallback OpenAI-compatible base URL.
`ASSISTANT_LLM_MODEL`	`gpt-4o-mini`	Fallback model id.
`ASSISTANT_LLM_API_KEY`	—	Fallback key (prefer per-integration `config.api_key`).
`ASSISTANT_INGEST_COLLECTIONS`	`` (off)	Comma list of collections auto-synced to KAOS on item change.
`ASSISTANT_RATE_LIMIT_PER_MIN`	`20`	Legacy assistant rate limit.
`KAOS_BASE_URL`	`http://localhost:7071`	KAOS RAG API base URL.
`KAOS_API_KEY`	—	KAOS bearer token.
`KAOS_IDENTITY_ID` / `KAOS_IDENTITY_KIND` / `KAOS_IDENTITY_SCOPES`	`odp` / `service` / `knowledge:jobs:write,knowledge:read`	KAOS identity used for ingest/read.
`AI_INGEST_MAX_CONTENT_BYTES`	`15 MB`	Per-document size guard (below KAOS's body limit).
`AI_HISTORY_TOKEN_BUDGET`	`3000`	Conversation-memory history budget (tokens).
`AI_MOCK_RETRIEVAL`	`false`	Dev-only synthetic retrieval; hard-disabled in production.
`AI_ENABLE_CODE_NODE`	`false`	Enable the `code` flow node (`Function()` eval — not a sandbox).
`REDIS_ENABLED`	—	Required for the background ingest queue.

AI Engine ​

Overview ​

AI App Studio ​

Flow graph (nodes + edges) ​

Node types (runtime/registry.ts) ​

Knowledge bases (ai_knowledge_bases) ​

Integrations (ai_integrations) — LLM/model providers only ​

MCP servers (ai_mcp_servers) — tool servers the app consumes ​

Data model ​

ai_* collections (App Studio — uuid PK, group ai) ​

odp_ai_* system tables (legacy assistant-config — int PK) ​

Endpoints ​

/ai/apps ​

/ai/knowledge ​

/ai/integrations and /ai/models ​

/ai/mcp-servers ​

/ai/sources ​

/assistant/* (legacy assistant-config + RAG ingest) ​

Runtime ​

Programmatic trigger (other extensions / AI workflows) ​

Ingest (RAG) — two paths ​

Invariants & gotchas ​

Conversation memory & logging status ​

Environment ​