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.

Source selectors use Dagster's logical layer.namespace.table identity. At activation, Frank resolves each selected source to the physical Trino catalog.namespace.table identity stored with the runnable transform. This keeps the execution query and Dagster dependency edge tied to the same table. When multiple selected sources have the same table name, use the qualified logical selector: Frank resolves that exact namespace before rendering SQL; only a bare ambiguous table name is rejected.

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 compiles one Pipeline policy across the current graph without creating a new Pipeline version, step, Transform, or Dagster asset identity:

• manual disables automation on every generated Transform. Run now selects the complete graph in one Dagster run.
• eager keeps the complete graph data-version-driven. A fan-in recomputes from the latest direct-parent snapshots when any parent changes; Dagster's native eager condition groups dependency-connected synchronous branches.
• cron and interval put the same wall-clock condition on the synchronous portion of the graph. Every dependency-connected Pipeline graph is kept in one native automation sensor shard. Dagster can select and group a SQL chain in one run, including a downstream step after a no-op root that produced no new Iceberg snapshot. A K8s-backed step is an explicit two-phase boundary: its submit joins the due run, its external output is materialized by the completion sensor, and all transitive descendants use eager from that event. A fan-in accumulates each direct-parent update since the Pipeline's cron tick, so a synchronous branch that finishes first is not forgotten while the K8s branch is still running. The K8s run stores its full output asset key when it is submitted; completion therefore still reports the launched asset if the Transform is renamed or moved between Silver and Gold while the job runs.

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. Each handled wall-clock tick requests at most one run. If the root already has a queued or running materialization, the due tick stays pending and produces one coalesced catch-up after that run becomes terminal; Frank does not replay every missed tick as a backlog.

Bronze freshness is published before eager evaluation. Native Dagster automation sensors evaluate stable, disjoint Bronze shards; observation runs read Iceberg snapshot IDs as DataVersion values. A native per-asset in_progress() guard prevents a new minute tick from overlapping queued or running observation work for that same Bronze asset. Only a changed value reaches the independently sharded native transform automation and requests the eager DAG. Bronze hashes independent observable assets individually; transform automation hashes each connected Silver/Gold graph by its stable root and never splits dependencies between sensors. Dagster retains condition cursors and grouped-run planning for synchronous assets; K8s outputs cross their explicit completion-event boundary without losing graph ownership. No Frank trigger endpoint or duplicate change-gate sensor sits between Source sync and transform automation. See Sources: Bronze freshness observation.

Bronze polling and Pipeline transforms share Dagster's bounded run queue but do not have the same latency contract. Automatic and product-triggered transform runs carry a higher Dagster run priority, so a waiting Pipeline takes the next available run slot ahead of queued Bronze polling. The policy does not interrupt work that is already running, reduce the deployment's global capacity limit, or give Frank a second scheduling owner. When no transform is waiting, Bronze remains free to use every available slot.

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.
• pending: the desired state is committed, but Dagster has not loaded it 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 browser uses the immediate 202 mode: PUT /api/v1/pipelines/{id}/schedule commits intent, requests the existing Temporal reconciliation coordinator for that tenant and Pipeline, and polls loaded-state readback. The request does not inspect unrelated Pipelines or wait for a Dagster code-location reload. Each browser observation has its own abort deadline; the API also bounds the complete readback across every root and Dagster GraphQL call. The defaults are 30 seconds in the browser and 20 seconds in the API, configurable with DAGSTER_SCHEDULE_READBACK_DEADLINE_SECONDS. A transient failed observation is retried until the bounded overall reconciliation deadline. frankctl uses the same non-blocking write and then polls the read endpoint for bounded loaded-state proof. The legacy ?wait_for_reconciliation=true query is accepted for compatibility but cannot perform a direct Dagster reload or bypass the serialized lane. The CLI treats anything other than confirmed synced state as an error and prints the desired/effective difference.

Automation health and execution evidence

Configuration confirmation is not an execution-health claim. Schedule readback therefore also returns automation_health, next_action_at, and last_automatic_run:

• healthy: every root has exactly one running owner sensor and a fresh, successful, skipped, or currently-running tick.
• degraded: an owner is stopped or its last tick is stale.
• unhealthy: ownership is missing/duplicated, the latest tick failed, or a running tick exceeded the freshness boundary.
• unknown: Dagster health, tick evidence, or automatic-run history could not be read completely.
• disabled: the Pipeline is manual, paused, archived, or not active.

Each root includes the exact owning sensor, latest tick status/time/error, a stable error_class when the evidence is invalid, and its last automatic Dagster run. Frank resolves ownership from the loaded sensors' real asset selections, then reads tick history only for those exact owners; unrelated sensor history is not part of a Pipeline's health readback. The corresponding Frank TransformRun is matched by both Dagster run ID and triggering Transform ID, so grouped runs are not attributed to the wrong root. The default stale boundary is 900 seconds; operators may set DAGSTER_AUTOMATION_SENSOR_STALE_AFTER_SECONDS to another value of at least 60 seconds.

If automatic-run history lookup fails, the root exposes last_automatic_run_error and last_automatic_run_error_class, and otherwise healthy automation becomes unknown. The UI reports “history unavailable” instead of “no run recorded,” and the CLI exits non-zero.

next_action_at is an upcoming wall-clock timestamp for cron and interval. It is null with an explicit reason for event-driven, manual, paused, or archived policies. The CLI exits non-zero for degraded, unhealthy, or unknown automation even when configuration is synced.

The reconciler retries pending or drifted Pipeline schedules every five minutes. Periodic global repair and interactive targeted repair both enter one stable Temporal Signal-With-Start coordinator. It executes one reconciliation activity at a time, coalesces repeated edits for the same Pipeline to the latest durable intent, splits fleet maintenance into one-Pipeline activities, and drains interactive targets before queued maintenance. The five-minute schedule remains the complete drift-repair fallback; it no longer runs a second reconciliation lane. All Frank code-location reloads also share one cross-process PostgreSQL advisory lock, so Source and Transform lifecycle paths cannot overlap the coordinator's reload. A reconciliation pass 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 that Pipeline's generated-Transform automation; it does not stop the shared Dagster sensor or unrelated Transform schedules. Resume restores the saved policy. Trigger selects the complete active Pipeline graph in one Dagster run and returns one correlation ID plus root-level 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