Aqueduct

The AI harness for data pipelines: the model proposes, the harness decides.
One blueprint. Spark or DuckDB. Your data never leaves your servers.

Why Aqueduct

A pipeline fails at 3 a.m. over a column rename upstream. Somebody gets paged, scrolls a four-kilobyte stack trace, and hand-writes a one-line fix. Aqueduct turns that night into a pull request waiting for review in the morning.

Wake up to a reviewable fix instead of a wall of errors.

• Declarative, not DAG code. Pipelines are YAML Blueprints. No PySpark boilerplate, no operator classes. Bring your own scheduler; Aqueduct is the control plane on top of the engine.
• Self-healing rather than alerting. On failure, an LLM agent diagnoses the root cause and emits a structured patch. The patch must clear guardrail, lineage, and sandbox gates before it touches your pipeline. The agent cannot generate code or touch a shell. A failure it has solved before heals from memory, with zero LLM tokens.
• Observable by construction. Every run, heal attempt, and column-lineage edge lands in a queryable store. Nothing is added to the hot path to make that happen.
• Model-agnostic and local-first. Anthropic natively, or any OpenAI-compatible endpoint: OpenRouter, DeepSeek, Groq, a local 7B on Ollama or LM Studio. Set provider and base_url, done. Point it at a model inside your perimeter and your data, schemas, and error traces never leave your servers. Multi-model cascades escalate to a bigger model only when the small one gets stuck.
• The harness owns correctness; the model only proposes. The LLM does exactly one thing deterministic code cannot: turn an unstructured failure into a structured hypothesis. Everything after that is deterministic. A constrained patch grammar with 15 operations and no code generation. Validation gates. Budget caps. A signature cache. That division of labor is why a small local model holds up here when raw code-generating assistants don't.

From failure to pull request

$ aqueduct run blueprints/orders.yml
  ✗ orders.load  PATH_NOT_FOUND: data/in.csv
  agent: root cause identified (confidence 0.92), patch staged
  → patches/pending/orders-20260824T031412.json

$ aqueduct patch pr orders-20260824T031412 --blueprint blueprints/orders.yml
  ✓ gates passed  guardrails · compile · lineage · sandbox
  ✓ PR opened  branch=aqueduct/heal/orders-20260824T031412
  https://github.com/you/your-repo/pull/128

The patch is a Git-diffable JSON document, never code:

// patches/pending/orders-20260824T031412.json (abridged)
{
  "category": "config_error",
  "root_cause": "Ingress 'load' reads data/in.csv, but the upstream job renamed it to data/input.csv.",
  "confidence": 0.92,
  "rationale": "PATH_NOT_FOUND on data/in.csv; a sibling data/input.csv exists with a matching schema, so the path is stale rather than the data missing.",
  "operations": [
    { "op": "set_module_config_key", "module_id": "load", "key": "path", "value": "data/input.csv" }
  ]
}

The trust boundaries behind all of this, and what each gate actually enforces, are written down in the Threat Model.

Table of contents

• Supported engines
• The healing flow
• Architecture
• Core concepts
• Getting started
• What you get
• References
• Contributing

Supported engines

Apache Spark Distributed batch at cluster scale. Delta Lake, JDBC, custom data sources, the full Blueprint grammar.

DuckDB Single-node production, zero JVM. Ships with the base install. Same blueprint, same CLI, same healing loop.

aqueduct run pipeline.yml --set deployment.engine=duckdb

One blueprint format, one CLI, per-engine execution. Each engine declares a capability table, and the compiler checks your blueprint against it. If the target engine can't do something, you get a compile error naming the exact capability instead of a runtime surprise.

Three promises, each enforced by machinery rather than review discipline:

• Every supported entry in the capability matrix is backed by a named test that runs on that engine in CI.
• Nothing that can change your results is ever silently ignored. The engine honors it, warns about it, or refuses to compile.
• Healed blueprints record which engine produced each patch. Deploy a DuckDB-healed blueprint to Spark and the compiler tells you.

The two engines do not promise identical values, and Aqueduct does not pretend they do. SQL dialects genuinely differ; the known differences are cataloged in the Compatibility Matrix. Your Assert rules run on both engines, so the properties you care about are the properties that get checked. That is your portability contract.

Polyglot pipelines

Engines can also mix inside one blueprint. Set engine: on a module and everything downstream inherits it until another module overrides:

modules:
  - id: extract        # heavy join across two warehouses
    type: Channel
    engine: spark
    config: { op: sql, query: "..." }

  - id: aggregate      # small result, no cluster needed
    type: Channel
    engine: duckdb
    config: { op: sql, query: "SELECT region, sum(amount) FROM extract GROUP BY region" }

The compiler partitions the DAG into engine islands and inserts a handoff at each boundary. The handoff materializes data as parquet at a configurable location, visible in observability like any other module, with bytes and duration recorded. Each handoff point is announced at compile time, because the extra I/O is a real cost you should see before the run. If an island fails, a rerun picks up the already-materialized upstream data instead of recomputing it.

The healing flow

When a pipeline fails, Aqueduct does not throw a stack trace at an LLM and hope. Healing is a staged, auditable pipeline. The model works inside a constrained grammar: it cannot write code, edit files, or run shell commands.

A generated patch clears five gates plus a compile-check before it ever touches the Blueprint. Gate 1 is guardrails: deterministic policy checks on paths, operations, and confidence. The compile-check follows immediately, and the patched Blueprint must still parse. Then Gate 2, lineage: does the patch break a downstream column consumer. Gate 3, sandbox: replay against representative data. Gate 4, plan regression. Gate 5, resolvability: any dependency the patch declares must actually be installable. They run in that order and the first failure wins:

✓ guardrails  →  ✓ compile-check  →  ✓ lineage  →  ✓ sandbox  →  ✓ plan-regression  →  ✓ resolvability  →  patch applied

aqueduct patch preview --sandbox runs the same pyramid on demand, before you decide to apply.

Approval modes

Who applies a generated patch. Deterministic guardrails (allowed paths, forbidden operations, minimum confidence) bound every patch regardless of mode.

Low-confidence patches and any guardrail violation auto-escalate to human review.

Why it holds up

• Every change is visible. A patch is a structured diff with a rationale and a confidence score. Low confidence escalates to a human.
• Live data stays safe. The sandbox validates each patch against representative data before any live write.
• Loops are bounded. A multi-axis budget caps wall-clock time, tokens, reprompts, and stuck-signature windows. A rolling rate limit caps heals per hour per blueprint.
• Injection is assumed, not ignored. Error text and data samples reach the model framed as untrusted data, and everything the model produces must still clear the deterministic gates. The full analysis is in the Threat Model.
• Decisions are auditable. Every LLM turn is recorded with the gate that rejected it, a stable error signature, and the prompt version. One run id joins every iteration of a heal.
• Reach beyond the blueprint, with the same discipline. Healing extends past pipeline definitions into engine and session config (allowlisted keys only) and dependency declarations, each behind its own gate. Data mutation stays off, period: the patch grammar contains no operation that can touch your data, and a test enforces that the grammar stays that way.

For the stage-by-stage detail, see the Blueprint & Engine Spec.

Architecture

Where Aqueduct sits in a data platform: the control plane between your scheduler and the engine.

Inside the box, Aqueduct is a single CLI that runs on the driver, with no servers and no daemons. Logic flows through four immutable layers:

• Parser validates YAML into an immutable AST.
• Compiler resolves context, expands Arcades and macros, extracts column lineage, gates the blueprint against the target engine's capabilities, and assembles a fully-resolved Manifest.
• Executor runs the Manifest on the target engine. Engines register through an entry-point protocol; Spark code is isolated under executor/spark/, DuckDB under executor/duckdb_/. The core never imports an engine by name.
• Surveyor records runs, failures, and lineage to pluggable stores and triggers the Agent on failure.

Core concepts

Full details in the References.

Getting started

Installation

pip install aqueduct-core              # DuckDB engine included, no JVM needed
pip install "aqueduct-core[spark]"     # adds Apache Spark + Delta Lake

Requires Python 3.11+ and, for the spark extra only, Java 17. Every release is CI-tested from Spark 3.5 through 4.1; the exact tested combos live in the Compatibility Matrix.

Extras exist for Airflow (airflow), cloud secret managers (secrets), networked stores (stores), and everything at once (all); compose them as needed, for example pip install "aqueduct-core[spark,airflow,aws]". The full list with rationale is in pyproject.toml and the Production Guide.

A first blueprint

aqueduct: "1.0"
id: hello.pipeline
name: Hello Pipeline

macros:
  active: "status = 'active' AND deleted_at IS NULL"

modules:
  - id: load
    type: Ingress
    label: Load orders
    config: { format: csv, path: "data/in.csv", options: { header: true } }

  - id: clean
    type: Channel
    label: Filter active
    config:
      op: sql
      query: "SELECT order_id, amount FROM load WHERE {{ macros.active }}"

  - id: save
    type: Egress
    label: Write parquet
    config: { format: parquet, path: "data/out/", mode: overwrite }

edges:
  - { from: load,  to: clean }
  - { from: clean, to: save }

agent:
  approval: human

Run it on either engine:

aqueduct run blueprints/hello.yml                                  # engine from aqueduct.yml
aqueduct run blueprints/hello.yml --set deployment.engine=duckdb   # no cluster, no JVM

Engine-wide defaults live in a separate aqueduct.yml (target engine, LLM provider, store backends, danger settings). Inline module tests live in *.aqtest.yml. Repeatable healing benchmarks live in *.aqscenario.yml. The Gallery has runnable examples of each.

Five commands to know

1. aqueduct doctor blueprints/hello.yml is the preflight check. It validates YAML, resolves paths, verifies LLM reachability, and opens stores.
2. aqueduct run blueprints/hello.yml executes the pipeline. On failure, the agent generates a patch under patches/pending/.
3. aqueduct patch pr <id> --blueprint blueprints/hello.yml turns a staged patch into a pull request on a heal branch (or review locally with patch apply).
4. aqueduct test blueprints/hello.aqtest.yml runs Channel / Junction / Funnel modules against inline data, without touching Ingress, Egress, or any external I/O.
5. aqueduct benchmark gallery/aqscenarios/ --model <a> --model <b> compares LLM models against simulated failures, so you can pick the cheapest model that heals your pipelines. No engine required.

Full reference in CLI Reference.

What you get

References

• Blueprint & Engine Spec: module types, configs, architecture, type system, healing loop
• SKILL.md: distilled Blueprint-authoring guide for LLMs (grammar, patterns, provider base_urls)
• CLI Reference: all commands and flags
• Threat Model: trust boundaries, injection surface, mitigations
• Spark Engine Guide: warnings, performance, tuning
• Observability Guide: schemas + diagnostic query cookbook
• Production Guide: cluster deployment, security, Delta operations
• Compatibility Matrix: supported Python × Spark versions, per-engine capability tables
• Extending Aqueduct: how to add an execution engine (ExecutorProtocol, capability declarations, entry points)
• Gallery: real working examples

Contributing

Contributions are welcome! See CONTRIBUTING.md.

Aqueduct is Apache 2.0 licensed: free and open source, with no telemetry and no lock-in.