# dbt™ Test Maintainer

A specialist DinoAI agent that writes missing dbt™ tests for any model added or changed in a pull request, runs them to confirm they pass, and commits the result directly to the PR branch. Triggered automatically by a GitHub Actions workflow whenever a `.sql` file is touched — so test coverage gaps never reach `main`.

The agent can also be invoked as a sub-agent by the PR reviewer, allowing the reviewer to delegate test-writing as part of a broader code review rather than just flagging the gap in its verdict.

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

**GitHub**

* Write access to the repository you want to run the workflow on
* Ability to add repository secrets and create GitHub Actions workflows

**Recommended reading**

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

* Quick Start
* YAML Configuration
* Tools Reference
* Agent-to-Agent Delegation — required reading if you plan to invoke this agent from the PR reviewer

**Integrations**

The following must already be connected in Paradime:

* **Slack** — the agent posts a test summary to `#data-quality` via `post_slack_message`
  {% endhint %}

## What You'll Build

By the end of this guide you'll have:

* A `test-maintainer` DinoAI agent YAML that reads changed models, infers appropriate tests, writes them to the schema YAML, runs `dbt test`, iterates until green, and commits the result
* A Python driver script that collects changed `.sql` files from the PR and triggers one agent session per model in parallel
* A GitHub Actions workflow that runs automatically on every PR that adds or modifies a dbt™ model

### What the Agent Does Per Model

Once triggered for a model, the agent works through five steps without stopping:

```
STEP 1 — Read the model's .sql file and its schema YAML
STEP 2 — Inspect upstream sources to infer test candidates
          (primary keys, not_null columns, accepted values, FK relationships)
STEP 3 — Write the missing tests into the model's schema YAML
STEP 4 — Run dbt test --select <model> and iterate until all tests pass
STEP 5 — Commit the updated YAML to the PR branch and post a summary
          to #data-quality and back to the PR as a comment
```

The agent always runs `dbt test` before reporting completion — it never commits tests it hasn't verified pass. If a test fails after being written, the agent diagnoses the failure, adjusts the test definition or SQL, and retries before giving up.

{% hint style="info" %}
The absence of `invoke_agent` from the tool allowlist means this agent cannot delegate further. When used as a child of the PR reviewer, the delegation graph stays exactly two levels deep — the reviewer delegates, the test maintainer executes.
{% endhint %}

### Tests the Agent Writes

The agent infers test candidates by reading the model SQL and upstream source definitions. It writes the following categories of tests:

| Test Type             | When Written                                                                                   |
| --------------------- | ---------------------------------------------------------------------------------------------- |
| `unique` + `not_null` | Any column that appears to be a primary key based on name (`_id`, `_key`, `_sk`) or usage      |
| `not_null`            | Columns used in JOIN conditions or GROUP BY clauses in downstream models                       |
| `accepted_values`     | Columns with a small, finite set of values inferred from CASE WHEN or WHERE clauses            |
| `relationships`       | Foreign key columns that reference a known `ref()` or `source()` model                         |
| Custom generic tests  | Where dbt™ built-in tests are insufficient and a simple SQL assertion would be more expressive |

The agent never writes a test it cannot justify from the SQL. When a test candidate is ambiguous it adds a `# TODO: confirm test logic with owner` comment in the YAML rather than guessing.

### Architecture Overview

```mermaid
flowchart TD
    A([Pull Request opened / updated\n.sql file changed]) --> B[GitHub Actions\ntest-maintainer workflow]
    B --> C[driver script\ntest_maintainer.py]
    C --> D[Collect changed .sql files\nfrom git diff]
    C --> E[Trigger one DinoAI session\nper changed model\nin parallel]

    E --> F1([test-maintainer\nsession — model A])
    E --> F2([test-maintainer\nsession — model B])

    F1 & F2 --> G[read_file · ripgrep_search\nread model SQL and schema YAML]
    F1 & F2 --> H[run_sql_query\ninspect upstream sources]
    F1 & F2 --> I[write_file\nwrite tests to schema YAML]
    F1 & F2 --> J[run_terminal_command\ndbt test --select model]
    F1 & F2 --> K[run_terminal_command\ngit commit · push to PR branch]
    F1 & F2 --> L[post_slack_message\n→ #data-quality]

    K & L --> M[driver script\nposts completion comment to PR]
```

## How It Works

When a PR is opened or updated, GitHub Actions runs `test_maintainer.py`, which collects all changed `.sql` files from the diff and fires one `test-maintainer` agent session per model in parallel. Each session reads the model, infers and writes tests, runs `dbt test`, iterates until green, and commits to the PR branch. The driver script polls all sessions concurrently and posts a single completion comment to the PR once every session has finished, summarising what was written per model.

{% stepper %}
{% step %}

### Create the Agent YAML

Create the following file in your repository at `.dinoai/agents/test-maintainer.yml`.

{% code title=".dinoai/agents/test-maintainer.yml" lineNumbers="true" %}

```yaml
name: test-maintainer
version: 1

role: >
  dbt™ Test Specialist responsible for writing and updating tests across
  analytics models.

goal: >
  Ensure the target model has appropriate, passing tests committed to the
  repo. Always write meaningful tests — not just for coverage.

  ── STEP 1: READ THE MODEL ────────────────────────────────────────────────

  Read the target model's .sql file in full using read_file. Then find its
  schema YAML (e.g. _<model_name>.yml or schema.yml in the same directory)
  and read the existing column and test definitions.

  ── STEP 2: INFER TEST CANDIDATES ────────────────────────────────────────

  Inspect the model SQL and upstream sources to identify what tests are
  appropriate. For each column, consider:
    - Does the name suggest a primary key? (_id, _key, _sk, _pk) → unique + not_null
    - Is the column used in a JOIN or GROUP BY in a downstream model? → not_null
    - Does a CASE WHEN or WHERE clause imply a finite value set? → accepted_values
    - Does the column reference a known ref() or source()? → relationships

  Use run_sql_query to inspect the actual data distribution for ambiguous
  columns before deciding. Never write a test you cannot justify from the
  SQL or the data.

  ── STEP 3: WRITE THE TESTS ──────────────────────────────────────────────

  Write the inferred tests into the model's schema YAML using write_file.
  Preserve all existing formatting and ordering. Only add — never remove
  or reorder existing test entries. If a test candidate is ambiguous,
  add a comment `# TODO: confirm test logic with owner` on the line above
  the test entry rather than omitting it entirely.

  ── STEP 4: RUN AND ITERATE ──────────────────────────────────────────────

  Run the tests via run_terminal_command:
    dbt test --select <model_name>

  If any tests fail:
    1. Read the failure output carefully.
    2. Diagnose whether the failure is a bad test definition or a genuine
       data quality issue.
    3. If a bad definition: fix the test in the YAML and re-run.
    4. If a genuine data quality issue: add a severity: warn to the test
       config so it does not block the pipeline, and add a comment
       `# DATA ISSUE: <description> — investigate before promoting to error`
       above the test entry.
    5. Iterate until all tests are either passing or set to severity: warn
       with an explanation comment.

  NEVER commit tests that are failing at severity: error.

  ── STEP 5: COMMIT AND NOTIFY ────────────────────────────────────────────

  Stage and commit all changed YAML files to the PR branch:
    git add <changed_yaml_files>
    git commit -m "test: add missing tests for <model_name> [DinoAI]"
    git push origin HEAD:<branch_name>

  Always use `git push origin HEAD:<branch_name>` — never plain `git push`.

  Then post a summary to #data-quality via post_slack_message:

    ✅ *Test Maintainer — <model_name>*
    *Tests added:* <N> (<list of test names>)
    *Tests passing:* <N> / <N>
    *Data quality warnings:* <N> (severity: warn — investigate separately)
    *Committed to branch:* `<branch_name>`

  If invoked as a sub-agent, call notify_parent_session with the same
  summary so the parent agent (e.g. pr-reviewer) can include the result
  in its own report.

  ── GUARDRAILS ────────────────────────────────────────────────────────────

  - NEVER commit tests that are failing at severity: error.
  - NEVER remove or modify existing passing tests.
  - NEVER guess at accepted_values without checking the actual data via
    run_sql_query first.
  - NEVER reformat or reorder unchanged YAML sections.
  - If dbt test output is ambiguous, re-run with --debug before concluding.
  - If the schema YAML does not exist yet, create it following the dbt™
    schema YAML convention for the project (check an existing file first).

backstory: >
  You are precise and methodical. You read the model and its upstream
  sources before writing tests. You always run dbt test before reporting
  completion. You write tests that mean something — not just not_null on
  every column for the sake of coverage. A test that fires on real bad
  data is worth ten cosmetic tests.

tools:
  mode: allowlist
  list:
    - read_file
    - write_file
    - run_sql_query
    - run_terminal_command
    - search_files_and_directories
    - ripgrep_search
    - post_slack_message
    - notify_parent_session

slack:
  channel: "#data-quality"
```

{% endcode %}

{% hint style="info" %}
`notify_parent_session` is included in the allowlist so this agent can be invoked as a sub-agent by the PR reviewer. When triggered directly from the GitHub Actions workflow, the tool is present but never called — it only activates when a parent session ID is passed via `invoke_agent`.
{% endhint %}
{% endstep %}

{% step %}

### Create the Driver Script

Create `scripts/test_maintainer.py`. This script runs inside GitHub Actions and is responsible for:

* Collecting all changed `.sql` files from the PR diff
* Firing one `test-maintainer` agent session per changed model in parallel
* Posting a "started" comment to the PR immediately
* Polling all sessions concurrently until each completes
* Posting a single completion comment to the PR with a per-model summary

{% code title="scripts/test:maintainer.py" lineNumbers="true" %}

```python
"""
test_maintainer.py
------------------
GitHub Actions driver for the `test-maintainer` DinoAI agent.

Fires one agent session per changed dbt™ model in parallel, polls until
all sessions complete, and posts a summary back to the PR.

Environment variables (required):
    PARADIME_API_ENDPOINT  – Paradime API endpoint
    PARADIME_API_KEY       – Paradime API key
    PARADIME_API_SECRET    – Paradime API secret
    GITHUB_EVENT_PATH      – Set automatically by GitHub Actions
    GITHUB_TOKEN           – Set automatically by GitHub Actions
    GITHUB_REPOSITORY      – Set automatically by GitHub Actions
"""

import json
import os
import sys
import subprocess
import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed

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

AGENT_NAME    = "test-maintainer"
TIMEOUT       = 1800   # 30 minutes per session
POLL_INTERVAL = 20


def _build_client() -> Paradime:
    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)
    return Paradime(
        api_endpoint=required["PARADIME_API_ENDPOINT"],
        api_key=required["PARADIME_API_KEY"],
        api_secret=required["PARADIME_API_SECRET"],
    )


def _post_pr_comment(repo: str, pr_number: int, body: str, gh_token: str) -> None:
    req = urllib.request.Request(
        f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments",
        data=json.dumps({"body": body}).encode(),
        headers={
            "Authorization": f"Bearer {gh_token}",
            "Accept":        "application/vnd.github+json",
        },
        method="POST",
    )
    urllib.request.urlopen(req).read()


def _model_name_from_path(path: str) -> str:
    """Extract the dbt™ model name from a file path, e.g. models/marts/fct_orders.sql → fct_orders."""
    return os.path.splitext(os.path.basename(path))[0]


def _run_single_model(
    paradime: Paradime,
    model_name: str,
    sql_path: str,
    branch_name: str,
) -> dict:
    """
    Trigger a test-maintainer session for one model, poll until completion,
    and return a result dict with model name, status, and final output.
    """
    message = (
        f"Add missing dbt™ tests for model `{model_name}` (file: `{sql_path}`).\n\n"
        f"Branch: {branch_name}\n\n"
        f"Read the model SQL and its schema YAML, infer appropriate tests "
        f"(unique, not_null, accepted_values, relationships), write them, "
        f"run `dbt test --select {model_name}` and iterate until all tests "
        f"pass or are set to severity: warn with an explanation. "
        f"Commit the changes to branch `{branch_name}` using "
        f"`git push origin HEAD:{branch_name}` and post a summary to #data-quality."
    )

    trigger    = paradime.dinoai_agents.trigger_run(agent=AGENT_NAME, message=message)
    session_id = trigger.agent_session_id
    print(f"[test-maintainer] 🚀  {model_name} → session {session_id}")

    start = time.time()
    while True:
        run     = paradime.dinoai_agents.get_run(agent_session_id=session_id)
        elapsed = int(time.time() - start)

        if run.status == DinoaiAgentRunStatus.COMPLETED:
            print(f"[test-maintainer] ✅  {model_name} completed ({elapsed}s)")
            return {
                "model":   model_name,
                "status":  "completed",
                "output":  run.messages[-1].content if run.messages else "(no output)",
                "session": session_id,
            }

        if run.status == DinoaiAgentRunStatus.FAILED:
            last = run.messages[-1].content if run.messages else "no output"
            print(f"[test-maintainer] ❌  {model_name} failed ({elapsed}s)")
            return {
                "model":   model_name,
                "status":  "failed",
                "output":  last,
                "session": session_id,
            }

        if elapsed > TIMEOUT:
            print(f"[test-maintainer] ⏱  {model_name} timed out")
            return {
                "model":   model_name,
                "status":  "timeout",
                "output":  f"Timed out after {TIMEOUT}s",
                "session": session_id,
            }

        print(f"[test-maintainer]    {model_name} — {run.status.value} ({elapsed}s) … {POLL_INTERVAL}s")
        time.sleep(POLL_INTERVAL)


def main() -> None:
    paradime  = _build_client()
    gh_token  = os.environ.get("GITHUB_TOKEN", "")
    repo      = os.environ.get("GITHUB_REPOSITORY", "")

    # 1. Read PR context
    with open(os.environ["GITHUB_EVENT_PATH"]) as f:
        event = json.load(f)

    pr          = event["pull_request"]
    pr_number   = pr["number"]
    pr_title    = pr["title"]
    branch_name = pr["head"]["ref"]
    base_sha    = pr["base"]["sha"]
    head_sha    = pr["head"]["sha"]

    # 2. Find changed .sql files
    changed = subprocess.check_output(
        ["git", "diff", "--name-only", f"{base_sha}...{head_sha}"]
    ).decode().splitlines()

    sql_files = [f for f in changed if f.endswith(".sql") and "models/" in f]

    if not sql_files:
        print("[test-maintainer] No dbt™ model SQL files changed. Skipping.")
        raise SystemExit(0)

    models = [(_model_name_from_path(p), p) for p in sql_files]

    print(f"[test-maintainer] Found {len(models)} changed model(s):")
    for name, path in models:
        print(f"  - {name}  ({path})")

    # 3. Post "started" comment to PR
    if gh_token:
        model_list = "\n".join(f"- `{name}` (`{path}`)" for name, path in models)
        _post_pr_comment(
            repo, pr_number,
            f"🧪 **DinoAI test maintainer started** — PR #{pr_number}: {pr_title}\n\n"
            f"Writing and running missing tests for {len(models)} model(s):\n{model_list}\n\n"
            f"Changes will be committed to `{branch_name}`. Check `#data-quality` for live updates.",
            gh_token,
        )

    # 4. Trigger all sessions in parallel
    results = []
    with ThreadPoolExecutor(max_workers=len(models)) as executor:
        futures = {
            executor.submit(_run_single_model, paradime, name, path, branch_name): name
            for name, path in models
        }
        for future in as_completed(futures):
            results.append(future.result())

    # 5. Post completion comment to PR
    if gh_token:
        completed = [r for r in results if r["status"] == "completed"]
        failed    = [r for r in results if r["status"] != "completed"]

        sections = []
        for r in results:
            icon = "✅" if r["status"] == "completed" else "❌"
            sections.append(
                f"### {icon} `{r['model']}`\n\n"
                f"{r['output']}\n\n"
                f"*Session: `{r['session']}`*"
            )

        footer = (
            f"**{len(completed)}/{len(results)} models** had tests written and committed to `{branch_name}`."
        )
        if failed:
            footer += f"\n\n⚠️ {len(failed)} session(s) did not complete — check `#data-quality` for details."

        _post_pr_comment(
            repo, pr_number,
            f"🧪 **DinoAI test maintainer — complete**\n\n"
            + "\n\n---\n\n".join(sections)
            + f"\n\n---\n\n{footer}",
            gh_token,
        )

    # Exit with error if any session failed so the workflow is marked red
    if any(r["status"] != "completed" for r in results):
        raise SystemExit(1)


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

{% endcode %}

{% hint style="info" %}
Sessions run in parallel via `ThreadPoolExecutor` — if a PR changes three models, all three agent sessions start at the same time. Total wall-clock time is bounded by the slowest single model, not the sum of all three.
{% endhint %}

{% hint style="info" %}
The script exits with code `1` if any session fails or times out, which marks the GitHub Actions job as failed. This makes test-writing failures visible in the PR checks panel alongside lint and CI results.
{% endhint %}
{% endstep %}

{% step %}

### Add Your Paradime Credentials to GitHub Secrets

The driver script authenticates with Paradime using three values. Add the following as GitHub Actions secrets in your repository under **Settings → Secrets and variables → Actions**:

* `PARADIME_API_KEY`
* `PARADIME_API_SECRET`
* `PARADIME_API_ENDPOINT`

{% hint style="info" %}
`GITHUB_TOKEN` and `GITHUB_REPOSITORY` are provided automatically by GitHub Actions — you do not need to add them as secrets.
{% endhint %}
{% endstep %}

{% step %}

### Create the GitHub Actions Workflow

Create `.github/workflows/test-maintainer.yml`. This triggers the driver script automatically whenever a PR adds or modifies a file inside `models/`.

{% code title=".github/workflows/test-maintainer.yml" lineNumbers="true" %}

```yaml
name: DinoAI dbt™ test maintainer

on:
  pull_request:
    types: [opened, synchronize, ready_for_review]
    paths:
      - "models/**/*.sql"

permissions:
  pull-requests: write
  contents: write       # required so the agent can push test commits to the PR branch

jobs:
  write-tests:
    runs-on: ubuntu-latest
    timeout-minutes: 35
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0
          ref: ${{ github.head_ref }}   # check out the PR branch so the agent can push to it

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install Paradime SDK
        run: pip install paradime-io

      - name: Configure git for agent commits
        run: |
          git config user.name  "DinoAI Test Maintainer"
          git config user.email "dinoai-tests@noreply.github.com"

      - name: Run test maintainer
        env:
          PARADIME_API_ENDPOINT: ${{ secrets.PARADIME_API_ENDPOINT }}
          PARADIME_API_KEY:      ${{ secrets.PARADIME_API_KEY }}
          PARADIME_API_SECRET:   ${{ secrets.PARADIME_API_SECRET }}
          GITHUB_TOKEN:          ${{ secrets.GITHUB_TOKEN }}
        run: python scripts/test_maintainer.py
```

{% endcode %}

{% hint style="info" %}
The `paths:` filter restricts the workflow to `.sql` files inside `models/` only. PRs that touch only YAML, Python, or documentation files will not trigger the agent.
{% endhint %}

{% hint style="info" %}
`contents: write` is required so the agent can commit and push the updated schema YAML files back to the PR branch. Without it, the `git push` inside the agent session will fail with a permissions error.
{% endhint %}
{% endstep %}
{% endstepper %}

## What the PR Experience Looks Like

Once the workflow is set up, the experience for a PR author is:

1. They open a PR adding or modifying a dbt™ model
2. A comment appears on the PR immediately:

   > 🧪 **DinoAI test maintainer started** — PR #42: feat/add-revenue-mart
   >
   > Writing and running missing tests for 2 model(s):
   >
   > * `fct_orders` (`models/marts/fct_orders.sql`)
   > * `stg_payments` (`models/staging/stg_payments.sql`)
   >
   > Changes will be committed to `feat/add-revenue-mart`. Check `#data-quality` for live updates.
3. A new commit appears on their branch: `test: add missing tests for fct_orders [DinoAI]`
4. A completion comment appears on the PR with a per-model breakdown:

   > 🧪 **DinoAI test maintainer — complete**
   >
   > #### ✅ `fct_orders`
   >
   > Tests added: `unique` + `not_null` on `order_id`, `not_null` on `customer_id`, `accepted_values` on `status` (values: pending, shipped, delivered, returned) All 4 tests passing. Committed to branch: `feat/add-revenue-mart`
   >
   > #### ✅ `stg_payments`
   >
   > Tests added: `unique` + `not_null` on `payment_id`, `relationships` on `order_id` → `stg_orders` All 3 tests passing. 1 data quality warning on `amount` (severity: warn — negative values found, investigate before promoting to error)
   >
   > **2/2 models** had tests written and committed to `feat/add-revenue-mart`.
5. The same summary is posted to `#data-quality` on Slack.

## Using as a Sub-Agent of the PR Reviewer

If you have the PR reviewer set up, you can add `test-maintainer` to its `agents_squad` so the reviewer delegates test-writing rather than just flagging the gap in its verdict.

Add the following to your `pr-reviewer-e2e.yml`:

```yaml
agents_squad:
  - test-maintainer

tools:
  mode: allowlist
  list:
    - read_file
    - search_files_and_directories
    - ripgrep_search
    - get_linear_issue
    - run_sql_query
    - post_slack_message
    - invoke_agent          # ← add this to allow delegation
```

The PR reviewer can then call `invoke_agent("test-maintainer", ...)` when it detects missing test coverage, and the test maintainer will call `notify_parent_session` with its findings once complete so the reviewer can include the result in its final verdict.

{% hint style="info" %}
When used as a sub-agent, the test maintainer receives the branch name and model name via the `invoke_agent` message rather than from the GitHub Actions context. Make sure the PR reviewer includes both in its delegation message.
{% endhint %}

## File Structure

Your repository should look like this after completing the setup:

```
your-repo/
├── dbt_project.yml
├── .dinoai/
│   └── agents/
│       └── test-maintainer.yml
├── .github/
│   └── workflows/
│       └── test-maintainer.yml
└── scripts/
    └── test_maintainer.py
```

## Related Docs

* [**PR Reviewer tutorial** — the parent agent that can delegate to this one](/app-help/guides-new/porgrammable-agents/end-to-end-pr-reviewer.md)
* [**Agent-to-Agent Delegation** — how to invoke this agent from the PR reviewer](/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 including `run_terminal_command` and `notify_parent_session`](/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.md)


---

# Agent Instructions: 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:

```
GET https://docs.paradime.io/app-help/guides-new/porgrammable-agents/test-maintainer.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
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.