Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 52 additions & 3 deletions scripts/ci/pr_review_fix_scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from __future__ import annotations

import argparse
import concurrent.futures
import json
import os
import re
Expand Down Expand Up @@ -133,7 +134,13 @@ def dispatch_autofix(repo: str, pr: dict[str, Any], *, workflow: str, dry_run: b
run(args)


def inspect_pr(repo: str, pr: dict[str, Any], args: argparse.Namespace) -> tuple[str, tuple[str, ...]]:
def inspect_pr(
repo: str,
pr: dict[str, Any],
args: argparse.Namespace,
*,
comments: list[dict[str, Any]] | None = None,
) -> tuple[str, tuple[str, ...]]:
"""Inspect one PR and optionally dispatch autofix."""
number = int(pr["number"])
if pr.get("isDraft"):
Expand All @@ -147,7 +154,9 @@ def inspect_pr(repo: str, pr: dict[str, Any], args: argparse.Namespace) -> tuple
if not needs_fix:
return "skip", ("no current-head change request or active unresolved review thread",)

comments = issue_comments(repo, number)
if comments is None:
comments = issue_comments(repo, number)

if recent_fix_marker_exists(comments, str(pr["headRefOid"]), args.retry_hours * 3600):
return "wait", ("recent autofix marker exists for this head",)

Expand All @@ -163,13 +172,53 @@ def process_queue(args: argparse.Namespace) -> int:
inspected = 0
decisions: list[dict[str, Any]] = []

prs_needing_comments = []
for pr in prs:
if pr.get("isDraft"):
continue
if pr.get("baseRefName") != args.base_branch:
continue
if not same_repository_head(args.repo, pr):
continue
needs_fix, _ = needs_autofix(pr)
if needs_fix:
prs_needing_comments.append(pr)

comments_by_pr: dict[int, list[dict[str, Any]]] = {}
if len(prs_needing_comments) <= 1:
# Fast path for single items
for pr in prs_needing_comments:
pr_number = int(pr["number"])
comments_by_pr[pr_number] = issue_comments(args.repo, pr_number)
else:
# ⚡ Bolt: Avoid N+1 API blocking by parallelizing independent issue_comments fetches
# Impact: Reduces wait time from O(N) API calls to O(N/max_workers) for queue scanning
max_workers = min(10, len(prs_needing_comments))
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
def fetch_comments(pr_number: int) -> tuple[int, list[dict[str, Any]]]:
return pr_number, issue_comments(args.repo, pr_number)

futures = [executor.submit(fetch_comments, int(pr["number"])) for pr in prs_needing_comments]
for future in concurrent.futures.as_completed(futures):
try:
pr_number, comments = future.result()
comments_by_pr[pr_number] = comments
except Exception:
pass

for pr in prs:
inspected += 1
if dispatched >= args.max_dispatches:
decisions.append({"pr": pr["number"], "action": "skip", "reasons": ["autofix dispatch limit reached"]})
continue
try:
action, reasons = inspect_pr(args.repo, pr, args)
pr_number = int(pr["number"])
action, reasons = inspect_pr(
args.repo,
pr,
args,
comments=comments_by_pr.get(pr_number),
)
except RuntimeError as exc:
action, reasons = "error", (str(exc),)
if action == "dispatch":
Expand Down
4 changes: 2 additions & 2 deletions tests/test_pr_review_fix_scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -275,14 +275,14 @@ def test_fix_inspect_skip_wait_and_error_paths(monkeypatch):
pr1 = make_pr(number=1)
pr2 = make_pr(number=2)
monkeypatch.setattr(fix, "fetch_open_prs", lambda repo, max_prs: [pr1, pr2])
monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args: ("dispatch", ("reason",)))
monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args, **kwargs: ("dispatch", ("reason",)))
payload_lines = []
monkeypatch.setattr("builtins.print", lambda *parts, **kwargs: payload_lines.append(" ".join(map(str, parts))))
assert fix.process_queue(args) == 0
assert "autofix dispatch limit reached" in payload_lines[-1]

monkeypatch.setattr(fix, "fetch_pr", lambda repo, number: [make_pr(number=number)])
monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args: (_ for _ in ()).throw(RuntimeError("boom")))
monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args, **kwargs: (_ for _ in ()).throw(RuntimeError("boom")))
args = fix.parse_args(["--repo", "owner/repo", "--base-branch", "main", "--pr-number", "3"])
payload_lines.clear()
assert fix.process_queue(args) == 0
Expand Down
55 changes: 55 additions & 0 deletions tests/test_pr_review_fix_scheduler_coverage.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
import pytest
import scripts.ci.pr_review_fix_scheduler as fix

def test_coverage_process_queue_skips_draft_and_wrong_base_and_external_repo(monkeypatch):
def make_pr(number=1, **kwargs):
pr = {
"number": number,
"headRefOid": "abc",
"baseRefName": "main",
"headRefName": "feature",
"isDraft": False,
"headRepository": {"nameWithOwner": "owner/repo"},
}
pr.update(kwargs)
return pr

args = fix.parse_args(["--repo", "owner/repo", "--base-branch", "main"])

pr1 = make_pr(number=1, isDraft=True)
pr2 = make_pr(number=2, baseRefName="other")
pr3 = make_pr(number=3, headRepository={"nameWithOwner": "fork/repo"})

monkeypatch.setattr(fix, "fetch_open_prs", lambda repo, max_prs: [pr1, pr2, pr3])
monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args, **kwargs: ("skip", ("skip reason",)))

assert fix.process_queue(args) == 0

def test_coverage_process_queue_exception_handling(monkeypatch):
def make_pr(number=1, **kwargs):
pr = {
"number": number,
"headRefOid": "abc",
"baseRefName": "main",
"headRefName": "feature",
"isDraft": False,
"headRepository": {"nameWithOwner": "owner/repo"},
}
pr.update(kwargs)
return pr

args = fix.parse_args(["--repo", "owner/repo", "--base-branch", "main"])

pr1 = make_pr(number=1)
pr2 = make_pr(number=2)

monkeypatch.setattr(fix, "fetch_open_prs", lambda repo, max_prs: [pr1, pr2])
monkeypatch.setattr(fix, "needs_autofix", lambda pr: (True, ("reason",)))

def raise_error(repo, number):
raise RuntimeError("boom")

monkeypatch.setattr(fix, "issue_comments", raise_error)
monkeypatch.setattr(fix, "inspect_pr", lambda repo, pr, args, **kwargs: ("skip", ("skip reason",)))

assert fix.process_queue(args) == 0
Loading