Linear Backlog Agent

Automate Linear backlog execution with a DinoAI agent that picks up Agent Ready issues, builds dbt™ models with tests and docs, opens PRs, and posts progress to Slack.

Automate your analytics engineering backlog with a DinoAI agent that reads every open Linear ticket labelled Agent Ready, spins up a Paradime development session, implements the required dbt™ models with tests and YAML docs, and opens a PR — all in parallel, one session per ticket.

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.

Linear

• A Linear personal API key — generate one under Settings → API → Personal API keys

• The team key for the team whose backlog you want to process (e.g. "DNA")

• Issues you want to automate must carry the label Agent Ready and have a state type that is not completed

Integrations

The following must already be connected in Paradime:

• Linear — the agent calls get_linear_issue to fetch linked tickets

• Slack — the agent posts status updates to #analytics-eng via post_slack_message

What You'll Build

By the end of this guide you'll have:

• A Poetry project wired to the Paradime SDK and the Linear GraphQL API

• A LinearClient that paginates through all non-completed, label-filtered issues

• An AgentInvoker that fires one DinoAI session per issue in parallel and polls until completion

• A run_notifier.py entry point that ties everything together

• A linear-backlog-agent DinoAI agent YAML that reads the ticket, implements the dbt™ models, and opens a PR

What the Agent Does Per Ticket

Once triggered, the agent follows this sequence for each Linear issue:

1. Posts a "🚀 Starting" message to #analytics-eng
2. Reads the Linear ticket via get_linear_issue
3. Identifies source tables and implied metrics
4. Drafts staging (and mart) dbt™ models with tests and YAML docs
5. Commits changes on a new branch
6. Opens a PR and posts a "✅ Done: <PR URL>" message to #analytics-eng

The agent never invents a column that isn't in the source. When a ticket is ambiguous it surfaces explicit Open questions in the PR description rather than guessing.

Architecture Overview

How It Works

When run_notifier.py is executed it loads environment variables, fetches all non-completed Linear issues labelled Agent Ready, then fires one DinoAI agent session per issue using a ThreadPoolExecutor. All sessions run concurrently — total wall-clock time is bounded by the slowest single ticket, not the sum of all tickets. Each session is polled every 20 seconds for up to 30 minutes.

1

Set Up the Poetry Project

Create the following pyproject.toml at the same directory level as your dbt_project.yml. This is required so the agent can locate and write dbt™ model files correctly.

pyproject.toml

[tool.poetry]
name = "linear-dinoai-orchestrator"
version = "0.1.0"
description = "Linear → DinoAI Agent Orchestrator"
authors = ["Your Name <[e-mail]>"]
package-mode = false

[tool.poetry.dependencies]
python = ">=3.11,<3.13"
requests = "^2.28.1"
paradime-io = "^5.3.0"

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

Install dependencies by running:

poetry install

The only two third-party dependencies are requests (for the Linear GraphQL API) and paradime-io (the Paradime SDK). All other imports — concurrent.futures, os, sys, time, typing — are part of the Python standard library.

2

Create the Agent YAML

Create the following file in your repository at .dinoai/agents/linear-backlog-agent.yml. This defines the agent's role, goal, tools, and Slack output channel.

.dinoai/agents/linear-backlog-agent.yml

name: linear-backlog-agent
version: 1

role: >
  Analytics Engineer translating product / business requests in Linear
  into well-scoped dbt™ model PRs.

goal: >
  For the Linear issue referenced in the trigger message: read the ticket,
  identify the source tables and metrics implied, draft a staging model
  (and a mart model when appropriate) with tests and YAML docs, commit
  the changes on a new branch, and post the PR link back to the Linear
  ticket and to Slack.

backstory: >
  You are a careful translator. You never invent a column that isn't in
  the source. When the ticket is ambiguous you call it out in the PR
  description with explicit "Open questions" rather than guessing.

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

slack:
  channel: "#analytics-eng"

tools.mode: allowlist means the agent can only call the tools explicitly listed. This keeps each session focused and prevents unintended side effects across your repository.

The Slack channel is set to #analytics-eng by default. Update slack.channel before committing if your team uses a different channel.

3

Create the Linear Client

Create python/linear_slack_notifier/linear_client.py. This module communicates with the Linear GraphQL API and handles automatic pagination so no issues are missed regardless of backlog size.

python/linear:slack:notifier/linear:client.py

"""
Linear API client for fetching issues by team and label.
"""
from typing import List, Dict, Any, Optional

import requests


LINEAR_API_URL = "https://api.linear.app/graphql"

ISSUES_QUERY = """
query($teamKey: String!, $labelName: String!, $cursor: String) {
  issues(
    first: 50
    after: $cursor
    filter: {
      team:   { key: { eq: $teamKey } }
      labels: { some: { name: { eq: $labelName } } }
      state:  { type: { neq: "completed" } }
    }
  ) {
    pageInfo {
      hasNextPage
      endCursor
    }
    nodes {
      id
      identifier
      title
      url
      description
      state {
        name
        type
      }
      assignee {
        id
        name
        displayName
      }
      labels {
        nodes {
          name
        }
      }
    }
  }
}
"""


class LinearClient:
    """Thin client for the Linear GraphQL API."""

    def __init__(self, api_key: str) -> None:
        self._headers = {
            "Authorization": api_key,   # Linear accepts the key directly (no "Bearer" prefix needed)
            "Content-Type":  "application/json",
        }

    def _run_query(self, query: str, variables: Dict[str, Any]) -> Dict[str, Any]:
        """Execute a GraphQL query and return the parsed response."""
        resp = requests.post(
            LINEAR_API_URL,
            json={"query": query, "variables": variables},
            headers=self._headers,
            timeout=30,
        )
        resp.raise_for_status()
        payload = resp.json()

        if "errors" in payload:
            for err in payload["errors"]:
                print(f"❌  Linear API error: {err.get('message')}")
            raise RuntimeError("Linear API returned errors — see above.")

        return payload["data"]

    def get_incomplete_issues_by_label(
        self,
        team_id: str,
        label_name: str,
    ) -> List[Dict[str, Any]]:
        """
        Fetch all non-completed issues in team_id that carry label_name.
        Paginates automatically.
        """
        issues: List[Dict[str, Any]] = []
        cursor: Optional[str] = None

        while True:
            variables: Dict[str, Any] = {
                "teamKey":   team_id,
                "labelName": label_name,
                "cursor":    cursor,
            }
            data       = self._run_query(ISSUES_QUERY, variables)
            page       = data["issues"]
            issues    += page["nodes"]
            page_info  = page["pageInfo"]

            if not page_info["hasNextPage"]:
                break

            cursor = page_info["endCursor"]

        return issues

The client accepts both a Linear team UUID and a short team key (e.g. "DNA"). Pagination is handled automatically via pageInfo.hasNextPage and endCursor — all matching issues are returned regardless of backlog size.

4

Create the Agent Invoker

Create python/linear_slack_notifier/slack_notifier.py. Despite the filename, this module no longer posts to Slack directly — it triggers and polls DinoAI agent sessions, one per issue, in parallel threads.

python/linear:slack:notifier/slack:notifier.py

"""
Agent Invoker — for each Linear issue, triggers a `linear-backlog-agent`
DinoAI session via the Paradime SDK and polls until completion.

All issues are processed in parallel — one thread per issue — so the total
wall-clock time is bounded by the slowest single ticket, not the sum of all.
"""

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Dict, List

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


AGENT_NAME    = "linear-backlog-agent"
TIMEOUT       = 1800   # seconds to wait per agent session
POLL_INTERVAL = 20     # seconds between status polls


def _build_trigger_message(issue: Dict[str, Any]) -> str:
    """
    Compose the opening message sent to the agent for a single Linear issue.
    Embeds all available issue metadata so the agent can start immediately.
    """
    identifier  = issue.get("identifier", issue.get("id", "???"))
    title       = issue.get("title", "(no title)")
    url         = issue.get("url", "")
    description = (issue.get("description") or "").strip()
    state       = issue.get("state", {}).get("name", "?")
    assignee    = (issue.get("assignee") or {}).get("displayName", "unassigned")
    labels      = ", ".join(
        lbl["name"] for lbl in (issue.get("labels") or {}).get("nodes", [])
    )

    desc_block = (
        f"\n**Description:**\n{description}\n"
        if description
        else "\n*(No description provided)*\n"
    )

    return (
        f"Please tackle the following Linear issue end-to-end as described.\n\n"
        f"**Issue:** [{identifier}] {title}\n"
        f"**URL:** {url}\n"
        f"**State:** {state}\n"
        f"**Assignee:** {assignee}\n"
        f"**Labels:** {labels or 'none'}\n"
        f"{desc_block}\n"
        f"IMPORTANT — follow these steps in order:\n\n"
        f"1. **Before doing anything else**, post this exact message to #analytics-eng:\n"
        f'   "🚀 [{identifier}] Starting: {title}"\n'
        f"   This is critical — multiple issues are being processed in parallel and the\n"
        f"   team needs to see each session announce itself immediately.\n\n"
        f"2. Read the ticket carefully, identify the source tables and metrics implied,\n"
        f"   implement the required dbt™ models with tests and YAML docs, and commit the\n"
        f"   changes on a new branch named `{identifier.lower()}-<short-description>-<5-char-random-string>`.\n\n"
        f"3. Open a PR, then post a completion message to #analytics-eng:\n"
        f'   "✅ [{identifier}] Done: <PR URL> — {title}"\n\n'
        f"Always prefix every Slack message you send with [{identifier}] so it is\n"
        f"distinguishable from the other parallel sessions running at the same time."
    )


class AgentInvoker:
    """
    Triggers one `linear-backlog-agent` session per Linear issue and polls
    each session concurrently until completion.
    """

    def __init__(self, paradime: Paradime) -> None:
        self._paradime = paradime

    def invoke_for_issues(self, issues: List[Dict[str, Any]]) -> None:
        """Trigger all agent sessions in parallel and print a summary."""
        if not issues:
            print("ℹ️   No issues to process.")
            return

        print(f"\n🤖  Triggering '{AGENT_NAME}' for {len(issues)} issue(s) in parallel...\n")

        success = 0
        failed  = 0

        with ThreadPoolExecutor(max_workers=len(issues)) as executor:
            future_to_issue = {
                executor.submit(self._run_single_issue, issue): issue
                for issue in issues
            }

            for future in as_completed(future_to_issue):
                issue      = future_to_issue[future]
                identifier = issue.get("identifier", issue.get("id", "???"))

                try:
                    last_msg = future.result()
                    print(f"{'─' * 60}")
                    print(f"✅  [{identifier}] Done.\n   Agent output:\n{last_msg}\n")
                    success += 1

                except (DinoaiAgentRunFailedException, TimeoutError) as exc:
                    print(f"{'─' * 60}")
                    print(f"❌  [{identifier}] {exc}\n")
                    failed += 1

        print(f"{'=' * 60}")
        print(f"📊  Done — ✅ {success} succeeded   ❌ {failed} failed")
        print(f"{'=' * 60}")

    def _run_single_issue(self, issue: Dict[str, Any]) -> str:
        identifier = issue.get("identifier", issue.get("id", "???"))
        title      = issue.get("title", "(no title)")

        print(f"{'─' * 60}")
        print(f"🚀  [{identifier}] {title}")

        session_id = self._trigger(issue)
        final_run  = self._poll_until_done(session_id, label=identifier)

        return (
            final_run.messages[-1].content
            if final_run.messages
            else "(no output returned)"
        )

    def _trigger(self, issue: Dict[str, Any]) -> str:
        message    = _build_trigger_message(issue)
        trigger    = self._paradime.dinoai_agents.trigger_run(
            agent=AGENT_NAME,
            message=message,
        )
        session_id = trigger.agent_session_id
        identifier = issue.get("identifier", "?")
        print(f"   📋  Session ID: {session_id} (issue: {identifier})")
        return session_id

    def _poll_until_done(self, session_id: str, label: str) -> object:
        start = time.time()

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

            if run.status == DinoaiAgentRunStatus.COMPLETED:
                return run

            if run.status == DinoaiAgentRunStatus.FAILED:
                last_msg = run.messages[-1].content if run.messages else "no messages returned"
                raise DinoaiAgentRunFailedException(
                    f"Agent session {session_id} FAILED for [{label}]. "
                    f"Last message: {last_msg}"
                )

            elapsed = time.time() - start
            if elapsed > TIMEOUT:
                raise TimeoutError(
                    f"Timed out after {TIMEOUT}s waiting for session "
                    f"{session_id} ([{label}]) to complete."
                )

            print(
                f"   ⏳  [{label}] status: {run.status.value} "
                f"(elapsed: {int(elapsed)}s) … retrying in {POLL_INTERVAL}s"
            )
            time.sleep(POLL_INTERVAL)

Sessions are polled every 20 seconds with a 30-minute timeout per issue. The ThreadPoolExecutor fires all sessions immediately — if you have 10 tickets, all 10 agent sessions start at the same time.

5

Create the Main Orchestrator

Create python/linear_slack_notifier/run_notifier.py. This is the entry point that loads configuration, fetches issues from Linear, and hands them to the AgentInvoker.

python/linear:slack:notifier/run:notifier.py

#!/usr/bin/env python3
"""
Linear → DinoAI Agent Orchestrator

Fetches all non-completed Linear issues with a given label in a team,
then triggers one `linear-backlog-agent` DinoAI session per issue via the
Paradime SDK — each session reads the ticket, implements the required dbt™
models, and opens a PR.

Usage (from the repo root):
    poetry run python python/linear_slack_notifier/run_notifier.py

Required environment variables:
    LINEAR_API_KEY         — Linear personal API key
    LINEAR_TEAM_ID         — Linear team key (e.g. "DNA")
    PARADIME_API_ENDPOINT  — e.g. https://api.paradime.io/api/v1/<token>/graphql
    PARADIME_API_KEY       — Paradime API key
    PARADIME_API_SECRET    — Paradime API secret

Optional environment variables:
    LINEAR_LABEL_NAME      — Label to filter by (default: "Agent Ready")
"""

import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from paradime import Paradime
from linear_client import LinearClient
from slack_notifier import AgentInvoker

DEFAULT_LABEL_NAME = "Agent Ready"


def load_config() -> dict:
    errors = []

    linear_api_key      = os.getenv("LINEAR_API_KEY", "")
    linear_team_id      = os.getenv("LINEAR_TEAM_ID", "")
    paradime_endpoint   = os.getenv("PARADIME_API_ENDPOINT", "")
    paradime_api_key    = os.getenv("PARADIME_API_KEY", "")
    paradime_api_secret = os.getenv("PARADIME_API_SECRET", "")

    if not linear_api_key:
        errors.append("LINEAR_API_KEY is not set.")
    if not linear_team_id:
        errors.append("LINEAR_TEAM_ID is not set.")
    if not paradime_endpoint:
        errors.append("PARADIME_API_ENDPOINT is not set.")
    if not paradime_api_key:
        errors.append("PARADIME_API_KEY is not set.")
    if not paradime_api_secret:
        errors.append("PARADIME_API_SECRET is not set.")

    if errors:
        for e in errors:
            print(f"❌  {e}")
        print(
            "\n💡  Set the missing variables before running:\n"
            "      export LINEAR_API_KEY=<your-linear-api-key>\n"
            "      export LINEAR_TEAM_ID=<your-linear-team-id>\n"
            "      export PARADIME_API_ENDPOINT=<your-paradime-endpoint>\n"
            "      export PARADIME_API_KEY=<your-paradime-api-key>\n"
            "      export PARADIME_API_SECRET=<your-paradime-api-secret>\n"
        )
        sys.exit(1)

    return {
        "linear_api_key":      linear_api_key,
        "linear_team_id":      linear_team_id,
        "label_name":          os.getenv("LINEAR_LABEL_NAME", DEFAULT_LABEL_NAME),
        "paradime_endpoint":   paradime_endpoint,
        "paradime_api_key":    paradime_api_key,
        "paradime_api_secret": paradime_api_secret,
    }


def main() -> None:
    print("🚀  Linear → DinoAI Agent Orchestrator starting...\n" + "=" * 60)

    cfg = load_config()
    print(f"📋  Config:")
    print(f"     Linear team  : {cfg['linear_team_id']}")
    print(f"     Label filter : {cfg['label_name']}")
    print(f"     Agent        : linear-backlog-agent (via Paradime SDK)")

    paradime = Paradime(
        api_endpoint=cfg["paradime_endpoint"],
        api_key=cfg["paradime_api_key"],
        api_secret=cfg["paradime_api_secret"],
    )

    print(
        f"\n🔍  Fetching Linear issues "
        f"(label='{cfg['label_name']}', team='{cfg['linear_team_id']}')..."
    )
    linear = LinearClient(cfg["linear_api_key"])

    try:
        issues = linear.get_incomplete_issues_by_label(
            team_id    = cfg["linear_team_id"],
            label_name = cfg["label_name"],
        )
    except Exception as e:
        print(f"❌  Failed to fetch Linear issues: {e}")
        sys.exit(1)

    print(f"📌  Found {len(issues)} incomplete issue(s) with label '{cfg['label_name']}'.")

    if not issues:
        print("✅  Nothing to process. Exiting.")
        sys.exit(0)

    print("\n   Issues to process:")
    for issue in issues:
        state = issue.get("state", {}).get("name", "?")
        print(f"   • [{issue.get('identifier')}] {issue.get('title')}  (state: {state})")

    invoker = AgentInvoker(paradime=paradime)
    invoker.invoke_for_issues(issues=issues)


if __name__ == "__main__":
    main()

6

Set Your Environment Variables

The orchestrator requires five environment variables. Set them in your shell before running:

export LINEAR_API_KEY=<your-linear-api-key>
export LINEAR_TEAM_ID=<your-linear-team-key>        # e.g. "DNA"
export PARADIME_API_ENDPOINT=<your-paradime-endpoint>
export PARADIME_API_KEY=<your-paradime-api-key>
export PARADIME_API_SECRET=<your-paradime-api-secret>

# Optional — defaults to "Agent Ready"
export LINEAR_LABEL_NAME="Agent Ready"

Variable	Description
LINEAR_API_KEY	Linear personal API key
LINEAR_TEAM_ID	Linear team key (e.g. "DNA")
PARADIME_API_ENDPOINT	Paradime GraphQL endpoint
PARADIME_API_KEY	Paradime API key
PARADIME_API_SECRET	Paradime API secret
LINEAR_LABEL_NAME	(optional) Label to filter by — defaults to "Agent Ready"

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

7

Run the Orchestrator

From the repository root, run:

poetry run python python/linear_slack_notifier/run_notifier.py

You should see output like:

🚀  Linear → DinoAI Agent Orchestrator starting...
============================================================
📋  Config:
     Linear team  : DNA
     Label filter : Agent Ready
     Agent        : linear-backlog-agent (via Paradime SDK)

🔍  Fetching Linear issues (label='Agent Ready', team='DNA')...
📌  Found 3 incomplete issue(s) with label 'Agent Ready'.

   Issues to process:
   • [DNA-42] Add revenue mart model  (state: In Progress)
   • [DNA-57] Backfill sessions table  (state: Todo)
   • [DNA-61] Create user_lifetime_value model  (state: Todo)

🤖  Triggering 'linear-backlog-agent' for 3 issue(s) in parallel...

Each issue gets its own parallel agent session. The orchestrator polls every 20 seconds and prints a summary once all sessions have reached a terminal state.

File Structure

Your repository should look like this after completing the setup:

your-repo/
├── dbt_project.yml
├── pyproject.toml                          ← same level as dbt_project.yml
├── .dinoai/
│   └── agents/
│       └── linear-backlog-agent.yml
└── python/
    └── linear_slack_notifier/
        ├── run_notifier.py                 ← entry point
        ├── linear_client.py                ← Linear GraphQL client
        └── slack_notifier.py               ← AgentInvoker (triggers DinoAI sessions)

pyproject.toml must sit at the same directory level as dbt_project.yml. The agent uses run_terminal_command and read_file to navigate the dbt™ project, so the working directory at session start must be the repo root.

Schedule with Bolt

Instead of running the orchestrator manually, you can have Bolt execute it on a schedule — every morning, once a week, or at any cadence that suits your team's workflow. Bolt runs the two commands in sequence: first poetry install to ensure dependencies are up to date, then poetry run to trigger the agent sessions.

Add Environment Variables to Paradime

Before creating the schedule, store your secrets as environment variables in Paradime so Bolt can access them at runtime. Go to Account Settings → Environment Variables and add the following:

Variable	Value
LINEAR_API_KEY	Your Linear personal API key
LINEAR_TEAM_ID	Your Linear team key (e.g. "DNA")
PARADIME_API_ENDPOINT	Your Paradime GraphQL endpoint
PARADIME_API_KEY	Your Paradime API key
PARADIME_API_SECRET	Your Paradime API secret
LINEAR_LABEL_NAME	(optional) Defaults to "Agent Ready"

Environment variables set in Workspace Settings → Environment Variables are available to all Bolt schedules in your workspace. You only need to set them once.

Create the Bolt Schedule

Go to Bolt → Schedules and click New Schedule. Give it a name like Linear Backlog Agent and configure the two commands under the Commands section.

The schedule should run two commands in order:

poetry install

poetry run python python/linear_slack_notifier/run_notifier.py

poetry install is included as the first command so that any dependency updates committed to pyproject.toml are automatically picked up on every run — no manual intervention needed.

Choose a Schedule Frequency

Pick the cadence that matches how often your team labels new tickets as Agent Ready. The table below covers the most common options:

Cadence	Cron expression	When it runs
Every day at 9 AM	0 9 * * *	Monday–Sunday, 9:00 AM
Weekdays at 9 AM	0 9 * * 1-5	Monday–Friday, 9:00 AM
Once a week (Monday 9 AM)	0 9 * * 1	Every Monday, 9:00 AM
Twice a week (Mon & Thu)	0 9 * * 1,4	Monday and Thursday, 9:00 AM
Every 6 hours	0 */6 * * *	12 AM, 6 AM, 12 PM, 6 PM

For most teams, weekdays at 9 AM (0 9 * * 1-5) is a good default. The orchestrator exits cleanly with ✅ Nothing to process when there are no Agent Ready tickets, so there is no cost to running it on days with an empty queue.

Once saved, Bolt will run both commands on your chosen schedule, picking up any new Agent Ready Linear issues and spinning up the corresponding DinoAI sessions automatically.

Related Docs

• Programmable Agents — Quick Start — getting started with DinoAI agents

• Programmable Agents — YAML Configuration — full reference for agent config options

• Programmable Agents — Tools Reference — all available tools including read_file, get_linear_issue, and post_slack_message

• Linear Integration — connecting Linear to Paradime

• Slack Integration — connecting Slack to Paradime

• Paradime API & Credentials — where to find your API endpoint, key, and secret