Linear Backlog Agent

[tool.poetry]
name = "linear-dinoai-orchestrator"
version = "0.1.0"
description = "Linear → DinoAI Agent Orchestrator"
authors = ["Your Name <you@example.com>"]
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"

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"

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

"""
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)

#!/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()