Self-heal Dagster pipelines

The Dagster pipeline healer is a DinoAI agent that dispatches every op failure to a background agent session. The moment a step fails, a Dagster @failure_hook captures the exception and stack trace and hands it to the agent, which posts a triage summary, applies the minimal fix on a new branch, and opens one pull request per failed run.

Prerequisites

GitHub connected so the healer can create branches and open PRs against the workspace repository.
Slack connected so the healer can post its triage summary and progress.
A Paradime API endpoint, key, and secret with DinoAI agent API capabilities enabled (Workspace Settings, API; requires Admin access). Keys are workspace-scoped: generate the key from the workspace whose connected repository contains your .dinoai/agents/ folder, because the agent always runs in the workspace that issued the key.
Familiarity with programmable agents.

Estimated time: 30 minutes.

Steps

Create the agent

Build the healer in the Agent UI: fill in the builder fields below, leave Model on Auto, then select Deploy and choose Open a pull request so the definition lands at .dinoai/agents/dagster-pipeline-healer.yml. (Prefer working as code? That file is the full definition, and you can author or edit it by hand from then on.)Name: dagster-pipeline-healerRole:

Dagster Run Recovery Specialist focused on diagnosing and fixing
failed Dagster runs of the F1 dbt + dlt pipeline in this analytics
repo.

Goal:

When a Dagster run fails, identify the failing asset/step, propose a
minimal fix, and open a PR with the change.

The triggering message from the Dagster failure sensor includes:
  - run_id
  - job_name
  - failed_steps
  - one stack trace block per failed step, copied verbatim from
    Dagster

1. Parse the failure message and extract the exact error for every
   failed step.
2. BEFORE touching any code, post a triage summary as your first
   message. It must include, for every distinct error found:
     - run_id and job_name
     - the failed step/asset
     - the exact error string from the stack trace, quoted verbatim
     - your diagnosis of the root cause and the fix you are about to
       apply
   Post this summary first, then proceed immediately — do not wait
   for a reply.
3. Map each error to its origin in the repo:
     - dbt errors (compilation, schema, BigQuery) → dbt model, test,
       or YAML under models/, macros/, seeds/, snapshots/
     - dlt errors → python/dlthub/f1_data/f1_api_pipeline.py
     - Dagster definition errors → dagster_project/
4. Create ONE branch for the whole failure run, apply the minimal fix
   for every distinct error on that branch, and open a single PR.
5. In the PR description, itemize each error separately: quote the
   exact error string from the stack trace, name the file changed,
   and explain the fix applied for it.

Never open more than one PR per failure run. All fixes for a run
belong on the same branch and in the same PR, each documented as its
own item.

Backstory:

You are fully autonomous and act immediately without asking for
confirmation. You never ask for permission or approval before
creating a branch, applying a fix, or opening a PR — you state what
you are doing and do it. If something is ambiguous, you make a
reasonable assumption, state it in the PR description, and proceed
immediately.

You know dbt and Dagster deeply. You prefer surgical edits over
rewrites and always cite the failing step's error message in your
reasoning. The Dagster failure context is your primary source of
truth.

Allowed tools: read_file, search_files_and_directories, ripgrep_search, run_sql_query, run_terminal_commandOutput: Slack channel #dagster-dinoai-self-healing

Merge the deploy pull request to your default branch before triggering. API-triggered agents are loaded from the default branch of the workspace repository, so a definition sitting on a feature branch, an open PR, or only in your local checkout is invisible to trigger_run, and the session fails with no messages. Any later edits also only take effect once merged.

Adapt the error-to-file mapping in step 3 of the Goal to your own repository layout. The mapping is what lets the agent go from a stack trace to the right file without exploring the whole repo.

Add the failure hook

The hook fires in-process the moment an op fails: no sensor polling, no daemon dependency. Create dagster_project/hooks.py:

dagster_project/hooks.py

import os
import textwrap
import traceback

from dagster import HookContext, failure_hook

DINOAI_AGENT_NAME = "dagster-pipeline-healer"
MAX_MESSAGE_CHARS = 12_000


def _format_failure_message(context: HookContext) -> str:
    exc = context.op_exception
    if exc is not None:
        trace = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
    else:
        trace = "No exception captured."

    header = textwrap.dedent(
        f"""
        Dagster step failure detected.

        - run_id: `{context.run_id}`
        - job_name: `{context.job_name}`
        - failed_step: `{context.op.name}`
        """
    ).strip()

    message = f"{header}\n\n### step `{context.op.name}`\n```\n{trace}\n```"
    if len(message) > MAX_MESSAGE_CHARS:
        message = message[:MAX_MESSAGE_CHARS] + "\n\n…(truncated)"
    return message


@failure_hook(name="dinoai_failure_triage_hook")
def dinoai_failure_triage_hook(context: HookContext) -> None:
    """On op failure, dispatch the stack trace to the pipeline-healer DinoAI agent."""
    api_endpoint = os.getenv("PARADIME_API_ENDPOINT")
    api_key = os.getenv("PARADIME_API_KEY")
    api_secret = os.getenv("PARADIME_API_SECRET")

    if not all([api_endpoint, api_key, api_secret]):
        context.log.warning("Skipping DinoAI dispatch: Paradime API env vars not set.")
        return

    from paradime import Paradime

    paradime = Paradime(api_endpoint=api_endpoint, api_key=api_key, api_secret=api_secret)
    result = paradime.dinoai_agents.trigger_run(
        agent=DINOAI_AGENT_NAME,
        message=_format_failure_message(context),
    )
    context.log.info(
        f"Triggered DinoAI agent '{DINOAI_AGENT_NAME}' for failed step "
        f"`{context.op.name}` in run {context.run_id}. Result: {result}"
    )

Then attach the hook to your jobs at definition time:

dagster_project/jobs.py

from dagster import AssetSelection, define_asset_job

from dagster_project.hooks import dinoai_failure_triage_hook

f1_dbt_only_job = define_asset_job(
    name="f1_dbt_only_job",
    selection=AssetSelection.all() - AssetSelection.groups("ingestion"),
    hooks={dinoai_failure_triage_hook},   # ← fires on any op failure
)

Finally, set the three environment variables wherever your Dagster code runs, dagster dev locally or your deployment’s environment configuration in production:

Variable	Description
`PARADIME_API_ENDPOINT`	Your Paradime API endpoint
`PARADIME_API_KEY`	Your Paradime API key
`PARADIME_API_SECRET`	Your Paradime API secret

The hook is a graceful no-op when the env vars are unset: it logs a warning and returns. Local development without credentials never fails a run because of the hook, and the original op failure is always surfaced normally in Dagster either way.

Materializing assets directly from the asset graph runs Dagster’s internal __ASSET_JOB, which carries no hooks. Always launch through a hooked job to get self-healing coverage.

Test it end-to-end

Verify the loop with a deliberate, controlled failure. Two rules make or break the test:

The break must be a runtime error, not a parse error. A typo’d ref() breaks dbt parsing, which breaks Dagster’s code-location load itself. The run dies with “Could not load job definition” before any op starts, and the hook never fires. Instead, reference a column that doesn’t exist in the source table:

-- models/staging/stg_f1__races.sql
SELECT
  raceid AS race_id,
  grand_prix_sponsor AS race_sponsor,   -- ← does not exist in the source table
  ...
FROM {{ source('f1', 'races') }}

This parses fine, Dagster loads, and the model fails in the warehouse with Unrecognized name: grand_prix_sponsor, a real op failure that fires the hook.

The break must be pushed to the branch your Paradime workspace tracks. The agent clones the remote repository, so a local-only break fires the hook but the agent finds healthy code and has nothing to fix.

Push the break, then launch the hooked job from the Dagster UI. Watch the run in three places, in the order things happen:

Dagster event log: right after STEP_FAILURE you’ll see the dispatch confirmation, Triggered DinoAI agent 'dagster-pipeline-healer' … Result: ok=True agent_session_id='…' status='queued'
Slack: #dagster-dinoai-self-healing receives the triage summary, then progress updates
GitHub: the fix PR appears a few minutes later, citing the exact error string

Each failed run produces exactly one PR that heals the pipeline, with every distinct error documented (the verbatim error string, the file changed, and the fix applied). Merge the PR to close the loop. If a session fails immediately with no messages, the API key likely belongs to the wrong workspace, because trigger_run always runs the agent in the workspace that issued the key. A long-running dagster dev process keeps the environment variables it started with, so restart it after rotating or switching keys.

How it works

The hook fires in-process the moment an op fails and dispatches the failure fire-and-forget: it returns immediately, so the Dagster run’s teardown is never blocked by the agent session. A typical healing run completes in 2 to 6 minutes from step failure to open PR. The agent posts a triage summary before touching any code, maps each error to its source file, then creates one branch and opens a single PR itemizing every distinct error.

Dagster run
    │
    ▼  op fails
@failure_hook dinoai_failure_triage_hook
    │  captures run_id, job_name, failed step + stack trace
    ▼
paradime.dinoai_agents.trigger_run(agent="dagster-pipeline-healer", message=...)
    │
    ▼
dagster-pipeline-healer (background agent session)
    ├─ 1. posts a triage summary to Slack BEFORE touching any code
    ├─ 2. maps each error to its source file
    ├─ 3. creates ONE branch, applies the minimal fixes
    └─ 4. opens a single PR with every error itemized

Each failed run produces one PR with every distinct error documented verbatim:

fix: remove nonexistent grand_prix_sponsor column from stg_f1__races

Why: Dagster run 09d469d9-… (job f1_dbt_only_job, step f1_dbt_assets)
failed with:

  Database Error in model stg_f1__races
  Unrecognized name: grand_prix_sponsor at [17:3]

Root cause: the underlying source table has no grand_prix_sponsor
column, verified directly against INFORMATION_SCHEMA.COLUMNS.

Next steps

Bolt pipeline healer

The same pattern driven by Bolt self-healing.

Pipeline incident commander

Multi-agent triage from Airflow failures.

Build an agent in the UI

Create this agent visually instead of by hand.

Programmable Agents reference

The agent YAML schema and tools.

Foundations

Automate with AI agents

Set up & configure

Orchestrate data pipelines

Migrate to Paradime

Develop in Paradime

Learn dbt™

Self-heal Dagster pipelines

Steps

How it works

Next steps

Bolt pipeline healer

Pipeline incident commander

Build an agent in the UI

Programmable Agents reference

​Steps

​How it works

​Next steps

Bolt pipeline healer

Pipeline incident commander

Build an agent in the UI

Programmable Agents reference

Steps

How it works

Next steps