iterate-pr

interface:
  display_name: "Iterate PR"
  short_description: "Fix PR feedback and CI until green"
  default_prompt: "Use $iterate-pr to address actionable review feedback and CI failures on the current pull request until the actionable checks pass."
policy:
  allow_implicit_invocation: true

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.9"
# ///
"""
Fetch PR CI checks and extract relevant failure snippets.

Usage:
    uv run fetch_pr_checks.py [--pr PR_NUMBER]

If --pr is not specified, uses the PR for the current branch.

Output: JSON to stdout with structured check data.
"""
from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from typing import Any

HUMAN_GATE_PATTERNS = [
    r"(?i)review\s+required",
    r"(?i)required\s+review",
    r"(?i)requires\s+review",
    r"(?i)required\s+approving\s+review",
    r"(?i)approval\s+required",
    r"(?i)waiting\s+for\s+approval",
    r"(?i)manual\s+approval",
    r"(?i)draft\s+(pull\s+request|pr)",
]


def run_gh(args: list[str]) -> dict[str, Any] | list[Any] | None:
    """Run a gh CLI command and return parsed JSON output."""
    try:
        result = subprocess.run(
            ["gh"] + args,
            capture_output=True,
            text=True,
            check=True,
        )
        return json.loads(result.stdout) if result.stdout.strip() else None
    except subprocess.CalledProcessError as e:
        print(f"Error running gh {' '.join(args)}: {e.stderr}", file=sys.stderr)
        return None
    except json.JSONDecodeError:
        return None


def get_pr_info(pr_number: int | None = None) -> dict[str, Any] | None:
    """Get PR info, optionally by number or for current branch."""
    args = [
        "pr",
        "view",
        "--json",
        "number,url,headRefName,baseRefName,isDraft,reviewDecision",
    ]
    if pr_number:
        args.insert(2, str(pr_number))
    return run_gh(args)


def get_checks(pr_number: int | None = None) -> list[dict[str, Any]]:
    """Get all checks for a PR."""
    args = ["gh", "pr", "checks"]
    if pr_number:
        args.append(str(pr_number))
    args.extend(["--json", "name,bucket,link,workflow,state,description,event"])
    try:
        result = subprocess.run(
            args,
            capture_output=True,
            text=True,
        )
        if not result.stdout.strip():
            return []
        try:
            checks = json.loads(result.stdout)
            return checks if isinstance(checks, list) else []
        except json.JSONDecodeError:
            pass

        checks = []
        for line in result.stdout.strip().split("\n"):
            if not line.strip():
                continue
            parts = line.split("\t")
            if len(parts) >= 2:
                checks.append({
                    "name": parts[0].strip(),
                    "bucket": parts[1].strip(),
                    "link": parts[3].strip() if len(parts) > 3 else "",
                    "workflow": "",
                })
        return checks
    except Exception:
        return []


def is_human_gate_check(check: dict[str, Any]) -> bool:
    """Return true when a pending entry is a human review/approval gate."""
    haystack = " ".join(
        str(check.get(field, ""))
        for field in ("name", "state", "description", "workflow")
    )
    return any(re.search(pattern, haystack) for pattern in HUMAN_GATE_PATTERNS)


def get_failed_runs(branch: str) -> list[dict[str, Any]]:
    """Get recent failed workflow runs for a branch."""
    result = run_gh([
        "run", "list",
        "--branch", branch,
        "--limit", "10",
        "--json", "databaseId,name,status,conclusion,headSha"
    ])
    if not isinstance(result, list):
        return []
    # Return runs that failed or are in progress
    return [r for r in result if r.get("conclusion") == "failure"]


def extract_failure_snippet(log_text: str, max_lines: int = 50) -> str:
    """Extract relevant failure snippet from log text.

    Looks for common failure markers and extracts surrounding context.
    """
    lines = log_text.split("\n")

    # Patterns that indicate failure points (case-insensitive via re.IGNORECASE)
    failure_patterns = [
        r"error[:\s]",
        r"failed[:\s]",
        r"failure[:\s]",
        r"traceback",
        r"exception",
        r"assert(ion)?.*failed",
        r"FAILED",
        r"panic:",
        r"fatal:",
        r"npm ERR!",
        r"yarn error",
        r"ModuleNotFoundError",
        r"ImportError",
        r"SyntaxError",
        r"TypeError",
        r"ValueError",
        r"KeyError",
        r"AttributeError",
        r"NameError",
        r"IndentationError",
        r"===.*FAILURES.*===",
        r"___.*___",  # pytest failure separators
    ]

    combined_pattern = "|".join(failure_patterns)

    # Find lines matching failure patterns
    failure_indices = []
    for i, line in enumerate(lines):
        if re.search(combined_pattern, line, re.IGNORECASE):
            failure_indices.append(i)

    if not failure_indices:
        # No clear failure point, return last N lines
        return "\n".join(lines[-max_lines:])

    # Extract context around first failure point
    # Include some context before and after
    first_failure = failure_indices[0]
    start = max(0, first_failure - 5)
    end = min(len(lines), first_failure + max_lines - 5)

    snippet_lines = lines[start:end]

    # If there are more failures after our snippet, note it
    remaining_failures = [i for i in failure_indices if i >= end]
    if remaining_failures:
        snippet_lines.append(f"\n... ({len(remaining_failures)} more error(s) follow)")

    return "\n".join(snippet_lines)


def get_run_logs(run_id: int) -> str | None:
    """Get failed logs for a workflow run."""
    try:
        result = subprocess.run(
            ["gh", "run", "view", str(run_id), "--log-failed"],
            capture_output=True,
            text=True,
            timeout=60,
        )
        return result.stdout if result.stdout else result.stderr
    except subprocess.TimeoutExpired:
        return None
    except subprocess.CalledProcessError:
        return None


def main():
    parser = argparse.ArgumentParser(description="Fetch PR CI checks with failure snippets")
    parser.add_argument("--pr", type=int, help="PR number (defaults to current branch PR)")
    args = parser.parse_args()

    # Get PR info
    pr_info = get_pr_info(args.pr)
    if not pr_info:
        print(json.dumps({"error": "No PR found for current branch"}))
        sys.exit(1)

    pr_number = pr_info["number"]
    branch = pr_info["headRefName"]

    # Get checks
    checks = get_checks(pr_number)

    # Process checks and add failure snippets
    processed_checks = []
    failed_runs = None  # Lazy load

    for check in checks:
        status = check.get("bucket", check.get("state", "unknown"))
        human_gate = status == "pending" and is_human_gate_check(check)
        processed = {
            "name": check.get("name", "unknown"),
            "status": status,
            "link": check.get("link", ""),
            "workflow": check.get("workflow", ""),
        }
        if check.get("state"):
            processed["state"] = check["state"]
        if check.get("description"):
            processed["description"] = check["description"]
        if human_gate:
            processed["human_gate"] = True

        # For failures, try to get log snippet
        if processed["status"] == "fail":
            if failed_runs is None:
                failed_runs = get_failed_runs(branch)

            # Find matching run by workflow name
            workflow_name = processed["workflow"] or processed["name"]
            matching_run = next(
                (r for r in failed_runs if workflow_name in r.get("name", "")),
                None
            )

            if matching_run:
                logs = get_run_logs(matching_run["databaseId"])
                if logs:
                    processed["log_snippet"] = extract_failure_snippet(logs)
                    processed["run_id"] = matching_run["databaseId"]

        processed_checks.append(processed)

    # Build output
    output = {
        "pr": {
            "number": pr_number,
            "url": pr_info.get("url", ""),
            "branch": branch,
            "base": pr_info.get("baseRefName", ""),
            "is_draft": bool(pr_info.get("isDraft")),
            "review_decision": pr_info.get("reviewDecision", ""),
        },
        "summary": {
            "total": len(processed_checks),
            "passed": sum(1 for c in processed_checks if c["status"] == "pass"),
            "failed": sum(1 for c in processed_checks if c["status"] == "fail"),
            "pending": sum(1 for c in processed_checks if c["status"] == "pending"),
            "actionable_pending": sum(
                1
                for c in processed_checks
                if c["status"] == "pending" and not c.get("human_gate")
            ),
            "human_gate_pending": sum(
                1
                for c in processed_checks
                if c["status"] == "pending" and c.get("human_gate")
            ),
            "skipped": sum(1 for c in processed_checks if c["status"] in ("skipping", "cancel")),
        },
        "checks": processed_checks,
    }

    if pr_info.get("isDraft") and not processed_checks:
        output["action_required"] = "Draft PR has no registered checks; do not wait for CI indefinitely"
    elif not processed_checks:
        output["action_required"] = "No registered checks; monitor before reporting NO_CHECKS_REGISTERED"
    elif output["summary"]["actionable_pending"]:
        output["action_required"] = "Wait for actionable checks to finish; poll feedback while waiting"
    elif output["summary"]["failed"]:
        output["action_required"] = "Address failed checks"
    elif output["summary"]["pending"] and not output["summary"]["actionable_pending"]:
        output["action_required"] = "Only human review or approval gates remain pending"

    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.9"
# ///
"""
Fetch and categorize PR review feedback.

Usage:
    uv run fetch_pr_feedback.py [--pr PR_NUMBER]

If --pr is not specified, uses the PR for the current branch.

Output: JSON to stdout with categorized feedback.

Categories (using LOGAF scale - see https://develop.sentry.dev/engineering-practices/code-review/#logaf-scale):
- high: Must address before merge (h:, blocker, changes requested)
- medium: Should address (m:, standard feedback)
- low: Optional suggestions (l:, nit, style)
- bot: Informational automated comments (Codecov, Dependabot, etc.)
- resolved: Already resolved threads

Bot classification:
- Review bots (Sentry, Warden, Cursor, Bugbot, etc.) provide actionable code
  feedback. Their comments are categorized by content into high/medium/low with
  a ``review_bot: true`` flag — they are NOT placed in the ``bot`` bucket.
- Info bots (Codecov, Dependabot, Renovate, etc.) post status reports and are
  placed in the ``bot`` bucket for silent skipping.
"""
from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from typing import Any


# Bots that provide actionable code review feedback (security issues, lint
# violations, bugs). Their comments are categorized by content, not skipped.
REVIEW_BOT_PATTERNS = [
    r"(?i)^sentry",
    r"(?i)^warden",
    r"(?i)^cursor",
    r"(?i)^bugbot",
    r"(?i)^seer",
    r"(?i)^copilot",
    r"(?i)^codex",
    r"(?i)^claude",
    r"(?i)^codeql",
]

# Bots that post informational status reports (coverage, dependency updates).
# These are placed in the ``bot`` bucket and skipped silently.
INFO_BOT_PATTERNS = [
    r"(?i)^codecov",
    r"(?i)^dependabot",
    r"(?i)^renovate",
    r"(?i)^github-actions",
    r"(?i)^mergify",
    r"(?i)^semantic-release",
    r"(?i)^sonarcloud",
    r"(?i)^snyk",
    r"(?i)bot$",
    r"(?i)\[bot\]$",
]


def run_gh(args: list[str]) -> dict[str, Any] | list[Any] | None:
    """Run a gh CLI command and return parsed JSON output."""
    try:
        result = subprocess.run(
            ["gh"] + args,
            capture_output=True,
            text=True,
            check=True,
        )
        return json.loads(result.stdout) if result.stdout.strip() else None
    except subprocess.CalledProcessError as e:
        print(f"Error running gh {' '.join(args)}: {e.stderr}", file=sys.stderr)
        return None
    except json.JSONDecodeError:
        return None


def get_repo_info() -> tuple[str, str] | None:
    """Get owner and repo name from current directory."""
    result = run_gh(["repo", "view", "--json", "owner,name"])
    if result:
        return result.get("owner", {}).get("login"), result.get("name")
    return None


def get_pr_info(pr_number: int | None = None) -> dict[str, Any] | None:
    """Get PR info, optionally by number or for current branch."""
    args = ["pr", "view", "--json", "number,url,headRefName,author,reviews,reviewDecision"]
    if pr_number:
        args.insert(2, str(pr_number))
    return run_gh(args)


def is_review_bot(username: str) -> bool:
    """Check if username matches a review bot that posts actionable feedback."""
    return any(re.search(p, username) for p in REVIEW_BOT_PATTERNS)


def is_info_bot(username: str) -> bool:
    """Check if username matches an informational bot (skip silently)."""
    return any(re.search(p, username) for p in INFO_BOT_PATTERNS)


def is_bot(username: str) -> bool:
    """Check if username matches any known bot pattern."""
    return is_review_bot(username) or is_info_bot(username)


def get_review_comments(owner: str, repo: str, pr_number: int) -> list[dict[str, Any]]:
    """Get inline code review comments via API."""
    result = run_gh([
        "api",
        f"repos/{owner}/{repo}/pulls/{pr_number}/comments",
        "--paginate",
    ])
    return result if isinstance(result, list) else []


def get_issue_comments(owner: str, repo: str, pr_number: int) -> list[dict[str, Any]]:
    """Get PR conversation comments (includes bot comments)."""
    result = run_gh([
        "api",
        f"repos/{owner}/{repo}/issues/{pr_number}/comments",
        "--paginate",
    ])
    return result if isinstance(result, list) else []


def get_review_threads(owner: str, repo: str, pr_number: int) -> list[dict[str, Any]]:
    """Get review threads with resolution status via GraphQL."""
    query = """
    query($owner: String!, $repo: String!, $pr: Int!) {
      repository(owner: $owner, name: $repo) {
        pullRequest(number: $pr) {
          reviewThreads(first: 100) {
            nodes {
              id
              isResolved
              isOutdated
              path
              line
              comments(first: 10) {
                nodes {
                  id
                  body
                  author {
                    login
                  }
                  createdAt
                }
              }
            }
          }
        }
      }
    }
    """
    try:
        result = subprocess.run(
            [
                "gh", "api", "graphql",
                "-f", f"query={query}",
                "-F", f"owner={owner}",
                "-F", f"repo={repo}",
                "-F", f"pr={pr_number}",
            ],
            capture_output=True,
            text=True,
            check=True,
        )
        data = json.loads(result.stdout)
        threads = data.get("data", {}).get("repository", {}).get("pullRequest", {}).get("reviewThreads", {}).get("nodes", [])
        return threads
    except (subprocess.CalledProcessError, json.JSONDecodeError):
        return []


def detect_logaf(body: str) -> str | None:
    """Detect LOGAF scale markers in comment body.

    LOGAF scale (https://develop.sentry.dev/engineering-practices/code-review/#logaf-scale):
    - l: / [l] / low: → low priority (optional)
    - m: / [m] / medium: → medium priority (should address)
    - h: / [h] / high: → high priority (must address)

    Returns 'high', 'medium', 'low', or None if no marker found.
    """
    # Check for LOGAF markers at start of comment (with optional whitespace)
    logaf_patterns = [
        # h: or [h] or high: patterns
        (r"^\s*(?:h:|h\s*:|high:|\[h\])", "high"),
        # m: or [m] or medium: patterns
        (r"^\s*(?:m:|m\s*:|medium:|\[m\])", "medium"),
        # l: or [l] or low: patterns
        (r"^\s*(?:l:|l\s*:|low:|\[l\])", "low"),
    ]

    for pattern, level in logaf_patterns:
        if re.search(pattern, body, re.IGNORECASE):
            return level

    return None


def categorize_comment(comment: dict[str, Any], body: str) -> str:
    """Categorize a comment based on content and author.

    Uses LOGAF scale: high (must fix), medium (should fix), low (optional).
    """
    author = comment.get("author", {}).get("login", "") or comment.get("user", {}).get("login", "")

    # Info bots are skipped silently; review bots fall through to content
    # categorization so their actionable feedback is not lost.
    if is_info_bot(author) and not is_review_bot(author):
        return "bot"

    # Check for explicit LOGAF markers first
    logaf_level = detect_logaf(body)
    if logaf_level:
        return logaf_level

    # Look for high-priority (blocking) indicators
    high_patterns = [
        r"(?i)must\s+(fix|change|update|address)",
        r"(?i)this\s+(is\s+)?(wrong|incorrect|broken|buggy)",
        r"(?i)security\s+(issue|vulnerability|concern)",
        r"(?i)will\s+(break|cause|fail)",
        r"(?i)critical",
        r"(?i)blocker",
    ]

    for pattern in high_patterns:
        if re.search(pattern, body):
            return "high"

    # Look for low-priority (suggestion) indicators
    low_patterns = [
        r"(?i)nit[:\s]",
        r"(?i)nitpick",
        r"(?i)suggestion[:\s]",
        r"(?i)consider\s+",
        r"(?i)could\s+(also\s+)?",
        r"(?i)might\s+(want\s+to|be\s+better)",
        r"(?i)optional[:\s]",
        r"(?i)minor[:\s]",
        r"(?i)style[:\s]",
        r"(?i)prefer\s+",
        r"(?i)what\s+do\s+you\s+think",
        r"(?i)up\s+to\s+you",
        r"(?i)take\s+it\s+or\s+leave",
        r"(?i)fwiw",
    ]

    for pattern in low_patterns:
        if re.search(pattern, body):
            return "low"

    # Default to medium for non-bot comments without clear indicators
    return "medium"


def extract_feedback_item(
    body: str,
    author: str,
    path: str | None = None,
    line: int | None = None,
    url: str | None = None,
    is_resolved: bool = False,
    is_outdated: bool = False,
    review_bot: bool = False,
    thread_id: str | None = None,
) -> dict[str, Any]:
    """Create a standardized feedback item."""
    # Truncate long bodies for summary
    summary = body[:200] + "..." if len(body) > 200 else body
    summary = summary.replace("\n", " ").strip()

    item = {
        "author": author,
        "body": summary,
        "full_body": body,
    }

    if path:
        item["path"] = path
    if line:
        item["line"] = line
    if url:
        item["url"] = url
    if is_resolved:
        item["resolved"] = True
    if is_outdated:
        item["outdated"] = True
    if review_bot:
        item["review_bot"] = True
    if thread_id:
        item["thread_id"] = thread_id

    return item


def main():
    parser = argparse.ArgumentParser(description="Fetch and categorize PR feedback")
    parser.add_argument("--pr", type=int, help="PR number (defaults to current branch PR)")
    args = parser.parse_args()

    # Get repo info
    repo_info = get_repo_info()
    if not repo_info:
        print(json.dumps({"error": "Could not determine repository"}))
        sys.exit(1)
    owner, repo = repo_info

    # Get PR info
    pr_info = get_pr_info(args.pr)
    if not pr_info:
        print(json.dumps({"error": "No PR found for current branch"}))
        sys.exit(1)

    pr_number = pr_info["number"]
    pr_author = pr_info.get("author", {}).get("login", "")

    # Get review decision
    review_decision = pr_info.get("reviewDecision", "")

    # Categorized feedback using LOGAF scale
    feedback = {
        "high": [],      # Must address before merge
        "medium": [],    # Should address
        "low": [],       # Optional suggestions
        "bot": [],
        "resolved": [],
    }

    # Process reviews for overall status
    reviews = pr_info.get("reviews", [])
    for review in reviews:
        if review.get("state") == "CHANGES_REQUESTED":
            author = review.get("author", {}).get("login", "")
            body = review.get("body", "")
            if body and author != pr_author:
                item = extract_feedback_item(body, author)
                item["type"] = "changes_requested"
                feedback["high"].append(item)

    # Get review threads (inline comments with resolution status)
    threads = get_review_threads(owner, repo, pr_number)
    seen_thread_ids = set()

    for thread in threads:
        if not thread.get("comments", {}).get("nodes"):
            continue

        first_comment = thread["comments"]["nodes"][0]
        author = first_comment.get("author", {}).get("login", "")
        body = first_comment.get("body", "")

        # Skip if author is PR author (self-comments)
        if author == pr_author:
            continue

        # Skip empty or very short comments
        if not body or len(body.strip()) < 3:
            continue

        is_resolved = thread.get("isResolved", False)
        is_outdated = thread.get("isOutdated", False)

        thread_id = thread.get("id")
        item = extract_feedback_item(
            body=body,
            author=author,
            path=thread.get("path"),
            line=thread.get("line"),
            is_resolved=is_resolved,
            is_outdated=is_outdated,
            thread_id=thread_id,
        )

        if thread_id:
            seen_thread_ids.add(thread_id)

        if is_resolved:
            feedback["resolved"].append(item)
        elif is_review_bot(author):
            category = categorize_comment(first_comment, body)
            item["review_bot"] = True
            feedback[category].append(item)
        elif is_info_bot(author):
            feedback["bot"].append(item)
        else:
            category = categorize_comment(first_comment, body)
            feedback[category].append(item)

    # Get issue comments (general PR conversation)
    issue_comments = get_issue_comments(owner, repo, pr_number)

    for comment in issue_comments:
        author = comment.get("user", {}).get("login", "")
        body = comment.get("body", "")

        # Skip if author is PR author
        if author == pr_author:
            continue

        # Skip empty comments
        if not body or len(body.strip()) < 3:
            continue

        item = extract_feedback_item(
            body=body,
            author=author,
            url=comment.get("html_url"),
        )

        if is_review_bot(author):
            category = categorize_comment(comment, body)
            item["review_bot"] = True
            feedback[category].append(item)
        elif is_info_bot(author):
            feedback["bot"].append(item)
        else:
            category = categorize_comment(comment, body)
            feedback[category].append(item)

    # Count review bot items across priority buckets
    review_bot_count = sum(
        1 for bucket in ("high", "medium", "low")
        for item in feedback[bucket]
        if item.get("review_bot")
    )

    # Build output
    output = {
        "pr": {
            "number": pr_number,
            "url": pr_info.get("url", ""),
            "author": pr_author,
            "review_decision": review_decision,
        },
        "summary": {
            "high": len(feedback["high"]),
            "medium": len(feedback["medium"]),
            "low": len(feedback["low"]),
            "bot_comments": len(feedback["bot"]),
            "resolved": len(feedback["resolved"]),
            "review_bot_feedback": review_bot_count,
            "needs_attention": len(feedback["high"]) + len(feedback["medium"]),
        },
        "feedback": feedback,
    }

    # Add actionable summary based on LOGAF priorities
    if feedback["high"]:
        output["action_required"] = "Address high-priority feedback before merge"
    elif feedback["medium"]:
        output["action_required"] = "Address medium-priority feedback"
    elif feedback["low"]:
        output["action_required"] = "Review low-priority suggestions - ask user which to address"
    else:
        output["action_required"] = None

    print(json.dumps(output, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.9"
# ///
"""
Monitor PR checks until they reach a terminal state.

Usage:
    uv run monitor_pr_checks.py [--pr PR_NUMBER]

If --pr is not specified, uses the PR for the current branch.

Output:
    - Prints `ALL_CHECKS_PASSED` when all checks finish without failures
    - Prints `CHECKS_DONE_WITH_FAILURES` when checks finish with failures
    - Prints `NO_CHECKS_REGISTERED` when checks do not appear after the grace period
    - Prints `DRAFT_PR_WITH_NO_CHECKS` when a draft PR has no checks after the grace period
    - Prints `CHECKS_BLOCKED_BY_REVIEW_GATE` when only human review/approval gates remain
    - Prints a tab-separated check summary after the terminal marker

The script stays quiet while polling so background monitor tools do not emit
unnecessary notifications on every iteration. Transient `gh` failures are
retried instead of terminating the monitor.
"""

from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
import time
from typing import Any

HUMAN_GATE_PATTERNS = [
    r"(?i)review\s+required",
    r"(?i)required\s+review",
    r"(?i)requires\s+review",
    r"(?i)required\s+approving\s+review",
    r"(?i)approval\s+required",
    r"(?i)waiting\s+for\s+approval",
    r"(?i)manual\s+approval",
    r"(?i)draft\s+(pull\s+request|pr)",
]


def run_gh_json(
    args: list[str],
    allowed_returncodes: tuple[int, ...] = (0,),
    empty_stdout_value: list[dict[str, Any]] | dict[str, Any] | None = None,
) -> list[dict[str, Any]] | dict[str, Any] | None:
    """Run a gh command that returns JSON."""
    result = subprocess.run(
        ["gh"] + args,
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode not in allowed_returncodes:
        return None

    if not result.stdout.strip():
        return empty_stdout_value

    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        return None


def get_pr_info(pr_number: int | None) -> dict[str, Any] | None:
    """Resolve the PR to monitor."""
    if pr_number is not None:
        pr_info = run_gh_json([
            "pr",
            "view",
            str(pr_number),
            "--json",
            "number,url,isDraft,reviewDecision",
        ])
    else:
        pr_info = run_gh_json(["pr", "view", "--json", "number,url,isDraft,reviewDecision"])

    if not isinstance(pr_info, dict):
        return None

    number = pr_info.get("number")
    return pr_info if isinstance(number, int) else None


def get_checks(pr_number: int) -> list[dict[str, Any]] | None:
    """Fetch the current check list for a PR."""
    checks = run_gh_json([
        "pr",
        "checks",
        str(pr_number),
        "--json",
        "name,bucket,link,workflow,state,description",
    ], allowed_returncodes=(0, 1, 8, 16), empty_stdout_value=[])
    return checks if isinstance(checks, list) else None


def is_human_gate_check(check: dict[str, Any]) -> bool:
    """Return true when a pending entry is a human review/approval gate."""
    haystack = " ".join(
        str(check.get(field, ""))
        for field in ("name", "state", "description", "workflow")
    )
    return any(re.search(pattern, haystack) for pattern in HUMAN_GATE_PATTERNS)


def print_check_summary(checks: list[dict[str, Any]], max_lines: int = 20) -> None:
    """Print a concise tab-separated check summary."""
    for check in checks[:max_lines]:
        name = str(check.get("name", "unknown"))
        bucket = str(check.get("bucket", "unknown"))
        link = str(check.get("link", ""))
        print(f"{name}\t{bucket}\t{link}".rstrip(), flush=True)


def print_no_checks_summary(pr_info: dict[str, Any]) -> None:
    number = pr_info.get("number", "unknown")
    url = pr_info.get("url", "")
    is_draft = str(bool(pr_info.get("isDraft"))).lower()
    review_decision = str(pr_info.get("reviewDecision") or "")
    print(f"PR #{number}\tno_checks\t{url}".rstrip(), flush=True)
    print(f"is_draft\t{is_draft}", flush=True)
    if review_decision:
        print(f"review_decision\t{review_decision}", flush=True)


def main() -> int:
    parser = argparse.ArgumentParser(description="Monitor PR checks until they finish")
    parser.add_argument("--pr", type=int, help="PR number (defaults to current branch PR)")
    parser.add_argument(
        "--poll-seconds",
        type=int,
        default=30,
        help="Polling interval while checks are pending or gh is transiently failing",
    )
    parser.add_argument(
        "--no-checks-seconds",
        type=int,
        default=15,
        help="Retry delay when a fresh push has not registered any checks yet",
    )
    parser.add_argument(
        "--no-checks-timeout-seconds",
        type=int,
        default=180,
        help="Maximum time to wait for checks to register before reporting no checks",
    )
    args = parser.parse_args()

    pr_info = get_pr_info(args.pr)
    if pr_info is None:
        print("No PR found for current branch", file=sys.stderr)
        return 1

    pr_number = pr_info["number"]
    no_checks_started_at: float | None = None

    while True:
        checks = get_checks(pr_number)
        if checks is None:
            time.sleep(args.poll_seconds)
            continue

        if not checks:
            now = time.monotonic()
            if no_checks_started_at is None:
                no_checks_started_at = now
            if now - no_checks_started_at >= args.no_checks_timeout_seconds:
                marker = "DRAFT_PR_WITH_NO_CHECKS" if pr_info.get("isDraft") else "NO_CHECKS_REGISTERED"
                print(marker, flush=True)
                print_no_checks_summary(pr_info)
                return 0
            time.sleep(args.no_checks_seconds)
            continue

        no_checks_started_at = None

        pending_checks = [check for check in checks if check.get("bucket") == "pending"]
        failed = sum(1 for check in checks if check.get("bucket") == "fail")
        actionable_pending = [
            check for check in pending_checks if not is_human_gate_check(check)
        ]
        if actionable_pending:
            time.sleep(args.poll_seconds)
            continue
        if failed:
            print("CHECKS_DONE_WITH_FAILURES", flush=True)
            print_check_summary(checks)
            return 0
        if pending_checks:
            print("CHECKS_BLOCKED_BY_REVIEW_GATE", flush=True)
            print_check_summary(checks)
            return 0

        print("ALL_CHECKS_PASSED", flush=True)
        print_check_summary(checks)
        return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.9"
# ///
"""
Reply to PR review threads.

Usage:
    uv run reply_to_thread.py THREAD_ID BODY [THREAD_ID BODY ...]

Accepts one or more (thread_id, body) pairs as positional arguments.
Batches all replies into a single GraphQL mutation for efficiency.

Example:
    uv run reply_to_thread.py PRRT_abc "Fixed the issue."
    uv run reply_to_thread.py PRRT_abc "Fixed." PRRT_def "Also fixed."
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys


def _normalize_body(body: str) -> str:
    """Normalize escaped newlines from shell input.

    Bash double quotes keep "\\n" literal, but reply bodies should contain
    actual newlines for readability.
    """
    return body.replace("\\r\\n", "\\n").replace("\\n", "\n")


def reply_to_threads(pairs: list[tuple[str, str]]) -> list[tuple[str, bool]]:
    """Reply to one or more review threads in a single GraphQL call.

    Returns a per-operation list of (thread_id, success) tuples.
    """
    # Build aliased mutation
    mutations = []
    for i, (thread_id, body) in enumerate(pairs):
        escaped_thread_id = json.dumps(thread_id)
        escaped_body = json.dumps(_normalize_body(body))  # handles newlines, quotes
        mutations.append(
            f"  r{i}: addPullRequestReviewThreadReply(input: {{"
            f"pullRequestReviewThreadId: {escaped_thread_id}, "
            f"body: {escaped_body}"
            f"}}) {{ clientMutationId }}"
        )

    query = "mutation {\n" + "\n".join(mutations) + "\n}"

    try:
        result = subprocess.run(
            ["gh", "api", "graphql", "-f", f"query={query}"],
            capture_output=True,
            text=True,
            timeout=30,
        )
        if result.returncode != 0:
            print(f"GraphQL error: {result.stderr}", file=sys.stderr)
            return [(tid, False) for tid, _ in pairs]

        # Parse response to detect per-alias GraphQL errors
        try:
            response = json.loads(result.stdout)
        except (json.JSONDecodeError, TypeError):
            print(f"Failed to parse GraphQL response: {result.stdout}", file=sys.stderr)
            return [(tid, False) for tid, _ in pairs]

        data = response.get("data") or {}
        errors = response.get("errors") or []

        # Build a set of alias indices that have errors
        error_paths = set()
        for err in errors:
            for segment in err.get("path") or []:
                if isinstance(segment, str) and segment.startswith("r"):
                    error_paths.add(segment)

        operation_results = []
        for i, (tid, _) in enumerate(pairs):
            alias = f"r{i}"
            if alias in error_paths or data.get(alias) is None:
                operation_results.append((tid, False))
            else:
                operation_results.append((tid, True))

        if any(not ok for _, ok in operation_results):
            failed = [tid for tid, ok in operation_results if not ok]
            print(f"GraphQL partial failure for threads: {failed}", file=sys.stderr)

        return operation_results
    except subprocess.TimeoutExpired:
        print("Request timed out", file=sys.stderr)
        return [(tid, False) for tid, _ in pairs]


def main():
    parser = argparse.ArgumentParser(
        description="Reply to PR review threads",
        usage="%(prog)s THREAD_ID BODY [THREAD_ID BODY ...]",
    )
    parser.add_argument(
        "args",
        nargs="+",
        help="Alternating thread_id and body pairs",
    )
    parsed = parser.parse_args()

    if len(parsed.args) % 2 != 0:
        print("Error: arguments must be (thread_id, body) pairs", file=sys.stderr)
        sys.exit(1)

    pairs = []
    for i in range(0, len(parsed.args), 2):
        pairs.append((parsed.args[i], parsed.args[i + 1]))

    results = reply_to_threads(pairs)

    # Output results
    success = all(ok for _, ok in results)
    by_thread = {}
    for tid, ok in results:
        by_thread.setdefault(tid, []).append(ok)

    output = {
        "replied": sum(1 for _, ok in results if ok),
        "failed": sum(1 for _, ok in results if not ok),
        "operations": [
            {"thread_id": tid, "status": "ok" if ok else "failed"}
            for tid, ok in results
        ],
        "threads": {
            tid: "ok" if all(statuses) else "failed"
            for tid, statuses in by_thread.items()
        },
    }
    print(json.dumps(output, indent=2))

    if not success:
        sys.exit(1)


if __name__ == "__main__":
    main()

---
name: iterate-pr
description: Iterate on a PR until actionable CI passes and high/medium review feedback is addressed. Use for PR CI failures, review feedback, or green-check loops; do not wait for human approval, draft status, or merge gates.
disable-model-invocation: false
---

# Iterate on PR Until CI Passes

Goal: fix actionable CI failures and high/medium review feedback. Stop and report human approval, draft-readiness, and merge-readiness gates.

Requires:
- authenticated `gh`
- `uv`
- target repository root as cwd
- skill-root-relative script paths, for example `scripts/fetch_pr_checks.py`

## Bundled Scripts

| Script | Run | Output |
|--------|-----|--------|
| `scripts/fetch_pr_checks.py` | `uv run scripts/fetch_pr_checks.py [--pr NUMBER]` | JSON: `pr`, `summary`, `checks`, failure snippets |
| `scripts/fetch_pr_feedback.py` | `uv run scripts/fetch_pr_feedback.py [--pr NUMBER]` | JSON buckets: `high`, `medium`, `low`, `bot`, `resolved` |
| `scripts/monitor_pr_checks.py` | `uv run scripts/monitor_pr_checks.py [--pr NUMBER]` | terminal marker plus tab-separated checks |
| `scripts/reply_to_thread.py` | `uv run scripts/reply_to_thread.py THREAD_ID BODY [...]` | JSON reply results |

Check summary fields include `failed`, `pending`, `actionable_pending`, and `human_gate_pending`.

Monitor markers:
- `ALL_CHECKS_PASSED`
- `CHECKS_DONE_WITH_FAILURES`
- `NO_CHECKS_REGISTERED`
- `DRAFT_PR_WITH_NO_CHECKS`
- `CHECKS_BLOCKED_BY_REVIEW_GATE`

## Workflow

### 1. Identify PR

Run:
```bash
gh pr view --json number,url,headRefName,isDraft,reviewDecision
```

Stop when:
- no PR exists
- draft PR has no checks after monitor grace period: report `DRAFT_PR_WITH_NO_CHECKS`

Draft rule: inspect existing checks/feedback only. Do not mark ready for review unless asked.

### 2. Handle Feedback

Run `uv run scripts/fetch_pr_feedback.py [--pr NUMBER]`.

| Bucket | Action |
|--------|--------|
| `high` | fix |
| `medium` | fix |
| `low` | ask user which to address |
| `bot` | skip informational comments |
| `resolved` | skip |

Feedback fix checklist:
- verify root cause
- search related code
- fix all instances
- for `review_bot: true`: fix real issues, explain false positives

Low-priority prompt format:
```text
Found 3 low-priority suggestions:
1. [l] "Consider renaming this variable" - @reviewer in api.py:42
2. [nit] "Could use a list comprehension" - @reviewer in utils.py:18
3. [style] "Add a docstring" - @reviewer in models.py:55

Which should I address? ("1,3", "all", or "none")
```

### 3. Check CI Status

Run `uv run scripts/fetch_pr_checks.py [--pr NUMBER]`.

| State | Action |
|-------|--------|
| `failed > 0` and `actionable_pending == 0` | fix failures |
| `actionable_pending > 0` | wait; poll feedback while waiting |
| `pending > 0` and `actionable_pending == 0` | report `CHECKS_BLOCKED_BY_REVIEW_GATE` |
| no checks after grace period | report `NO_CHECKS_REGISTERED` or `DRAFT_PR_WITH_NO_CHECKS` |
| all actionable checks passed | run post-CI feedback check |

Wait for actionable review bots: sentry, warden, cursor, bugbot, seer, codeql.
Do not wait for approval, `isDraft`, `REVIEW_REQUIRED`, Codecov, or informational bots.

### 4. Fix CI Failures

For each failure:
1. read full log: `gh run view <run-id> --log-failed`
2. trace from assertion/exception/lint rule to source
3. state the cause before editing: "fails because X, affected by Y"
4. search related call sites/patterns
5. fix root cause, not symptom
6. add focused test coverage when needed

### 5. Verify Locally, Then Commit and Push

Before commit:
- test fix: rerun specific test
- lint/type fix: rerun affected checker
- code fix: rerun covering tests
- local failure: fix before pushing

```bash
git add <files>
git commit -m "fix: <descriptive message>"
git push
```

### 6. Monitor CI and Address Feedback

Loop:
1. run `uv run scripts/fetch_pr_checks.py`
2. handle table in step 3
3. while `actionable_pending > 0`, run `uv run scripts/fetch_pr_feedback.py`
4. fix new high/medium feedback immediately
5. if changed, verify, commit, push, restart loop
6. otherwise sleep 30 seconds and repeat
7. after checks pass, wait 10 seconds, fetch feedback once more
8. if new high/medium feedback exists, return to step 4

Claude Code optional: run `uv run scripts/monitor_pr_checks.py` through `MonitorTool` with `persistent: false`; set timeout to normal repo CI duration. Restart the monitor after every push.

## Exit Conditions

| Exit | Conditions |
|------|------------|
| Success | actionable CI passed; post-CI feedback clean; low-priority choice handled |
| Ask user | same failure after 2 attempts; feedback unclear; infrastructure issue |
| Stop | no PR; branch needs rebase; no checks; draft no-checks; only human gates remain |

## Fallback

If scripts fail, use `gh` CLI directly:
- `gh pr view --json number,url,headRefName,isDraft,reviewDecision`
- `gh pr checks --json name,state,bucket,description,link`
- `gh run view <run-id> --log-failed`
- `gh api repos/{owner}/{repo}/pulls/{number}/comments`

# Iterate PR Specification

## Intent

The `iterate-pr` skill drives a pull request through actionable CI failures and actionable review feedback until the work is locally fixed, pushed, and rechecked.

Its purpose is CI and feedback iteration, not merge readiness. It must not wait indefinitely for human approvals, required review decisions, draft PR state changes, or other gates that an agent cannot resolve by editing code.

## Scope

In scope:

- Identifying the PR for the current branch.
- Fetching and categorizing PR review feedback.
- Fixing high and medium priority review feedback.
- Asking the user which low priority suggestions to address.
- Fetching CI checks, failed logs, and failure snippets.
- Fixing CI failures with local verification before pushing.
- Monitoring checks until they pass, fail, or reach a non-actionable stop state.
- Reporting draft/no-checks and human review/approval gates without polling forever.

Out of scope:

- Waiting for or requesting human approval.
- Marking draft PRs ready for review unless the user explicitly asks.
- Merging PRs.
- Rebasing branches without user direction.
- Treating Codecov, Dependabot, or other informational comments as review feedback.

## Users And Trigger Context

- Primary users: engineers and coding agents iterating on existing pull requests.
- Invocation: model-invocable in Claude Code and Codex when the request or current PR state matches this skill's trigger context.
- Common user requests: fix CI on this PR, iterate on this PR until checks pass, address PR feedback, keep pushing fixes until green.
- Should not trigger for: creating a PR, writing commits without a PR, reviewing unrelated code, or monitoring merge approval state only.

## Runtime Contract

- Required first actions: resolve the current PR, read `isDraft` and `reviewDecision`, fetch current review feedback, and fetch current CI state before editing.
- Required outputs: concise progress updates, commits and pushes when fixes are made, and a final state that distinguishes passing CI from non-actionable review/draft/approval gates.
- Non-negotiable constraints: investigate failures before editing, verify locally before pushing, do not push known-broken fixes, do not wait for human approval, and do not treat draft PRs with no checks as pending forever.
- Expected bundled files loaded at runtime: `SKILL.md` and, when needed, scripts under `scripts/`.

## Source And Evidence Model

Authoritative sources:

- GitHub CLI PR and checks output.
- Sentry LOGAF review guidance.
- Repository-level agent instructions.
- Bundled script behavior documented in `SKILL.md`.

Useful improvement sources:

- positive examples: PRs where CI failures were fixed and checks passed after the loop.
- negative examples: PRs where the agent waited on draft status, required review, or approval gates.
- issue or PR feedback: reviewer comments about missing fixes, false positives, or feedback categorization.
- validation results: structural skill validation and script syntax checks.

Data that must not be stored:

- secrets
- customer data
- private URLs or identifiers not needed for reproduction
- full CI logs when small failure snippets are enough

## Reference Architecture

- `SKILL.md` contains the runtime workflow, script contracts, feedback handling rules, CI loop, and exit conditions.
- `SPEC.md` contains this maintenance contract.
- `references/` contains no files currently; add focused troubleshooting or evidence references only if runtime guidance becomes noisy.
- `references/evidence/` contains no files currently; use it for durable positive or negative PR-loop examples if regressions recur.
- `scripts/` contains non-interactive helpers for PR checks, PR feedback, check monitoring, and review-thread replies.
- `assets/` contains no files currently.

## Validation

- Lightweight validation: run `uv run skills/skill-writer/scripts/quick_validate.py skills/iterate-pr`.
- Script validation: run `uv run -m py_compile skills/iterate-pr/scripts/*.py` after script changes.
- Holdout examples: include a draft PR with no registered checks, a PR with `reviewDecision: REVIEW_REQUIRED` but passing checks, a PR with an actionable pending CI bot check, and a PR with failed CI logs.
- Acceptance gates: validator passes, scripts compile, draft/no-check states terminate with a report, human review gates are not treated as actionable pending CI, and actionable CI failures still route back to investigation and fixes.

## Known Limitations

- Human-gate detection depends on check names, states, and descriptions exposed by GitHub or CI integrations.
- Some repositories may intentionally model deployment or approval workflows as status checks; this skill reports those as blocked/non-actionable unless the user asks to manage that gate.
- The helper scripts use GitHub CLI output and can drift if `gh pr checks` changes its JSON schema.

## Maintenance Notes

- Update `SKILL.md` when the runtime loop, script contracts, feedback policy, or exit conditions change.
- Update `SPEC.md` when the skill's scope, validation expectations, or non-actionable gate policy changes.
- Add focused reference files only when repeated troubleshooting guidance would make `SKILL.md` hard to scan.
- Keep public inventories pointed at the canonical `skills/iterate-pr` skill, not mirrors or aliases.