> ## Documentation Index
> Fetch the complete documentation index at: https://docs.paradime.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Self-heal Dagster pipelines

> Build a DinoAI agent that recovers from failed Dagster ops by triaging the exception, patching the source, and opening a single fix PR per failed run.

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.

<Note>
  **Prerequisites**

  * [GitHub connected](/integrations/github/index) so the healer can create branches and open PRs against the workspace repository.
  * [Slack connected](/integrations/slack) 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](/products/dino-ai/programmable-agents/index).

  Estimated time: 30 minutes.
</Note>

## Steps

<Steps>
  <Step title="Create the agent">
    Build the healer [in the Agent UI](/guides/programmable-agents/build-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-healer`

    **Role**:

    ```text theme={"system"}
    Dagster Run Recovery Specialist focused on diagnosing and fixing
    failed Dagster runs of the F1 dbt + dlt pipeline in this analytics
    repo.
    ```

    **Goal**:

    ```text theme={"system"}
    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**:

    ```text theme={"system"}
    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_command`

    **Output**: Slack channel `#dagster-dinoai-self-healing`

    <Warning>
      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.
    </Warning>

    <Info>
      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.
    </Info>
  </Step>

  <Step title="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`:

    ````python title="dagster_project/hooks.py" lineNumbers theme={"system"}
    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:

    ```python title="dagster_project/jobs.py" lineNumbers theme={"system"}
    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   |

    <Info>
      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.
    </Info>

    <Warning>
      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.
    </Warning>
  </Step>

  <Step title="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:

    ```sql theme={"system"}
    -- 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.

    2. **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:

    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
  </Step>
</Steps>

<Check>
  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.
</Check>

## 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

<CardGroup cols={2}>
  <Card title="Bolt pipeline healer" href="/guides/programmable-agents/bolt-pipeline-healer" icon="workflow">
    The same pattern driven by Bolt self-healing.
  </Card>

  <Card title="Pipeline incident commander" href="/guides/programmable-agents/incident-commander" icon="bell">
    Multi-agent triage from Airflow failures.
  </Card>

  <Card title="Build an agent in the UI" href="/guides/programmable-agents/build-in-the-agent-ui" icon="bot">
    Create this agent visually instead of by hand.
  </Card>

  <Card title="Programmable Agents reference" href="/products/dino-ai/programmable-agents/index" icon="code">
    The agent YAML schema and tools.
  </Card>
</CardGroup>


## Related topics

- [DinoAI Bolt Pipeline Agent Self-Healing](/products/dino-ai/bolt-pipeline-agent/self-healing.md)
- [Set up self-healing](/guides/orchestrate-data-pipelines/set-up-self-healing.md)
- [Dagster](/integrations/dagster.md)
- [Bolt pipeline healer](/guides/programmable-agents/bolt-pipeline-healer.md)
- [Self-Healing settings for a Bolt schedule](/products/bolt/creating-schedules/self-healing.md)