dbt™ impact assessor

name: DinoAI impact assessor

on:
  pull_request:
    types: [opened, synchronize, ready_for_review]
  push:
    branches: [main]
    paths:
      - "models/**"

permissions:
  pull-requests: write
  contents: read

jobs:
  assess:
    runs-on: ubuntu-latest
    timeout-minutes: 35
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0   # need full history for git diff between base and head

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

      - name: Install Paradime SDK
        run: pip install "paradime-io>=6.0.0"

      - name: Run impact assessor
        env:
          PARADIME_API_ENDPOINT: ${{ secrets.PARADIME_API_ENDPOINT }}
          PARADIME_API_SECRET: ${{ secrets.PARADIME_API_SECRET }}
          PARADIME_WORKSPACE_UID: ${{ secrets.PARADIME_WORKSPACE_UID }}
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        run: python scripts/pr_impact.py

import json
import os
import subprocess
import time
import urllib.request

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

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

# 1. Read the event payload and work out the commit range to diff
gh_event_name = os.environ.get("GITHUB_EVENT_NAME", "")
gh_token = os.environ.get("GITHUB_TOKEN")
repo = os.environ.get("GITHUB_REPOSITORY")
pr_number = None

with open(os.environ["GITHUB_EVENT_PATH"]) as f:
    event = json.load(f)

if gh_event_name == "pull_request":
    pr = event["pull_request"]
    pr_number = pr["number"]
    diff_range = f"{pr['base']['sha']}...{pr['head']['sha']}"
else:
    # push to main: compare against the state before the push
    diff_range = f"{event['before']}...{event['after']}"

# 2. Get the list of changed model files in this PR / push
diff = subprocess.check_output(
    ["git", "diff", "--name-only", diff_range, "--", "models/"]
).decode().splitlines()

models = [
    os.path.basename(f).removesuffix(".sql")
    for f in diff if f.endswith(".sql")
]

if not models:
    print("No model changes detected, skipping impact assessment.")
    raise SystemExit(0)

print(f"Assessing downstream impact for: {', '.join(models)}")

# 3. Trigger the agent and capture the session ID before blocking
trigger = paradime.dinoai_agents.trigger_run(
    agent="impact-assessor",
    message=(
        f"Assess the downstream impact of changes to these dbt models: "
        f"{', '.join(models)}.\n\n"
        f"For each model, identify:\n"
        f"- Direct and indirect downstream dependants\n"
        f"- Dashboards, BI tools, or other consumers potentially affected\n"
        f"- Risk level (HIGH / MEDIUM / LOW) with a short rationale\n\n"
        f"Structure your report under these sections:\n"
        f"CHANGED MODELS / DOWNSTREAM IMPACT / RISK ASSESSMENT / RECOMMENDATIONS"
    ),
)
session_id = trigger.agent_session_id
print(f"Impact assessment session started: {session_id}")

# 4. Block until the agent finishes by polling get_run (up to 30 minutes)
timeout = 1800
poll_interval = 10
start = time.time()

while True:
    run = paradime.dinoai_agents.get_run(agent_session_id=session_id)

    if run.status == DinoaiAgentRunStatus.COMPLETED:
        print(f"Impact assessment session completed: {session_id}")
        break

    if run.status == DinoaiAgentRunStatus.FAILED:
        last = run.messages[-1].content if run.messages else "no messages"
        raise DinoaiAgentRunFailedException(f"Agent run failed. Last message: {last}")

    if time.time() - start > timeout:
        raise TimeoutError(f"Timed out waiting for session {session_id} to complete.")

    print(f"  status: {run.status.value}, waiting {poll_interval}s")
    time.sleep(poll_interval)

# 5. Extract the full report from the last agent message
final_report = run.messages[-1].content if run.messages else "(no report returned)"
print(final_report)

# 6. Post the full report as a PR comment (only on pull_request events)
if gh_token and pr_number and repo:
    comment_body = (
        f"## 🦕 DinoAI Impact Assessment (session `{session_id}`)\n\n"
        f"**Models changed:** `{'`, `'.join(models)}`\n\n"
        f"{final_report}"
    )

    req = urllib.request.Request(
        f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments",
        data=json.dumps({"body": comment_body}).encode(),
        headers={
            "Authorization": f"Bearer {gh_token}",
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        },
        method="POST",
    )
    urllib.request.urlopen(req).read()
    print(f"Impact report posted to PR #{pr_number}")