> ## 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.

# Flow Streaming Runtime

> Streaming scaffold gated behind FLOW_STREAMING_ENABLED — the flag plus a broker-agnostic checkpoint helper. Broker source consumers (Kafka / Kinesis / Pulsar) are not yet wired. Self-hosted opt-in, off by default. (cdc_mirror no longer sits behind this gate — it is a batch op-column change apply.)

<Warning>
  Streaming is **gated behind the `FLOW_STREAMING_ENABLED` env flag**, which
  is **off by default** — the playground deployment runs with it off (we do
  not provision a Kafka/Kinesis/Pulsar broker). Self-hosters opt in by setting
  the flag. What ships today is a **scaffold**: the flag itself and a
  broker-agnostic checkpoint helper. The actual broker source consumers are
  not built yet — see the table below for exactly what is and isn't wired.
  (`cdc_mirror` used to sit behind this gate; since 2026-07-27 it is a live
  batch op-column change apply — see
  [Loading Modes & Validation](/data/pipelines/loading-modes-validation).)
</Warning>

## What `FLOW_STREAMING_ENABLED` gates today

Nothing runnable is wired to the flag right now. The broker **source**
consumers are **not yet part of the pipeline IR** — the source-node
connector enum is file/object-store only (`s3`, `gcs`, `azure_blob`,
`sftp`, `ftp`, `local_fs`, `http_pull` — see
`lib/flow/ir/nodes/source.ts`), so a Kafka/Kinesis/Pulsar consumer node
cannot be authored or dispatched today. There is no streaming source
executor; the only streaming code that ships is the env flag and the
checkpoint helper below.

| Capability                                                | When env unset (default + playground)                                       | When env = `"true"` (self-hosted)                                                                   |
| --------------------------------------------------------- | --------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------- |
| Kafka / Kinesis / Pulsar source consumers                 | Not authorable — not in the source-node IR enum yet, and no executor exists | Still scaffold-only; self-hosters build the consumer (there is no shipped implementation to enable) |
| Checkpoint helper (`writeCheckpoint` / `readCheckpoints`) | Present but unused — no consumer calls it yet                               | Consumers persist offsets to `pipeline_runs.metadata.streamingCheckpoints[]`                        |

## Checkpoint persistence (lands the moment any broker connects)

The Phase 6.4 scaffold ships the helper at `lib/flow/runtime/streaming/checkpoint.ts`:

```ts theme={null}
import { writeCheckpoint, readCheckpoints } from "@/lib/flow/runtime/streaming/checkpoint";

// After every successful batch
await writeCheckpoint({
  prisma, runId, tenantId,
  checkpoint: { pipelineId, partition: 0, offset: 1234 },
});

// On consumer restart
const checkpoints = await readCheckpoints({ prisma, runId, tenantId });
```

Stored in `pipeline_runs.metadata.streamingCheckpoints[]`. At-least-once
delivery semantics — write checkpoint **before** the next batch's side
effects. Any consumer (Kafka / Kinesis / Pulsar) plugs into this same
helper so the persistence layer is broker-agnostic.

## Self-hoster setup

When you're ready to enable streaming on a self-hosted deployment:

1. Set `FLOW_STREAMING_ENABLED=true` in the API + worker env. On its own
   this does not add any broker source consumer — it only flips the gate
   future consumers will check.
2. Add a streaming source consumer — because the consumer node kind is not
   in the source-node IR enum yet, this means extending
   `lib/flow/ir/nodes/source.ts` and adding an executor, then installing
   `kafkajs` (or `@aws-sdk/client-kinesis` / `pulsar-client`) and
   implementing the broker-specific consumer in `kaireon-worker`.
3. Restart the worker process — the existing `kaireon-worker` ECR repo
   is unwired in playground (no App Runner service); self-hosters point
   their orchestrator at it.
4. Each consumer run reads `streamingCheckpoints` on startup, seeks the
   broker to the saved offset, processes, and `writeCheckpoint`s after
   each successful batch.

## Honest limits carried into Phase 6.6 hardening

* **Per-partition lag metrics** on the per-node metrics dashboard
* **Backpressure semantics** beyond at-least-once
* **UI hide for streaming node kinds** when `FLOW_STREAMING_ENABLED=false`
  — the Visual canvas + Schedule tab don't yet hide them with a
  "self-host required" CTA

## Spec reference

`specs/2026-04-24-flow-design.md` §6 Runtime Engine, streaming
subsection. Phase 6.4 SPEC at
`.planning/phases/06.4-streaming-runtime-cdc-mirror/06.4-SPEC.md`.