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)
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
Node kinds (Phase 1)
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
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
Optional safety fields on target nodes
Outcomes node
Theoutcomes 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
_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 outboundInteractionHistory 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 whoseoutcomeKey is classified
negative in the OutcomeType registry and listed in
suppressionRules.outcomeKeys) writes a suppression. Two mechanisms:
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 columnsrecipient_id, kaireon_rid, event, occurred_at, event_id. A
file_arrival trigger fires the pipeline when
the file appears; the source parses it and the outcomes node ingests it:
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
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:- Authoring time — Zod schemas in the UI / AI prompt
- Save time — server re-validates on POST and stores as
pipeline_ir_versionsrow - Runtime — interpreter re-checks before executing