Self-Healing Pipelines - Dagster

Automatically recover from failed Dagster runs by dispatching every op failure to a DinoAI background agent — the pipeline healer. The moment a step fails, a Dagster @failure_hook captures the exception and stack trace and hands it to the agent. The agent posts a triage summary, applies the minimal fix on a new branch, and opens a pull request — no manual intervention required.

Before You Start

Paradime

• Your Paradime API endpoint, API key, and API secret — generate these under Workspace Settings → API. Make sure to enable DinoAI agent API capabilities. Requires Admin access.

• API keys are workspace-scoped: generate the key from the workspace whose connected repository contains your .dinoai/agents/ folder — the agent always runs in the workspace that issued the key.

Recommended reading

Before proceeding, read the Programmable Agents section under Products → DinoAI: Quick Start, YAML Configuration, and Tools Reference.

Integrations

The following must already be connected in Paradime:

• GitHub - the healer creates branches and opens PRs against the workspace repository

• Slack - the healer posts its triage summary and progress to #dagster-dinoai-self-healing

What You'll Build

• One DinoAI agent YAML — the dagster-pipeline-healer — that triages failures, posts a summary, and opens a single fix PR per failed run

• A Dagster @failure_hook that fires the healer automatically when any op fails, pre-populated with the run ID, job name, failed step, and full stack trace

• Jobs with the hook attached at definition time — no UI toggle, every run is covered

What Happens During a Healing Run

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

The hook is fire-and-forget: it dispatches the failure and returns immediately, so the Dagster run's teardown is never blocked by the agent session. A typical healing run completes in 2–6 minutes from step failure to open PR.

The Fix PR

Each failed run produces exactly one PR, with every distinct error documented — the verbatim error string, the file changed, and the fix applied:

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.

1

Create the Agent YAML

Create the healer agent in .dinoai/agents/. The file must exist on the branch your Paradime workspace tracks (usually main) before the first trigger — the agent is loaded from the remote repository, not your local checkout.

.dinoai/agents/dagster-pipeline-healer.yml

name: dagster-pipeline-healer
version: 1

role: >
  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.

tools:
  mode: allowlist
  list:
    - read_file
    - search_files_and_directories
    - ripgrep_search
    - run_sql_query
    - run_terminal_command

slack:
  channel: "#dagster-dinoai-self-healing"

Merge the agent YAML to your default branch before triggering. DinoAI loads agent definitions from the default branch of the workspace repository — an agent YAML sitting on a feature branch, an open PR, or only in your local checkout is invisible to trigger_run, and the session will fail with no messages. Any later edits to the YAML also only take effect once they are merged.

Adapt the error → 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.

2

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.

3

Test It End-to-End

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

1. 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.

1. The break must be pushed to the branch your Paradime workspace tracks. The agent clones the remote repository — 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:

1. 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'

2. Slack — #dagster-dinoai-self-healing receives the triage summary, then progress updates

3. GitHub — the fix PR appears a few minutes later, citing the exact error string

Merge the PR to heal the pipeline and close the loop.

File Structure

your-repo/
├── dbt_project.yml
├── .dinoai/
│   └── agents/
│       └── dagster-pipeline-healer.yml   ← the healer agent
└── dagster_project/
    ├── definitions.py
    ├── jobs.py                           ← hooks attached at definition time
    └── hooks.py                          ← @failure_hook → trigger_run

If an agent session fails immediately with no messages, the API key likely belongs to the wrong workspace — trigger_run always runs the agent in the workspace that issued the key. Also note that a long-running dagster dev process keeps the environment variables it started with; restart it after rotating or switching keys.

Related Docs

• Agent-to-Agent Delegation — how invoke_agent and notify_parent_session work

• Programmable Agents — Quick Start — getting started with DinoAI agents

• Programmable Agents — YAML Configuration — full reference for agent config options

• Programmable Agents — Tools Reference — all available tools

• Slack Integration — connecting Slack to Paradime

• Paradime API & Credentials — where to find your API endpoint, key, and secret