Pipelines - KaireonAI

Flow Pipelines list view in the Data module — The Flow Pipelines page in the Data module.

Pipeline execution modes. batch and micro_batch are fully implemented. streaming is a planned placeholder — it is disabled in the UI and will not spawn a long-lived consumer at the API layer. See Data Platform → Execution Config for details.

Coming-soon connectors. Pipelines whose source connector is amazon_kinesis or braze will no-op at execution time. The executor logs a message and returns zero rows until ingestion support is wired. See Connectors for the full status table.

GET /api/v1/pipelines

List all pipelines with their connector and schema references.

Query Parameters

Parameter	Type	Default	Description
`limit`	integer	`50`	Max results per page (max 100)
`cursor`	string	—	Cursor for pagination

Response

{
  "data": [
    {
      "id": "pipe_001",
      "tenantId": "tenant_001",
      "name": "Customer Import",
      "description": "Daily customer data sync from Snowflake",
      "status": "draft",
      "schedule": "0 2 * * *",
      "connectorId": "conn_001",
      "schemaId": "schema_001",
      "executionConfig": { "batchSize": 5000, "parallelism": 4 },
      "irVersion": "1.0",
      "publishedIrVersion": null,
      "connector": { "id": "conn_001", "name": "Production Snowflake", "type": "snowflake" },
      "schema": { "id": "schema_001", "name": "customers", "displayName": "Customers" },
      "lastRunAt": null,
      "lastRunStatus": null,
      "createdAt": "2026-01-15T09:00:00.000Z",
      "updatedAt": "2026-01-15T09:00:00.000Z"
    }
  ],
  "pagination": {
    "total": 3,
    "hasMore": false,
    "limit": 50,
    "cursor": null
  }
}

POST /api/v1/pipelines

Create a new IR-native pipeline. Every pipeline is IR-native — the request must include irVersion: "1.0" and an ir document. The legacy nodes/edges creation format was removed on 2026-04-28; a request without irVersion: "1.0" returns 422 with a message pointing to the IR schema.

IR-native creation requires the tenant flag flowIrEnabled. If it is not enabled the endpoint returns 403 with { "error": "flow_ir_disabled" }. Enable it via PUT /api/v1/tenant-settings { "flowIrEnabled": true }.

Request Body

Field	Type	Required	Description
`irVersion`	string	Yes	Must be the literal `"1.0"`.
`ir`	object	Yes	The full Pipeline IR document (`{ version, id, metadata, nodes, ... }`). Validated with Zod + structural checks. See Pipeline IR.
`name`	string	Yes	Pipeline name.
`connectorId`	string	Yes	Source connector ID — must exist in the tenant.
`schemaId`	string	Yes	Target schema ID — must exist in the tenant.
`description`	string	No	Description.
`schedule`	string \| null	No	Legacy cron column. Defaults to `null`; the IR’s own `schedule` is the source of truth.
`executionConfig`	object	No	Execution settings (batchSize, parallelism, partitioning).
`authoredBy`	string \| null	No	Operator id recorded on IR version 1.
`comment`	string	No	Optional comment recorded on IR version 1.

The validated IR is stored as version 1 in the pipeline_ir_versions table. Subsequent runs via POST /pipelines/{id}/run are dispatched to the in-process batch interpreter.

Example

curl -X POST https://playground.kaireonai.com/api/v1/pipelines \
  -H "Content-Type: application/json" \
  -H "X-Tenant-Id: my-tenant" \
  -d '{
    "name": "Customer Import",
    "connectorId": "conn_001",
    "schemaId": "schema_001",
    "schedule": "0 2 * * *",
    "executionConfig": { "batchSize": 5000, "parallelism": 4 },
    "irVersion": "1.0",
    "ir": {
      "version": "1.0",
      "id": "customer-import",
      "metadata": { "name": "Customer Import" },
      "nodes": []
    }
  }'

Response: 201 Created — the created pipeline object plus the stored ir.

Error Responses

Status	Cause
`400`	Invalid JSON, or the `ir` failed Zod/structural validation (response body includes an `errors` array).
`403`	`flowIrEnabled` is not set for the tenant (`{ "error": "flow_ir_disabled" }`).
`409`	A pipeline with the same `name` already exists.
`422`	Request omits `irVersion: "1.0"`; or `name`/`connectorId`/`schemaId` missing; or `connectorId`/`schemaId` not found in the tenant.
`429`	Rate limit exceeded (200 requests / 60s).

PUT /api/v1/pipelines

Update a pipeline’s metadata. IR (node/edge) mutations do not go through this endpoint — save a new IR version with POST /api/v1/pipelines/{id}/ir instead.

Request Body

Field	Type	Required	Description
`id`	string	Yes	Pipeline ID or name
`name`	string	No	Updated name
`description`	string	No	Updated description
`schedule`	string \| null	No	Updated legacy cron column
`executionConfig`	object	No	Updated execution config
`status`	string	No	Updated status (`draft`, `active`, `paused`, `error`)

Every body field other than id is applied directly to the Pipeline row’s scalar columns. Sending removed fields such as nodes or edges fails — those tables no longer exist.

Response: 200 OK — the updated pipeline with its connector and schema references.

Error Responses

Status	Cause
`400`	Missing `id` or invalid JSON
`404`	Pipeline not found for the caller’s tenant (includes ids owned by another tenant)

The update is scoped by the (tenantId, id) compound key, so a PUT targeting a pipeline owned by another tenant returns 404 — never 200 or 500.

DELETE /api/v1/pipelines

Delete a pipeline and its nodes/edges.

Parameter	Type	Required	Description
`id`	string	Yes	Pipeline ID (query parameter)

Response: 204 No Content

Error Responses

Status	Cause
`400`	Missing `id`
`404`	Pipeline not found for the caller’s tenant (includes ids owned by another tenant)

POST /api/v1/pipelines//run

Run a pipeline synchronously, in-process. The route creates a pipeline_run row (status running), loads the published-or-latest IR version, executes it via the batch interpreter, then finalizes the run row to completed or failed. It does not enqueue to an external worker queue, and the response is returned only after the run finishes.

Response (200 — success)

{
  "runId": "8f2c1a4e-…",
  "irVersion": 3,
  "ok": true,
  "nodeResults": []
}

On failure the same shape is returned with ok: false, an error string, and failedNodeId, but with HTTP status 500.

Status Codes

Status	When
`200`	Run completed successfully (`ok: true`)
`404`	Pipeline not found for the tenant
`409`	Pipeline is not IR-native (row has no `irVersion`)
`500`	Run executed but failed (`ok: false`), or the pipeline is IR-native but has no stored IR versions

GET /api/v1/pipelines//runs

List recent execution runs for a pipeline (last 20 runs, newest first).

Response

[
  {
    "id": "run_001",
    "pipelineId": "pipe_001",
    "status": "completed",
    "createdAt": "2026-03-16T02:00:00.000Z",
    "completedAt": "2026-03-16T02:05:30.000Z"
  }
]

GET /api/v1/pipelines//ir/versions

List the IR version history for a pipeline, descending by version number. Each save through POST /api/v1/pipelines/{id}/ir writes a new row to the IR version log with the next version number — this endpoint reads them back for diff/rollback UIs. The endpoint runs a tenant-scoped existence check BEFORE the version lookup so a caller from a different tenant cannot probe pipeline ids by reading an empty [] (200) — they get a 404 instead.

Path Parameters

Parameter	Type	Description
`id`	string	`Pipeline.id` — must belong to the tenant.

Response

Returned at versions/route.ts:33-41. The full IR is intentionally NOT included — only the version metadata. Fetch a specific IR via GET /api/v1/pipelines/{id}/ir?version=N.

[
  {
    "version": 7,
    "authoredBy": "alice@example.com",
    "comment": "Add filter for high-value customers",
    "createdAt": "2026-04-30T14:00:00.000Z"
  },
  {
    "version": 6,
    "authoredBy": "bob@example.com",
    "comment": null,
    "createdAt": "2026-04-29T10:30:00.000Z"
  }
]

number

Monotonically increasing per pipeline. The first save produces version: 1.

string | null

Operator id supplied to savePipelineIr at the time of the save (pipeline-ir-repo.ts:18).

string | null

Optional human-readable comment attached at save time.

string

ISO timestamp of when the version row was inserted.

Status codes

Code	When	Source
200	Returns the version array (possibly empty for pipelines with no IR yet)	`versions/route.ts:33`
401	Caller is not authenticated	`requireRole`
403	Caller is not `viewer`, `editor`, or `admin`	`versions/route.ts:20`
404	Pipeline not found for tenant	`versions/route.ts:30`

Roles

admin, editor, viewer.

GET /api/v1/pipelines/

Fetch a single pipeline’s metadata (used by the flow editor topbar). Tenant-scoped; a cross-tenant id reads as 404.

Response `200`

{
  "id": "pipe_001",
  "name": "Customer Import",
  "description": "Daily customer data sync from Snowflake",
  "status": "active",
  "irVersion": "1.0",
  "lastRunAt": "2026-03-16T02:00:00.000Z",
  "lastRunStatus": "completed",
  "connectorId": "conn_001",
  "schemaId": "schema_001",
  "createdAt": "2026-01-15T09:00:00.000Z",
  "updatedAt": "2026-03-16T02:05:00.000Z"
}

Returns 404 when the pipeline does not exist for the tenant. Roles: admin, editor, viewer.

GET /api/v1/pipelines//ir

Return the latest IR document for a pipeline.

Response `200`

{
  "version": 3,
  "ir": { "version": "1.0", "id": "customer-import", "metadata": { "name": "Customer Import" }, "nodes": [] },
  "createdAt": "2026-04-30T14:00:00.000Z"
}

Status Codes

Status	When
`200`	Returns the latest IR version
`404`	Pipeline not found for the tenant
`409`	Pipeline uses the legacy node/edge format (not IR-native)
`500`	Pipeline is marked IR-native but has no stored IR versions

Roles: admin, editor, viewer.

POST /api/v1/pipelines//ir

Save a new IR version for a pipeline. The IR is validated (Zod + structural) before persistence, and a legacy pipeline is promoted to IR-native on first save.

Request Body

Field	Type	Required	Description
`ir`	object	Yes	The full Pipeline IR document.
`comment`	string	No	Human-readable comment recorded on the version.
`authoredBy`	string	No	Operator id recorded on the version.

Response `201`

{ "version": 4, "createdAt": "2026-05-01T09:00:00.000Z" }

Status Codes

Status	When
`201`	New IR version saved
`400`	`ir` missing, or failed Zod/structural validation (body includes `errors`)
`404`	Pipeline not found for the tenant

Roles: admin, editor.

POST /api/v1/pipelines//publish

Pin publishedIrVersion to a specific IR version. The scheduler and the run route both prefer the pinned version, so authors can save draft IR versions that don’t go live until they publish.

Request Body

Field	Type	Required	Description
`version`	number	No	Version to publish. Defaults to the latest saved version.

Response `200`

{ "publishedIrVersion": 4, "latestVersion": 4, "isLatest": true }

Status Codes

Status	When
`200`	Version pinned
`400`	Pipeline has no IR versions to publish, or `version` is invalid (`< 1`)
`404`	Pipeline not found, or the requested version does not exist

Roles: admin, editor.

POST /api/v1/pipelines//ir/versions//restore

Copy a prior IR version forward as a new version (the audit trail stays intact — the source version is not mutated). The new version’s comment records the source.

Path Parameters

Parameter	Type	Description
`id`	string	Pipeline ID
`version`	integer	Source version to restore from

Response `201`

{ "sourceVersion": 2, "newVersion": 5, "ir": { "version": "1.0", "id": "customer-import", "metadata": { "name": "Customer Import" }, "nodes": [] } }

Status Codes

Status	When
`201`	Source version re-appended as a new version
`400`	`version` is not a finite integer `>= 1`
`404`	Pipeline not found, or the source version does not exist
`422`	The stored IR for that version no longer parses

Roles: admin, editor.

GET /api/v1/pipelines//sql-preview

For each target node in the pipeline’s latest IR, compute the exact SQL the runtime would execute (INSERT ... SELECT, optionally with TRUNCATE or ON CONFLICT). Reuses the runtime’s own SQL builders so the preview matches behavior 1:1.

Response `200`

{
  "pipelineId": "pipe_001",
  "version": 3,
  "sourceTable": "_flow_src_preview",
  "targets": [
    {
      "nodeId": "load_customers",
      "loadMode": "append",
      "targetSchema": "public",
      "targetTable": "customers",
      "sql": ["INSERT INTO \"public\".\"customers\" (...) SELECT ... FROM \"_flow_src_preview\""],
      "warnings": []
    }
  ]
}

blue_green, incremental_watermark, and cdc_mirror targets return sql: null with an unsupported note — those modes render only at runtime via their dedicated load-mode helpers, so no static preview is fabricated.

Returns 404 when the pipeline is not found or has no IR yet. Roles: admin, editor, viewer.

Roles

Endpoint	Allowed Roles
`GET /pipelines`	admin, editor, viewer
`POST /pipelines`	admin, editor
`PUT /pipelines`	admin, editor
`DELETE /pipelines`	admin, editor
`GET /pipelines/{id}`	admin, editor, viewer
`POST /pipelines/{id}/run`	admin, editor
`GET /pipelines/{id}/runs`	any authenticated
`GET /pipelines/{id}/ir`	admin, editor, viewer
`POST /pipelines/{id}/ir`	admin, editor
`GET /pipelines/{id}/ir/versions`	admin, editor, viewer
`POST /pipelines/{id}/ir/versions/{version}/restore`	admin, editor
`POST /pipelines/{id}/publish`	admin, editor
`GET /pipelines/{id}/sql-preview`	admin, editor, viewer

​GET /api/v1/pipelines

​Query Parameters

​Response

​POST /api/v1/pipelines

​Request Body

​Example

​Error Responses

​PUT /api/v1/pipelines

​Request Body

​Error Responses

​DELETE /api/v1/pipelines

​Error Responses

​POST /api/v1/pipelines//run

​Response (200 — success)

​Status Codes

​GET /api/v1/pipelines//runs

​Response

​GET /api/v1/pipelines//ir/versions

​Path Parameters

​Response

​Status codes

​Roles

​GET /api/v1/pipelines/

​Response 200

​GET /api/v1/pipelines//ir

​Response 200

​Status Codes

​POST /api/v1/pipelines//ir

​Request Body

​Response 201

​Status Codes

​POST /api/v1/pipelines//publish

​Request Body

​Response 200

​Status Codes

​POST /api/v1/pipelines//ir/versions//restore

​Path Parameters

​Response 201

​Status Codes

​GET /api/v1/pipelines//sql-preview

​Response 200

​Roles

GET /api/v1/pipelines

Query Parameters

Response

POST /api/v1/pipelines

Request Body

Example

Error Responses

PUT /api/v1/pipelines

Request Body

Error Responses

DELETE /api/v1/pipelines

Error Responses

POST /api/v1/pipelines//run

Response (200 — success)

Status Codes

GET /api/v1/pipelines//runs

Response

GET /api/v1/pipelines//ir/versions

Path Parameters

Response

Status codes

Roles

GET /api/v1/pipelines/

Response `200`

GET /api/v1/pipelines//ir

Response `200`

Status Codes

POST /api/v1/pipelines//ir

Request Body

Response `201`

Status Codes

POST /api/v1/pipelines//publish

Request Body

Response `200`

Status Codes

POST /api/v1/pipelines//ir/versions//restore

Path Parameters

Response `201`

Status Codes

GET /api/v1/pipelines//sql-preview

Response `200`

Roles