Pipelines

Pipelines compose transforms into versioned DAGs. They are the product layer for repeatable multi-step data work.

What a pipeline stores

A pipeline owns:

• Metadata: tenant, name, description, created by.
• Status: draft, active, paused, error, archived.
• Source tables and target hints.
• Current version pointer.
• Schedule config.

Each pipeline version stores:

• Immutable DAG spec.
• Content hash.
• Created timestamp and author.
• Optional AI session ID.
• Steps and edges.

Pipeline status

draft -> active -> paused -> active
  |        |          |
  |        +-> error -+
  +-> archived
active -> archived
error -> draft | active | archived

Archived is terminal.

Steps

A step can reference:

• An existing transform.
• A transform pattern and params.
• Custom SQL.
• Custom Python content.
• Generated code that is being published to an external pattern repository.

Important step fields:

Field	Meaning
step_name	Stable name inside the version.
transform_id	Existing transform to execute.
pattern_id	Pattern to materialize as a transform during activation.
pattern_params	Params for the pattern.
step_type	catalog_pattern, field_mapping, custom_sql, or custom_python.
is_intermediate	Whether this step is an internal stage.
is_terminal	Whether this step is an output of the DAG.
error_policy	Step failure policy, including halt, skip, and retry:N.
config.output_layer	Usually silver for intermediate and gold for curated outputs.
pattern_status	Tracks external pattern publishing, such as available or building.
pr_url	Pull request for generated/published custom patterns.

Edges and validation

Edges are directed dependencies between steps. Frank validates every DAG before sandboxing or activation:

• Every edge endpoint must reference an existing step.
• Self-edges are rejected.
• Cycles are rejected with a cycle path.
• Topological order is computed deterministically.
• Roots, terminal steps, intermediate steps, and fan-in steps are classified.

This lets the UI and CLI catch broken structures before a runtime job starts.

Versioning

Pipeline versions are immutable snapshots. Frank computes a SHA-256 content hash from sorted step content and edge pairs. If the DAG content has not changed, duplicate versions are avoided.

This gives you:

• A clean audit trail.
• Safe roll-forward through new versions.
• Stable run history tied to the actual DAG that ran.
• A clear boundary between editing and activation.

Sandbox validation

Sandbox runs are the pre-activation safety gate:

frankctl pipelines validate <pipeline-id> --sample-limit 1000 --timeout 600

The CLI starts:

POST /api/v1/pipelines/{pipeline_id}/sandbox

Then polls:

GET /api/v1/pipelines/{pipeline_id}/sandbox/{workflow_id}/status

Step badges stream to stderr and final JSON is emitted to stdout. Completed exits 0; failed or partial failure exits 5.

Activation

Activation turns a draft/version into runnable transforms:

1. Validate the DAG.
2. Create or link transforms for each step.
3. Compute output table names through Frank naming helpers.
4. Set the pipeline current version.
5. Move pipeline status to active.
6. Trigger downstream synchronization with Dagster where needed.

Pipeline step output table names follow the pipeline + step naming convention, then layer into Silver or Gold.

Scheduling and execution ownership

A Source schedule and a Pipeline schedule control different stages. Temporal owns Source extraction into Bronze. Dagster owns the transforms generated from the current Pipeline version and materializes Silver or Gold. Scheduling one does not silently schedule the other.

Pipeline schedule intent is one of four strict shapes:

# Run only through Run now, the trigger API, or frankctl.
schedule_config: { type: manual }

# Normal continuous mode: react to a new upstream materialization.
schedule_config: { type: eager }

# Independent wall clock in an IANA timezone.
schedule_config:
  type: cron
  value: "0 6 * * *"
  timezone: Europe/Lisbon

# Exact interval shorthand. Frank compiles it to a five-field Dagster cron.
schedule_config:
  type: interval
  value: 15m
  timezone: UTC

Intervals must be exactly representable by a five-field cron expression: minute values must divide 60, hour values must divide 24, and the only day interval is 1d. Use cron for other cadences. manual and eager do not accept a value or timezone.

Frank applies one Pipeline policy to every root step in the current version. Downstream steps remain dependency-driven, so one accepted root execution produces one DAG chain rather than independently clocking every step. A schedule change updates generated Transform automation in place: it does not create another Pipeline version, step, Transform, or Dagster asset identity.

eager is the normal Source-to-Pipeline mode because it follows actual Bronze freshness. cron and interval are independent wall clocks: they can run against unchanged input or before a Source refresh completes.

Desired state and loaded state

The schedule readback exposes both the durable Pipeline intent (desired) and the effective configuration loaded by Dagster for each root (roots[].effective). It also returns deterministic desired and observed fingerprints.

• synced: every loaded root fingerprint matches the desired fingerprint.
• pending_activation: the Pipeline has no active generated roots yet.
• drifted: Dagster loaded a different effective policy.
• error: reconciliation or loaded-state readback failed.

A mutating API returns 200 only when readback confirms the owning Dagster location. It returns 202 when the durable intent was accepted but runtime confirmation is still pending. The CLI treats anything other than confirmed synced state as an error and prints the desired/effective difference.

The reconciler retries pending or drifted Pipeline schedules every five minutes. Re-applying an identical schedule is a no-op when the fingerprints already match.

Lifecycle controls

frankctl pipelines schedule get <pipeline-id>
frankctl pipelines schedule set <pipeline-id> --type eager
frankctl pipelines schedule set <pipeline-id> --type cron \
  --value "0 6 * * *" --timezone Europe/Lisbon
frankctl pipelines schedule set <pipeline-id> --type interval \
  --value 15m --timezone UTC
frankctl pipelines pause <pipeline-id>
frankctl pipelines resume <pipeline-id>
frankctl pipelines trigger <pipeline-id>

Pause disables only that Pipeline's generated root automation. It does not stop the shared Dagster sensor or unrelated Transform schedules. Resume restores the saved policy. Trigger runs every active root once and returns a correlation ID plus Dagster and Transform-run receipts. Archived Pipelines are terminal, read-only, and cannot trigger or schedule new work; history and existing data are retained.

Generated Transforms remain visible for execution evidence, but their schedule is Pipeline-owned. Direct schedule edits on those Transforms return 409 with the owning Pipeline route and CLI command. Standalone Transforms keep their own schedule controls.

AI composition

The pipeline composer calls Martha workflow frank_compose_pipeline. Input:

• Source tables.
• Target description.
• Optional target schema or SDM ID.
• Pipeline context.
• Pipeline name.

Output can include proposed steps, pattern choices, params, dependencies, reasoning, and confidence. The UI keeps the human in control: AI composes a draft; users review, edit, sandbox, and activate.

CLI:

frankctl ai compose-pipeline -f pipeline-intent.yaml --timeout 600

Common pipeline shapes

Staging to mart

raw.orders -> stg_orders -> fct_daily_sales
raw.products -> stg_products -/

Customer 360

raw.postgres_customers  \
raw.salesforce_contacts -> dim_customer_360
raw.stripe_customers    /

Geospatial enrichment

raw.events -> geo_parse_wkt -> h3_enrich -> h3_aggregate

Semantic publication

raw.source -> stg_clean -> dim_entity -> backing dataset -> ontology sync

Related APIs

POST /api/v1/pipelines
GET  /api/v1/pipelines
GET  /api/v1/pipelines/{pipeline_id}
PUT  /api/v1/pipelines/{pipeline_id}
POST /api/v1/pipelines/validate-dag
POST /api/v1/pipelines/{pipeline_id}/versions
POST /api/v1/pipelines/{pipeline_id}/activate
POST /api/v1/pipelines/{pipeline_id}/pause
GET  /api/v1/pipelines/{pipeline_id}/schedule
PUT  /api/v1/pipelines/{pipeline_id}/schedule
POST /api/v1/pipelines/{pipeline_id}/trigger
DELETE /api/v1/pipelines/{pipeline_id}  # archive; data and history retained
POST /api/v1/pipelines/{pipeline_id}/sandbox
GET  /api/v1/pipelines/{pipeline_id}/runs