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

# Pipeline Incident Commander –– Airflow

> Automatically trigger a multi-agent incident triage workflow from Airflow task failures or a manual CLI, then post a unified Slack summary with root cause, impact, owner, and next action.

Automatically triage a data pipeline incident by spawning three specialist sub-agents in parallel — a log analyser, a query profiler, and an owner notifier — then composing their findings into a single structured Slack post. The orchestrator runs the moment a failure is detected, with no manual intervention required.

This tutorial covers two trigger paths: an **Airflow `on_failure_callback`** that fires the incident commander automatically when any DAG task fails, and a **manual CLI trigger** for teams who want to invoke triage on demand.

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

  **Recommended reading**

  Before proceeding, read the Programmable Agents section under **Products → DinoAI**:

  * Quick Start
  * YAML Configuration
  * Tools Reference
  * Agent-to-Agent Delegation

  This tutorial uses `invoke_agent`, `notify_parent_session`, and `child_session_ids` — all covered in the Agent-to-Agent Delegation guide.

  **Integrations**

  The following must already be connected in Paradime:

  * **Slack** — the orchestrator and sub-agents post to `#incidents` via `post_slack_message`
</Info>

## What You'll Build

By the end of this guide you'll have:

* Four DinoAI agent YAMLs — one orchestrator (`incident-commander`) and three specialists (`log-analyzer`, `query-profiler`, `owner-notifier`)
* An Airflow `on_failure_callback` function that fires the incident commander automatically when any task in a DAG fails, pre-populated with the DAG name, task ID, run ID, and execution date
* A manual trigger script for ad-hoc incident triage

### What Happens During a Triage

Once the incident commander is triggered:

```
incident-commander (session A)
    ├─ invoke_agent("log-analyzer")    ─► session B  ─► pulls Bolt run logs
    ├─ invoke_agent("query-profiler")  ─► session C  ─► queries warehouse history
    └─ invoke_agent("owner-notifier")  ─► session D  ─► pings model owner in Slack

    B, C, D run in parallel
        ├─ each posts updates into the #incidents Slack thread
        └─ each calls notify_parent_session with their findings

    incident-commander resumes once all three have reported
        └─ posts unified ROOT CAUSE / IMPACT / OWNER / NEXT ACTION to #incidents
```

The commander never investigates itself — it only delegates and composes. None of the sub-agents have `invoke_agent` in their tool allowlist, so the graph is exactly two levels deep. This keeps incident response predictable and prevents runaway delegation chains.

### The Slack Incident Post

Once all three sub-agents have reported back, the incident commander posts a single structured message to `#incidents`:

```
🚨 INCIDENT TRIAGE — hourly_marts / fct_orders

ROOT CAUSE
  [log-analyzer findings]

IMPACT
  [query-profiler findings — top 3 slow/failing queries]

OWNER
  [owner-notifier findings — who was pinged and confirmed]

NEXT ACTION
  [commander synthesis — one concrete action item]
```

## Architecture Overview

```mermaid theme={"system"}
flowchart TD
    AF([Airflow on_failure_callback\nor manual CLI trigger]) --> IC

    IC([incident-commander\nsession A])
    IC --> LA([log-analyzer\nsession B])
    IC --> QP([query-profiler\nsession C])
    IC --> ON([owner-notifier\nsession D])

    LA --> LA1[list_bolt_schedules\nget_bolt_run_logs]
    LA --> LA2[notify_parent_session\n→ session A]

    QP --> QP1[run_sql_query\nlist_all_snowflake_databases]
    QP --> QP2[notify_parent_session\n→ session A]

    ON --> ON1[read_file · ripgrep_search\nfind meta.owner in schema YAML]
    ON --> ON2[post_slack_message\nping owner in #incidents]
    ON --> ON3[notify_parent_session\n→ session A]

    LA2 & QP2 & ON3 --> IC2([incident-commander resumes\nposts unified summary → #incidents])
```

<Steps>
  <Step title="Create the Agent YAMLs">
    Create all four agent files. The orchestrator and three sub-agents must all exist in `.dinoai/agents/` before the first trigger.

    **Orchestrator — `incident-commander`**

    ```yaml title=".dinoai/agents/incident-commander.yml" lineNumbers theme={"system"}
    name: incident-commander
    version: 1

    role: >
      Data Platform Incident Commander coordinating triage across log,
      warehouse, and people axes during a pipeline incident.

    goal: >
      When triggered with an incident summary: spawn the log-analyzer,
      query-profiler, and owner-notifier sub-agents in parallel via
      invoke_agent. Wait for their findings via the callback pattern and
      post a unified incident summary to the incident Slack thread with
      ROOT CAUSE / IMPACT / OWNER / NEXT ACTION.

      Structure the final Slack post as follows:

      🚨 INCIDENT TRIAGE — <schedule or DAG name> / <suspect model>

      ROOT CAUSE
        <log-analyzer findings verbatim>

      IMPACT
        <query-profiler findings verbatim — top 3 queries>

      OWNER
        <owner-notifier findings — who was pinged and confirmed>

      NEXT ACTION
        <one concrete action item synthesised from all three findings>

    backstory: >
      You delegate aggressively — you do not investigate yourself. You only
      compose the final report. You always include the three sub-agent
      findings verbatim under their respective headings before writing the
      synthesis. You never post a partial report — wait for all three
      sub-agents to notify before composing.

    tools:
      mode: allowlist
      list:
        - post_slack_message
        - invoke_agent

    agents_squad:
      - log-analyzer
      - query-profiler
      - owner-notifier

    slack:
      channel: "#incidents"
    ```

    **Sub-agent — `log-analyzer`**

    ```yaml title=".dinoai/agents/log-analyzer.yml" lineNumbers theme={"system"}
    name: log-analyzer
    version: 1

    role: Bolt log specialist.

    goal: >
      Pull the relevant Bolt run logs for the failing schedule and report the
      failing step plus a one-line root cause. Notify the parent session with
      your findings once complete.

    backstory: >
      You are concise. One paragraph maximum. You only report facts from the
      logs — never speculation.

    tools:
      mode: allowlist
      list:
        - list_bolt_schedules
        - get_bolt_run_logs
        - notify_parent_session
    ```

    **Sub-agent — `query-profiler`**

    ```yaml title=".dinoai/agents/query-profiler.yml" lineNumbers theme={"system"}
    name: query-profiler
    version: 1

    role: Warehouse query specialist.

    goal: >
      Run information-schema queries to identify slow or failing warehouse
      queries in the last hour. Report the top 3 with their owner model.
      Notify the parent session with your findings once complete.

    backstory: >
      You only return facts from the warehouse, never speculation. Always
      include query ID, execution time, and model name in your findings.

    tools:
      mode: allowlist
      list:
        - run_sql_query
        - list_all_snowflake_databases
        - notify_parent_session
    ```

    **Sub-agent — `owner-notifier`**

    ```yaml title=".dinoai/agents/owner-notifier.yml" lineNumbers theme={"system"}
    name: owner-notifier
    version: 1

    role: People-routing specialist.

    goal: >
      For the model named in the trigger, read its schema YAML to find the
      owner field (meta.owner) and post a direct Slack ping in the incident
      channel. Notify the parent session with the owner you contacted.

    backstory: >
      You always confirm the owner exists in the YAML before pinging. If no
      owner is found, report that explicitly rather than pinging a guess.

    tools:
      mode: allowlist
      list:
        - read_file
        - search_files_and_directories
        - ripgrep_search
        - post_slack_message
        - notify_parent_session
    ```

    <Info>
      None of the sub-agents have `invoke_agent` in their tool allowlist. This keeps the delegation graph exactly two levels deep — the commander is the only delegator. Sub-agents cannot spawn further children, which makes incident response predictable and prevents runaway chains.
    </Info>
  </Step>

  <Step title="Trigger Path A — Airflow `on_failure_callback`">
    This is the recommended path for teams running Airflow. The callback fires automatically when any task in your DAG fails, extracts the task and DAG context from Airflow's `context` dict, and hands it directly to the incident commander so the triage message is pre-populated with real incident details.

    Create `dags/callbacks/incident_commander.py`:

    ```python title="dags/callbacks/incident:commander.py" lineNumbers theme={"system"}
    """
    incident_commander.py
    ---------------------
    Airflow on_failure_callback that triggers the `incident-commander`
    DinoAI agent when a DAG task fails.

    Attach to any DAG or individual task:

        # Whole DAG — fires on any task failure
        with DAG(
            "hourly_marts",
            on_failure_callback=trigger_incident_commander,
            ...
        ) as dag:
            ...

        # Single task — fires only when this task fails
        run_fct_orders = DbtRunOperator(
            task_id="fct_orders",
            on_failure_callback=trigger_incident_commander,
            ...
        )

    Environment variables (required — set in Airflow Connections or Variables):
        PARADIME_API_ENDPOINT  – Paradime API endpoint
        PARADIME_API_KEY       – Paradime API key
        PARADIME_API_SECRET    – Paradime API secret

    Optional:
        INCIDENT_SLACK_CHANNEL – Slack channel for incident posts (default: #incidents)
        SUSPECT_MODEL          – dbt™ model to focus triage on (default: extracted
                                 from failed task_id if it matches a model name)
    """

    from __future__ import annotations

    import os
    import threading
    from typing import Any

    from paradime import Paradime

    AGENT_NAME     = "incident-commander"
    SLACK_CHANNEL  = os.environ.get("INCIDENT_SLACK_CHANNEL", "#incidents")


    def _build_client() -> Paradime:
        return Paradime(
            api_endpoint=os.environ["PARADIME_API_ENDPOINT"],
            api_key=os.environ["PARADIME_API_KEY"],
            api_secret=os.environ["PARADIME_API_SECRET"],
        )


    def _build_incident_message(context: dict[str, Any]) -> str:
        """
        Build the incident trigger message from Airflow task context.
        Airflow passes a rich context dict to on_failure_callback — we extract
        the most useful fields for the triage agent.
        """
        dag_id        = context.get("dag").dag_id if context.get("dag") else "unknown"
        task_id       = context.get("task_instance").task_id if context.get("task_instance") else "unknown"
        run_id        = context.get("run_id", "unknown")
        execution_date = str(context.get("execution_date", "unknown"))
        exception     = str(context.get("exception", "unknown"))
        log_url       = context.get("task_instance").log_url if context.get("task_instance") else "unavailable"

        # Try to infer the suspect dbt™ model from the task ID or env override
        suspect_model = (
            os.environ.get("SUSPECT_MODEL")
            or task_id  # Airflow task IDs often match dbt™ model names directly
        )

        return (
            f"INCIDENT: Airflow task failure detected.\n\n"
            f"DAG:            {dag_id}\n"
            f"Failed task:    {task_id}\n"
            f"Run ID:         {run_id}\n"
            f"Execution date: {execution_date}\n"
            f"Exception:      {exception}\n"
            f"Log URL:        {log_url}\n"
            f"Suspect model:  {suspect_model}\n\n"
            f"Coordinate triage across log-analyzer, query-profiler, and "
            f"owner-notifier. Post a unified ROOT CAUSE / IMPACT / OWNER / "
            f"NEXT ACTION summary to {SLACK_CHANNEL}."
        )


    def trigger_incident_commander(context: dict[str, Any]) -> None:
        """
        Airflow on_failure_callback. Fires asynchronously so it does not block
        Airflow's task runner thread while the agent session runs.

        Attach to a DAG:   on_failure_callback=trigger_incident_commander
        Attach to a task:  on_failure_callback=trigger_incident_commander
        """
        def _run() -> None:
            try:
                paradime = _build_client()
                message  = _build_incident_message(context)

                dag_id  = context.get("dag").dag_id if context.get("dag") else "unknown"
                task_id = context.get("task_instance").task_id if context.get("task_instance") else "unknown"

                print(f"[incident-commander] 🚨  Triggering triage for {dag_id}/{task_id} …")

                run = paradime.dinoai_agents.trigger_run_and_wait(
                    agent=AGENT_NAME,
                    message=message,
                    slack_channel=SLACK_CHANNEL,
                    timeout=1800,   # 30 minutes max
                )

                print(
                    f"[incident-commander] ✅  Triage complete for {dag_id}/{task_id}.\n"
                    f"{run.messages[-1].content}"
                )

            except Exception as exc:  # noqa: BLE001
                # Log but never re-raise — a callback failure must not mask the original error
                print(f"[incident-commander] ❌  Failed to trigger triage: {exc}")

        # Run in a daemon thread so Airflow doesn't wait for it
        thread = threading.Thread(target=_run, daemon=True)
        thread.start()
    ```

    Then attach the callback to your DAG:

    ```python title="dags/hourly_marts.py" lineNumbers theme={"system"}
    from airflow import DAG
    from airflow.utils.dates import days_ago
    from callbacks.incident_commander import trigger_incident_commander

    with DAG(
        dag_id="hourly_marts",
        schedule_interval="@hourly",
        start_date=days_ago(1),
        on_failure_callback=trigger_incident_commander,   # ← fires on any task failure
        catchup=False,
    ) as dag:
        ...
    ```

    <Info>
      The callback runs in a **daemon thread** so it does not block Airflow's task runner while the agent session runs (which can take several minutes). The original task failure is always surfaced normally in Airflow regardless of whether the triage agent succeeds or fails.
    </Info>

    <Info>
      To attach the callback to a **single task** rather than the whole DAG, set `on_failure_callback=trigger_incident_commander` on the operator directly. This is useful when only certain high-priority models should trigger a full triage.
    </Info>

    <Info>
      `PARADIME_API_ENDPOINT`, `PARADIME_API_KEY`, and `PARADIME_API_SECRET` should be stored as **Airflow Variables** or in an **Airflow Connection**, not hardcoded. Retrieve them with `Variable.get("PARADIME_API_KEY")` if you prefer the Airflow Variables pattern over environment variables.
    </Info>
  </Step>

  <Step title="Trigger Path B — Manual CLI Trigger">
    For teams not running Airflow, or for ad-hoc incident triage when a failure is spotted manually, create `scripts/trigger_incident.py`:

    ```python title="scripts/trigger:incident.py" lineNumbers theme={"system"}
    """
    trigger_incident.py
    -------------------
    Manual trigger for the `incident-commander` DinoAI agent.

    Usage:
        python scripts/trigger_incident.py \
            --schedule "hourly_marts" \
            --model fct_orders \
            --context "Failed 3 times in the last 30 minutes"

    Environment variables (required):
        PARADIME_API_ENDPOINT  – Paradime API endpoint
        PARADIME_API_KEY       – Paradime API key
        PARADIME_API_SECRET    – Paradime API secret
    """

    import argparse
    import os
    import sys

    from paradime import Paradime

    AGENT_NAME    = "incident-commander"
    SLACK_CHANNEL = os.environ.get("INCIDENT_SLACK_CHANNEL", "#incidents")


    def _parse_args() -> argparse.Namespace:
        parser = argparse.ArgumentParser(description="Trigger the incident-commander agent.")
        parser.add_argument("--schedule", required=True, help="Bolt schedule or DAG name that is failing")
        parser.add_argument("--model",    required=True, help="Suspect dbt™ model name (e.g. fct_orders)")
        parser.add_argument("--context",  default="",    help="Any additional context about the failure")
        return parser.parse_args()


    def main() -> None:
        args = _parse_args()

        required = {
            "PARADIME_API_ENDPOINT": os.environ.get("PARADIME_API_ENDPOINT"),
            "PARADIME_API_KEY":      os.environ.get("PARADIME_API_KEY"),
            "PARADIME_API_SECRET":   os.environ.get("PARADIME_API_SECRET"),
        }
        missing = [k for k, v in required.items() if not v]
        if missing:
            print(f"ERROR: Missing env vars: {', '.join(missing)}", file=sys.stderr)
            raise SystemExit(1)

        paradime = Paradime(
            api_endpoint=required["PARADIME_API_ENDPOINT"],
            api_key=required["PARADIME_API_KEY"],
            api_secret=required["PARADIME_API_SECRET"],
        )

        message = (
            f"INCIDENT: schedule '{args.schedule}' is failing. "
            f"Suspect model is {args.model}. "
            f"{args.context}\n\n"
            f"Coordinate triage across log-analyzer, query-profiler, and "
            f"owner-notifier. Post a unified ROOT CAUSE / IMPACT / OWNER / "
            f"NEXT ACTION summary to {SLACK_CHANNEL}."
        )

        print(f"[incident-commander] 🚨  Triggering triage for {args.schedule}/{args.model} …")

        run = paradime.dinoai_agents.trigger_run_and_wait(
            agent=AGENT_NAME,
            message=message,
            slack_channel=SLACK_CHANNEL,
            timeout=1800,
        )

        print("\n" + "=" * 72)
        print("INCIDENT TRIAGE REPORT")
        print("=" * 72)
        print(run.messages[-1].content)
        print("=" * 72)


    if __name__ == "__main__":
        main()
    ```

    Run it from your terminal:

    ```bash theme={"system"}
    export PARADIME_API_ENDPOINT=<your-endpoint>
    export PARADIME_API_KEY=<your-key>
    export PARADIME_API_SECRET=<your-secret>

    python scripts/trigger_incident.py \
      --schedule "hourly_marts" \
      --model fct_orders \
      --context "Failed 3 times in the last 30 minutes. Suspect partition filter issue."
    ```
  </Step>

  <Step title="Watch Sub-Agents in Flight">
    Both trigger paths block until the full triage is complete. If you want to stream sub-agent progress in real time as each specialist reports back, use the non-blocking `trigger_run` pattern and poll `child_session_ids`:

    ```python title="scripts/watch:incident.py" lineNumbers theme={"system"}
    """
    watch_incident.py
    -----------------
    Triggers the incident commander and streams sub-agent progress in real time.
    Each child session is polled independently so you see findings as they arrive
    rather than waiting for all three to complete.
    """

    import os
    import time

    from paradime import Paradime
    from paradime.apis.dinoai_agents.types import DinoaiAgentRunStatus

    paradime = Paradime(
        api_endpoint=os.environ["PARADIME_API_ENDPOINT"],
        api_key=os.environ["PARADIME_API_KEY"],
        api_secret=os.environ["PARADIME_API_SECRET"],
    )

    # Trigger without waiting — returns immediately with the parent session ID
    result    = paradime.dinoai_agents.trigger_run(
        agent="incident-commander",
        message=(
            "INCIDENT: schedule 'hourly_marts' has failed 3 times in the last "
            "30 minutes. Suspect model is fct_orders. Coordinate the triage "
            "and post a unified summary in #incidents."
        ),
    )
    parent_id = result.agent_session_id
    print(f"[commander] 🚨  Session started: {parent_id}\n")

    # Poll the parent — as sub-agents are spawned, child_session_ids populates
    seen_children: set[str] = set()

    while True:
        parent = paradime.dinoai_agents.get_run(agent_session_id=parent_id)

        # Stream any newly spawned child sessions
        for child_id in (parent.child_session_ids or []):
            if child_id in seen_children:
                continue
            seen_children.add(child_id)

            child = paradime.dinoai_agents.get_run(agent_session_id=child_id)
            print(f"--- sub-agent spawned: {child_id} ({child.status.value}) ---")
            for msg in child.messages[-3:]:
                print(f"  [{msg.role}] {msg.content[:200]}")

        if parent.status in (DinoaiAgentRunStatus.COMPLETED, DinoaiAgentRunStatus.FAILED):
            break

        time.sleep(15)

    print("\n" + "=" * 72)
    print("FINAL TRIAGE REPORT")
    print("=" * 72)
    print(parent.messages[-1].content)
    ```

    <Info>
      `child_session_ids` populates progressively as the orchestrator spawns each sub-agent. The first poll may return an empty list — this is expected. The three child sessions will appear within the first 30–60 seconds as the commander issues its `invoke_agent` calls.
    </Info>
  </Step>
</Steps>

## Execution Flow

When the incident commander receives the trigger message:

1. It calls `invoke_agent` three times in parallel — one for each specialist — passing its own session ID as the callback target
2. Each sub-agent runs independently, posting updates to the `#incidents` Slack thread as it works
3. Each sub-agent calls `notify_parent_session` with its findings when complete
4. The commander resumes once all three callbacks have arrived, then composes and posts the unified triage report

The entire triage typically completes in **3–8 minutes** depending on log volume and warehouse query history depth.

### Set Your Environment Variables

The callback and trigger scripts require three variables. For the Airflow path, set these as Airflow Variables or in your Airflow environment. For the manual path, export them in your shell.

| Variable                 | Description                                                                                 |
| ------------------------ | ------------------------------------------------------------------------------------------- |
| `PARADIME_API_ENDPOINT`  | Your Paradime API endpoint                                                                  |
| `PARADIME_API_KEY`       | Your Paradime API key                                                                       |
| `PARADIME_API_SECRET`    | Your Paradime API secret                                                                    |
| `INCIDENT_SLACK_CHANNEL` | *(optional)* Slack channel for incident posts — defaults to `#incidents`                    |
| `SUSPECT_MODEL`          | *(optional, Airflow only)* Override the suspect model name — defaults to the failed task ID |

<Info>
  Your Paradime API endpoint, key, and secret are available under Workspace Settings → API. Make sure the key has `DinoAI agent API` capabilities enabled.
</Info>

### File Structure

Your repository should look like this after completing the setup:

```
your-repo/
├── dbt_project.yml
├── .dinoai/
│   └── agents/
│       ├── incident-commander.yml
│       ├── log-analyzer.yml
│       ├── query-profiler.yml
│       └── owner-notifier.yml
├── dags/
│   ├── hourly_marts.py                   ← example DAG with callback attached
│   └── callbacks/
│       └── incident_commander.py         ← Airflow on_failure_callback
└── scripts/
    ├── trigger_incident.py               ← manual CLI trigger
    └── watch_incident.py                 ← real-time sub-agent streaming
```

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

- [Trigger Datadog Incidents](/products/bolt/creating-schedules/trigger-integrations/trigger-datadog-incidents.md)
- [Airflow](/integrations/airflow.md)
- [Datadog Alerts](/integrations/elementary-data/sending-alerts/datadog-alerts.md)
- [Incident Templates](/products/bolt/creating-schedules/templates/incident-templates/index.md)
- [Create Incident in incident.io](/products/bolt/creating-schedules/templates/incident-templates/create-incident-in-incident.io.md)