> ## Documentation Index
> Fetch the complete documentation index at: https://docs.kaireonai.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Pipeline IR (Flow)
> The typed JSON-AST format that defines every KaireonAI Flow pipeline. Authored by humans or AI; validated before it can run.
The Pipeline IR is the new source of truth for ETL pipelines starting in
KaireonAI Flow Phase 1. It replaces the legacy `PipelineNode + PipelineEdge`
relational format with a single typed JSON document that the runtime
interprets directly. Pipelines authored in IR mode are authored once and
read by:
* The visual canvas (renders the IR)
* The AI assistant (proposes diffs against the IR)
* The MCP server (accepts IR via tool calls)
* The runtime interpreter (executes the IR)
Same source of truth, four surfaces.
See [Transforms](/data/transforms/transforms) for the full reference of the
column ops a `transform` node can carry. The IR schema accepts 19 op
types; the visual editor exposes the 14 most-used ones in its toolbar
(complex ops like `summarize`, `vector_embed`, `geo_resolve`,
`sentiment_score`, `language_detect` are authored via the JSON IR tab
or the AI assistant).
## Why IR-first
Generative AI is reliable when it produces structured data, not code.
Constrained JSON output validated against a Zod schema cannot drift
from the contract. This is what makes "AI builds the pipeline" actually
work — every proposal is schema-valid by construction.
## Top-level shape
```json theme={null}
{
"kind": "pipeline",
"version": "1.0",
"id": "retailco-daily-orders",
"metadata": {
"name": "Daily orders ingestion",
"owner": "data-team",
"tags": ["batch", "orders"]
},
"schedule": {
"kind": "cron",
"expression": "0 2 * * *",
"timezone": "America/Los_Angeles"
},
"nodes": [ /* see node kinds below */ ],
"errorHandling": {
"dlq": { "enabled": false },
"retry": { "maxAttempts": 3, "backoff": "exponential" }
}
}
```
## Node kinds (Phase 1)
| Kind | Purpose |
| ----------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `source` | Read from external system (file/DB/API/stream). Built-in **atomic staging** moves the source file through `.processing/` → `.archive/{YYYY}/{MM}/{DD}/` on success or `.failed/` on parse failure |
| `transform` | Column ops (per-row + set-level) — 19 op types ([Transforms](/data/transforms/transforms)) |
| `validate` | Row-level + dataset-level validation |
| `target` | Write to internal schema table |
| `branch` | Conditional routing |
| `join` | Multi-input join |
| `enrich` | Per-row external call (LLM tag, geocode, ML score). **Beta** — node is validated and recorded, but its providers (`llm_tag` / `geocode` / `ml_score`) require tenant credentials that are not wired in this build. Without them the executor **throws at run time** (it does not pass rows through) — remove the `enrich` node from a pipeline before running it. This pipeline `enrich` node is distinct from the Decision Flow **Enrich stage**, which is wired |
| `outcomes` | **Inbound** sink. Ingests a third-party response file (opens / clicks / bounces / conversions / unsubscribes) into `InteractionHistory` **and** feeds online model learning via the shared `applyOutcome` core. No output table — each row is mapped to a customer outcome and recorded idempotently. Negative outcomes optionally write a suppression. See [Outcomes node](#outcomes-node) below |
**Archival is handled exclusively by the source node's `atomicity` config.**
There is no separate `archive` node kind — it was removed entirely from
the IR, the executor registry, the structural validator, the visual
editor, and the AI prompt. The source moves the source file to
`successFolder` after parsing succeeds and to `failureFolder` after parse
errors. Date tokens `{YYYY}`, `{MM}`, `{DD}`, `{YYYY-MM-DD}`, `{YYYYMMDD}`,
`{HH}`, `{mm}`, `{ss}` are expanded at run time, so a nested folder
layout like `.archive/{YYYY}/{MM}/{DD}/` produces `.archive/2026/05/12/`.
Authoring an IR with a node of `kind: "archive"` will now fail schema
validation at save time — file-level atomicity covers the common case
and the previous row-level archive executor only ever recorded intent
without actually moving data.
## Source node example
```json theme={null}
{
"id": "src",
"kind": "source",
"connector": "local_fs",
"config": {
"path": "/data/orders/",
"pattern": { "type": "glob", "value": "orders_*.csv" },
"ordering": "latest_by_mtime",
"waitPolicy": { "maxRetries": 6, "intervalMinutes": 10, "onMissAction": "alert" },
"atomicity": {
"stagingFolder": ".processing/",
"successFolder": ".archive/{YYYY}/{MM}/{DD}/",
"failureFolder": ".failed/{YYYY}/{MM}/{DD}/"
},
"format": "csv"
}
}
```
The IR schema `source.kind` field accepts a fixed set of cloud-store
and filesystem kinds: `s3`, `gcs`, `azure_blob`, `sftp`, `ftp`,
`local_fs`, `http_pull`. Runtime executors for `local_fs`, `s3`,
`gcs`, `azure_blob`, `sftp`, and `http_pull` are all live. `ftp` is
documented as deprecated (plaintext) and never reaches a runtime
executor. Streaming kinds (`kafka`, `kinesis`, `pulsar`) are gated by
`FLOW_STREAMING_ENABLED` and only land when self-hosters provide their
own broker. Note that `source.kind` is a separate concept from the full
connector registry (80 types); pipelines reference a saved Connector
record by `connectorId`, not by kind directly.
## Target node load modes
| Mode | Status | Notes |
| ----------------------- | ------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `append` | Live | Explicit-column INSERT with NULLIF + per-column CAST |
| `truncate` | Live | TRUNCATE + INSERT wrapped in `prisma.$transaction`. Empty-source guard (`failOnEmptySource: true` default) aborts before the destructive statement when upstream produced 0 rows |
| `upsert` | Live | Requires `upsertKey`; throws if every projected column is in `upsertKey` (no updatable columns) |
| `blue_green` | Live | Loads to `_new`, atomic 3-step rename, drops `_old`. Empty-source guard applies |
| `incremental_watermark` | Live | High-water persisted in `pipeline_watermarks`; INSERT + watermark-upsert run in a single transaction. Requires `watermarkColumn` |
| `cdc_mirror` | Live | Op-column-driven change apply: each row's `cdcOpColumn` value (case-insensitive `insert`/`update`/`upsert`/`I`/`U`/`c`/`r` → upsert; `delete`/`D` → delete) is applied by `upsertKey` in one transaction. Unknown op values fail the run before any DML. Requires `cdcOpColumn` + `upsertKey` |
### Optional safety fields on target nodes
| Field | Type | Behavior |
| ----------------------- | ----------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `failOnEmptySource` | `boolean` (default `true`) | Aborts `truncate` / `blue_green` BEFORE the destructive statement when upstream produced 0 rows. Set `false` only when "empty file means clear the table" is the genuine intent |
| `expectedRowCountDelta` | `{ minPct?, maxPct?, windowRuns? }` | Emits a `warning` System Health alert when today's `rowsLoaded` is outside the band vs. the running mean over the last N successful runs (default 7). Run still completes |
| `backupBeforeLoad` | `{ enabled, retainCount? }` | Snapshots the target via `CREATE TABLE … AS TABLE …` before destructive load. Only acts on `truncate` and `blue_green`. Backups older than `retainCount` (default 3) are pruned post-success. On failure, the backup is left in place and the run record's `meta.backupTable` carries the snapshot table name |
## Outcomes node
The `outcomes` node closes the learning loop. Where a `target` writes rows
*out* to a schema table, an `outcomes` node reads a vendor **response file**
*in* — the opens, clicks, bounces, conversions, and unsubscribes an email or
SMS provider hands back after a send — and turns each row into a recorded
customer outcome. It has **no output table**: its side effect is an
`InteractionHistory` row plus online model learning through the shared
`applyOutcome` core, exactly the same path the real-time `/respond` API uses.
### Config
| Field | Required | Purpose |
| -------------------------------------------------------- | ---------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `input` | yes | Upstream node whose rows are the response file (usually the `source`, or a `transform` that cleans the vendor columns first) |
| `columnMapping.customerId` | yes | Column holding the external customer identifier |
| `columnMapping.outcomeKey` | yes | Column holding the event name (`open` / `click` / `unsubscribe` …). Must match an `OutcomeType` key for the tenant; unknown keys are **skipped, not fatal** |
| `columnMapping.offerId` **or** `columnMapping.ridColumn` | one of | How the offer is resolved. `ridColumn` names the column carrying the `_kaireon_rid` creative token; `offerId` names a column holding the offer id or name. **Exactly one is required** (enforced by the schema's `superRefine`) |
| `columnMapping.channelId` | no | Delivery channel column. Also resolves from the creative when using `ridColumn` |
| `columnMapping.timestamp` | no | Event time column. Defaults to ingestion time when absent |
| `columnMapping.conversionValue` | no | Revenue / conversion value column |
| `columnMapping.eventId` | no | Vendor event id — the **preferred idempotency key** |
| `attributionWindowDays` | no (default `7`) | How far back to look for the originating **outbound** send when attributing the outcome to a campaign. `1`–`365` |
| `suppressionRules` | no | See below |
**`_kaireon_rid` vs `offerId`.** `_kaireon_rid` is the creative token Kaireon
stamps on every outbound row (it is `Creative.id`). Mapping `ridColumn` is the
recommended path because the offer, creative, **and** channel all fall out of
that one column — so vendor files that echo the token back need only three
mapped columns (`customerId`, `ridColumn`, `outcomeKey`). Use `offerId`
(id-or-name, tenant-scoped) only when the vendor file cannot carry the token.
An unresolved token/offer still records the outcome, just unattributed to an
offer.
### Idempotency
`applyOutcome` is always called with an explicit idempotency key: the vendor
`eventId` when mapped, otherwise a stable content hash of
`(customerId, offerId, channelId, outcomeKey, timestamp, conversionValue)`.
**Re-processing the same file never double-counts** — the dedupe and
anti-double-reward guarantee lives in `applyOutcome`; the node only supplies
the key. Rows that dedupe are counted separately from rows newly recorded.
### Attribution
For a resolved offer, the node looks for the most recent **outbound**
`InteractionHistory` impression for `(customer, offer[, channel])` whose
timestamp falls inside `[eventTime − attributionWindowDays, eventTime]` and
attaches that impression's `campaignId`. No matching send → the outcome is
still recorded, marked **unattributed**.
### Negative outcomes → suppression
A **newly-recorded** negative outcome (one whose `outcomeKey` is classified
`negative` in the `OutcomeType` registry **and** listed in
`suppressionRules.outcomeKeys`) writes a suppression. Two mechanisms:
| `mechanism` | Effect |
| -------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `consent_revoke` (default) | Writes a `ConsentRecord` revocation for each purpose in `revokePurposes` (default `["marketing"]`). Soft, reversible; enforced by the recommend **consent stage** |
| `do_not_contact` | Appends the customer to the tenant's canonical `do_not_contact` ContactPolicy — a hard, global block across every channel |
Suppression is best-effort / fail-open: a suppression-write failure warns but
never fails the run or blocks the outcome from being recorded. Defaults:
`enabled: true`, `outcomeKeys: ["unsubscribe", "spam_complaint"]`,
`mechanism: "consent_revoke"`, `revokePurposes: ["marketing"]`.
### Worked example — email vendor bounce/click file
A daily file lands from an email provider with columns
`recipient_id, kaireon_rid, event, occurred_at, event_id`. A
[`file_arrival` trigger](/data/pipelines/flow-schedule) fires the pipeline when
the file appears; the `source` parses it and the `outcomes` node ingests it:
```json theme={null}
{
"kind": "pipeline",
"version": "1.0",
"id": "email-outcomes",
"metadata": { "name": "Email response ingestion", "owner": "growth" },
"trigger": {
"kind": "file_arrival",
"sourceId": "src",
"controlFilePattern": "events_*.csv",
"debounceSeconds": 60
},
"nodes": [
{
"id": "src",
"kind": "source",
"connector": "s3",
"config": {
"path": "esp/outcomes/",
"pattern": { "type": "glob", "value": "events_*.csv" },
"ordering": "fifo_by_name",
"format": "csv"
}
},
{
"id": "ingest",
"kind": "outcomes",
"input": "src",
"columnMapping": {
"customerId": "recipient_id",
"ridColumn": "kaireon_rid",
"outcomeKey": "event",
"timestamp": "occurred_at",
"eventId": "event_id"
},
"attributionWindowDays": 7,
"suppressionRules": {
"enabled": true,
"outcomeKeys": ["unsubscribe", "spam_complaint"],
"mechanism": "consent_revoke",
"revokePurposes": ["marketing"]
}
}
],
"errorHandling": { "dlq": { "enabled": false }, "retry": { "maxAttempts": 0, "backoff": "fixed" } }
}
```
Running it records one inbound outcome per row: a `click` within 7 days of the
send is attributed to that campaign and rewards the model; an `unsubscribe`
revokes the customer's `marketing` consent; a duplicate `event_id` on tomorrow's
overlapping file is deduped rather than double-counted.
## Tenant gate
Pipeline IR is **on by default** — `tenant_settings.flowIrEnabled`
defaults to `true`. New pipelines created via `POST /api/v1/pipelines`
with `irVersion: "1.0"` and an `ir` body field are stored in the
`pipeline_ir_versions` table.
The flag remains in the schema as an explicit **kill-switch**: setting
`flowIrEnabled = false` makes pipeline create/run/AI-author endpoints
return `403 flow_ir_disabled`. The legacy ETL editor and runtime were
removed in the 2026-04-28 cleanup, so disabling the flag effectively
disables pipelines entirely.
## API
`POST /api/v1/pipelines` — send `{ name, connectorId, schemaId, irVersion: "1.0", ir: { ... } }`
to create an IR-native pipeline. See [Pipelines API](/api-reference/pipelines)
for the full shape.
`POST /api/v1/pipelines/:id/run` — when the pipeline has `irVersion`
set, the request is handed to the in-process batch interpreter and
returns the per-node result synchronously. Legacy pipelines continue
going to the BullMQ worker queue.
## Validation layers
The IR is enforced at three layers:
1. **Authoring time** — Zod schemas in the UI / AI prompt
2. **Save time** — server re-validates on POST and stores as `pipeline_ir_versions` row
3. **Runtime** — interpreter re-checks before executing
This makes "AI generated invalid IR" structurally impossible.