> For the complete documentation index, see [llms.txt](https://docs.paradime.io/app-help/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.paradime.io/app-help/guides-new/programmable-agents/self-healing-pipelines-dagster.md).

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

{% hint style="info" icon="compass" %}
**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`
  {% endhint %}

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

{% stepper %}
{% step %}

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

{% code title=".dinoai/agents/dagster-pipeline-healer.yml" lineNumbers="true" %}

```yaml
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"
```

{% endcode %}

{% hint style="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.
{% endhint %}

{% hint style="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.
{% endhint %}
{% endstep %}

{% step %}

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

{% code title="dagster\_project/hooks.py" lineNumbers="true" %}

````python
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}"
    )
````

{% endcode %}

Then attach the hook to your jobs at definition time:

{% code title="dagster\_project/jobs.py" lineNumbers="true" %}

```python
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
)
```

{% endcode %}

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   |

{% hint style="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.
{% endhint %}

{% hint style="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.
{% endhint %}
{% endstep %}

{% step %}

#### 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
-- 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.
{% endstep %}
{% endstepper %}

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

{% hint style="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.
{% endhint %}

**Related Docs**

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


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://docs.paradime.io/app-help/guides-new/programmable-agents/self-healing-pipelines-dagster.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.