Iterate Pr

#!/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.\n\n*— Claude Code*"
    uv run reply_to_thread.py PRRT_abc "Fixed." PRRT_def "Also fixed."
"""

from __future__ import annotations

import argparse
import json
import re
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/signatures.
    """
    normalized = body.replace("\\r\\n", "\\n").replace("\\n", "\n")
    
    # Add Claude Code attribution if not already present
    # Check if the last line matches the bot signature pattern: *— Bot Name* or *- Bot Name*
    lines = normalized.rstrip().split("\n")
    last_line = lines[-1] if lines else ""
    
    # Match bot signatures like "*— Claude Code*", "*- Any Bot*", etc.
    bot_signature_pattern = r"^\*[—-]\s+.+\*$"
    
    if not re.match(bot_signature_pattern, last_line.strip()):
        # Ensure proper spacing before attribution
        if normalized and not normalized.endswith("\n"):
            normalized += "\n"
        if normalized and not normalized.endswith("\n\n"):
            normalized += "\n"
        normalized += "*— Claude Code*"
    
    return normalized


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