Introduction to Conduit

What is Conduit?

Conduit is a Rust-native data pipeline orchestrator designed from the ground up to solve problems that existing tools like Airflow, Dagster, and Prefect cannot solve architecturally.

Unlike traditional orchestrators that treat pipelines as mutable entities that change over time, Conduit treats them as immutable artifacts. Every DAG is compiled once, versioned, deployed via a Terraform-style plan/apply workflow, and executed in isolated virtual environments. This fundamental shift enables capabilities that are impossible in existing systems:

• Virtual pipeline environments (instant, zero data copy)
• Time-travel debugging (replay any run from the event log)
• Sub-second compilation (tree-sitter parses Python without executing it)
• Fingerprint-based change detection (only re-execute what changed)
• Zero external dependencies (everything in a single binary)

Why Conduit Exists

The Airflow Problem

Airflow is the market leader, but it has fundamental architectural limitations:

1. Mutable state model: Airflow stores DAG definitions and variables in a mutable database. When you change a DAG, Airflow re-executes it entirely. There's no way to safely roll back or test changes.

2. Polling-based scheduling: Airflow polls the database every 5–30 seconds looking for ready tasks. In microsecond-latency systems, this is unacceptable.

3. No pipeline environments: Airflow has no concept of "staging" vs "production" pipelines. You deploy a single DAG and hope nothing breaks.

4. Runtime validation: DAGs are validated at runtime via Python execution. Syntax errors and dependency cycles slip past development.

5. Opaque lineage: SQL lineage is either manual or tied to specific data systems. Column-level lineage is not standardized.

Why Existing Alternatives Fall Short

• Dagster is more principled than Airflow but still uses mutable state and Python execution for DAG validation.
• Prefect focuses on flow composition, not structural validation or environment isolation.
• dbt is a great data transformation tool, but orchestration is a second-class citizen.
• Temporal is excellent for long-running workflows but not designed for batch data pipelines.

Key Innovations

1. Fingerprint-Based Change Detection

Every task in a DAG is assigned a content-addressable fingerprint based on:

• The task definition (code, dependencies, configuration)
• Upstream task fingerprints (cascading invalidation)
• Schedule and trigger rules

When you deploy to production, Conduit compares the new fingerprints against the current environment. Only tasks that changed are marked for re-execution. Unchanged tasks automatically reuse cached outputs.

Task A (hash: abc123)
├─ Task B (hash: def456) — input changed
│  └─ Task C (hash: ghi789) — upstream changed, cascade invalidates
└─ Task D (hash: jkl012) — no changes

This is fundamentally different from Airflow, where changing one task triggers full re-execution.

2. Virtual Environments (SQLMesh Inspired)

Creating a new environment (staging, test, feature branch) is O(1) and consumes zero data:

# Instant, no copy
conduit env create staging --from production

# Make changes to DAG definitions
vim dags/etl.py

# Plan what would change
conduit plan staging

# Apply changes
conduit apply staging -y

# Promote to production when ready
conduit env promote staging production

Environments are snapshots: pointers to immutable compiled DAGs. Promoting an environment is just moving a pointer. Rolling back is instant.

3. Event-Sourced State

Conduit stores an append-only event log instead of mutable database state:

1. DAGCompiled(dag_id, fingerprint, timestamp)
2. EnvironmentCreated(staging, fork_of=production)
3. SnapshotDeployed(staging, fingerprint, timestamp)
4. DAGScheduled(etl_pipeline, run_id, scheduled_time)
5. TaskStarted(run_id, task_id, timestamp)
6. TaskCompleted(run_id, task_id, xcom_output, timestamp)
7. DAGCompleted(run_id, status, timestamp)

From this immutable log, you can:

• Time-travel: Replay any run from the original event
• Rollback: Point production back to a previous environment snapshot
• Debug: Inspect exactly what changed and when
• Audit: Complete history of all pipeline mutations

4. Sub-Second Compilation

Conduit uses tree-sitter to parse Python DAG definitions without executing Python code. This is radically faster and safer than Airflow's approach:

# This won't execute during compilation
@dag
def my_dag():
    t1 = python_task("import os; os.system('rm -rf /')")  # Safe!
    t2 = python_task("requests.get('http://evil.com')")   # Safe!
    return t1 >> t2

Tree-sitter extracts the DAG structure, task dependencies, schedules, and retry policies without ever running untrusted code. Compilation takes milliseconds for DAGs with hundreds of tasks.

5. Process-Based Execution

Each task runs as an isolated child process with:

• Timeout enforcement (no hung tasks)
• Exit-code based success/failure detection
• Enforced per-task timeouts (declared CPU/memory limits are not yet enforced)
• Structured protocol for logging, metrics, and XCom

Tasks communicate via stdout:

LOG|INFO|Starting task
XCOM|key1|value1
PROGRESS|50
METRIC|rows_processed|1000

The executor parses these messages in real-time, no database polling.

6. Type-Safe DAG Definition

The Python SDK enforces structure at definition time:

from conduit.sdk import dag, task, Pool

@dag(
    schedule="0 9 * * *",  # 9 AM daily
    retries=2,
    timeout=3600,
    pool=Pool.name("api_requests", size=5)
)
def etl_pipeline():
    raw = extract()
    clean = transform(raw)
    load(clean)
    return clean

No magic strings, no dictionary literals, full IDE autocomplete.

Architecture Overview

Conduit is built as 9 specialized crates:

graph LR
    CLI["conduit-cli<br/>(Entry Point)"]
    COMPILER["conduit-compiler<br/>(Tree-sitter Parser)"]
    SCHEDULER["conduit-scheduler<br/>(Tokio Event Loop)"]
    EXECUTOR["conduit-executor<br/>(Task Runtime)"]
    PLANNER["conduit-planner<br/>(Fingerprint Diffing)"]
    STATE["conduit-state<br/>(Event Store + Snapshots)"]
    LINEAGE["conduit-lineage<br/>(Column-Level)"]
    API["conduit-api<br/>(REST + WebSocket)"]
    COMMON["conduit-common<br/>(Types + Errors)"]

    CLI --> COMPILER
    CLI --> PLANNER
    CLI --> SCHEDULER
    COMPILER --> COMMON
    SCHEDULER --> EXECUTOR
    SCHEDULER --> STATE
    PLANNER --> STATE
    PLANNER --> COMMON
    EXECUTOR --> STATE
    LINEAGE --> COMMON
    API --> STATE
    API --> COMMON

Each crate is responsible for one concern:

• conduit-common: Shared type definitions (DAG model, errors, events)
• conduit-compiler: Tree-sitter DAG parsing + dependency resolution
• conduit-state: Event store (RocksDB) + snapshot manager
• conduit-scheduler: Cron, trigger rules, pool management, run state machine
• conduit-executor: Process-based task execution with timeouts and retries
• conduit-planner: Fingerprint diffing and impact analysis
• conduit-lineage: Column-level lineage (Phase 4)
• conduit-api: REST and WebSocket API (Phase 2)
• conduit-python: PyO3 native bindings (compiler/planner/lineage/state) — the authoring SDK lives in sdk/python

Comparison Matrix

Next Steps

• Installation: Set up Conduit
• Quick Start: Create your first DAG in 5 minutes
• Concepts: Understand DAG definitions, environments, and the plan/apply workflow

Installation

This guide walks you through installing Conduit and setting up your first project.

System Requirements

Rust

Conduit requires Rust 1.75 or later (2021 edition).

Check your Rust version:

rustc --version

If you don't have Rust installed, install it via rustup:

curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
source $HOME/.cargo/env

Python

Python is required for writing and running task definitions. Conduit supports Python 3.9+.

Check your Python version:

python3 --version

System Libraries

Conduit uses RocksDB for the event store and tree-sitter for DAG parsing. You'll need development headers.

Ubuntu/Debian:

apt update
apt install -y build-essential clang librocksdb-dev

macOS:

brew install llvm rocksdb

Fedora/RHEL:

dnf groupinstall -y "Development Tools"
dnf install -y clang rocksdb-devel

Installing Conduit

Option 1: From Source (Recommended)

Clone the repository and build:

git clone https://github.com/jayhere1/conduit.git
cd conduit
cargo install --path conduit-cli

Verify the installation:

conduit --version

Option 2: From Crates.io (When Available)

Once Conduit reaches 1.0, it will be available on crates.io:

cargo install conduit-cli

Initializing Your First Project

Create a new Conduit project:

conduit init my-project
cd my-project

This scaffolds a project directory with the following structure:

my-project/
├── .conduit/
│   ├── state.db              # Event store (RocksDB)
│   ├── snapshots.json        # Snapshot index
│   └── environments.json      # Environment pointers
├── dags/
│   └── hello_world.py        # Example DAG
├── tasks/
│   └── common.py             # Shared task utilities
├── .gitignore
└── README.md

Project Structure

dags/

This directory contains your DAG definitions. Each file should define one or more @dag decorated functions:

# dags/etl.py
from conduit.sdk import dag, task

@task
def extract():
    print("Extracting data...")
    return "data.csv"

@task
def transform(data):
    print(f"Transforming {data}...")
    return "clean.csv"

@dag(schedule="0 9 * * *")
def etl_pipeline():
    raw = extract()
    clean = transform(raw)
    return clean

tasks/

Shared task logic and utilities. This is useful for DRY-ing up common patterns:

# tasks/common.py
from conduit.sdk import task

@task
def log_status(message):
    print(f"Status: {message}")

.conduit/

Internal state directory (version control with .gitignore):

• state.db: RocksDB event store containing all pipeline runs
• snapshots.json: Content-addressable index of compiled DAG snapshots
• environments.json: Pointers to production, staging, etc.

Configuration

Optional: Create a .conduit.toml file in your project root to customize behavior:

[project]
name = "my-project"
version = "0.1.0"

[execution]
# Default timeout for all tasks (seconds)
timeout = 3600

# Max concurrent task pool size
max_concurrency = 10

# Event store path (relative to project root)
state_dir = ".conduit"

[scheduler]
# How often to check for scheduled DAGs (seconds)
check_interval = 5

[ui]
# Web UI port
port = 8080

Verifying Installation

Compile the example DAG to verify everything works:

conduit compile

You should see output like:

Compiling DAGs in dags/...
✓ hello_world.py
  - Tasks: 2
  - Dependencies: 1
  - Schedule: None
  - Fingerprint: abc123def456

Compilation successful: 1 DAG compiled

Next Steps

• Quick Start: Build a 3-task DAG in 5 minutes
• DAG Concepts: Learn the full DAG definition syntax
• Virtual Environments: Understand production, staging, and feature branches

Quick Start: 5-Minute Tutorial

In this guide, you'll create a complete data pipeline, compile it, run it, deploy it to a virtual environment, make changes, plan the diff, and apply it.

1. Create a Project

conduit init analytics-pipeline
cd analytics-pipeline

2. Write Your First DAG

Replace the contents of dags/etl.py with a 3-task pipeline:

from conduit.sdk import dag, task, Pool
from datetime import datetime

@task(timeout=300)
def extract():
    """Download raw data from the API."""
    print("Extracting user data from API...")
    # In real usage, this would call an API or read a database
    users_count = 1000
    print(f"xcom|users_count|{users_count}")
    return "raw_users.csv"

@task(timeout=300, pool=Pool.name("transforms", size=3))
def transform(raw_data):
    """Clean and deduplicate the data."""
    print(f"Transforming {raw_data}...")
    # Deduplicate, validate schemas, etc.
    users_count = 950  # 5% duplicates removed
    print(f"xcom|users_count|{users_count}")
    return "clean_users.csv"

@task(timeout=600, retries=2)
def load(clean_data):
    """Load cleaned data to the warehouse."""
    print(f"Loading {clean_data} to warehouse...")
    # This task has 2 retries in case of transient warehouse errors
    print("xcom|rows_loaded|950")
    return "success"

@dag(
    schedule="0 2 * * *",  # Daily at 2 AM UTC
    description="User analytics ETL pipeline"
)
def daily_analytics_etl():
    """
    Extract → Transform → Load pipeline.

    Typical run time: ~2 minutes
    Latest run: 2024-03-21 02:15 UTC
    """
    raw = extract()
    clean = transform(raw)
    result = load(clean)
    return result

3. Compile the DAG

Conduit uses tree-sitter to parse and validate your DAG without executing Python:

conduit compile

Expected output:

Compiling DAGs in dags/...

✓ etl.py
  - DAG: daily_analytics_etl
  - Tasks: 3
  - Dependencies: extract → transform → load
  - Schedule: 0 2 * * * (daily at 2 AM)
  - Fingerprint: f1e2d3c4b5a6

Compilation successful: 1 DAG, 3 tasks

The fingerprint is a content-addressable hash of the entire DAG. If you change even one character, the fingerprint changes, triggering a recompilation.

4. Run the DAG Locally

Test the complete pipeline end-to-end in your development environment:

conduit run daily_analytics_etl

Watch the output:

[2024-03-22 14:32:10.123] DAG run started: run_id=abc123def456
[2024-03-22 14:32:10.456] Task extract started
Extracting user data from API...
xcom|users_count|1000
[2024-03-22 14:32:12.123] Task extract completed (1.67s)
[2024-03-22 14:32:12.456] Task transform started
Transforming raw_users.csv...
xcom|users_count|950
[2024-03-22 14:32:15.789] Task transform completed (3.33s)
[2024-03-22 14:32:16.012] Task load started
Loading clean_users.csv to warehouse...
xcom|rows_loaded|950
[2024-03-22 14:32:18.567] Task load completed (2.56s)
[2024-03-22 14:32:18.890] DAG run completed successfully

Total time: 8.77s
XCom outputs:
  - extract.users_count: 1000
  - transform.users_count: 950
  - load.rows_loaded: 950

Conduit runs tasks sequentially or in parallel based on dependencies. In this case, extract → transform → load is a linear chain, so they run sequentially.

5. Check Status

View the history of all runs:

conduit status

Output:

Environment: development

Last 10 runs:
┌─────────┬────────────────┬──────────┬────────────┐
│ Run ID  │ DAG            │ Status   │ Started    │
├─────────┼────────────────┼──────────┼────────────┤
│ abc123  │ daily_analytics_etl │ success  │ 14:32:10   │
└─────────┴────────────────┴──────────┴────────────┘

6. Deploy to Production

Create a "production" environment and deploy:

# Create a production environment (forked from development)
conduit env create production --from development

# See what would change
conduit plan production

Expected output:

Planning changes for environment: production

Added:
  ✓ daily_analytics_etl (tasks: 3)
    - extract (fingerprint: f1a2b3c4d5e6)
    - transform (fingerprint: g2h3i4j5k6l7)
    - load (fingerprint: m3n4o5p6q7r8)

Summary: 1 DAG added, 3 tasks added
Snapshot would use 1.2 KB

Now apply the changes:

conduit apply production -y

Output:

Applying plan for environment: production

✓ Deployed daily_analytics_etl (3 tasks)
  - extract: new snapshot
  - transform: new snapshot
  - load: new snapshot

Environment updated: production
Snapshot ID: prod-snap-20240322-143215

Your DAG is now deployed to production!

7. Make a Change and Deploy Again

Imagine you want to increase the retry count for the load task. Edit dags/etl.py:

@task(timeout=600, retries=3)  # Changed from 2 to 3
def load(clean_data):
    """Load cleaned data to the warehouse."""
    print(f"Loading {clean_data} to warehouse...")
    print("xcom|rows_loaded|950")
    return "success"

Recompile:

conduit compile

Now plan the changes for production:

conduit plan production

Output:

Planning changes for environment: production

Modified:
  ⚠ daily_analytics_etl
    - extract: unchanged
    - transform: unchanged
    - load: modified (retries 2 → 3)

Summary: 1 task modified
Impact analysis:
  - Blast radius: 1 task (load only)
  - Cascading changes: 0 tasks
  - Safe to apply immediately

Snapshot would reuse 2/3 tasks (extract, transform)

Only the load task fingerprint changed. The other tasks can be reused from the previous snapshot.

Apply the changes:

conduit apply production -y

Output:

Applying plan for environment: production

✓ Updated daily_analytics_etl
  - extract: reused (fingerprint: f1a2b3c4d5e6)
  - transform: reused (fingerprint: g2h3i4j5k6l7)
  - load: updated (retries 2 → 3)

Environment updated: production
Snapshot ID: prod-snap-20240322-143456

Notice that extract and transform were reused from the previous snapshot. Only load had to be recompiled. This is the power of fingerprint-based change detection.

8. Create a Feature Branch

Now create a staging environment to test new changes before production:

conduit env create staging --from production

Output:

Environment created: staging
Forked from: production
Snapshot: prod-snap-20240322-143456 (shared)

Make a bigger change in your DAG (e.g., add a validation task):

@task(timeout=300)
def validate(clean_data):
    """Validate data quality before loading."""
    print(f"Validating {clean_data}...")
    # Check for null values, schema mismatches, etc.
    print("xcom|validation_passed|true")
    return True

@dag(
    schedule="0 2 * * *",
    description="User analytics ETL pipeline"
)
def daily_analytics_etl():
    raw = extract()
    clean = transform(raw)
    validated = validate(clean)
    result = load(validated)
    return result

Compile and plan:

conduit compile
conduit plan staging

Output:

Planning changes for environment: staging

Modified:
  ⚠ daily_analytics_etl
    - extract: unchanged
    - transform: unchanged
    - validate: added
    - load: modified (input changed)

Summary: 1 task added, 1 task modified
Impact analysis:
  - Blast radius: 2 tasks (validate, load)
  - Cascading changes: 1 task (load invalidated by new validate)

Snapshot would reuse 2/3 tasks (extract, transform)

Test in staging:

conduit run daily_analytics_etl --env staging

When ready, promote staging to production:

conduit env promote staging production

Output:

Promoting staging → production

Snapshot: staging-snap-20240322-144123 → production
Previous production snapshot archived: prod-snap-20240322-143456

Environment updated: production

Production now has the new validate task. Rollback is instant if needed:

conduit env rollback production --to prod-snap-20240322-143456

Summary

You now understand the Conduit workflow:

1. Define DAGs in Python with @dag and @task decorators
2. Compile to validate syntax and extract dependencies (tree-sitter, no execution)
3. Run locally to test end-to-end
4. Plan changes against an environment (fingerprint diffing)
5. Apply to deploy (snapshot reuse optimization)
6. Promote between environments (staging → production)
7. Rollback instantly if needed

Next steps:

• DAG Concepts: Learn task types, schedules, retries, and pools
• Virtual Environments: Deep dive into fork/promote/rollback
• Plan/Apply Workflow: Understand fingerprinting and change detection

DAG Definition and Task Types

A DAG (Directed Acyclic Graph) in Conduit is a Python function decorated with @dag that orchestrates a sequence of tasks. Each task is a unit of work decorated with @task.

Basic DAG Structure

from conduit.sdk import dag, task

@task
def task_a():
    print("Running task A")
    return "a_output"

@task
def task_b(input_a):
    print(f"Running task B with input: {input_a}")
    return "b_output"

@dag
def my_first_dag():
    """
    A simple two-task pipeline.

    This DAG runs task_a, passes its output to task_b, and completes.
    """
    output_a = task_a()
    output_b = task_b(output_a)
    return output_b

When you call a task function within a @dag, you're not executing the task immediately. Instead, you're declaring a dependency in the DAG graph.

DAG Configuration

Use the @dag decorator to configure execution behavior:

@dag(
    schedule="0 9 * * *",           # Cron: daily at 9 AM
    description="Sales ETL",        # Human-readable description
    retries=1,                       # Retry entire DAG on failure
    timeout=3600,                    # Timeout in seconds (1 hour)
    pool=Pool.name("etl", size=2),  # Named pool for concurrency control
    tags=["production", "daily"],    # Metadata tags
)
def sales_etl():
    raw = extract()
    clean = transform(raw)
    load(clean)

Schedule (Cron Expressions)

Conduit uses standard 5-field cron syntax:

┌─────────── minute (0 - 59)
│ ┌─────────── hour (0 - 23)
│ │ ┌─────────── day of month (1 - 31)
│ │ │ ┌─────────── month (1 - 12)
│ │ │ │ ┌─────────── day of week (0 - 6, 0=Sunday)
│ │ │ │ │
│ │ │ │ │
* * * * *

Common patterns:

• 0 9 * * * — Daily at 9:00 AM
• 0 0 * * * — Daily at midnight
• */15 * * * * — Every 15 minutes
• 0 9 * * 1-5 — Weekdays at 9 AM
• 0 0 1 * * — First of every month at midnight
• 0 */6 * * * — Every 6 hours

Pools (Concurrency Control)

Named pools limit how many tasks can run concurrently:

from conduit.sdk import Pool

@task(pool=Pool.name("api_requests", size=5))
def call_api():
    # Max 5 tasks with this pool can run in parallel
    requests.get("https://api.example.com/data")

@task(pool=Pool.name("database", size=2))
def query_db():
    # Max 2 database queries at once
    conn.execute("SELECT * FROM users")

Pools are global across all DAGs. If you have two DAGs both using pool_database, they share the same concurrency limit.

Task Types

Conduit supports five built-in task types. The type is inferred from the task implementation.

1. Python Tasks

Execute arbitrary Python code:

@task
def process_data():
    import pandas as pd
    df = pd.read_csv("data.csv")
    df = df.dropna()
    print(f"Processed {len(df)} rows")
    return df.to_json()

@task(timeout=600, retries=2)
def api_call(url):
    import requests
    response = requests.get(url, timeout=10)
    return response.json()

Python tasks are executed via python -c "<task_code>". The task can import any package available in the Python environment.

2. Shell Tasks

Execute bash commands:

from conduit.sdk import shell_task

@shell_task
def backup_database():
    # This is executed as a bash script
    pg_dump my_database > backup.sql
    gzip backup.sql
    aws s3 cp backup.sql.gz s3://backups/$(date +%Y%m%d).sql.gz

@shell_task(timeout=1800)
def run_dbt():
    cd /projects/analytics
    dbt run --profiles-dir /etc/dbt
    dbt test

3. SQL Tasks

Execute SQL against a configured data warehouse:

from conduit.sdk import sql_task

@sql_task(dialect="postgres")
def create_staging_table():
    CREATE TABLE staging_users AS
    SELECT * FROM raw_users
    WHERE created_at >= current_date - interval '1 day'

@sql_task(dialect="postgres")
def aggregate_metrics(table_name: str):
    INSERT INTO metrics (user_id, transaction_count, total_amount)
    SELECT
        user_id,
        COUNT(*) as transaction_count,
        SUM(amount) as total_amount
    FROM {table_name}
    GROUP BY user_id

SQL tasks are templated, allowing parameter substitution. The dialect (postgres, mysql, snowflake, bigquery) determines how the task parses and validates the SQL.

4. Sensor Tasks

Block until a condition is met:

from conduit.sdk import sensor_task
import time

@sensor_task(timeout=3600, poke_interval=60)
def wait_for_file():
    # Check every 60 seconds if file exists
    if os.path.exists("/data/export.csv"):
        return True  # Unblock
    return False  # Retry

@sensor_task(timeout=7200, poke_interval=300)
def wait_for_api():
    import requests
    response = requests.get("https://api.example.com/status")
    if response.json()["status"] == "ready":
        return True
    return False

Sensors poll indefinitely until they return True or hit the timeout.

5. Generic Executables

Run any binary or script:

from conduit.sdk import executable_task

@executable_task
def run_custom_binary():
    /opt/custom-tool/process data.txt > output.txt

@executable_task(timeout=600)
def run_ruby_script():
    ruby /scripts/aggregate.rb

Task Configuration

All tasks accept common configuration parameters:

@task(
    timeout=300,                          # Timeout in seconds
    retries=2,                            # Number of retries on failure
    retry_delay=60,                       # Delay between retries (seconds)
    retry_exponential_base=2,             # Exponential backoff multiplier
    pool=Pool.name("transforms", size=3), # Concurrency pool
    tags=["data", "critical"],            # Metadata tags
)
def my_task():
    print("Task with full config")

Timeouts

If a task exceeds its timeout, it's terminated and marked as failed:

@task(timeout=10)
def long_running():
    # This will be killed after 10 seconds
    import time
    time.sleep(30)  # Timeout!

Retries

Failed tasks are retried with exponential backoff:

@task(
    retries=3,                    # Retry up to 3 times
    retry_delay=5,                # Wait 5s before first retry
    retry_exponential_base=2,     # 5s, 10s, 20s, ...
)
def flaky_api_call():
    requests.get("https://flaky-api.example.com/data")

The retry sequence is: initial attempt, then 5s, 10s, 20s delays between retries.

Trigger Rules

Control when downstream tasks start based on upstream status:

from conduit.sdk import TriggerRule

@task
def extract():
    return "data"

@task(trigger_rule=TriggerRule.ALL_SUCCESS)  # Default
def transform(data):
    # Runs only if extract succeeded
    return f"transformed: {data}"

@task(trigger_rule=TriggerRule.ALL_DONE)
def log_result():
    # Runs whether extract succeeded or failed
    print("Extract finished")

@task(trigger_rule=TriggerRule.ONE_FAILED)
def alert_ops():
    # Runs only if extract failed
    print("Alert: extract failed")

Available trigger rules:

• ALL_SUCCESS (default) — All upstream tasks succeeded
• ALL_DONE — All upstream tasks finished (success or failure)
• ONE_SUCCESS — At least one upstream task succeeded
• ONE_FAILED — At least one upstream task failed

Data Exchange (XCom)

Tasks communicate via XCom (Cross-Communication). There are two ways:

Implicit Return Values

Return values are automatically captured:

@task
def task_a():
    return {"count": 42, "status": "ok"}

@task
def task_b(data):
    # data = {"count": 42, "status": "ok"}
    print(f"Count: {data['count']}")
    return data["count"] * 2

@dag
def my_dag():
    output = task_a()
    result = task_b(output)
    return result

Explicit XCom Output

For structured logging and metrics:

@task
def extract():
    print("xcom|row_count|1000")
    print("xcom|file_size|2.5GB")
    return "data.csv"

@dag
def my_dag():
    data = extract()
    return data

The executor parses xcom|key|value messages and stores them in the event log for later retrieval.

Task Dependencies

Dependencies are declared implicitly through function calls:

@task
def a(): return "a"
@task
def b(): return "b"
@task
def c(in_a, in_b): return f"{in_a}{in_b}"
@task
def d(in_c): return f"final: {in_c}"

@dag
def complex_dag():
    output_a = a()
    output_b = b()
    output_c = c(output_a, output_b)  # Depends on both a and b
    output_d = d(output_c)              # Depends on c
    return output_d

Conduit builds the DAG by tracing these function call chains.

Complex Workflows

Branching

Tasks can have multiple downstream dependencies:

@task
def split_data(data):
    return data

@task
def process_for_warehouse(data):
    return f"warehouse_ready: {data}"

@task
def process_for_analytics(data):
    return f"analytics_ready: {data}"

@task
def merge_results(warehouse, analytics):
    return f"merged: {warehouse} + {analytics}"

@dag
def parallel_processing():
    data = split_data("input.csv")
    w = process_for_warehouse(data)
    a = process_for_analytics(data)
    merged = merge_results(w, a)
    return merged

This creates a diamond-shaped DAG: split → {warehouse, analytics} → merge

Conditional Execution

Use Python conditionals to create dynamic DAGs:

@task
def check_condition():
    import random
    return random.choice([True, False])

@task
def process_path_a():
    return "A"

@task
def process_path_b():
    return "B"

@task
def join_paths(result):
    return f"final: {result}"

@dag
def conditional_dag():
    condition = check_condition()

    # This is Python conditional logic at DAG definition time
    # (not runtime, so it's evaluated once during compilation)
    if condition:
        result = process_path_a()
    else:
        result = process_path_b()

    final = join_paths(result)
    return final

Important: Conduit evaluates conditionals at compile time, not runtime. For runtime branching, use trigger rules instead.

Best Practices

1. Keep tasks small: Each task should do one thing well.
2. Use pools: Prevent resource exhaustion by limiting concurrent tasks.
3. Set meaningful timeouts: Catch hung tasks early.
4. Retry judiciously: Don't retry non-idempotent operations.
5. Document with tags: Use tags for categorization and filtering.
6. Return structured data: Use dictionaries or JSON for multi-value returns.
7. Log progress: Use xcom|metric|value for monitoring.

Next Steps

• Virtual Environments: Deploy DAGs to isolated environments
• Plan/Apply Workflow: Understand change detection and deployment
• Python SDK: Advanced SDK features and patterns

YAML Workflow Definitions

Conduit supports defining DAGs in YAML as an alternative to Python. YAML workflows are ideal for configuration-driven pipelines — SQL queries, shell commands, sensors — where the full expressiveness of Python isn't needed.

Both formats produce the same compiled ConduitPlan and are treated identically by the scheduler, executor, and environment system. You can freely mix .py and .yaml DAGs in the same dags/ directory.

Basic Structure

id: my_pipeline
description: A daily data pipeline
schedule: "0 6 * * *"
tags: [etl, warehouse]
max_active_runs: 1

tasks:
  extract:
    type: sql
    connection: warehouse
    query: "SELECT * FROM source.orders WHERE date = '{{ ds }}'"
    retries: 3
    timeout: 10m

  transform:
    type: python
    module: transforms.orders
    function: enrich
    depends_on: [extract]

  load:
    type: shell
    command: "python scripts/load.py --date {{ ds }}"
    depends_on: [transform]

Task Types

python

Runs a Python function. Fields: module (default: "tasks"), function (default: task ID).

my_task:
  type: python
  module: my_module
  function: my_function

shell / bash

Runs a shell command. Field: command (required).

my_task:
  type: shell
  command: "echo hello world"

sql

Runs a SQL query against a named connection. Fields: connection (default: "default"), query (required).

my_task:
  type: sql
  connection: warehouse
  query: |
    INSERT INTO target.table
    SELECT * FROM staging.table

sensor

Waits for an external condition. Fields: sensor_type (default: "file"), poke_interval.

my_task:
  type: sensor
  sensor_type: file
  poke_interval: 60s

executable / exec

Runs a binary with arguments. Fields: command (required), args (optional list).

my_task:
  type: executable
  command: /usr/local/bin/processor
  args: ["--input", "data.csv", "--format", "parquet"]

Common Task Fields

Every task type supports these optional fields:

Incremental Configuration

YAML tasks support the same incremental strategies as the Python SDK:

my_task:
  type: sql
  query: "SELECT * FROM orders"
  incremental:
    strategy: append           # append, merge_on_key, delete_insert, snapshot_diff, full_refresh
    time_column: created_at
    lookback: 2h
    batch_size: 50000

Strategies

• append — Append new rows based on a time column. Fields: time_column, lookback
• merge_on_key / upsert — Merge rows by unique key. Fields: unique_key, time_column, invalidate_hard_deletes
• delete_insert — Delete and reinsert by partition. Fields: partition_column, partition_granularity (hour/day/week/month/year)
• snapshot_diff / scd — SCD Type 2 snapshots. Fields: unique_key, check_columns, scd_type_2
• full_refresh — Always recompute everything.

Template Variables

YAML workflows support Jinja-style template variables that are expanded at runtime:

• {{ ds }} — The logical execution date (YYYY-MM-DD)
• {{ ds_nodash }} — Execution date without dashes
• {{ ts }} — Full ISO timestamp

Special Files

The YAML parser automatically skips conduit.yaml and conduit.yml files, which are reserved for project configuration.

Python vs YAML: When to Use Which

Virtual Environments

Virtual environments in Conduit are isolated, independent versions of your pipeline. Unlike traditional systems that have a single mutable "production," Conduit lets you create multiple environments (production, staging, testing) that fork, diverge, and converge independently.

Why Virtual Environments?

In Airflow or Dagster, testing changes before production means:

1. Deploy to a separate machine/cluster (expensive)
2. Or modify variables/connections (risky, error-prone)

In Conduit, creating a new environment is instantaneous and zero-copy:

# Fork production → staging (instant, no data copy)
conduit env create staging --from production

Staging now has its own independent pipeline state, schedule, and run history. Changes in staging don't affect production until you promote them.

How Environments Work

Environments are snapshot pointers, not data copies:

graph TB
    Snapshot["Content-addressable Snapshot<br/>f1e2d3c4b5a6"]

    ProdPointer["production<br/>(points to snap-v1)"]
    StagingPointer["staging<br/>(points to snap-v1 + local changes)"]
    DevPointer["dev<br/>(points to snap-v1 + dev changes)"]

    Snapshot -->|referenced by| ProdPointer
    Snapshot -->|referenced by| StagingPointer
    Snapshot -->|referenced by| DevPointer

When you create an environment, you create a new pointer to the same snapshot. No data is copied. Environment operations are O(1).

Environment Lifecycle

1. Create an Environment

Fork an existing environment:

conduit env create staging --from production

Output:

Environment created: staging
Forked from: production (snapshot: prod-snap-20240322-143215)
Current snapshot: prod-snap-20240322-143215 (shared)

Staging now has an identical pipeline to production, but is completely independent. They share the snapshot, but have separate run histories and schedules.

2. Make Changes

Modify your DAG files:

# Edit your DAG definitions
vim dags/etl.py

Compile the changes:

conduit compile

3. Plan Changes

See what would change when you apply to the environment:

conduit plan staging

Output:

Planning changes for environment: staging
Current snapshot: prod-snap-20240322-143215

Added:
  - task_x (fingerprint: abc123...)

Modified:
  - task_y (timeout 300 → 600)

Unchanged:
  ✓ task_z (will be reused from snapshot)

Summary: 1 task added, 1 task modified, 1 task reused
Estimated snapshot size: 1.2 KB (35% smaller than full)

The plan shows exactly what changed and what will be reused from the previous snapshot.

4. Apply Changes

Deploy the plan to the environment:

conduit apply staging -y

Output:

Applying plan for environment: staging

✓ Updated daily_analytics_etl
  - task_x: compiled from source (new)
  - task_y: compiled from source (modified)
  - task_z: reused from snapshot (unchanged)

Snapshot ID: staging-snap-20240322-145123
Environment pointer: staging → staging-snap-20240322-145123

The environment now points to a new snapshot with your changes.

5. Promote to Production

When ready, promote the environment to production:

conduit env promote staging production

Output:

Promoting staging → production

Source: staging (staging-snap-20240322-145123)
Target: production (prod-snap-20240322-143215)
Backup: prod-snap-20240322-143215 archived as prod-snap-backup-20240322-145456

Environment updated: production
production → staging-snap-20240322-145123

Production now runs the new snapshot. The previous production snapshot is archived for rollback.

6. Rollback (Instant)

If something goes wrong, roll back instantly:

conduit env rollback production --to prod-snap-backup-20240322-145456

Output:

Rollback: production
From: staging-snap-20240322-145123
To: prod-snap-backup-20240322-145456

New snapshot: prod-snap-backup-20240322-145456

This is instantaneous. No data migration, no complex recovery. Just a pointer swap.

Environment Operations

List Environments

conduit env list

Output:

Available environments:

Name          Status      Snapshot                          DAGs  Tasks
──────────────────────────────────────────────────────────────────────
production    active      prod-snap-20240322-143215         5     32
staging       active      staging-snap-20240322-145123      5     34
dev           inactive    dev-snap-20240318-092345          5     35

View Environment Details

conduit env info staging

Output:

Environment: staging
Status: active
Snapshot: staging-snap-20240322-145123
Snapshot size: 1.2 KB
Created: 2024-03-22 14:51:23 UTC
Last modified: 2024-03-22 14:51:23 UTC
Forked from: production
DAGs: daily_analytics_etl, hourly_metrics, user_segmentation, ...
Tasks: 34
Runs in last 24h: 0 (not scheduled)

Archive Environment

Archive an unused environment to save metadata space:

conduit env archive dev

Output:

Archived environment: dev
Snapshot dev-snap-20240318-092345 archived

Snapshot Management

Snapshots are the immutable artifacts that environments point to. They live in the durable snapshot store at .conduit/snapshots_db and are created by conduit apply (one per executed task).

There is no standalone snapshot CLI; you inspect snapshots through the environments that reference them:

# Per-task snapshot pointers for an environment (API)
curl localhost:8080/api/v1/environments/production

# Compare the pointers of two environments
conduit env diff staging production

# See how the pointers changed over time
conduit env history production

Snapshots are never deleted automatically — old versions stay available for conduit env rollback.

Note: You cannot delete a snapshot that's referenced by an active environment.

Multi-Environment Workflows

Example: Staging + Canary Release

# 1. Create a staging environment for testing
conduit env create staging --from production

# 2. Make changes
vim dags/etl.py
conduit compile
conduit plan staging
conduit apply staging -y

# 3. Test staging for a few days
# (run the scheduler, monitor metrics, etc.)

# 4. Create a canary: run 1% of traffic to new version
conduit env create canary --from production
# (same changes as staging)
conduit apply canary -y

# 5. Monitor canary for 24 hours
# If successful:

# 6. Roll out to 10% of production via gradual promotion
conduit env promote canary production

# If issues arise:
conduit env rollback production --to prod-snap-backup-20240322-145456

Example: Feature Branch Environment

# 1. Create dev environment for feature work
conduit env create feature-new-transform --from production

# 2. Experiment freely
vim dags/transforms.py
conduit compile && conduit apply feature-new-transform -y

# 3. When ready, compare to production
conduit diff production feature-new-transform

# 4. Promote to staging first
conduit env promote feature-new-transform staging

# 5. Then to production
conduit env promote staging production

Comparing Environments

See how two environments differ:

conduit diff production staging

Output:

Comparing: production vs staging

DAGs:
  ✓ daily_analytics_etl (both present)
  ✓ hourly_metrics (both present)
  + user_segmentation (in staging only)

Tasks in daily_analytics_etl:
  ✓ extract (unchanged)
  ✓ transform (unchanged)
  ⚠ load (timeout 600 → 1200)

Summary:
  Added: 1 DAG
  Modified: 1 task
  Unchanged: 9 tasks

Schedules Are Per-DAG

Schedules are declared on the DAG itself (schedule: cron expression in the DAG definition) and executed by the scheduler inside conduit serve. There is no per-environment schedule override: cron-initiated runs execute against the default (production) environment. To exercise a DAG in another environment, trigger it explicitly:

conduit run daily_analytics_etl --env staging

or over the API with {"environment": "staging"}.

Run History Carries the Environment

Every run records which environment it ran against, and history is filterable by it:

# CLI status for one environment
conduit status --env production

# API: runs filtered by environment
curl 'localhost:8080/api/v1/runs?environment=staging'

All runs share one durable event log — the environment is a recorded field on each run, not a separate stream.

Best Practices

1. Always test in staging first: Catch issues before production.
2. Use descriptive environment names: staging, canary, feature-{name}, not test1, test2.
3. Keep production unchanged until ready: Use plan/apply workflow, never direct edits.
4. Guard production promotions with a policy: conduit env set-policy production --require-source staging.
5. Document promotions: Leave notes when promoting major changes.

Limitations

• Environments are pipeline-only: They don't isolate data or external systems. A task in staging that writes to production still writes to production.
• Snapshots are immutable: Once created, a snapshot cannot be modified. Create a new one instead.
• No partial rollback: You roll back the entire environment, not individual tasks.

If you need data isolation (staging database separate from production), configure that at the task level:

@task
def write_to_db(data):
    if os.getenv("CONDUIT_ENVIRONMENT") == "production":
        db_connection = "prod.example.com"
    else:
        db_connection = "staging.example.com"
    # Write to appropriate database

Next Steps

• Plan/Apply Workflow: Deep dive into fingerprinting and change detection
• CLI Reference: All environment commands
• Architecture: How snapshots and events work internally

Plan/Apply Workflow and Change Detection

Conduit uses a Terraform-style plan/apply workflow for deploying DAG changes. Instead of applying changes directly, you first generate a plan showing exactly what would change, then apply it after review.

Overview

The workflow is:

1. Compile — Parse and validate DAG definitions (tree-sitter)
2. Plan — Compare compiled DAGs against environment state, compute fingerprints, detect changes
3. Apply — Execute the plan, update environment snapshots, reuse unchanged tasks

This ensures that deployments are safe, predictable, and auditable.

Fingerprinting: The Core Innovation

A fingerprint is a content-addressable hash of a task and all its upstream dependencies.

Fingerprint Computation

Conduit computes fingerprints in topological order:

@task
def extract():
    return "data"

@task
def transform(data):
    return f"clean: {data}"

@task
def load(data):
    return f"loaded: {data}"

@dag
def etl():
    d = extract()
    c = transform(d)
    load(c)

Fingerprints are computed as:

extract fingerprint:
  hash(task_code, timeout, retries, pool, schedule)
  = "f1a2b3c4d5e6"

transform fingerprint:
  hash(task_code, timeout, retries, pool, upstream_fingerprints=[extract])
  = "g2h3i4j5k6l7"

load fingerprint:
  hash(task_code, timeout, retries, pool, upstream_fingerprints=[transform])
  = "m3n4o5p6q7r8"

Key insight: If extract changes, both transform and load fingerprints change automatically, even though their code didn't change. This is upstream cascade invalidation.

What Goes Into a Fingerprint?

• Task code (the function body)
• Task configuration (timeout, retries, pool, tags)
• Schedule (cron expression)
• Trigger rules
• Data type annotations
• Upstream task fingerprints (recursive)

Fingerprints do not include:

• Comments or docstrings
• Variable names (unless they affect code)
• Comments in logs

Why Content-Addressable?

Content-addressable means the hash is deterministic. The same DAG definition always produces the same fingerprint. This enables:

1. Snapshot reuse: If a task hasn't changed, reuse it from the previous snapshot
2. Change detection: Different fingerprint = something changed
3. Deduplication: Store only one copy of unchanged tasks

Change Classification

When you plan a deployment, Conduit classifies each task as one of:

Added

A new task that doesn't exist in the current environment:

conduit plan production

Output:

Added:
  + extract (fingerprint: f1a2b3c4d5e6)
  + transform (fingerprint: g2h3i4j5k6l7)
  + load (fingerprint: m3n4o5p6q7r8)

Unchanged

Task code, config, and all upstream tasks are identical:

Unchanged:
  ✓ extract (fingerprint: f1a2b3c4d5e6 → f1a2b3c4d5e6)
  ✓ transform (fingerprint: g2h3i4j5k6l7 → g2h3i4j5k6l7)
  ✓ load (fingerprint: m3n4o5p6q7r8 → m3n4o5p6q7r8)

Modified

Task code or config changed, but the DAG structure is the same:

Modified:
  ⚠ extract (timeout 300 → 600)
    Fingerprint: f1a2b3c4d5e6 → f1a2b3c4d5f7

UpstreamInvalidated

An upstream task changed, so this task must be recompiled even if its code is unchanged:

Modified:
  ⚠ extract (timeout 300 → 600)
    Fingerprint: f1a2b3c4d5e6 → f1a2b3c4d5f7

UpstreamInvalidated:
  ⚠ transform (no code changes, but extract changed)
    Fingerprint: g2h3i4j5k6l7 → g2h3i4j5k7m8
  ⚠ load (no code changes, but transform changed)
    Fingerprint: m3n4o5p6q7r8 → m3n4o5q7r9s0

When extract changes, transform and load fingerprints cascade, even though their code is identical.

Removed

A task was deleted from the DAG:

Removed:
  - old_task (fingerprint: x1y2z3a4b5c6)

The Plan Output

conduit plan production

Complete plan example:

Planning changes for environment: production
Current snapshot: prod-snap-20240322-143215

Modified:
  ⚠ daily_analytics_etl.extract
    Timeout: 300 → 600
    Fingerprint: f1a2... → f1b3...

UpstreamInvalidated:
  ⚠ daily_analytics_etl.transform
    No code changes, but upstream extract changed
    Fingerprint: g2h3... → g2i4...

  ⚠ daily_analytics_etl.load
    No code changes, but upstream transform changed
    Fingerprint: m3n4... → m3o5...

Unchanged:
  ✓ hourly_metrics (all tasks)
  ✓ user_segmentation (all tasks)

Impact Analysis:
  Blast radius: 3 tasks (extract, transform, load)
  Cascading changes: 2 tasks (transform, load)
  Safe to apply: Yes

Snapshot Optimization:
  New tasks to compile: 3 (extract, transform, load)
  Reusable from snapshot: 7 (hourly_metrics, user_segmentation)
  Snapshot size savings: 35% (reusing 7 of 10 tasks)
  New snapshot size: 1.2 KB

Ready to apply?
  conduit apply production -y

The Apply Process

conduit apply production -y

During apply:

1. Validate: Confirm no new conflicts since plan was generated
2. Compile: Recompile only modified and invalidated tasks
3. Snapshot: Create new snapshot with compiled tasks + reused tasks
4. Update: Move environment pointer to new snapshot
5. Broadcast: Notify scheduler of new snapshot

Output:

Applying plan for environment: production

Compiling modified tasks:
  ✓ daily_analytics_etl.extract (compiled from source)
  ✓ daily_analytics_etl.transform (compiled from source)
  ✓ daily_analytics_etl.load (compiled from source)

Reusing unchanged tasks:
  ✓ hourly_metrics.extract
  ✓ hourly_metrics.transform
  ✓ hourly_metrics.load
  ✓ user_segmentation.extract
  ✓ user_segmentation.transform
  ✓ user_segmentation.load
  ✓ user_segmentation.aggregate

Creating snapshot:
  ✓ Fingerprint index: 10 tasks
  ✓ Serialization: 1.2 KB
  ✓ Snapshot ID: prod-snap-20240322-145456

Updating environment:
  production → prod-snap-20240322-145456
  Previous snapshot: prod-snap-20240322-143215 (archived)

Deployment complete!

Snapshot Reuse Optimization

The key insight: Only compile what changed. Everything else is reused from the previous snapshot.

Example

Start with this DAG:

@dag
def etl():
    d = extract()
    c = transform(d)
    load(c)

Initial deployment creates snapshot with 3 tasks.

Now change only the load task's timeout:

@dag
def etl():
    d = extract()
    c = transform(d)
    load(c)  # No code change

Change config:

@task(timeout=600)  # Was 300
def load(data):
    ...

Plan shows:

Modified:
  ⚠ load (timeout 300 → 600)
    Fingerprint: m3n4... → m3o5...

UpstreamInvalidated:
  (none, load has no downstream tasks)

Snapshot Optimization:
  New tasks to compile: 1 (load)
  Reusable from snapshot: 2 (extract, transform)
  Savings: 67% smaller snapshot

Only load is recompiled. Extract and transform are byte-for-byte identical, so they're reused. This is content-addressable snapshots.

Conflict Detection

Every saved plan records base_environment_version — the environment's revision counter (Environment.current_version) at the moment the plan was generated. That counter is bumped by every apply, promote, and rollback. If the environment changes while you're holding a saved plan, Conduit detects the conflict at apply time and refuses to apply it:

# Generate a plan
conduit plan production --output plan.json

# Someone else applies a different change
# (in another shell)

# Try to apply your (now stale) plan
conduit apply production --plan-file plan.json -y

Output:

Error: stale plan — environment 'production' changed since this plan was generated.
  Current environment version: 4
  Plan was based on version:    3

Recommended action:
  conduit plan production --output plan.json   # regenerate against current state
  conduit apply production --plan-file plan.json -y

Conduit also rejects a plan file applied against the wrong environment: if a plan's target_environment doesn't match the environment named on the apply command line, the apply fails immediately with an error telling you which environment the plan actually targets.

Conduit prevents applying stale plans.

Rollback Restores a Recorded Version

Every apply, promote, and rollback records a version in the environment's history. Rolling back restores the snapshot pointers captured at a prior version:

# See the environment's recorded versions
conduit env history production

# Undo the most recent mutation
conduit env rollback production --yes

# Or restore a specific version
conduit env rollback production --to-version 3 --yes

Rollback is instant because snapshots are immutable — only the environment's pointers move.

Safety Guarantees

The plan/apply workflow provides several safety guarantees:

1. Read-Only Plans

Plans are read-only. Running conduit plan never modifies state:

conduit plan production  # Safe to run anytime

2. Deterministic Fingerprints

Same code always produces same fingerprint:

conduit plan production     # Fingerprint: f1a2b3c4d5e6
# Wait 1 hour
conduit plan production     # Fingerprint: f1a2b3c4d5e6 (identical)

3. Atomic Snapshots

Snapshots are created atomically. Partial uploads or corruption is impossible:

conduit apply production -y
# If killed mid-apply, environment remains unchanged
# (No half-applied snapshots)

4. Complete Audit Trail

Every apply is logged to the durable event store (PlanApplied events), and every environment mutation is recorded in its version history:

conduit replay --events-only          # walk the raw event log
conduit env history production        # per-environment version history

Over the API: GET /api/v1/events?event_type=PlanApplied.

Best Practices

1. Always plan before apply: Never skip the plan step
2. Review plans in peer reviews: Share plan output with team
3. Apply during low-traffic windows: Minimize impact if issues arise
4. Keep snapshots around: Don't immediately delete old snapshots
5. Test in staging first: Apply changes to staging before production
6. Monitor after apply: Check metrics for 1-2 hours post-deployment

Complex Scenarios

Zero-Downtime Deployments

# 1. Create canary environment
conduit env create canary --from production

# 2. Apply changes to canary
vim dags/etl.py
conduit compile
conduit plan canary
conduit apply canary -y

# 3. Verify canary works (run a few test executions)
conduit run daily_analytics_etl --env canary

# 4. Switch traffic gradually
# (This is application-level routing, not Conduit)

# 5. Promote canary to production when stable
conduit env promote canary production

A/B Testing Different Implementations

# Branch A: New implementation
conduit env create impl-a --from production
vim dags/etl.py  # Implement new transform
conduit apply impl-a -y

# Branch B: Keep old implementation
conduit env create impl-b --from production
# (no changes, same as production)

# Run both for 1 week, compare metrics
# Winner becomes new production
conduit env promote impl-a production

Next Steps

• Virtual Environments: Understand how environments and snapshots work
• Column-Level Lineage: Track data flow through pipelines
• CLI Reference: Plan and apply command details

Column-Level Lineage (Beta)

Conduit provides column-level SQL lineage powered by sqlparser-rs AST analysis. This feature is functional and tested (67 tests) but labeled beta due to known limitations with template SQL and view resolution.

What Is Lineage?

Lineage answers the question: "Where did this column come from?"

For example, if a dashboard shows a metric customer_lifetime_value, lineage traces its source:

dashboard.revenue_dashboard.customer_ltv
  ← analytics.metrics.customer_metrics.customer_ltv
    ← warehouse.transformed.customer_summary.total_spent
      ← warehouse.raw.transactions.amount
        ← (origin: payment API)

This tracing works across multiple DAGs, systems, and technologies.

Why Column-Level Lineage Matters

1. Data Governance

Understand data dependencies for compliance:

• GDPR: Find all places customer PII flows
• HIPAA: Audit medical data access
• CCPA: Identify where personal data is stored

2. Impact Analysis

When a source system changes, understand cascading effects:

• Source column deleted → which dashboards break?
• Data quality issue upstream → which reports are affected?

3. Debugging

Trace data quality issues to their root cause:

• Dashboard shows wrong numbers → which query is wrong?
• ETL pipeline failed → which upstream task caused it?

4. Documentation

Automatic documentation of how data flows:

• No more "I think this column comes from…"
• Executable source of truth

How Conduit Tracks Lineage

1. SQL Parsing (sqlparser-rs AST)

For SQL tasks, Conduit parses the query AST to extract column dependencies. This uses sqlparser-rs for proper parsing (not regex), supporting:

• SELECT with aliases, expressions, and qualified references (schema.table.column)
• JOINs (INNER, LEFT, RIGHT, FULL, CROSS)
• CTEs (WITH clauses, including chained CTEs with column propagation)
• UNIONs / INTERSECT / EXCEPT
• Subqueries in FROM and scalar positions
• Window functions (PARTITION BY, ORDER BY)
• INSERT INTO...SELECT and CREATE TABLE AS SELECT
• WHERE clause dependency tracking

@sql_task(dialect="postgres")
def create_customer_summary():
    CREATE TABLE analytics.customer_summary AS
    SELECT
        customer_id,
        name,           -- FROM raw.customers
        email,          -- FROM raw.customers
        SUM(amount),    -- FROM raw.transactions
        COUNT(*)        -- FROM raw.transactions
    FROM raw.customers c
    JOIN raw.transactions t ON c.id = t.customer_id
    GROUP BY customer_id, name, email

Conduit's SQL parser extracts:

Output: analytics.customer_summary
  - customer_id → FROM raw.customers.customer_id
  - name → FROM raw.customers.name
  - email → FROM raw.customers.email
  - sum → FROM raw.transactions.amount
  - count → FROM raw.transactions (COUNT(*))

TableCatalog Integration

When a TableCatalog is provided (populated from connected providers via information_schema), lineage gains additional capabilities:

• Bare column resolution: SELECT active FROM orders o JOIN customers c correctly maps active to customers (instead of guessing the first table)
• Wildcard expansion: SELECT * is expanded to actual column names
• CTE column propagation: Output columns of CTEs are resolved through to their source tables

2. Python Task Lineage

For Python tasks, lineage is specified via annotations:

from conduit.sdk import task, Lineage

@task
def transform_data(input_df):
    """
    Lineage annotations specify where outputs come from.
    """
    output_df = input_df.copy()
    output_df['revenue_category'] = pd.cut(
        output_df['amount'],
        bins=[0, 100, 1000, float('inf')],
        labels=['low', 'medium', 'high']
    )
    return output_df

# Specify lineage explicitly
transform_data.set_lineage({
    'revenue_category': ['amount']  # Column created from amount
})

Or with decorators:

from conduit.sdk import task, lineage

@task
@lineage.input_columns(['customer_id', 'amount', 'date'])
@lineage.output_columns({'revenue_category': ['amount']})
def categorize_revenue(df):
    df['revenue_category'] = pd.cut(df['amount'], ...)
    return df[['customer_id', 'revenue_category', 'date']]

Lineage Graph API

Query the lineage graph:

Upstream Lineage

Find the origin of a column:

from conduit import lineage_client

# What sources feed into this column?
upstream = lineage_client.upstream(
    table='analytics.metrics.customer_metrics',
    column='customer_ltv'
)

# Output:
# [
#   {
#     'source': 'warehouse.transformed.customer_summary',
#     'column': 'total_spent',
#     'task': 'daily_analytics_etl.aggregate_metrics',
#     'dag': 'daily_analytics_etl'
#   },
#   {
#     'source': 'warehouse.raw.transactions',
#     'column': 'amount',
#     'task': 'daily_analytics_etl.extract',
#     'dag': 'daily_analytics_etl'
#   }
# ]

Downstream Lineage

Find all dependents of a column:

# What downstream objects use this column?
downstream = lineage_client.downstream(
    table='warehouse.raw.transactions',
    column='amount'
)

# Output:
# [
#   {
#     'target': 'warehouse.transformed.customer_summary',
#     'column': 'total_spent',
#     'task': 'daily_analytics_etl.aggregate_metrics',
#     'dag': 'daily_analytics_etl'
#   },
#   {
#     'target': 'analytics.metrics.customer_metrics',
#     'column': 'customer_ltv',
#     'task': 'daily_analytics_etl.create_metrics',
#     'dag': 'daily_analytics_etl'
#   }
# ]

Lineage Graph Visualization

# Get complete lineage graph
graph = lineage_client.graph(
    start_table='analytics.metrics.customer_metrics',
    start_column='customer_ltv',
    direction='both'  # upstream and downstream
)

# Outputs:
# nodes: [
#   {'id': 'analytics.metrics.customer_ltv', 'type': 'column'},
#   {'id': 'warehouse.transformed.customer_summary.total_spent', 'type': 'column'},
#   {'id': 'warehouse.raw.transactions.amount', 'type': 'column'},
#   {'id': 'daily_analytics_etl.aggregate_metrics', 'type': 'task'},
#   {'id': 'daily_analytics_etl.extract', 'type': 'task'}
# ]
# edges: [
#   {'source': 'analytics.metrics.customer_ltv', 'target': 'warehouse.transformed.customer_summary.total_spent'},
#   ...
# ]

Schema Contracts

Define expected schemas to detect breaking changes:

from conduit.sdk import task, SchemaContract
from pydantic import BaseModel

class RawTransactionsSchema(BaseModel):
    transaction_id: str
    customer_id: int
    amount: float
    timestamp: str
    status: str  # NEW: This is a breaking change

@task
@SchemaContract.input('transactions', RawTransactionsSchema)
def process_transactions(df):
    # If upstream doesn't provide 'status', fail at compile time
    return df[df['status'] == 'completed']

Conduit verifies schema contracts at compile time:

conduit compile

Output:

Error: Schema contract violation
  Task: process_transactions
  Contract: RawTransactionsSchema
  Missing: status
  Provided by upstream: transaction_id, customer_id, amount, timestamp

The upstream task doesn't provide the 'status' column.
Update the upstream task to include 'status', or update the contract.

Lineage Queries via REST API

Query lineage via HTTP:

# Get upstream lineage
curl http://localhost:8080/api/v1/lineage/upstream \
  -H "Content-Type: application/json" \
  -d '{
    "table": "analytics.metrics.customer_metrics",
    "column": "customer_ltv"
  }'

# Response:
# [
#   {
#     "source": "warehouse.transformed.customer_summary",
#     "column": "total_spent",
#     "task": "daily_analytics_etl.aggregate_metrics",
#     "dag": "daily_analytics_etl"
#   }
# ]

Data Contracts and Breaking Changes

Detect breaking changes automatically:

from conduit.sdk import task, DataContract

class CustomerMetricsContract(DataContract):
    """Schema for customer metrics table."""
    customer_id: int
    total_spent: float
    last_purchase: str
    segment: str

@task(contract=CustomerMetricsContract)
def create_metrics():
    # If output columns don't match contract, fail at runtime
    return {
        'customer_id': 123,
        'total_spent': 999.99,
        'last_purchase': '2024-03-22',
        'segment': 'high_value'
    }

# If you remove a column:
@task(contract=CustomerMetricsContract)
def create_metrics_v2():
    return {
        'customer_id': 123,
        'total_spent': 999.99,
        'last_purchase': '2024-03-22'
        # Missing: segment
    }

Conduit detects the contract violation:

Error: Contract violation
  Task: create_metrics_v2
  Contract: CustomerMetricsContract
  Missing columns: segment
  Impact: 3 downstream tasks depend on this column
    - dashboard_builder.segment_analysis
    - reporting_etl.customer_segments
    - ml_pipeline.customer_clustering

Advanced Use Cases

Finding Stale Columns

Identify columns that are produced but never consumed:

stale = lineage_client.unused_columns(
    since='30d'  # Not used in past 30 days
)

# Output:
# [
#   {
#     'table': 'warehouse.transformed.legacy_metrics',
#     'column': 'old_metric',
#     'produced_by': 'daily_etl.legacy_aggregate',
#     'days_unused': 45
#   }
# ]

Impact Analysis for Schema Changes

When you remove a column, see what breaks:

impact = lineage_client.impact_of_removal(
    table='warehouse.raw.transactions',
    column='amount'
)

# Output:
# {
#   'direct_consumers': [
#     'daily_etl.aggregate_metrics'
#   ],
#   'indirect_consumers': [
#     'daily_etl.create_metrics',
#     'reporting_etl.dashboard_data'
#   ],
#   'total_affected_tasks': 3,
#   'total_affected_dashboards': 5
# }

Compliance Audits

Find all places PII flows:

pii_columns = ['email', 'phone', 'ssn', 'address']

for col in pii_columns:
    paths = lineage_client.downstream(
        table='raw.customers',
        column=col
    )
    print(f"Column {col} flows to:")
    for path in paths:
        print(f"  {path['target']}.{path['column']}")

Current Status (Beta)

SQL lineage is functional and tested:

Implemented

• SQL parsing via sqlparser-rs AST (PostgreSQL, MySQL, BigQuery, Snowflake, and other dialects)
• TableCatalog integration for bare column resolution and wildcard expansion
• CTE column propagation
• OpenLineage RunEvent generation with output columnLineage facets
• Explicit lineage annotations for Python tasks
• REST API endpoints (/lineage/sql, /lineage/catalog/refresh)
• Lineage graph traversal
• 79 lineage tests covering real ETL patterns, Jinja stripping, views, and OpenLineage emission

Known Limitations

• Jinja/template SQL is stripped enough for surrounding SQL to parse, but dbt-aware semantic rendering (ref, source, vars) is not implemented yet
• View registration exposes view columns, but full recursive source-table substitution is still limited
• SELECT * EXCEPT(col) (BigQuery syntax) not supported
• LATERAL joins, PIVOT/UNPIVOT, UNNEST not handled
• Python task lineage is annotation-based only (no static analysis)

Planned

• OpenLineage HTTP transport / CLI export
• dbt-aware template rendering
• Runtime schema capture via CONDUIT::SCHEMA:: protocol
• Schema evolution tracking
• Lineage visualization in the Web UI

Next Steps

• Events Architecture: How events enable lineage tracking
• REST API: Lineage API endpoints
• Architecture: How lineage integrates with the system

Event-Sourced Architecture

Conduit uses event sourcing instead of mutable database state. Every action in the system is recorded as an immutable event in an append-only log. From this log, you can reconstruct any state, replay any execution, and time-travel through history.

What Is Event Sourcing?

Traditional databases store the current state:

Database State (Mutable):
  DAG daily_analytics_etl
    status: running
    last_run: 2024-03-22 14:32
    next_run: 2024-03-22 15:32

If you want to know what the status was 1 hour ago, you're out of luck. The state has been mutated.

Event sourcing stores every state change as an immutable event:

Event Log (Append-only):
  1. DAGScheduled(dag_id: daily_analytics_etl, time: 2024-03-22 14:00)
  2. SnapshotDeployed(env: production, snapshot: prod-v5)
  3. DAGRunStarted(run_id: run123, dag: daily_analytics_etl, time: 2024-03-22 14:32)
  4. TaskStarted(run_id: run123, task: extract, time: 2024-03-22 14:32:10)
  5. TaskProgressEvent(run_id: run123, task: extract, progress: 50%)
  6. TaskProgressEvent(run_id: run123, task: extract, progress: 100%)
  7. TaskCompleted(run_id: run123, task: extract, time: 2024-03-22 14:32:45)
  8. TaskStarted(run_id: run123, task: transform, time: 2024-03-22 14:32:46)
  9. TaskCompleted(run_id: run123, task: transform, time: 2024-03-22 14:35:12)
  10. TaskStarted(run_id: run123, task: load, time: 2024-03-22 14:35:13)
  11. TaskCompleted(run_id: run123, task: load, time: 2024-03-22 14:37:20)
  12. DAGRunCompleted(run_id: run123, status: success, time: 2024-03-22 14:37:20)

From this log, you can:

• Reconstruct state: Apply all events to find current state at any point
• Time-travel: Query state at event 6 vs event 12
• Replay: Re-execute the same run using the same inputs
• Debug: See exactly what happened and when

Event Types

Conduit defines structured event types:

1. Compilation Events

class DAGCompiled(Event):
    dag_id: str
    fingerprint: str
    num_tasks: int
    compile_duration_ms: int
    timestamp: datetime

Fired when conduit compile succeeds.

2. Scheduling Events

class DAGScheduled(Event):
    dag_id: str
    run_id: str
    scheduled_time: datetime
    trigger: str  # "cron" | "manual" | "dependency" | "event"
    timestamp: datetime

class DAGUnscheduled(Event):
    dag_id: str
    run_id: str
    reason: str
    timestamp: datetime

Fired when the scheduler determines a DAG should run.

3. Execution Events

class DAGRunStarted(Event):
    run_id: str
    dag_id: str
    environment: str
    timestamp: datetime

class TaskStarted(Event):
    run_id: str
    task_id: str
    timestamp: datetime
    worker_id: str

class TaskProgressEvent(Event):
    run_id: str
    task_id: str
    progress_percent: int
    timestamp: datetime

class TaskCompleted(Event):
    run_id: str
    task_id: str
    status: str  # "success" | "failed" | "skipped"
    exit_code: int
    duration_ms: int
    xcom_output: Dict[str, Any]
    timestamp: datetime

class DAGRunCompleted(Event):
    run_id: str
    status: str
    total_duration_ms: int
    failed_tasks: List[str]
    timestamp: datetime

4. Deployment Events

class SnapshotCompiled(Event):
    snapshot_id: str
    dag_id: str
    num_tasks: int
    snapshot_size_bytes: int
    timestamp: datetime

class SnapshotDeployed(Event):
    snapshot_id: str
    environment: str
    previous_snapshot: str
    num_reused_tasks: int
    timestamp: datetime

class EnvironmentCreated(Event):
    environment: str
    forked_from: Optional[str]
    snapshot_id: str
    timestamp: datetime

class EnvironmentPromoted(Event):
    source_env: str
    target_env: str
    snapshot_id: str
    previous_snapshot: str
    timestamp: datetime

5. Error Events

class TaskFailed(Event):
    run_id: str
    task_id: str
    exit_code: int
    stderr: str
    timestamp: datetime

class DAGRunFailed(Event):
    run_id: str
    failed_task_id: str
    reason: str
    timestamp: datetime

Event Storage

Events are stored in RocksDB, Conduit's embedded key-value store:

[Event Log in RocksDB]
  Key: event_sequence_000001
  Value: {serialized DAGCompiled event}

  Key: event_sequence_000002
  Value: {serialized DAGScheduled event}

  Key: event_sequence_000003
  Value: {serialized TaskStarted event}

  ... (100+ events)

Each event has a monotonic sequence number. Range queries are O(log n).

Durability

RocksDB uses write-ahead logging, ensuring:

• Crash safety: Even if the process dies, events are persisted
• No data loss: Every event is durable before returning to caller
• Fast writes: Batch writes reduce I/O

Time-Travel Debugging

Replay the event log to reconstruct system state as of any point in history:

# List the raw events (sequence, timestamp, type)
conduit replay --events-only

# Reconstruct state as of event 7
conduit replay --to 7

# Machine-readable reconstruction
conduit replay --to 7 --json

The reconstruction shows the environments that existed, every run and its status, and per-task success/failure counts — all derived purely from events up to the chosen sequence number. Because events are immutable, the same replay always produces the same state.

Event Streaming

Subscribe to events in real-time over the WebSocket endpoint (/ws/events, outside the /api/v1 prefix) with any WebSocket client:

import asyncio, json, websockets

async def watch():
    async with websockets.connect("ws://localhost:8080/ws/events") as ws:
        async for message in ws:
            event = json.loads(message)
            print(f"[{event['type']}] {event}")

asyncio.run(watch())

Event Queries

Query the event log over the API:

# All events for a run
curl 'localhost:8080/api/v1/events?run_id=run123'

# Failed tasks
curl 'localhost:8080/api/v1/events?event_type=TaskFailed&limit=50'

# Deployment history
curl 'localhost:8080/api/v1/events?event_type=PlanApplied'

# A sequence range
curl 'localhost:8080/api/v1/events?from=100&to=200'

## Snapshots

Snapshots are **derived state** computed from the event log:

Event Log: [1] DAGCompiled(fingerprint: f1a2b3...) [2] DAGCompiled(fingerprint: f1b3c4...) [3] SnapshotDeployed(snapshot: v2, uses fingerprints: f1b3c4...) [4-100] ... (other events) [101] DAGCompiled(fingerprint: f1c4d5...) [102] SnapshotDeployed(snapshot: v3, uses fingerprints: f1c4d5...)

Computed Snapshots: v2: snapshot_id → {dag_id → compiled_dag} v3: snapshot_id → {dag_id → compiled_dag}

Snapshots are **read-only caches** of compiled DAGs. They're computed from the event log on-demand.

### Snapshot Coherency

A snapshot is **coherent** if all events leading to it are present in
the event log. You can verify what the log supports by reconstructing
state from it:

```bash
conduit replay --json

Whatever replay reconstructs is exactly what the event log can prove happened; anything else was lost or never recorded.

Retention Policies

The event store supports retention limits (maximum event age, maximum event count, and a minimum number of events always kept). This is currently a library-level policy — RetentionPolicy in conduit-state, with standard() (7 days / 100k events) and extended() (30 days / 1M events) presets — and the default keeps everything. There is no user-facing configuration file setting for it yet.

Event-Driven Triggers

There is no built-in webhook mechanism. To trigger external actions (Slack alerts, PagerDuty, …), subscribe to the WebSocket stream at /ws/events and forward the events you care about — see Event Streaming above.

Audit Trail

Every apply is recorded in the event log, and every environment mutation (apply, promote, rollback) is recorded in the environment's version history:

# Environment-level audit: version history with reasons
conduit env history production

# Raw event log
conduit replay --events-only

Over the API: GET /api/v1/events?event_type=PlanApplied.

Consistency Guarantees

Event sourcing provides strong consistency guarantees:

1. Monotonic consistency: Events are applied in order
2. Durability: Events are persisted before returning
3. Completeness: All state changes are captured
4. Auditability: Complete trace of what happened when

Performance Implications

Event sourcing has a cost: latency. Reconstructing state from 100,000 events is slower than reading from a database.

Conduit mitigates this with:

1. Snapshots: Cache compiled DAGs to avoid replaying compile events
2. Event indexing: Index by run_id, task_id, timestamp for fast queries
3. Batching: Write multiple events in one RocksDB transaction

Typical latencies:

• Query recent events: < 1 ms (in-memory cache)
• Reconstruct state: < 10 ms (RocksDB range query)
• Time-travel to event: < 50 ms (index lookup + replay)

Next Steps

• Plan/Apply Workflow: How events enable safe deployments
• Architecture: How the event store integrates with other crates
• API Reference: Query and subscribe to events via REST/WebSocket

Incremental Computation

Conduit includes a SQLMesh-inspired incremental computation engine that avoids reprocessing data that hasn't changed. Instead of full refreshes every run, tasks can process only new or modified rows using watermark-based change tracking.

Strategies

Conduit supports five incremental strategies, each suited to different data patterns:

Full Refresh

Reprocesses all data every run. Use when data is small or transformations are non-deterministic.

incremental:
  strategy: full_refresh

Append

Processes only rows newer than the last watermark. Ideal for immutable, time-ordered event streams (clicks, logs, transactions).

incremental:
  strategy: append
  time_column: created_at
  lookback: 2h           # safety overlap
  batch_size: 100000

Merge on Key

Upserts rows by a unique key — detects new and changed rows. Ideal for dimension tables and mutable records.

incremental:
  strategy: merge_on_key
  unique_key: [user_id]
  time_column: updated_at
  invalidate_hard_deletes: true

Delete + Insert

Replaces entire partitions. Ideal for date-partitioned fact tables where late-arriving data may update a whole day's partition.

incremental:
  strategy: delete_insert
  partition_column: event_date
  partition_granularity: day
  max_partitions_per_run: 7

Snapshot Diff (SCD Type 2)

Tracks historical changes by comparing current and previous snapshots. Creates new versions when monitored columns change.

incremental:
  strategy: snapshot_diff
  unique_key: [product_id]
  check_columns: [name, price, status]
  scd_type_2: true

Watermarks

Conduit maintains watermarks that record how far each task has processed. Watermark types include:

• Timestamp — a datetime high-water mark (used by append)
• Sequence — an integer offset (used by event streams)
• Partition — a set of processed partitions (used by delete_insert)

Watermarks are persisted to .conduit/watermarks.json in the state directory and survive restarts. conduit run and conduit apply both load this file before executing tasks. A task's watermark only advances once it exits successfully and emits a new value; conduit run writes the file back once the whole run finishes (even if some other task in the DAG failed), while conduit apply only writes it back once the apply reaches its success path — a blocked or failed apply discards any in-memory watermark advances instead of persisting them.

SQL Rewriting

For SQL tasks with incremental config, Conduit automatically rewrites queries to filter only unprocessed data. For example, an append strategy adds:

WHERE created_at > '2024-01-15T10:30:00Z'
  AND created_at <= '2024-01-16T06:00:00Z'

This happens transparently — your SQL stays clean and the engine handles the filtering.

Full Refresh Override

Force a complete reprocess of any task by passing --full-refresh:

conduit run my_dag --full-refresh
conduit apply production --full-refresh

This ignores all watermarks and reprocesses from scratch — useful after schema changes or to fix data quality issues.

Python SDK

The Python SDK provides helpers for reading incremental context in Python tasks:

from conduit_sdk.incremental import get_incremental_context, emit_watermark

ctx = get_incremental_context()
if ctx.is_full_refresh:
    process_all()
else:
    process_since(ctx.low_watermark)
    emit_watermark(ctx.high_watermark)

Data Quality Contracts

Contracts are declarative assertions about your pipeline's data. They live alongside your DAG definitions — in YAML or Python — and are validated automatically during conduit plan and conduit apply. Error-severity violations block deployment. Warning-severity violations are reported but don't block.

Evidence-Based Design

Unlike tools that assume SQL access, Conduit contracts are evidence-based. Tasks emit measurements (evidence) via the stdout protocol, and contracts assert against those measurements. This works uniformly across SQL, Python, shell, API, and any other task type.

Task (any language)
  └─ emits CONDUIT::METRIC::row_count::5000
  └─ emits CONDUIT::METRIC::data_age_seconds::3600
  └─ emits CONDUIT::METRIC::null_rate.email::0.02

Executor
  └─ collects metrics into Evidence { metrics: HashMap<String, f64> }

ContractEvaluator
  └─ takes (Evidence, TaskContracts) → ValidationResult
  └─ each contract check knows which metric to look up
  └─ missing evidence = contract failure

The task is the only thing that knows its own output. The contract doesn't know or care how the measurement was produced — only that it was emitted.

Emitting Evidence

Python Tasks

from conduit_sdk import emit_metric, emit_row_count, emit_freshness

def extract_orders():
    rows = fetch_data()

    # Convenience helpers
    emit_row_count(len(rows))
    emit_freshness(rows[-1]["created_at"])

    # Generic metric emission
    emit_metric("duplicate_count", count_dupes(rows))
    emit_metric("null_rate.email", null_fraction(rows, "email"))
    emit_metric("accuracy", model.evaluate())

    return rows

Shell Tasks

#!/bin/bash
ROW_COUNT=$(wc -l < output.csv)
echo "CONDUIT::METRIC::row_count::$ROW_COUNT"
echo "CONDUIT::METRIC::data_age_seconds::3600"

SQL Tasks

Conduit's built-in SQL executor auto-emits common metrics (row_count, data_age_seconds, duplicate_count, null_rate.*) for SQL tasks. You don't need to emit them manually — contracts just work.

Contract Types

Row Count

Assert that the output has a minimum, maximum, or exact number of rows. Expects metric: row_count

contracts:
  - type: row_count
    min: 1
    max: 10000000

Freshness

Assert that data is recent. Expects metric: data_age_seconds

contracts:
  - type: freshness
    max_age: 24h

Unique

Assert no duplicate values across columns. Expects metric: duplicate_count

contracts:
  - type: unique
    columns: [id]

Not Null

Assert that a column has a minimum fraction of non-null values. Expects metric: null_rate.{column}

contracts:
  - type: not_null
    column: customer_id
    min_rate: 0.99  # allow up to 1% nulls

Accepted Values

Assert that a column's values are within a known set. Expects metric: invalid_value_count.{column}

contracts:
  - type: accepted_values
    column: status
    values: [pending, shipped, delivered, cancelled]

Value Range

Assert that a numeric column falls within bounds. Expects metric: out_of_range_count.{column}

contracts:
  - type: value_range
    column: amount
    min: 0
    max: 1000000

Referential Integrity

Assert that every value in a column exists in another task's output. Expects metric: orphan_count.{column}

contracts:
  - type: references
    column: customer_id
    ref_task: extract_customers
    ref_column: id

Row Count Delta

Assert that the row count doesn't change too dramatically between runs. Expects metric: row_count_delta_pct

contracts:
  - type: row_count_delta
    max_percent_change: 0.1    # flag if >10% change
    allow_decrease: false       # any decrease is an error

Metric (Generic)

The universal contract — assert any named metric against bounds. This is the escape hatch for any custom measurement.

contracts:
  - type: metric
    metric_name: accuracy
    min: 0.95
  - type: metric
    metric_name: latency_ms
    max: 500
  - type: metric
    metric_name: enrichment_rate
    min: 0.90
    max: 1.0

Custom Assertion

A named pass/fail check. The task emits pass.{name}::1 or pass.{name}::0.

contracts:
  - type: custom
    assertion_name: no_orphan_orders
    description: "Every order must have a valid customer"

Severity

Every contract defaults to error severity (blocks deployment). Set severity: warning to report without blocking:

contracts:
  - type: row_count_delta
    max_percent_change: 0.1
    severity: warning
    description: "Alert if customer count changes significantly"

YAML Example

id: daily_etl
tasks:
  extract_orders:
    type: sql
    query: "SELECT * FROM source.orders"
    contracts:
      - type: row_count
        min: 1
      - type: freshness
        max_age: 24h
      - type: unique
        columns: [id]
      - type: not_null
        column: customer_id
      - type: accepted_values
        column: status
        values: [pending, shipped, delivered]
      - type: value_range
        column: amount
        min: 0

  train_model:
    type: python
    module: ml.training
    function: train
    contracts:
      - type: metric
        metric_name: accuracy
        min: 0.90
      - type: metric
        metric_name: training_loss
        max: 0.1
      - type: custom
        assertion_name: model_convergence

Python SDK

The Python SDK supports both decorator and imperative styles:

from conduit_sdk import task, emit_metric, emit_row_count
from conduit_sdk.contracts import contract, check

# Decorator style — declare what to assert
@task(retries=3)
@contract(
    check.row_count(min=1),
    check.freshness(max_age="24h"),
    check.unique(["id"]),
    check.not_null("customer_id"),
    check.metric("accuracy", min=0.95),
)
def extract_orders():
    rows = do_work()
    # Emit evidence for the contracts to validate against
    emit_row_count(len(rows))
    emit_metric("data_age_seconds", compute_age(rows))
    emit_metric("duplicate_count", count_dupes(rows))
    emit_metric("null_rate.customer_id", null_frac(rows, "customer_id"))
    emit_metric("accuracy", evaluate())
    return rows

# Imperative style — contracts emitted at runtime
from conduit_sdk.contracts import Contracts

def transform():
    result = do_work()
    c = Contracts("transform")
    c.row_count(min=1, max=1_000_000)
    c.metric("accuracy", min=0.95)
    c.emit()  # sends to executor via stdout protocol

Plan/Apply Integration

When you run conduit plan, the output shows which contracts will be validated and what metrics each task is expected to emit:

Contracts: 11 checks across 3 tasks (validated during apply)
  daily_etl.extract_orders — 6 checks
    expected metrics: row_count, data_age_seconds, duplicate_count,
                      null_rate.customer_id, invalid_value_count.status,
                      out_of_range_count.amount
  daily_etl.train_model — 2 checks
    expected metrics: accuracy, pass.model_convergence

During conduit apply, immediately after each task executes, its evidence is validated against that task's contracts. A passing task prints an inline [CHK ] line; a violation prints [CVIO] plus the failing check(s) to stderr:

  [EXEC]  contract_ok.emit
  [CHK ] contract_ok.emit contracts: 1/1 checks passed
  [OK]    contract_ok.emit (5ms)

  [EXEC]  contract_bad.emit
  [CVIO] contract_bad.emit contracts: 0/1 checks passed
          ! row_count:emit: 5 rows, expected at least 1000

If any Error-severity contract fails, the apply stops before the environment is updated and prints the DeploymentValidation summary (this is the actual output from conduit apply against a task whose row_count contract requires min: 1000 but the task only emitted CONDUIT::METRIC::row_count::5):

Contract Validation Summary
─────────────────────────────
Contracts for 'contract_bad.emit': FAILED (0/1 checks passed, 1 errors, 0 warnings)
  [ERROR] row_count:emit: 5 rows, expected at least 1000

Result: BLOCKED — 1 errors must be fixed before deployment (0 warnings)

Error: apply blocked: contract validation failed for contract_bad.emit — environment not updated

The environment's snapshot pointers are left untouched — conduit status / conduit env list show no change, and a subsequent conduit plan still reports the task as pending execution. Warning-severity failures print the same way but do not block; DeploymentValidation.can_deploy (and DeploymentPlan::can_apply) is only false when at least one Error-severity check fails.

API

The REST API exposes contract information:

• GET /api/v1/contracts — list all contracts across all DAGs
• GET /api/v1/contracts/:dag_id — contracts for a specific DAG
• GET /api/v1/contracts/:dag_id/:task_id — contracts for a specific task

CLI Reference

Complete reference for all Conduit CLI commands. Every command and flag on this page is taken from the CLI's own --help output — run conduit <command> --help for the authoritative version.

Global Options

All commands accept these options:

--verbose, -v         Show detailed output
--help, -h            Show help for this command
--version, -V         Show Conduit version (top-level only)

Project Commands

init

Initialize a new Conduit project.

conduit init <NAME>

Arguments:
  <NAME>                Project name

Example:

conduit init analytics-pipeline
cd analytics-pipeline

compile

Compile DAGs and report results.

conduit compile [PATH] [options]

Arguments:
  [PATH]                Path to DAG definitions (default: ./dags)

Options:
  --output, -o <path>   Output compiled plan to file
  --check               Check only (don't write output)

Example:

conduit compile                          # Compile ./dags
conduit compile ./my-dags                # Compile different directory
conduit compile --check                  # Validate only
conduit compile --output compiled.json   # Save compiled plan

Execution Commands

run

Run a DAG (compile, schedule, and execute).

conduit run <DAG_ID> [options]

Arguments:
  <DAG_ID>              DAG ID to run

Options:
  --dags-path, -d <path>   Path to DAG definitions (default: ./dags)
  --date <date>            Logical date override (default: now)
  --max-tasks <n>          Maximum tasks to execute concurrently (default: 16)
  --full-refresh           Force full refresh on all incremental tasks
                           (ignore watermarks)
  --env <name>             Target environment recorded for this run
                           (default: production; context only — snapshots
                           are managed by plan/apply)
  --distributed            Run via the distributed coordinator; workers
                           must connect (see conduit worker)
  --bind <addr>            Coordinator bind address for distributed mode
                           (default: 0.0.0.0:9400)

Example:

conduit run daily_etl                       # Run now
conduit run daily_etl --date 2024-03-01     # Run for specific date
conduit run daily_etl --env staging         # Record run against staging
conduit run daily_etl --max-tasks 4         # Cap concurrency
conduit run daily_etl --full-refresh        # Ignore watermarks
conduit run daily_etl --distributed         # Dispatch to connected workers

status

Show system status.

conduit status [options]

Options:
  --env, -e <name>         Show status for a specific environment
  --dags-path, -d <path>   Path to DAG definitions (default: ./dags)

Example:

conduit status                      # Overall status
conduit status --env production     # Production status

backfill

Backfill a DAG across a range of dates/partitions.

conduit backfill <DAG_ID> --start <START> --end <END> [options]

Arguments:
  <DAG_ID>              DAG ID to backfill

Options:
  --start <date>           Start date (inclusive, YYYY-MM-DD) [required]
  --end <date>             End date (exclusive, YYYY-MM-DD) [required]
  --granularity <g>        Partition granularity (default: day)
  --max-concurrent <n>     Maximum partitions to execute concurrently
                           (default: 1)
  --full-refresh           Force full refresh on all partitions
  --dry-run                Show what would run without executing
  --env <name>             Target environment (default: production)
  --dags-path, -d <path>   Path to DAG definitions (default: ./dags)

Example:

conduit backfill daily_etl --start 2024-03-01 --end 2024-03-08
conduit backfill daily_etl --start 2024-03-01 --end 2024-03-08 --max-concurrent 4
conduit backfill daily_etl --start 2024-03-01 --end 2024-03-08 --dry-run

query

Run SQL queries locally (powered by DuckDB).

conduit query <SQL> [options]

Arguments:
  <SQL>                 SQL query to execute

Options:
  --connection, -c <name>  Named connection from conduit.yaml
                           (default: ephemeral in-memory DuckDB)
  --file, -f <path>        Query a local file (Parquet, CSV, JSON) —
                           registers it as a table
  --setup, -s <sql>        Run setup SQL before the main query
  --format <fmt>           Output format: table, json, csv (default: table)
  --limit <n>              Maximum rows to return (default: 50)
  --config <path>          Path to conduit.yaml (for connection resolution)

Example:

conduit query "SELECT 1 AS answer"
conduit query "SELECT * FROM data" --file events.parquet
conduit query "SELECT count(*) FROM orders" --connection warehouse

preview

Preview a SQL task's output locally.

conduit preview <TASK_REF> [options]

Arguments:
  <TASK_REF>            Task reference: dag_id.task_id

Options:
  --dags-path, -d <path>   Path to DAG definitions (default: ./dags)
  --connection, -c <name>  Override connection (default: ephemeral DuckDB)
  --format <fmt>           Output format: table, json, csv (default: table)
  --limit <n>              Maximum rows to return (default: 50)

Example:

conduit preview daily_etl.transform
conduit preview daily_etl.transform --connection warehouse --limit 10

Deployment Commands

plan

Show changes between local state and an environment.

conduit plan [ENVIRONMENT] [options]

Arguments:
  [ENVIRONMENT]         Target environment (default: production)

Options:
  --dags-path, -d <path>   Path to DAG definitions (default: ./dags)
  --output, -o <path>      Save the plan to a file (for later apply)

Saved plans record the environment version they were generated against (base_environment_version); apply rejects a saved plan if the environment has changed since (see Plan & Apply — Conflict Detection).

Example:

conduit plan                             # Plan against production
conduit plan staging                     # Plan against staging
conduit plan production -o plan.json     # Save plan for later apply

apply

Apply a deployment plan to an environment. Executes changed tasks for real, validates their data contracts, and updates the environment's snapshot pointers. Exits non-zero if any task fails, errors, or violates a contract — safe to use as a CI gate.

conduit apply [ENVIRONMENT] [options]

Arguments:
  [ENVIRONMENT]         Target environment (default: production)

Options:
  --dags-path, -d <path>   Path to DAG definitions (default: ./dags)
  --plan-file <path>       Load a saved plan file instead of generating
                           a new one (stale plans are rejected)
  --auto-approve, -y       Skip confirmation prompt
  --full-refresh           Force full refresh on all incremental tasks
  --only <DAG.TASK>        Apply only the named tasks (repeatable).
                           Upstream Execute/Reuse/Remove actions in the
                           same plan are auto-included so dependencies
                           stay consistent

Example:

conduit apply production -y
conduit apply production --plan-file plan.json -y
conduit apply production --only etl.load -y

Environment Commands

env create

Create a new environment.

conduit env create <NAME> [options]

Arguments:
  <NAME>                Environment name

Options:
  --from <name>            Base environment to fork from (default: production)
  --dags-path, -d <path>   Path to DAG definitions (default: ./dags)

env list

List all environments.

conduit env list [options]

Options:
  --dags-path, -d <path>   Path to DAG definitions (default: ./dags)

env promote

Promote one environment into another.

conduit env promote <SOURCE> <TARGET> [options]

Arguments:
  <SOURCE>              Source environment
  <TARGET>              Target environment

env diff

Diff two environments — show added/removed/changed snapshots.

conduit env diff <A> <B> [options]

Arguments:
  <A>                   Left environment (the "from" side)
  <B>                   Right environment (the "to" side)

env history

Show version history for an environment.

conduit env history <NAME> [options]

Arguments:
  <NAME>                Environment name

env rollback

Roll back an environment to a prior history version.

conduit env rollback <NAME> [options]

Arguments:
  <NAME>                Environment name

Options:
  --to-version <n>         Specific version to restore. Defaults to the
                           env's current_version (which restores the state
                           captured before the most recent mutation)
  --yes                    Skip the confirmation prompt

env set-policy

Set or clear the promotion policy on an environment.

conduit env set-policy <NAME> [options]

Arguments:
  <NAME>                Environment name (target of the policy)

Options:
  --require-source <name>  Only allow promotions whose source matches
                           this env name
  --min-age-secs <n>       Newest snapshot in the source must be at
                           least N seconds old
  --clear                  Clear the policy (overrides the other flags)

Example workflow:

conduit env create staging --from production
conduit apply staging -y
conduit env diff staging production
conduit env promote staging production
conduit env history production
conduit env rollback production --yes

Debugging Commands

replay

Replay events to reconstruct historical state.

conduit replay [options]

Options:
  --from <n>               Replay from this sequence number (default: 1)
  --to <n>                 Replay up to this sequence number
  --dags-path, -d <path>   Path to DAG definitions (for resolving state dir)
  --json                   Output the reconstructed state as JSON
  --events-only            Show events only (don't reconstruct state)

Example:

conduit replay                        # Replay the full event log
conduit replay --to 50                # State as of event 50
conduit replay --events-only          # List events without reconstruction

Lineage Commands

lineage extract

Extract SQL lineage for a single task (native JSON output, or an OpenLineage RunEvent under --openlineage).

conduit lineage extract <TASK_REF> [options]

Arguments:
  <TASK_REF>            Task reference in the form dag_id.task_id

Options:
  --dags-path, -d <path>       Path to DAG definitions (default: ./dags)
  --openlineage                Emit an OpenLineage RunEvent instead of
                               Conduit's native lineage JSON
  --output-dataset <name>      OpenLineage output dataset name
  --dataset-namespace <ns>     OpenLineage dataset namespace
  --job-namespace <ns>         OpenLineage job namespace (default: conduit)
  --job-name <name>            OpenLineage job name (default: dag_id.task_id)
  --run-id <uuid>              OpenLineage run UUID (default: generated)
  --event-time <ts>            OpenLineage event timestamp (default: now)
  --event-type <type>          OpenLineage event type (default: COMPLETE)

lineage trace

Trace a column's lineage across task boundaries via the cross-task stitched graph (Python → SQL → Python).

conduit lineage trace --dag <DAG> --column <COLUMN> [options]

Options:
  --dag <dag>              DAG to trace within [required]
  --column <col>           Column to trace, as task_id.column_name [required]
  --dags-path, -d <path>   Path to DAG definitions (default: ./dags)
  --direction <dir>        upstream or downstream (default: upstream)
  --format <fmt>           text or json (default: text)
  --dbt-manifest <path>    dbt target/manifest.json to resolve
                           {{ ref('x') }} / {{ source('s','x') }} against

Example:

conduit lineage extract daily_etl.transform
conduit lineage trace --dag daily_etl --column load.revenue --direction upstream

impact

Schema impact between two DAG versions — diffs task output schemas and traces the downstream blast radius through cross-task lineage. This is the CI gate behind .github/workflows/conduit-impact.yml.

conduit impact [options]

Options:
  --base <ref>             Base side: git ref (git mode; pair with --head)
  --head <ref>             Head side: git ref, or the literal WORKING for
                           the uncommitted working tree
  --base-plan <path>       Base side: compiled plan JSON or DAGs directory
                           (file mode; pair with --head-plan)
  --head-plan <path>       Head side: compiled plan JSON or DAGs directory
  --dags-path <path>       DAGs directory relative to the repo root
                           (git mode only; default: dags)
  --format <fmt>           markdown or json (default: markdown)
  --output <path>          Write the report to this file instead of stdout

Example:

conduit impact --base main --head WORKING
conduit impact --base v1.0 --head v1.1 --format json

Distributed Commands

Distributed execution uses a coordinator (started by conduit run --distributed) and one or more workers connected over gRPC. Remote workers currently execute bash/python tasks; SQL tasks require a local provider registry and fail loudly on remote workers.

worker

Start a distributed worker node.

conduit worker [options]

Options:
  --coordinator, -c <addr>  Coordinator address to connect to
                            (default: localhost:9400)
  --capacity, -n <n>        Maximum concurrent tasks this worker can run
                            (default: 4)
  --pools, -p <pools>       Resource pools this worker handles,
                            comma-separated (default: default)
  --id <id>                 Worker ID (auto-generated if omitted)
  --labels, -l <k=v>        Labels for worker selection (key=value pairs)

cluster status

Show cluster status (workers, running tasks, health).

conduit cluster status [options]

Options:
  --coordinator, -c <addr>  Coordinator address (default: localhost:9400)
  --json                    Output as JSON

cluster drain

Drain a worker (finish current tasks, then stop).

conduit cluster drain <WORKER_ID> [options]

Arguments:
  <WORKER_ID>           Worker ID to drain

Options:
  --coordinator, -c <addr>  Coordinator address (default: localhost:9400)

Example workflow:

# Terminal 1: start a long-lived coordinator run
conduit run daily_etl --distributed --bind 0.0.0.0:9400

# Terminal 2: connect a worker
conduit worker --coordinator localhost:9400 --capacity 8

# Inspect and manage
conduit cluster status
conduit cluster drain worker-1

All three commands fail with a clear error when no coordinator is reachable — there is no simulated output.

API Commands

serve

Start the API server.

conduit serve [options]

Options:
  --host <host>            Host to bind to (default: 0.0.0.0)
  --port, -p <port>        Port to listen on (default: 8080)
  --dags-path, -d <path>   Path to DAG definitions (default: ./dags)
  --state-dir <path>       Path to state directory (default: ./.conduit)
  --auth-enabled           Enable API key authentication
  --cors-origin <origin>   Origin allowed to call the API cross-origin
                           (repeatable; default: same-origin only)
  --demo                   Seed fabricated demo run history (for trying
                           out the UI; never enabled by default)

Example:

conduit serve                            # Serve on 0.0.0.0:8080
conduit serve --port 9090 --auth-enabled
conduit serve --cors-origin http://localhost:3000

See the REST API Reference for the endpoints.

Migration Commands

migrate

Migrate Airflow DAGs to Conduit format.

conduit migrate <SOURCE> [options]

Arguments:
  <SOURCE>              Path to Airflow DAGs directory

Options:
  --output, -o <path>   Output directory for Conduit DAGs (default: ./dags)
  --dry-run             Dry run (show what would be converted)

Example:

conduit migrate ~/airflow/dags --dry-run
conduit migrate ~/airflow/dags --output ./dags

Exit Codes

• 0: Success
• 1: Error (failed task, contract violation, stale plan, compilation error, unreachable coordinator, …) — the message on stderr says which
• 2: Invalid command-line arguments

Common Workflows

Deploy to Production

# 1. Compile locally
conduit compile

# 2. Plan changes (optionally save the plan)
conduit plan production -o plan.json

# 3. Review plan output, then apply exactly what was reviewed
conduit apply production --plan-file plan.json -y

# 4. Verify status
conduit status --env production

Create and Test Staging

# 1. Create staging environment
conduit env create staging --from production

# 2. Make changes
vim dags/etl.py
conduit compile

# 3. Plan and apply
conduit plan staging
conduit apply staging -y

# 4. Test
conduit run daily_etl --env staging

# 5. Promote to production
conduit env promote staging production

Debug a Failed Run

# 1. View status
conduit status

# 2. Replay the event log around the failure
conduit replay --events-only
conduit replay --to 50 --json

# 3. Roll the environment back if a bad apply went out
conduit env history production
conduit env rollback production --yes

REST API Reference

Conduit provides a comprehensive REST API for all operations. Start the API server with conduit serve.

Base URL

http://localhost:8080/api/v1

Authentication

API-key authentication is built in but disabled by default. Enable it with:

conduit serve --auth-enabled

On first start with auth enabled, a bootstrap admin key is created and printed once. Send the key on every request:

Authorization: Bearer <api-key>

Keys are managed via POST/GET /auth/keys, GET/DELETE /auth/keys/{key_id}, and GET /auth/me. When auth is disabled, all endpoints are public — secure them at the network/infrastructure level.

Response Format

Success responses are plain endpoint-specific JSON (no envelope) — see each endpoint below.

Errors share one shape:

{
  "error": {
    "type": "not_found",
    "message": "DAG 'daily_etl' not found"
  }
}

Health Endpoints

GET /health

Health check.

Response:

{
  "status": "ok",
  "service": "conduit",
  "version": "0.2.0"
}

GET /metrics (root path, not under /api/v1)

Prometheus metrics, served at http://localhost:8080/metrics. (GET /api/v1/metrics is a different endpoint — it lists task metrics.)

Response:

# HELP conduit_dags_total Total DAGs compiled
# TYPE conduit_dags_total counter
conduit_dags_total 5

DAG Endpoints

GET /dags

List all compiled DAGs. Compiles the DAGs directory on each call; no query parameters.

Response:

{
  "dags": [
    {
      "id": "daily_etl",
      "name": "daily_etl",
      "description": "Daily ETL pipeline",
      "schedule": "0 2 * * *",
      "tags": [],
      "taskCount": 3,
      "sourceFile": "dags/daily_etl.yaml"
    }
  ],
  "total": 5
}

GET /dags/

Get DAG details.

Response:

{
  "id": "daily_etl",
  "name": "daily_etl",
  "description": "Daily ETL pipeline",
  "schedule": "0 2 * * *",
  "tags": [],
  "maxActiveRuns": 1,
  "taskCount": 2,
  "sourceFile": "dags/daily_etl.yaml",
  "executionOrder": ["extract", "transform"],
  "tasks": [
    {
      "id": "extract",
      "name": "extract",
      "type": "Python",
      "dependencies": [],
      "retries": 1,
      "retryDelay": null,
      "pool": null,
      "timeout": 300,
      "priority": 0,
      "triggerRule": "AllSuccess"
    }
  ]
}

POST /dags/compile

Recompile the whole DAGs directory. No request body.

Response:

{
  "success": true,
  "dagsCompiled": 5,
  "tasksTotal": 32,
  "errors": [],
  "warnings": [],
  "durationMs": 45
}

Run Endpoints

POST /dags/{dag_id}/runs

Start a new DAG run. The run is dispatched to the scheduler which coordinates task execution via the executor. Task state changes are broadcast over WebSocket in real-time.

Request (all fields optional):

{
  "logical_date": "2024-03-22T14:00:00Z",
  "config": { "batch_size": "1000" },
  "environment": "staging"
}

environment defaults to "production" and is recorded on the run and threaded into task execution context.

Response:

{
  "runId": "run_daily_etl_20240322_143210_123",
  "dagId": "daily_etl",
  "environment": "staging",
  "status": "dispatched",
  "taskStates": {
    "extract": "pending",
    "transform": "pending",
    "load": "pending"
  },
  "message": "DAG run 'run_daily_etl_20240322_143210_123' dispatched to scheduler (3 tasks)"
}

If no scheduler is attached, status will be "queued" instead of "dispatched".

GET /runs

List recent runs across all DAGs. (GET /dags/{dag_id}/runs returns the same shape scoped to one DAG.)

Query Parameters:

• dag_id (optional): Filter by DAG
• status (optional): Filter by status (pending, running, success, failed)
• environment (optional): Filter by environment
• limit (optional): Max results (default: 100)

Response:

{
  "runs": [
    {
      "id": "run_abc123def456",
      "dagId": "daily_etl",
      "status": "success",
      "startedAt": "2024-03-22T14:32:10Z",
      "endedAt": "2024-03-22T14:37:45Z",
      "taskStates": { "extract": "success" },
      "taskLogs": {},
      "triggeredBy": "api",
      "environment": "production"
    }
  ],
  "total": 142
}

GET /runs/

Get run details. Captured task output (truncated stdout/stderr per task) is returned in taskLogs — there is no separate log-streaming endpoint; live events stream over the WebSocket (see below).

Response:

{
  "id": "run_abc123def456",
  "dagId": "daily_etl",
  "status": "success",
  "startedAt": "2024-03-22T14:32:10Z",
  "endedAt": "2024-03-22T14:37:45Z",
  "taskStates": {
    "extract": "success",
    "transform": "success"
  },
  "taskLogs": {
    "extract": "Extracting data from API...\n1000 rows written"
  },
  "triggeredBy": "api",
  "environment": "production"
}

Environment Endpoints

GET /environments

List all environments.

Response:

{
  "environments": [
    {
      "id": "production",
      "name": "production",
      "snapshotCount": 32,
      "updatedAt": "2024-03-22T14:32:00Z",
      "basedOn": null,
      "currentVersion": 4,
      "promotionPolicy": {
        "requireSource": null,
        "minAgeSecs": null
      }
    }
  ]
}

POST /environments

Create a new environment.

Request:

{
  "name": "staging",
  "based_on": "production"
}

Response:

{
  "id": "staging",
  "name": "staging",
  "snapshotCount": 32,
  "basedOn": "production",
  "message": "Environment 'staging' created"
}

GET /environments/

Get environment details, including its per-task snapshot pointers.

Response:

{
  "id": "production",
  "name": "production",
  "snapshotCount": 32,
  "updatedAt": "2024-03-22T14:32:00Z",
  "basedOn": null,
  "currentVersion": 4,
  "promotionPolicy": {
    "requireSource": null,
    "minAgeSecs": null
  },
  "snapshots": [
    {
      "dagId": "daily_etl",
      "taskId": "extract",
      "snapshotId": "snap_extract_20240322143215123"
    }
  ]
}

POST /environments/promote

Promote one environment's state to another. Source and target are given in the body, not the path.

Request:

{
  "source": "staging",
  "target": "production"
}

Response:

{
  "source": "staging",
  "target": "production",
  "snapshotChanges": 4,
  "message": "Promoted 'staging' -> 'production' (4 snapshot changes)"
}

POST /environments/{env_name}/rollback

Roll an environment back to a previous recorded version (see GET /environments/{env_name}/history). Omit to_version to roll back one step.

Request:

{
  "to_version": 3
}

Response:

{
  "environment": "production",
  "rolledBackTo": 3,
  "newVersion": 5,
  "snapshotChanges": 2,
  "message": "Rolled back 'production' (new version 5, 2 snapshot changes)"
}

Plan/Apply Endpoints

POST /plan

Generate a deployment plan against an environment's current state. Generated plans are cached server-side (in-memory, most recent 50) so a later POST /apply can apply exactly the plan that was reviewed, by plan_id. Cached plans do not survive a server restart — regenerate if in doubt.

Request:

{
  "environment": "production"
}

Response:

{
  "plan_id": "plan_abc123",
  "environment": "production",
  "created_at": "2024-03-22T14:52:00Z",
  "actions": [
    {
      "dag_id": "daily_etl",
      "task_id": "extract",
      "action": "Execute",
      "reason": "fingerprint changed",
      "fingerprint": "f1a2b3c4d5e6"
    }
  ],
  "stats": {
    "total_tasks": 32,
    "to_execute": 2,
    "to_reuse": 30,
    "to_skip": 0,
    "to_remove": 0,
    "critical_path_depth": 3,
    "blast_radius": 2
  },
  "compilation": {
    "dags_compiled": 5,
    "tasks_total": 32,
    "duration_ms": 45
  }
}

POST /apply

Execute a deployment plan synchronously: each Execute action runs for real (through the provider registry for SQL tasks), contracts are validated against the emitted evidence, snapshots are stored, and the environment is updated with a history-recorded, rollbackable version bump.

With plan_id, the cached plan is applied only if it still matches reality:

• 404 not_found — unknown or expired plan_id
• 400 bad_request — plan targets a different environment than requested
• 409 conflict — stale plan: the environment's version has moved since the plan was generated; regenerate the plan
• 422 apply_failed — a task failed, errored, or violated a contract; the environment is not updated

Without plan_id, a fresh plan is generated against current state and applied immediately.

Request:

{
  "plan_id": "plan_abc123",
  "environment": "production"
}

Response:

{
  "plan_id": "plan_abc123",
  "environment": "production",
  "status": "applied",
  "tasks_executed": 2,
  "tasks_reused": 30,
  "tasks_removed": 0,
  "environment_version": 4
}

If there is nothing to execute or remove, status is "noop" and the environment is left untouched.

Lineage Endpoints

POST /lineage/sql

Analyze SQL for column-level lineage. Optionally provide inline table schemas for bare column resolution and wildcard expansion. When openlineage metadata is provided, the response also includes an OpenLineage RunEvent with a columnLineage output dataset facet.

Request:

{
  "sql": "SELECT c.id, c.name, SUM(o.amount) as total FROM customers c JOIN orders o ON c.id = o.customer_id GROUP BY c.id, c.name",
  "source_task_id": "daily_customer_totals",
  "tables": [
    {
      "table": "customers",
      "columns": [
        { "name": "id", "data_type": "integer" },
        { "name": "name", "data_type": "string" }
      ]
    },
    {
      "table": "orders",
      "columns": [
        { "name": "customer_id", "data_type": "integer" },
        { "name": "amount", "data_type": "float" }
      ]
    }
  ],
  "openlineage": {
    "output_dataset": "analytics.customer_totals",
    "dataset_namespace": "warehouse",
    "job_namespace": "conduit",
    "job_name": "daily_etl.daily_customer_totals",
    "run_id": "550e8400-e29b-41d4-a716-446655440000",
    "event_type": "COMPLETE"
  }
}

Response:

{
  "source_task_id": "daily_customer_totals",
  "catalog_used": true,
  "output_columns": [
    {
      "name": "id",
      "expression": "c.id",
      "is_computed": false
    },
    {
      "name": "total",
      "expression": "sum(o.amount)",
      "is_computed": true
    }
  ],
  "source_tables": [
    { "name": "customers", "alias": "c", "schema": null },
    { "name": "orders", "alias": "o", "schema": null }
  ],
  "column_mappings": [
    {
      "output": "total",
      "inputs": [{ "task_id": "orders", "column_name": "amount" }]
    }
  ],
  "openlineage": {
    "eventType": "COMPLETE",
    "run": { "runId": "550e8400-e29b-41d4-a716-446655440000" },
    "job": { "namespace": "conduit", "name": "daily_etl.daily_customer_totals" },
    "inputs": [
      { "namespace": "warehouse", "name": "customers" },
      { "namespace": "warehouse", "name": "orders" }
    ],
    "outputs": [
      {
        "namespace": "warehouse",
        "name": "analytics.customer_totals",
        "facets": {
          "columnLineage": {
            "fields": {
              "total": {
                "inputFields": [
                  {
                    "namespace": "warehouse",
                    "name": "orders",
                    "field": "amount"
                  }
                ]
              }
            }
          }
        }
      }
    ]
  }
}

POST /lineage/catalog/refresh

Refresh the schema catalog by introspecting connected providers.

Response:

{
  "status": "refreshed",
  "tables_cataloged": 42,
  "providers_queried": 3
}

POST /lineage/trace/upstream

Get upstream lineage.

Request:

{
  "table": "analytics.metrics.customer_metrics",
  "column": "customer_ltv",
  "depth": 10
}

Response:

{
  "column": "customer_ltv",
  "sources": [
    {
      "table": "warehouse.transformed.customer_summary",
      "column": "total_spent",
      "task": "daily_etl.aggregate_metrics",
      "dag": "daily_etl"
    }
  ]
}

POST /lineage/trace/downstream

Get downstream lineage.

Request:

{
  "table": "warehouse.raw.transactions",
  "column": "amount",
  "depth": 10
}

Response:

{
  "column": "amount",
  "targets": [
    {
      "table": "analytics.metrics.customer_metrics",
      "column": "customer_ltv",
      "task": "daily_etl.create_metrics",
      "dag": "daily_etl"
    }
  ]
}

POST /lineage/graph

Get complete lineage graph.

Request:

{
  "start_table": "analytics.metrics.customer_metrics",
  "start_column": "customer_ltv",
  "direction": "both"
}

Response:

{
  "nodes": [
    {
      "id": "analytics.metrics.customer_ltv",
      "type": "column",
      "table": "analytics.metrics.customer_metrics"
    }
  ],
  "edges": [
    {
      "source": "analytics.metrics.customer_ltv",
      "target": "warehouse.transformed.customer_summary.total_spent"
    }
  ]
}

Event Endpoints

GET /events

Query event log.

Query Parameters:

• from / to (optional): Sequence number range (inclusive)
• event_type (optional): Filter to one event type (e.g. TaskFailed, DagRunCompleted)
• run_id, dag_id, task_id (optional): Scope to a run, DAG, or task
• limit (optional): Max results (default: 100)

Response:

{
  "events": [
    {
      "sequence": 12345,
      "type": "TaskCompleted",
      "run_id": "run_abc123",
      "task_id": "extract",
      "timestamp": "2024-03-22T14:32:45Z",
      "data": {
        "duration_ms": 35000,
        "exit_code": 0
      }
    }
  ],
  "total": 5432
}

WebSocket GET /ws/events

Stream events in real-time. Note the path is /ws/events, outside the /api/v1 prefix.

Subscribe:

const ws = new WebSocket('ws://localhost:8080/ws/events');
ws.addEventListener('message', (event) => {
  const data = JSON.parse(event.data);
  console.log(`${data.type}: ${data.task_id}`);
});

Messages:

{
  "type": "TaskStarted",
  "run_id": "run_abc123",
  "task_id": "extract",
  "timestamp": "2024-03-22T14:32:10Z"
}

Error Codes

Errors are returned as {"error": {"type": "...", "message": "..."}}:

Rate-limited requests receive 429 Too Many Requests.

Rate Limiting

Requests are rate limited per client IP: 10 requests/second with a burst capacity of 50. Exceeding the limit returns 429 Too Many Requests.

Pagination

List endpoints accept a limit query parameter:

GET /runs?limit=50       # Most recent 50 runs
GET /events?limit=200    # Most recent 200 events

Other Endpoints

Also routed but not detailed here (see GET /api/v1/docs for the live OpenAPI spec and Swagger UI):

• GET /api/v1/info — system info
• POST/GET /api/v1/auth/keys, GET/DELETE /api/v1/auth/keys/{key_id}, GET /api/v1/auth/me — API-key management
• GET /api/v1/dags/{dag_id}/graph — DAG graph for visualization
• GET /api/v1/environments/{env_name}/diff/{other_env}, GET .../history, GET .../history/{version}, PUT .../policy — environment diff, history, and promotion policy
• POST /api/v1/lineage/schema/diff, POST /api/v1/lineage/contracts/validate — schema diff and contract validation
• POST /api/v1/openlineage/v1/lineage, GET /api/v1/openlineage/events, GET /api/v1/openlineage/datasets/{namespace}/{name}, GET /api/v1/openlineage/stats, GET /api/v1/lineage/datasets/{namespace}/{name}/unified, GET /api/v1/lineage/cache/stats, POST /api/v1/lineage/cache/invalidate — OpenLineage ingest and unified dataset views
• GET /api/v1/contracts, GET /api/v1/contracts/{dag_id}, GET /api/v1/contracts/{dag_id}/{task_id} — contract inventory
• GET /api/v1/metrics, GET /api/v1/metrics/{dag_id}/{task_id} — task metrics
• GET /api/v1/connections, GET /api/v1/connections/providers, GET /api/v1/connections/{name}, POST /api/v1/connections/{name}/test — provider connections
• POST /api/v1/backfill — backfill runs
• GET /api/v1/cluster/status, POST /api/v1/cluster/workers/{id}/drain — distributed cluster operations

Next Steps

• Python SDK: Use Python client library
• CLI Reference: Command-line access
• Architecture: How API integrates with system

Python SDK Guide

The Conduit Python SDK provides decorators and utilities for defining DAGs and tasks. All code is statically analyzed via tree-sitter, not executed.

Installation

The SDK is bundled with Conduit. Import it:

from conduit.sdk import dag, task, Task, DAG

Basic Usage

Defining a DAG

from conduit.sdk import dag, task

@task
def extract():
    print("Extracting data...")
    return "data.csv"

@task
def transform(data):
    print(f"Transforming {data}...")
    return "clean.csv"

@task
def load(data):
    print(f"Loading {data}...")
    return "success"

@dag(schedule="0 2 * * *")
def etl_pipeline():
    """Daily ETL pipeline."""
    raw = extract()
    clean = transform(raw)
    result = load(clean)
    return result

Task Decorators

All functions decorated with @task or @dag must be defined at module level (not nested).

# Valid
@task
def my_task():
    pass

@dag
def my_dag():
    my_task()

# Invalid - nested
def outer():
    @task
    def inner():  # ERROR: nested task
        pass

Task Types

Python Tasks

Default task type, executes Python code:

@task(timeout=300)
def python_task():
    import pandas as pd
    df = pd.read_csv("data.csv")
    return len(df)

Shell Tasks

Execute bash commands:

from conduit.sdk import shell_task

@shell_task(timeout=600)
def bash_task():
    set -e  # Exit on error
    aws s3 ls s3://my-bucket/
    dbt run --profiles-dir /etc/dbt

SQL Tasks

Execute SQL against configured data warehouse:

from conduit.sdk import sql_task

@sql_task(dialect="postgres")
def create_table():
    CREATE TABLE staging_users AS
    SELECT * FROM raw_users
    WHERE created_at >= current_date - interval '1 day'

@sql_task(dialect="postgres")
def parametrized_query(start_date: str, limit: int):
    SELECT * FROM transactions
    WHERE date >= '{start_date}'
    LIMIT {limit}

Sensor Tasks

Poll until condition is met:

from conduit.sdk import sensor_task
import os
import time

@sensor_task(timeout=3600, poke_interval=60)
def wait_for_file():
    if os.path.exists("/data/export.csv"):
        return True  # Unblock
    return False  # Retry in 60s

@sensor_task(timeout=7200, poke_interval=300)
def wait_for_api():
    import requests
    response = requests.get("https://api.example.com/status")
    return response.json()["ready"] == True

Task Configuration

from conduit.sdk import task, Pool, TriggerRule

@task(
    timeout=300,                              # Timeout in seconds
    retries=2,                                # Retry count
    retry_delay=60,                           # Initial retry delay (seconds)
    retry_exponential_base=2,                 # Exponential backoff multiplier
    pool=Pool.name("api_calls", size=5),     # Concurrency pool
    tags=["production", "critical"],          # Metadata tags
    trigger_rule=TriggerRule.ALL_SUCCESS,    # When to run
)
def configured_task():
    print("Task with full configuration")

Pools

Limit concurrent task execution:

from conduit.sdk import Pool

api_pool = Pool.name("api_requests", size=3)

@task(pool=api_pool)
def api_call_1():
    requests.get("https://api.example.com/data")

@task(pool=api_pool)
def api_call_2():
    requests.get("https://api.example.com/users")

@task(pool=api_pool)
def api_call_3():
    requests.get("https://api.example.com/events")

# Only 3 of these run in parallel

Trigger Rules

Control when downstream tasks execute:

from conduit.sdk import TriggerRule

@task
def may_fail():
    import random
    if random.random() < 0.5:
        raise Exception("Random failure")
    return "success"

@task(trigger_rule=TriggerRule.ALL_SUCCESS)
def run_on_success():
    # Runs only if may_fail succeeded
    print("May fail succeeded")

@task(trigger_rule=TriggerRule.ALL_DONE)
def run_always():
    # Runs regardless of may_fail status
    print("May fail completed (success or failure)")

@task(trigger_rule=TriggerRule.ONE_FAILED)
def run_on_failure():
    # Runs only if may_fail failed
    print("May fail failed")

@dag
def conditional_dag():
    result = may_fail()
    run_on_success(result)
    run_always(result)
    run_on_failure(result)

Data Exchange (XCom)

Implicit Returns

Return values flow automatically:

@task
def extract():
    return {"count": 1000, "file": "data.csv"}

@task
def transform(data):
    # data = {"count": 1000, "file": "data.csv"}
    print(f"Processing {data['file']}")
    return {"clean_count": 950, "file": "clean.csv"}

@task
def load(data):
    # data = {"clean_count": 950, "file": "clean.csv"}
    print(f"Loading {data['count']} rows")

@dag
def etl():
    extracted = extract()
    transformed = transform(extracted)
    load(transformed)

Explicit XCom Output

For structured logging:

@task
def extract():
    print("xcom|row_count|1000")
    print("xcom|file_size|2.5GB")
    print("Starting extraction...")
    return "data.csv"

@dag
def my_dag():
    data = extract()
    # XCom values: row_count=1000, file_size="2.5GB"

Dependency Resolution

Dependencies are inferred from function arguments:

@task
def a(): return "a"
@task
def b(): return "b"
@task
def c(in_a, in_b): return f"{in_a}{in_b}"
@task
def d(in_c): return f"final: {in_c}"

@dag
def complex_dag():
    output_a = a()
    output_b = b()
    # c depends on both a and b
    output_c = c(output_a, output_b)
    # d depends on c
    output_d = d(output_c)
    return output_d

Tree-sitter parses the function calls and builds the dependency graph automatically.

Task Context

Access information about the current run:

from conduit.sdk import TaskContext

@task
def contextual_task():
    ctx = TaskContext()
    print(f"Run ID: {ctx.run_id}")
    print(f"Task ID: {ctx.task_id}")
    print(f"Environment: {ctx.environment}")
    print(f"Attempt: {ctx.attempt}")  # 1, 2, 3... on retries
    return "done"

Available context:

• ctx.run_id — Unique run ID
• ctx.task_id — Task ID within DAG
• ctx.dag_id — DAG ID
• ctx.environment — Environment name
• ctx.attempt — Attempt number (1-based)
• ctx.xcom_pull(key) — Retrieve XCom from previous task

Advanced Patterns

Shared Utilities

Keep utilities in tasks/ directory:

# tasks/api.py
import requests

def fetch_json(url):
    return requests.get(url).json()

def batch_api_calls(urls, batch_size=5):
    for i in range(0, len(urls), batch_size):
        batch = urls[i:i+batch_size]
        yield [fetch_json(url) for url in batch]

Import in DAGs:

# dags/etl.py
from conduit.sdk import dag, task
from tasks.api import batch_api_calls

@task
def fetch_data():
    urls = ["https://api.example.com/page/1", ...]
    for batch in batch_api_calls(urls):
        print(f"Fetched {len(batch)} items")
    return "done"

@dag
def my_dag():
    fetch_data()

Conditional DAGs

Use Python conditionals for dynamic DAGs:

import os

@task
def check_env():
    return os.getenv("ENVIRONMENT", "dev")

@task
def prod_task():
    return "production path"

@task
def dev_task():
    return "development path"

@task
def finalize(result):
    return f"done: {result}"

@dag
def conditional_dag():
    env = check_env()

    # This is evaluated at COMPILE time
    if env == "production":
        result = prod_task()
    else:
        result = dev_task()

    finalize(result)

Important: This DAG is evaluated at compile time, not runtime. If you need runtime branching, use trigger rules instead.

Parameterized DAGs

Pass parameters via environment variables:

import os

@task
def extract():
    limit = int(os.getenv("EXTRACTION_LIMIT", "1000"))
    print(f"Extracting {limit} rows...")
    return "data.csv"

@dag
def configurable_etl():
    extract()

Run with parameters:

EXTRACTION_LIMIT=5000 conduit run configurable_etl

Dynamic Task Graphs

Generate multiple tasks dynamically:

@task
def extract(partition):
    print(f"Extracting partition {partition}")
    return f"data_{partition}.csv"

@task
def transform(data):
    return f"clean_{data}"

@dag
def multi_partition_etl():
    partitions = ["2024-01", "2024-02", "2024-03"]

    # This is evaluated at COMPILE time
    extracted = [extract(p) for p in partitions]
    transformed = [transform(e) for e in extracted]

    return transformed

This creates 6 tasks total (3 extract + 3 transform).

Error Handling

Task-Level Retries

Configured via task decorator:

@task(retries=3, retry_delay=60)
def flaky_api():
    requests.get("https://flaky-api.example.com")

DAG-Level Handling

Use trigger rules for error handling:

@task
def critical_task():
    if something_wrong:
        raise Exception("Critical failure")

@task(trigger_rule=TriggerRule.ALL_DONE)
def notify_ops(result):
    # Runs regardless of critical_task status
    print(f"Task completed: {result}")
    # Check actual status and alert if needed

Logging Best Practices

Standard Output

Everything printed goes to logs:

@task
def logged_task():
    print("INFO: Starting extraction")
    print("ERROR: Connection failed")  # Treated as regular output
    return "result"

Structured Logging

Use XCom for metrics:

@task
def extract():
    rows = 1000
    size_mb = 2.5

    print("xcom|rows_extracted|1000")
    print("xcom|size_mb|2.5")
    print(f"Extracted {rows} rows ({size_mb}MB)")

    return "data.csv"

SDK Functions

Import All

from conduit.sdk import (
    dag,
    task,
    shell_task,
    sql_task,
    sensor_task,
    executable_task,
    DAG,
    Task,
    TaskContext,
    Pool,
    TriggerRule,
    Lineage,
    SchemaContract,
    DataContract,
)

Type Hints (Optional)

Conduit doesn't enforce types, but hints are allowed:

from typing import Dict, List

@task
def extract() -> Dict[str, int]:
    return {"count": 1000}

@task
def transform(data: Dict) -> List[str]:
    return ["a", "b", "c"]

@dag
def typed_dag():
    d = extract()
    transform(d)

Type hints are ignored during compilation and execution. They're purely for IDE assistance.

Debugging

Print Debugging

Standard Python print works:

@task
def debug_task():
    x = 42
    print(f"DEBUG: x = {x}")
    return x

Output appears in logs.

Introspection

Use TaskContext for debugging:

@task
def introspect():
    ctx = TaskContext()
    print(f"Run: {ctx.run_id}")
    print(f"Attempt: {ctx.attempt}")
    return "done"

Performance Notes

• Compilation: Tree-sitter parses in milliseconds
• Task execution: Standard Python execution, no overhead
• Data transfer: XCom is serialized as JSON
• Memory: Large return values are written to disk automatically

Next Steps

• DAG Concepts: Advanced DAG patterns
• Python Task Examples: Real-world examples
• REST API: Programmatic API access

Providers & Connections

Conduit ships with a typed provider system: every connector implements a category trait (SqlProvider, StorageProvider, HttpProvider, StreamProvider, SaasProvider, DocumentProvider) on top of the base Provider trait. Configure connections in conduit.yaml and validate them with conduit test-connection <name>.

32 provider types: 12 production, 20 experimental. Experimental providers expose the trait interface but their operations return NotImplemented; conduit compile warns when a DAG routes through one. This table is generated from each provider's ProviderInfo.is_stub flag — it cannot drift from the code.

SQL Databases

Object Storage

HTTP / Webhooks

Streaming

SaaS Platforms

Document / NoSQL

Configuration

Connections live under connections: in conduit.yaml. Secrets can be injected from environment variables or a secrets backend rather than committed inline. Example:

connections:
  warehouse:
    type: postgres
    host: db.internal
    port: 5432
    database: analytics
    # credentials resolved from CONDUIT_CONN_WAREHOUSE or a secrets backend

Validate connectivity before running pipelines:

conduit test-connection warehouse

System Architecture

This document describes Conduit's internal architecture, data flow, and design decisions.

Crate Layout

Conduit is organized into 11 specialized crates plus 3 optional extension crates:

conduit/
├── conduit-common/        Shared types, errors, events, config, contracts
├── conduit-compiler/      Tree-sitter DAG parser + Kahn's algorithm
├── conduit-state/         RocksDB event store + snapshots + environments
├── conduit-scheduler/     Tokio-based event loop + task scheduling
├── conduit-executor/      Process isolation + task runtime
├── conduit-planner/       Fingerprint diffing + impact analysis
├── conduit-lineage/       SQL parsing + column-level lineage
├── conduit-api/           REST API (37 endpoints) + WebSocket
├── conduit-providers/     32-provider ecosystem + secrets backends
├── conduit-distributed/   Leader-worker distributed execution (gRPC)
├── conduit-cli/           CLI entry point
│
├── conduit-bench/         (excluded) Benchmarks
├── conduit-python/        (excluded) PyO3 native bindings (the SDK is sdk/python)
└── conduit-wasm/          (excluded) WebAssembly target

Each crate is independently testable and has a single responsibility.

Data Flow: Compile → Schedule → Execute → Store

graph LR
    A["Python DAG Files"]
    B["conduit-compiler<br/>tree-sitter"]
    C["Compiled DAG<br/>JSON"]
    D["conduit-state<br/>Event Store"]
    E["conduit-scheduler<br/>Tokio"]
    F["conduit-executor<br/>Process Pool"]
    G["Event Log<br/>RocksDB"]

    A -->|parse| B
    B -->|generate| C
    C -->|create snapshot| D
    D -->|schedule runs| E
    E -->|execute tasks| F
    F -->|emit events| G
    G -->|drive scheduler| E

1. Compilation Phase (conduit-compiler)

Input: Python source files with @dag and @task decorators

Process:

1. Tree-sitter parsing: Extract AST without executing Python
2. Function extraction: Find all @task and @dag definitions
3. Dependency resolution: Build call graph using Kahn's algorithm
4. Validation: Detect cycles, missing tasks, circular dependencies
5. Fingerprint generation: Compute content-addressable hash

Output: Compiled DAG structure

#![allow(unused)]
fn main() {
pub struct CompiledDAG {
    pub dag_id: String,
    pub tasks: Vec<CompiledTask>,
    pub dependencies: HashMap<String, Vec<String>>,
    pub fingerprint: String,
    pub schedule: Option<String>,
    pub created_at: DateTime,
}
}

2. Snapshot Creation (conduit-state)

Input: Compiled DAG from compiler

Process:

1. Fingerprint indexing: Map fingerprint → compiled task
2. Snapshot serialization: Convert compiled DAG to JSON
3. Content addressing: Store by fingerprint, not by name
4. Reuse detection: Check if fingerprint already exists

Output: Immutable snapshot ID

#![allow(unused)]
fn main() {
pub struct Snapshot {
    pub snapshot_id: String,
    pub fingerprint: String,
    pub tasks: HashMap<String, CompiledTask>,
    pub size_bytes: usize,
    pub created_at: DateTime,
}
}

3. Environment Pointer (conduit-state)

Input: Snapshot ID

Process:

1. Create environment record: Link name → snapshot
2. Schedule lookup: Get cron schedule for DAG
3. Trigger registration: Register event triggers

Output: Environment pointer

#![allow(unused)]
fn main() {
pub struct Environment {
    pub name: String,
    pub snapshot_id: String,
    pub created_at: DateTime,
    pub schedules: HashMap<String, String>, // dag_id → cron
}
}

4. Scheduling Phase (conduit-scheduler)

Input: Event log (DAG compiled, environment created)

Process:

1. Event observation: Tokio channels watch for state changes
2. Cron evaluation: For each scheduled DAG, check if should run
3. Trigger rule evaluation: Check upstream task statuses
4. Pool enforcement: Respect concurrency limits
5. Task queuing: Insert into run queue with dependencies

Output: Scheduled run

#![allow(unused)]
fn main() {
pub struct ScheduledRun {
    pub run_id: String,
    pub dag_id: String,
    pub scheduled_time: DateTime,
    pub tasks: Vec<ScheduledTask>,
}
}

5. Execution Phase (conduit-executor / conduit-distributed)

Input: Scheduled task

Local execution (conduit-executor):

1. Isolation: Spawn child process with isolated environment
2. Input injection: Pass task inputs via stdin JSON
3. Timeout enforcement: Kill process if exceeds limit
4. Protocol parsing: Read stdout for XCOM/LOG/PROGRESS/METRIC
5. Retry logic: On failure, schedule retry with backoff
6. Event emission: Emit TaskStarted, TaskCompleted, etc.

Distributed execution (conduit-distributed):

1. Task routing: Coordinator assigns task to available worker via gRPC
2. Pool affinity: Match task pool requirements to worker capabilities
3. Remote execution: Worker executes task with same protocol parsing
4. Result forwarding: Worker reports result back through coordinator
5. Health monitoring: Heartbeat-based worker health tracking
6. Reassignment: Dead worker tasks are automatically requeued

Output: Task completion event

#![allow(unused)]
fn main() {
pub struct TaskCompletion {
    pub task_id: String,
    pub run_id: String,
    pub status: TaskStatus, // success | failed | skipped
    pub exit_code: i32,
    pub duration_ms: u64,
    pub xcom: HashMap<String, String>,
}
}

6. Event Storage (conduit-state)

Input: Task completion event

Process:

1. Event serialization: Convert to JSON
2. Write-ahead logging: Pre-write to RocksDB log
3. Monotonic sequencing: Assign event sequence number
4. Snapshot materialization: Update cached DAG state

Output: Durable event in append-only log

Provider Ecosystem (conduit-providers)

Conduit supports 32 data providers across 6 categories, all implementing a unified async trait hierarchy:

#![allow(unused)]
fn main() {
#[async_trait]
pub trait Provider: Send + Sync {
    fn info(&self) -> ProviderInfo;
    async fn test_connection(&self) -> Result<ConnectionTestResult, ProviderError>;
    async fn close(&self) -> Result<(), ProviderError>;
}
}

Specialized traits extend the base: SqlProvider, StorageProvider, HttpProvider, StreamProvider, SaasProvider, and DocumentProvider.

Provider Categories

SQL Databases (12): PostgreSQL, MySQL, SQLite, DuckDB, Snowflake, BigQuery, Redshift, Databricks, ClickHouse, Oracle, SQL Server, TimescaleDB, CockroachDB

Object Storage (3): AWS S3, Google Cloud Storage, Azure Blob Storage

HTTP/API (3): Generic REST, GraphQL, Webhook

Streaming (4): Apache Kafka, RabbitMQ, AWS Kinesis, Google Pub/Sub, Redis Streams

SaaS (5): Salesforce, Stripe, Slack, GitHub, Twilio

Document/NoSQL (6): MongoDB, DynamoDB, Cassandra, Elasticsearch, Redis, Neo4j

Secrets Management

The provider registry integrates with pluggable secrets backends for credential resolution, supporting environment variables, file-based secrets, and external vaults.

Distributed Execution (conduit-distributed)

Architecture

  ┌─────────────────────────────────────────┐
  │  Coordinator (leader node)              │
  │  ├── WorkerPool (tracks workers)        │
  │  ├── Task queue (pending assignments)   │
  │  └── gRPC server (:9400)                │
  └────────────┬───────────────┬────────────┘
               │ gRPC          │ gRPC
         ┌─────▼─────┐  ┌─────▼─────┐
         │ Worker-1  │  │ Worker-2  │
         │ cap: 4    │  │ cap: 8    │
         │ pool: gpu │  │ pool: *   │
         └───────────┘  └───────────┘

gRPC Protocol (5 RPCs)

1. Register: Worker joins the cluster with capacity and pool affinity
2. ReportResult: Worker sends task completion with metrics and XCom
3. Heartbeat: Bidirectional health check and directive delivery
4. StreamLogs: Real-time log streaming from workers to coordinator
5. ClusterStatus: Query cluster health, worker states, pending/inflight counts

Routing Strategies

The WorkerPool supports three task routing strategies:

• LeastLoaded: Assign to the worker with the fewest active tasks (default)
• BinPack: Fill workers to capacity before using the next
• RoundRobin: Cycle through workers in order

Execution Modes

• Local: All tasks run on the scheduler node (single-node, backward compatible)
• Distributed: All tasks routed through the coordinator to remote workers
• Hybrid: Tasks with pool affinity go to workers; others run locally

Health Monitoring

Workers send periodic heartbeats. The coordinator tracks three states:

• Active: Heartbeat received within 30 seconds
• Disconnected: No heartbeat for 30–120 seconds (no new tasks assigned)
• Dead: No heartbeat for 120+ seconds (tasks reassigned to healthy workers)

Data Contracts (conduit-common)

Data contracts provide schema validation, SLA enforcement, and quality checks:

#![allow(unused)]
fn main() {
pub struct DataContract {
    pub checks: Vec<ContractCheck>,
    pub severity: Severity, // Warning | Error | Critical
}
}

Contract evaluators validate data at DAG boundaries, with results surfaced in the UI and API.

Core Concepts

Fingerprints

A fingerprint is a deterministic hash of a task and all upstream dependencies:

Task: transform(data)
  Code hash: sha256(function_body) = abc123
  Timeout: 600 seconds
  Retries: 2
  Upstream fingerprints: [extract_task_fingerprint]

Computed fingerprint: sha256(abc123 + "600" + "2" + upstream_fingerprints) = f1a2b3c4d5e6

Properties:

• Deterministic: Same code always produces same fingerprint
• Cascading: If upstream fingerprint changes, downstream cascades
• Content-addressable: Fingerprint is the stable identifier, not task name

Snapshots

A snapshot is an immutable, versioned collection of compiled DAGs:

Snapshot: prod-snap-20240322-143215
  Tasks:
    - daily_analytics_etl.extract (fingerprint: f1a...)
    - daily_analytics_etl.transform (fingerprint: g2h...)
    - daily_analytics_etl.load (fingerprint: m3n...)
  Size: 1.2 KB
  Created: 2024-03-22 14:32:15 UTC

Snapshots are immutable once created. To deploy changes, create a new snapshot.

Environments

An environment is a pointer to a snapshot plus metadata:

Environment: production
  Snapshot: prod-snap-20240322-143215
  Schedules:
    - daily_analytics_etl: "0 2 * * *"
  Run history: [run1, run2, run3, ...]

Promoting an environment is just changing the pointer: O(1), zero data copy.

Events

An event is an immutable record of something that happened:

#![allow(unused)]
fn main() {
pub enum Event {
    DAGCompiled {
        dag_id: String,
        fingerprint: String,
        timestamp: DateTime,
    },
    TaskStarted {
        run_id: String,
        task_id: String,
        timestamp: DateTime,
    },
    TaskCompleted {
        run_id: String,
        task_id: String,
        status: TaskStatus,
        xcom: Map<String, String>,
        timestamp: DateTime,
    },
    SnapshotDeployed {
        snapshot_id: String,
        environment: String,
        timestamp: DateTime,
    },
    // ... 20+ other event types
}
}

All state is derived from events. Events are the source of truth.

State Model

graph TB
    Events["Append-only Event Log<br/>(RocksDB)"]
    Snapshots["Snapshot Index<br/>(Content-addressable)"]
    Environments["Environment Pointers<br/>(Mutable)"]
    Scheduler["Scheduler State<br/>(In-memory)"]

    Events -->|materialize| Snapshots
    Snapshots -->|referenced by| Environments
    Events -->|drive| Scheduler
    Scheduler -->|emit events to| Events

Event Store (RocksDB)

Key-Value Store:
  event_sequence_000001 → {DAGCompiled event}
  event_sequence_000002 → {TaskStarted event}
  event_sequence_000003 → {TaskCompleted event}
  ...

Index:
  run:run123 → [event_seq_10, event_seq_11, event_seq_12, ...]
  task:extract → [event_seq_2, event_seq_8, event_seq_15, ...]

Properties:

• Append-only: Never overwrite, only append
• Durable: Write-ahead logging ensures no data loss
• Indexed: Range queries for efficient lookups
• Compactable: Delete old events after retention period

Change Detection: Plan/Apply

The plan phase compares two snapshots:

Current: prod-snap-20240322-143215 (old)
Proposed: (recompile from source)

Tasks in DAG:
  extract:
    Old fingerprint: f1a2b3c4d5e6
    New fingerprint: f1b3c4d5e6f7  ← Changed (timeout 300 → 600)
    Status: Modified

  transform:
    Old fingerprint: g2h3i4j5k6l7
    New fingerprint: g2i4j5k6l7m8  ← Changed (upstream extract changed)
    Status: UpstreamInvalidated

  load:
    Old fingerprint: m3n4o5p6q7r8
    New fingerprint: m3n4o5p6q7r8  ← Unchanged
    Status: Unchanged (will be reused)

The apply phase creates a new snapshot and updates the environment pointer.

Scheduler Architecture

graph TB
    EventLog["Event Log<br/>(RocksDB)"]
    Scheduler["Scheduler<br/>(Tokio Task)"]
    Channels["Channel Network"]
    Local["Local Executor<br/>(Task Pool)"]
    Distributed["Distributed Executor<br/>(Coordinator → Workers)"]

    EventLog -->|change events| Scheduler
    Scheduler -->|enqueue runs| Channels
    Channels -->|local tasks| Local
    Channels -->|distributed tasks| Distributed
    Local -->|completion events| EventLog
    Distributed -->|completion events| EventLog
    EventLog -->|feedback| Scheduler

The scheduler is a single Tokio task listening to events. In distributed mode, the DistributedExecutor is a drop-in replacement for the local executor, using the same MPSC channel interface.

Lineage Tracking (conduit-lineage)

graph LR
    SQLTask["SQL Task"]
    Parser["SQL Parser"]
    Edges["Lineage Edges"]
    Graph["Lineage Graph"]

    SQLTask -->|parse| Parser
    Parser -->|extract columns| Edges
    Edges -->|build| Graph

Column-level lineage for SQL tasks, with upstream/downstream tracing exposed via the API.

REST API (conduit-api)

The API layer exposes 37 REST endpoints organized by domain:

• DAGs: list, get, graph, compile
• Runs: trigger, list, get, list-all
• Environments: list, create, get, delete, promote, diff
• Lineage: extract SQL, trace upstream/downstream, graph, schema diff
• Contracts: validate, list, per-dag, per-task
• Providers: list connections, get, test, list provider types
• Plan/Apply: generate plan, apply plan
• Metrics: list, get per-task
• Cluster: status, drain worker
• Backfill: create
• System: health check, system info, events list/get

Web UI (conduit-ui)

A React-based dashboard with 16 pages:

Dashboard, DAG List, DAG Detail, DAG Graph, Runs, Run Detail, Run Execution, Task Logs, Environments, Plan/Apply, Lineage, Contracts Dashboard, Metric Explorer, Connections, Events, Cluster

Performance Characteristics

Durability Guarantees

1. Append-only log: Events are never mutated
2. Write-ahead logging: Events are persisted before acknowledgment
3. Atomic snapshots: Snapshot creation is atomic
4. Consistent pointers: Environment pointers are atomic
5. Crash recovery: On restart, replay events to recover state

Current Limitations and Trade-offs

Limitations

• No partial rollback: Rollback is all-or-nothing for an environment
• No authentication: API key auth planned for future phase
• Event log grows unbounded: Retention policies can prune old events
• gRPC transport only: No REST fallback for distributed workers

Trade-offs

• Event sourcing latency: Reconstructing state takes time, mitigated by snapshots
• Snapshot size: Larger DAGs create larger snapshots (but still O(n) where n=tasks)
• Memory overhead: Keeping scheduler state in-memory (can add persistence later)

Testing

The workspace contains 558+ tests across all crates:

cargo test --workspace

Test coverage spans unit tests (inline), integration tests (per-crate tests/ directories), and property-based tests using proptest.

Benchmarks

cargo bench -p conduit-compiler

Benchmarks for parsing DAGs (10–1,000 tasks), dependency resolution, fingerprint computation, and event log queries.

Next Steps

• Event-Sourced Architecture: How events drive the system
• Plan/Apply Workflow: Change detection in detail
• CLI Reference: User-facing commands that use this architecture

WASM Plugin Sandbox

Conduit supports extending its functionality through WebAssembly (WASM) plugins. Plugins run in a sandboxed Wasmtime runtime with fine-grained permission controls, fuel-based execution limits, and memory caps — ensuring plugins can't crash the orchestrator or access resources they shouldn't.

Why WASM?

Traditional plugin systems use shared libraries (.so/.dll) or embedded scripting (Lua, Python). Both have serious tradeoffs — shared libraries can crash the host process and have full system access, while embedded scripting is slow and language-specific.

WASM plugins offer:

• Sandboxing — plugins can't access files, network, or memory outside their sandbox
• Performance — near-native execution speed via Wasmtime's optimizing compiler
• Language-agnostic — write plugins in Rust, Go, C/C++, AssemblyScript, or anything that compiles to WASM
• Deterministic — fuel metering prevents infinite loops; memory limits prevent OOM

Plugin Manifest

Every plugin ships as a .wasm binary alongside a plugin.toml manifest:

[plugin]
name = "my-custom-transform"
version = "1.0.0"
description = "A custom data transformation plugin"
author = "Your Name"
license = "Apache-2.0"

[permissions]
log = true
read_xcom = true
write_xcom = true
read_watermark = true
write_watermark = true
read_params = true

[limits]
max_fuel = 1000000000    # execution fuel (prevents infinite loops)
max_memory_bytes = 67108864  # 64 MB memory cap

Permissions

Plugins must declare which host functions they need. The runtime enforces these at call time — if a plugin tries to call a function it doesn't have permission for, the call returns an error.

Writing a Plugin (Rust)

Here's a minimal Rust plugin:

#![allow(unused)]
fn main() {
// lib.rs
extern "C" {
    fn host_log(level: i32, ptr: *const u8, len: i32);
    fn host_xcom_get(ptr: *const u8, len: i32) -> i64;
    fn host_xcom_set(key_ptr: *const u8, key_len: i32, val_ptr: *const u8, val_len: i32);
}

#[no_mangle]
pub extern "C" fn run() -> i32 {
    let msg = "Hello from WASM plugin!";
    unsafe {
        host_log(1, msg.as_ptr(), msg.len() as i32);
    }
    0  // return 0 for success
}
}

Compile with:

cargo build --target wasm32-wasi --release

Plugin Registry

Conduit maintains a local plugin registry at .conduit/plugins/. Install plugins with:

# Install from a local .wasm file
conduit plugin install ./my-plugin.wasm --manifest ./plugin.toml

# List installed plugins
conduit plugin list

# Run a plugin directly
conduit plugin run my-custom-transform --params '{"key": "value"}'

Runtime Configuration

Configure the WASM runtime globally in conduit.yaml:

wasm:
  max_fuel: 1000000000       # default fuel limit per execution
  max_memory_mb: 64          # default memory limit per plugin
  enable_epoch_interruption: true
  plugin_dir: .conduit/plugins

Security Model

The WASM sandbox provides defense-in-depth:

1. Memory isolation — each plugin gets its own linear memory space, completely separate from the host process
2. Fuel metering — every WASM instruction consumes fuel; when fuel runs out, execution halts with an error
3. Permission gates — host functions check the plugin's declared permissions before executing
4. No filesystem access — plugins cannot read or write files on the host
5. No network access — plugins cannot make network calls
6. Deterministic execution — given the same inputs and fuel, plugins produce the same outputs

Migration Guide: Airflow to Conduit

This guide helps you migrate existing Airflow DAGs to Conduit. While the orchestration model is different, the core concepts map cleanly.

Concept Mapping

High-Level Migration Path

1. Audit Airflow DAGs — Understand your existing setup
2. Rewrite DAGs — Convert to Conduit syntax
3. Test locally — Use conduit run to validate
4. Parallel run — Run both Airflow and Conduit for a period
5. Switchover — Point production to Conduit
6. Decommission Airflow — Clean up old infrastructure

Typical timeline: 2–8 weeks depending on DAG complexity.

Example: Rewriting an Airflow DAG

Airflow Original

# airflow/dags/sales_etl.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from airflow.utils.dates import days_ago

default_args = {
    'owner': 'analytics',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'sales_etl',
    default_args=default_args,
    description='Daily sales ETL',
    schedule_interval='0 2 * * *',
    start_date=days_ago(1),
    catchup=False,
    tags=['production', 'sales'],
)

def extract_data():
    import pandas as pd
    df = pd.read_sql("SELECT * FROM raw_sales", conn)
    print(f"Extracted {len(df)} rows")

def transform_data():
    import pandas as pd
    df = pd.read_sql("SELECT * FROM raw_sales", conn)
    df = df.dropna()
    print(f"Transformed {len(df)} rows")

def load_data():
    print("Loading to warehouse...")

extract_task = PythonOperator(
    task_id='extract',
    python_callable=extract_data,
    op_kwargs={},
    dag=dag,
)

transform_task = PythonOperator(
    task_id='transform',
    python_callable=transform_data,
    op_kwargs={},
    dag=dag,
)

load_task = PythonOperator(
    task_id='load',
    python_callable=load_data,
    op_kwargs={},
    dag=dag,
)

extract_task >> transform_task >> load_task

Conduit Converted

# dags/sales_etl.py
from conduit.sdk import dag, task

@task(retries=2, retry_delay=300)
def extract():
    import pandas as pd
    # Credentials from environment
    conn_str = os.getenv("DATABASE_URL")
    df = pd.read_sql("SELECT * FROM raw_sales", conn_str)
    print(f"Extracted {len(df)} rows")
    print(f"xcom|row_count|{len(df)}")
    return "raw_sales.csv"

@task(retries=2, retry_delay=300)
def transform(raw_file):
    import pandas as pd
    conn_str = os.getenv("DATABASE_URL")
    df = pd.read_sql("SELECT * FROM raw_sales", conn_str)
    df = df.dropna()
    clean_count = len(df)
    print(f"Transformed {clean_count} rows")
    print(f"xcom|clean_count|{clean_count}")
    return "clean_sales.csv"

@task(retries=2, retry_delay=300)
def load(clean_file):
    print("Loading to warehouse...")
    return "success"

@dag(schedule="0 2 * * *", tags=["production", "sales"])
def sales_etl():
    """Daily sales ETL pipeline."""
    raw = extract()
    clean = transform(raw)
    load(clean)

Key differences:

• Decorators instead of operator classes
• No separate default_args dict
• Return values flow automatically (no explicit XCom sets)
• Schedule as string parameter, not schedule_interval
• Credentials from environment, not Airflow Connections

Feature-by-Feature Migration

Schedules

Airflow:

dag = DAG(
    'my_dag',
    schedule_interval='0 2 * * *',
    start_date=datetime(2024, 1, 1),
    catchup=False,
)

Conduit:

@dag(schedule="0 2 * * *")
def my_dag():
    pass

Conduit uses standard 5-field cron syntax. No separate start_date or catchup behavior.

Operators → Tasks

Airflow:

from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator

task1 = PythonOperator(
    task_id='python_task',
    python_callable=my_function,
    provide_context=True,
    dag=dag,
)

task2 = BashOperator(
    task_id='bash_task',
    bash_command='ls -la /data',
    dag=dag,
)

Conduit:

from conduit.sdk import task, shell_task

@task
def python_task():
    my_function()

@shell_task
def bash_task():
    ls -la /data

Dependencies

Airflow:

task1 >> task2 >> task3  # Sequential
[task1, task2] >> task3  # task1 and task2 → task3

Conduit:

@dag
def my_dag():
    t1 = task1()
    t2 = task2(t1)  # task2 depends on task1 via input
    t3 = task3(t2)  # task3 depends on task2

Dependencies are inferred from function arguments.

Retries and Timeouts

Airflow:

task = PythonOperator(
    task_id='my_task',
    python_callable=my_func,
    retries=3,
    retry_delay=timedelta(minutes=5),
    execution_timeout=timedelta(hours=1),
    dag=dag,
)

Conduit:

@task(
    retries=3,
    retry_delay=300,      # seconds
    timeout=3600,         # seconds
)
def my_task():
    my_func()

XCom

Airflow:

def task1(context):
    context['task_instance'].xcom_push(
        key='my_key',
        value='my_value'
    )

def task2(context):
    value = context['task_instance'].xcom_pull(
        task_ids='task1',
        key='my_key'
    )
    print(value)

task1 = PythonOperator(task_id='task1', python_callable=task1, dag=dag)
task2 = PythonOperator(task_id='task2', python_callable=task2, dag=dag)
task1 >> task2

Conduit:

@task
def task1():
    return {'my_key': 'my_value'}

@task
def task2(data):
    print(data['my_key'])

@dag
def my_dag():
    data = task1()
    task2(data)

Or explicit XCom:

@task
def task1():
    print("xcom|my_key|my_value")
    return "result"

Trigger Rules

Airflow:

from airflow.utils.trigger_rule import TriggerRule

task = PythonOperator(
    task_id='my_task',
    python_callable=my_func,
    trigger_rule=TriggerRule.ALL_DONE,
    dag=dag,
)

Conduit:

from conduit.sdk import TriggerRule

@task(trigger_rule=TriggerRule.ALL_DONE)
def my_task():
    my_func()

Available rules: ALL_SUCCESS, ALL_DONE, ONE_SUCCESS, ONE_FAILED

Context Access

Airflow:

def my_task(context):
    dag_run = context['dag_run']
    task_instance = context['task_instance']
    print(f"Run ID: {dag_run.run_id}")
    print(f"Attempt: {task_instance.try_number}")

Conduit:

from conduit.sdk import TaskContext

@task
def my_task():
    ctx = TaskContext()
    print(f"Run ID: {ctx.run_id}")
    print(f"Attempt: {ctx.attempt}")

Connections and Secrets

Airflow:

from airflow.models import Variable
from airflow.hooks.base import BaseHook

def my_task():
    # Get connection
    conn = BaseHook.get_connection('my_db')
    user = conn.login
    password = conn.password

    # Get variable
    api_key = Variable.get('API_KEY')

Conduit: Use environment variables:

import os

@task
def my_task():
    user = os.getenv('DB_USER')
    password = os.getenv('DB_PASSWORD')
    api_key = os.getenv('API_KEY')

Set environment variables in your deployment:

export DB_USER=admin
export DB_PASSWORD=secret
export API_KEY=xyz123

conduit run my_dag

Or in .conduit.toml:

[env]
DB_USER = "admin"
DB_PASSWORD = "secret"
API_KEY = "xyz123"

Conditional Execution

Airflow:

from airflow.operators.python import PythonOperator
from airflow.operators.branch import BranchPythonOperator

def choose_path(context):
    if context['execution_date'].day % 2 == 0:
        return 'even_task'
    else:
        return 'odd_task'

branch = BranchPythonOperator(
    task_id='branch',
    python_callable=choose_path,
    dag=dag,
)

even_task = PythonOperator(task_id='even_task', ...)
odd_task = PythonOperator(task_id='odd_task', ...)

branch >> [even_task, odd_task]

Conduit:

import os

@task
def choose_path():
    day = int(os.getenv('DAY', '1'))
    return day % 2 == 0

@task
def even_task():
    print("Even day")

@task
def odd_task():
    print("Odd day")

@dag
def my_dag():
    is_even = choose_path()

    if is_even:
        even_task()
    else:
        odd_task()

Note: Conditionals are evaluated at compile time in Conduit. For runtime branching, use trigger rules instead.

Sensors

Airflow:

from airflow.sensors.filesystem import FileSensor

wait = FileSensor(
    task_id='wait_for_file',
    filepath='/data/export.csv',
    poke_interval=60,
    timeout=3600,
    dag=dag,
)

Conduit:

from conduit.sdk import sensor_task
import os

@sensor_task(timeout=3600, poke_interval=60)
def wait_for_file():
    return os.path.exists('/data/export.csv')

Automated Migration Tool

Conduit provides a migration tool for simple DAGs:

conduit migrate airflow --config ~/airflow.cfg --output dags/

Capabilities:

• Converts PythonOperator → @task
• Converts BashOperator → @shell_task
• Extracts schedules
• Converts trigger rules
• Generates basic XCom mappings

Limitations:

• Cannot convert complex operators (custom subclasses)
• Cannot infer context usage
• May require manual cleanup

Step-by-Step Migration Plan

Phase 1: Preparation (1 week)

1. Audit Airflow DAGs

   # List all DAGs
   airflow dags list
   
   # Find complex DAGs
   grep -r "BranchOperator\|SubDagOperator\|TriggerRuleSensor" dags/
   

2. Categorize by complexity

   • Tier 1: Simple linear DAGs (extract → transform → load)
   • Tier 2: Conditional DAGs (branching, dynamic tasks)
   • Tier 3: Complex DAGs (subDAGs, multiple pools, custom operators)

3. Identify dependencies

   • Which DAGs depend on which?
   • Which use shared connections, variables, or pools?

Phase 2: Rewrite (2–4 weeks)

1. Start with Tier 1 DAGs

   # Rewrite simple DAGs
   # Run through migration tool for baseline
   conduit migrate airflow --output dags/
   
   # Manually cleanup and test
   cd dags
   vim extracted_dag.py
   

2. Validate compilation

   conduit compile
   

3. Test locally

   conduit run my_dag
   

4. Handle Tier 2 DAGs

   • Rewrite conditional logic as Python if/else
   • Convert custom operators to @task

5. Handle Tier 3 DAGs

   • Break subDAGs into multiple DAGs or inline them
   • Rewrite custom operators as @task + subprocess

Phase 3: Testing (1–2 weeks)

1. Create staging environment

   conduit env create staging --from production
   

2. Deploy Conduit DAGs

   vim dags/*.py  # Make sure all DAGs are ready
   conduit compile
   conduit plan staging
   conduit apply staging -y
   

3. Run test executions

   # Run each DAG manually
   conduit run my_dag --env staging
   
   # Verify outputs match Airflow
   

4. Monitor for 1 week

   • Check logs
   • Verify XCom outputs
   • Validate downstream systems receive correct data

Phase 4: Parallel Run (1–2 weeks)

1. Keep Airflow running for production DAGs
2. Run Conduit in staging for the same DAGs
3. Compare outputs — Are results identical?
4. Build confidence — Run both systems side-by-side

Phase 5: Switchover (1 day)

1. Disable Airflow DAGs

   # In Airflow UI: Set DAGs to off
   

2. Enable Conduit DAGs

   conduit env promote staging production
   

3. Monitor closely for 24 hours

   • Watch scheduler logs
   • Check task execution times
   • Verify downstream systems

4. Rollback plan ready if needed

   conduit env rollback production --to previous-snapshot
   

Phase 6: Decommission (ongoing)

1. Delete old Airflow DAGs after 30 days of successful Conduit runs
2. Archive Airflow database for audit trail
3. Notify teams of migration completion

Common Gotchas

1. Credentials and Secrets

Problem: Airflow uses Connections, Conduit uses environment variables.

Solution:

• Export all Airflow connections to environment variables
• Use a secrets management tool (HashiCorp Vault, AWS Secrets Manager)
• Set in .conduit.toml or CI/CD system

2. Dynamic Task Generation

Problem: Airflow allows dynamic tasks via loops. Conduit's DAG structure is static at compile time.

Solution:

# Airflow: Dynamic tasks
for i in range(10):
    task = PythonOperator(task_id=f'task_{i}', ...)

# Conduit: Static DAG, dynamic execution
@dag
def static_dag():
    tasks = [
        extract(i)
        for i in range(10)  # Generated at compile time
    ]
    return tasks

Both compile statically, but Conduit declares all tasks upfront.

3. Custom Operators

Problem: Your organization has custom operators that don't have Conduit equivalents.

Solution:

• Rewrite custom operator logic as a @task function
• Or use @executable_task to call the operator binary

4. Backfill and Catchup

Problem: Airflow supports backfill and catchup. Conduit does not.

Solution:

• Use Conduit's replay feature for historical debugging
• For backfill-like behavior, run DAG manually for past dates:

  conduit run my_dag --env production --date 2024-01-01
  

5. Pools and Resource Limits

Problem: Airflow has multiple pools. Conduit uses a single Pool abstraction.

Solution:

• Create a Pool for each resource type
• Map Airflow pools to Conduit pools 1:1

Performance Expectations

Compilation

• Airflow: Seconds (DAG parsing + full execution)
• Conduit: Milliseconds (tree-sitter parsing only)

Scheduling

• Airflow: Polling every 5–30 seconds
• Conduit: Event-driven, microseconds latency

Deployment

• Airflow: Restart webserver, parse all DAGs
• Conduit: Plan/apply, only recompile changed DAGs

Typical improvements

• Faster feedback cycle: seconds → milliseconds
• Lower latency: polls → events
• Better visibility: immutable event log

Next Steps

• Installation: Set up Conduit
• Quick Start: Write your first DAG
• DAG Concepts: Full DAG definition guide
• Migration Tool: Automated conversion