> ## 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-healing Dagster pipelines with DinoAI

> Automatically recover from failed Dagster ops by dispatching every failure to a DinoAI programmable agent that triages, patches, and opens a fix PR on GitHub.

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.

<Info>
  **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`
</Info>

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

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

    ```yaml title=".dinoai/agents/dagster-pipeline-healer.yml" lineNumbers theme={"system"}
    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"
    ```

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

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

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

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

**Related Docs**

* [**Agent-to-Agent Delegation** — how `invoke_agent` and `notify_parent_session` work](/products/dino-ai/programmable-agents/agent-to-agent-delegation)
* [**Programmable Agents — Quick Start** — getting started with DinoAI agents](/products/dino-ai/programmable-agents/quick-start)
* [**Programmable Agents — YAML Configuration** — full reference for agent config options](/products/dino-ai/programmable-agents/yaml-configuration)
* [**Programmable Agents — Tools Reference** — all available tools](/products/dino-ai/programmable-agents/tools-reference)
* [**Slack Integration** — connecting Slack to Paradime](/integrations/slack)
* [**Paradime API & Credentials** — where to find your API endpoint, key, and secret](/developers/generate-api-keys-legacy)


## Related topics

- [DinoAI Bolt Pipeline Agent Self-Healing](/products/dino-ai/bolt-pipeline-agent/self-healing.md)
- [Self-Healing settings for a Bolt schedule](/products/bolt/creating-schedules/self-healing.md)
- [Bolt Pipeline Healer](/guides/programmable-agents/bolt-pipeline-healer.md)
- [Bolt Pipeline Agent](/products/dino-ai/bolt-pipeline-agent/index.md)
- [Pipeline Insights](/products/dino-ai/bolt-pipeline-agent/production-pipelines.md)