AI Pipeline Authoring

​

How the AI stays honest

1. Constrained JSON generation — the provider’s output is bound by a Zod schema via Vercel AI SDK generateObject. Invalid shapes are rejected at the wire.
2. Patch application — the AI emits an RFC 6902 JSON Patch, not a full IR. Our applier rejects paths that do not exist in the current IR.
3. Validate-then-regenerate — the patched IR is fed through the Phase 1 parsePipelineIR validator. If structural checks fail, the server re-prompts the AI with the exact error and retries (up to 3 times). You only ever see IR that the runtime can execute.

​

Prerequisites

• Tenant setting flowIrEnabled must be on (defaults to true; the flag exists as a kill-switch only).
• AI provider configured in platform settings. Any provider that supports structured output works — Anthropic, OpenAI, Google, AWS Bedrock. If the provider errors or can’t return schema-valid JSON, the endpoint responds 200 with the ok: false failure shape (see below) and the provider message in errors — not a distinct HTTP error status.

​

Rate limits

• 100 proposals per tenant per hour. Each proposal may use up to 3 internal retries against your provider. Your provider’s per-request rate limits still apply.

​

Phase 1 scope

• Sources — runtime executors are live for local_fs, s3, gcs, azure_blob, sftp, and http_pull. ftp is documented as deprecated (plaintext) and unreachable from runtime. Streaming consumer kinds (kafka, kinesis, pulsar) are gated by FLOW_STREAMING_ENABLED — disabled by default in playground.
• Target load modes — all six are live: append, truncate, upsert, blue_green, incremental_watermark, and cdc_mirror (op-column change apply — requires cdcOpColumn + upsertKey; unknown op values fail the run).
• Validate node — five dataset-level checks live (rowCount, freshness, fkIntegrity, cardinality, duplicateKey).

​

API

{
  "pipelineId": "optional-uuid-or-name",
  "message": "describe the change you want"
}

{
  "ok": true,
  "rationale": "Adds a validate node that aborts when fewer than 100 rows arrive.",
  "patch": [
    {
      "op": "add",
      "path": "/nodes/-",
      "value": {
        "id": "v",
        "kind": "validate",
        "input": "src",
        "datasetLevel": { "rowCount": { "min": 100, "onFail": "abort" } }
      }
    }
  ],
  "resultingIr": { /* full pipeline IR after the patch */ },
  "warnings": [],
  "retries": 0,
  "tokensUsed": 1842
}

{
  "ok": false,
  "rationale": "...",
  "errors": ["...", "..."],
  "retries": 2,
  "tokensUsed": 4012
}

{
  "pipelineId": "p1",
  "rationale": "<the rationale that was rejected>",
  "patch": [ /* the patch that was rejected */ ],
  "reason": "wrong target table"
}

​

Related

• Pipeline IR — the underlying typed document the AI edits.
• Pipelines API — how to apply an IR via POST /api/v1/pipelines with irVersion: "1.0".