diff --git a/.github/scripts/detect_overlaps.py b/.github/scripts/detect_overlaps.py
index eab2b856ec..8d8e5f4392 100644
--- a/.github/scripts/detect_overlaps.py
+++ b/.github/scripts/detect_overlaps.py
@@ -16,102 +16,774 @@ from dataclasses import dataclass
from typing import Optional
-@dataclass
-class ChangedFile:
- path: str
- additions: list[tuple[int, int]] # List of (start_line, end_line) ranges
- deletions: list[tuple[int, int]]
- is_rename: bool = False
- old_path: str = None
+# =============================================================================
+# MAIN ENTRY POINT
+# =============================================================================
+
+def main():
+ """Main entry point for PR overlap detection."""
+ import argparse
+
+ parser = argparse.ArgumentParser(description="Detect PR overlaps and potential merge conflicts")
+ parser.add_argument("pr_number", type=int, help="PR number to check")
+ parser.add_argument("--base", default=None, help="Base branch (default: auto-detect from PR)")
+ parser.add_argument("--skip-merge-test", action="store_true", help="Skip actual merge conflict testing")
+ parser.add_argument("--discord-webhook", default=os.environ.get("DISCORD_WEBHOOK_URL"), help="Discord webhook URL for notifications")
+ parser.add_argument("--dry-run", action="store_true", help="Don't post comments, just print")
+
+ args = parser.parse_args()
+
+ owner, repo = get_repo_info()
+ print(f"Checking PR #{args.pr_number} in {owner}/{repo}")
+
+ # Get current PR info
+ current_pr = fetch_pr_details(args.pr_number)
+ base_branch = args.base or current_pr.base_ref
+
+ print(f"PR #{current_pr.number}: {current_pr.title}")
+ print(f"Base branch: {base_branch}")
+ print(f"Files changed: {len(current_pr.files)}")
+
+ # Find overlapping PRs
+ overlaps, all_changes = find_overlapping_prs(
+ owner, repo, base_branch, current_pr, args.pr_number, args.skip_merge_test
+ )
+
+ if not overlaps:
+ print("No overlaps detected!")
+ return
+
+ # Generate and post report
+ comment = format_comment(overlaps, args.pr_number, current_pr.changed_ranges, all_changes)
+
+ if args.dry_run:
+ print("\n" + "="*60)
+ print("COMMENT PREVIEW:")
+ print("="*60)
+ print(comment)
+ else:
+ if comment:
+ post_or_update_comment(args.pr_number, comment)
+ print("Posted comment to PR")
+
+ if args.discord_webhook:
+ send_discord_notification(args.discord_webhook, current_pr, overlaps)
+
+ # Report results and exit
+ report_results(overlaps)
-# Files that are auto-generated or rarely cause real conflicts
-IGNORE_FILES = {
- "autogpt_platform/frontend/src/app/api/openapi.json", # Auto-generated from backend
- "poetry.lock", # Lock file, conflicts are usually trivial
- "pnpm-lock.yaml",
- "package-lock.json",
- "yarn.lock",
-}
+# =============================================================================
+# HIGH-LEVEL WORKFLOW FUNCTIONS
+# =============================================================================
+
+def fetch_pr_details(pr_number: int) -> "PullRequest":
+ """Fetch details for a specific PR including its diff."""
+ result = run_gh(["pr", "view", str(pr_number), "--json", "number,title,url,author,headRefName,baseRefName,files"])
+ data = json.loads(result.stdout)
+
+ pr = PullRequest(
+ number=data["number"],
+ title=data["title"],
+ author=data["author"]["login"],
+ url=data["url"],
+ head_ref=data["headRefName"],
+ base_ref=data["baseRefName"],
+ files=[f["path"] for f in data["files"]],
+ changed_ranges={}
+ )
+
+ # Get detailed diff
+ diff = get_pr_diff(pr_number)
+ pr.changed_ranges = parse_diff_ranges(diff)
+
+ return pr
-@dataclass
-class PullRequest:
- number: int
- title: str
- author: str
- url: str
- head_ref: str
- base_ref: str
- files: list[str]
- changed_ranges: dict[str, ChangedFile] # path -> ChangedFile
- updated_at: str = None # ISO timestamp
+def find_overlapping_prs(
+ owner: str,
+ repo: str,
+ base_branch: str,
+ current_pr: "PullRequest",
+ current_pr_number: int,
+ skip_merge_test: bool
+) -> tuple[list["Overlap"], dict[int, dict[str, "ChangedFile"]]]:
+ """Find all PRs that overlap with the current PR."""
+ # Query other open PRs
+ all_prs = query_open_prs(owner, repo, base_branch)
+ other_prs = [p for p in all_prs if p["number"] != current_pr_number]
+
+ print(f"Found {len(other_prs)} other open PRs targeting {base_branch}")
+
+ # Find file overlaps (excluding ignored files)
+ candidates = find_file_overlap_candidates(current_pr.files, other_prs)
+
+ print(f"Found {len(candidates)} PRs with file overlap (excluding ignored files)")
+
+ if not candidates:
+ return [], {}
+
+ # Analyze each candidate for line overlaps and conflicts
+ overlaps = []
+ all_changes = {}
+
+ for pr_data, shared_files in candidates:
+ overlap, pr_changes = analyze_pr_overlap(
+ owner, repo, base_branch, current_pr, pr_data, shared_files, skip_merge_test
+ )
+ if overlap:
+ overlaps.append(overlap)
+ all_changes[pr_data["number"]] = pr_changes
+
+ return overlaps, all_changes
-@dataclass
-class ConflictInfo:
- """Info about a single conflicting file."""
- path: str
- conflict_count: int = 0 # Number of conflict regions
- conflict_lines: int = 0 # Total lines in conflict regions
- conflict_type: str = "content" # content, added, deleted, renamed, binary
+def analyze_pr_overlap(
+ owner: str,
+ repo: str,
+ base_branch: str,
+ current_pr: "PullRequest",
+ other_pr_data: dict,
+ shared_files: list[str],
+ skip_merge_test: bool
+) -> tuple[Optional["Overlap"], dict[str, "ChangedFile"]]:
+ """Analyze overlap between current PR and another PR."""
+ # Filter out ignored files
+ non_ignored_shared = [f for f in shared_files if not should_ignore_file(f)]
+ if not non_ignored_shared:
+ return None, {}
+
+ other_pr = PullRequest(
+ number=other_pr_data["number"],
+ title=other_pr_data["title"],
+ author=other_pr_data["author"],
+ url=other_pr_data["url"],
+ head_ref=other_pr_data["head_ref"],
+ base_ref=other_pr_data["base_ref"],
+ files=other_pr_data["files"],
+ changed_ranges={},
+ updated_at=other_pr_data.get("updated_at")
+ )
+
+ # Get diff for other PR
+ other_diff = get_pr_diff(other_pr.number)
+ other_pr.changed_ranges = parse_diff_ranges(other_diff)
+
+ # Check line overlaps
+ line_overlaps = find_line_overlaps(
+ current_pr.changed_ranges,
+ other_pr.changed_ranges,
+ shared_files
+ )
+
+ overlap = Overlap(
+ pr_a=current_pr,
+ pr_b=other_pr,
+ overlapping_files=non_ignored_shared,
+ line_overlaps=line_overlaps
+ )
+
+ # Test for actual merge conflicts if we have line overlaps
+ if line_overlaps and not skip_merge_test:
+ print(f"Testing merge conflict with PR #{other_pr.number}...", flush=True)
+ has_conflict, conflict_files, conflict_details, error_type = test_merge_conflict(
+ owner, repo, base_branch, current_pr, other_pr
+ )
+ overlap.has_merge_conflict = has_conflict
+ overlap.conflict_files = conflict_files
+ overlap.conflict_details = conflict_details
+ overlap.conflict_type = error_type
+
+ return overlap, other_pr.changed_ranges
-@dataclass
-class Overlap:
- pr_a: PullRequest
- pr_b: PullRequest
- overlapping_files: list[str]
- line_overlaps: dict[str, list[tuple[int, int]]] # file -> overlapping line ranges
- has_merge_conflict: bool = False
- conflict_files: list[str] = None
- conflict_details: list[ConflictInfo] = None # Detailed conflict info per file
- conflict_type: str = None # None, 'pr_a_conflicts_base', 'conflict'
-
- def __post_init__(self):
- if self.conflict_files is None:
- self.conflict_files = []
- if self.conflict_details is None:
- self.conflict_details = []
+def find_file_overlap_candidates(
+ current_files: list[str],
+ other_prs: list[dict]
+) -> list[tuple[dict, list[str]]]:
+ """Find PRs that share files with the current PR."""
+ current_files_set = set(f for f in current_files if not should_ignore_file(f))
+ candidates = []
+
+ for pr_data in other_prs:
+ other_files = set(f for f in pr_data["files"] if not should_ignore_file(f))
+ shared = current_files_set & other_files
+
+ if shared:
+ candidates.append((pr_data, list(shared)))
+
+ return candidates
-def run_gh(args: list[str], check: bool = True) -> subprocess.CompletedProcess:
- """Run a gh CLI command."""
- result = subprocess.run(
- ["gh"] + args,
- capture_output=True,
- text=True,
+def report_results(overlaps: list["Overlap"]):
+ """Report results and exit with appropriate code."""
+ conflicts = [o for o in overlaps if o.has_merge_conflict]
+ if conflicts:
+ print(f"\n⚠️ Found {len(conflicts)} merge conflict(s)")
+ sys.exit(1)
+
+ line_overlap_count = len([o for o in overlaps if o.line_overlaps])
+ if line_overlap_count:
+ print(f"\n⚠️ Found {line_overlap_count} PR(s) with line overlap")
+
+ print("\n✅ Done")
+
+
+# =============================================================================
+# COMMENT FORMATTING
+# =============================================================================
+
+def format_comment(
+ overlaps: list["Overlap"],
+ current_pr: int,
+ changes_current: dict[str, "ChangedFile"],
+ all_changes: dict[int, dict[str, "ChangedFile"]]
+) -> str:
+ """Format the overlap report as a PR comment."""
+ if not overlaps:
+ return ""
+
+ lines = ["## 🔍 PR Overlap Detection"]
+ lines.append("")
+ lines.append("This check compares your PR against all other open PRs targeting the same branch to detect potential merge conflicts early.")
+ lines.append("")
+
+ # Check if current PR conflicts with base branch
+ format_base_conflicts(overlaps, lines)
+
+ # Classify and sort overlaps
+ classified = classify_all_overlaps(overlaps, current_pr, changes_current, all_changes)
+
+ # Group by risk
+ conflicts = [(o, r) for o, r in classified if r == 'conflict']
+ high_risk = [(o, r) for o, r in classified if r == 'high']
+ medium_risk = [(o, r) for o, r in classified if r == 'medium']
+ low_risk = [(o, r) for o, r in classified if r == 'low']
+
+ # Format each section
+ format_conflicts_section(conflicts, current_pr, lines)
+ format_high_risk_section(high_risk, current_pr, lines)
+ format_medium_risk_section(medium_risk, current_pr, changes_current, all_changes, lines)
+ format_low_risk_section(low_risk, current_pr, lines)
+
+ # Summary
+ total = len(overlaps)
+ lines.append(f"\n**Summary:** {len(conflicts)} conflicts, {len(high_risk)} high risk, {len(medium_risk)} medium risk, {len(low_risk)} low risk (out of {total} PRs with file overlap)")
+ lines.append("\n---\n*Auto-generated on push. Ignores: `openapi.json`, lock files.*")
+
+ return "\n".join(lines)
+
+
+def format_base_conflicts(overlaps: list["Overlap"], lines: list[str]):
+ """Format base branch conflicts section."""
+ base_conflicts = [o for o in overlaps if o.conflict_type == 'pr_a_conflicts_base']
+ if base_conflicts:
+ lines.append("### ⚠️ This PR has conflicts with the base branch\n")
+ lines.append("Conflicts will need to be resolved before merging:\n")
+ first = base_conflicts[0]
+ for f in first.conflict_files[:10]:
+ lines.append(f"- `{f}`")
+ if len(first.conflict_files) > 10:
+ lines.append(f"- ... and {len(first.conflict_files) - 10} more files")
+ lines.append("\n")
+
+
+def format_conflicts_section(conflicts: list[tuple], current_pr: int, lines: list[str]):
+ """Format the merge conflicts section."""
+ pr_conflicts = [(o, r) for o, r in conflicts if o.conflict_type != 'pr_a_conflicts_base']
+
+ if not pr_conflicts:
+ return
+
+ lines.append("### 🔴 Merge Conflicts Detected")
+ lines.append("")
+ lines.append("The following PRs have been tested and **will have merge conflicts** if merged after this PR. Consider coordinating with the authors.")
+ lines.append("")
+
+ for o, _ in pr_conflicts:
+ other = o.pr_b if o.pr_a.number == current_pr else o.pr_a
+ format_pr_entry(other, lines)
+ format_conflict_details(o, lines)
+ lines.append("")
+
+
+def format_high_risk_section(high_risk: list[tuple], current_pr: int, lines: list[str]):
+ """Format the high risk section."""
+ if not high_risk:
+ return
+
+ lines.append("### 🟠 High Risk — Significant Line Overlap")
+ lines.append("")
+ lines.append("These PRs modify many of the same lines (>20 lines). While not yet tested for conflicts, they have high potential to conflict.")
+ lines.append("")
+
+ for o, _ in high_risk:
+ other = o.pr_b if o.pr_a.number == current_pr else o.pr_a
+ format_pr_entry(other, lines)
+ format_line_overlaps(o.line_overlaps, lines)
+ lines.append("")
+
+
+def format_medium_risk_section(
+ medium_risk: list[tuple],
+ current_pr: int,
+ changes_current: dict,
+ all_changes: dict,
+ lines: list[str]
+):
+ """Format the medium risk section."""
+ if not medium_risk:
+ return
+
+ lines.append("### 🟡 Medium Risk — Some Line Overlap\n")
+ lines.append("These PRs have some overlapping changes:\n")
+
+ for o, _ in medium_risk:
+ other = o.pr_b if o.pr_a.number == current_pr else o.pr_a
+ other_changes = all_changes.get(other.number, {})
+ format_pr_entry(other, lines)
+
+ # Note if rename is involved
+ for file_path in o.overlapping_files:
+ file_a = changes_current.get(file_path)
+ file_b = other_changes.get(file_path)
+ if (file_a and file_a.is_rename) or (file_b and file_b.is_rename):
+ lines.append(f" - ⚠️ `{file_path}` is being renamed/moved")
+ break
+
+ if o.line_overlaps:
+ for file_path, ranges in o.line_overlaps.items():
+ range_strs = [f"L{r[0]}-{r[1]}" if r[0] != r[1] else f"L{r[0]}" for r in ranges]
+ lines.append(f" - `{file_path}`: {', '.join(range_strs)}")
+ else:
+ non_ignored = [f for f in o.overlapping_files if not should_ignore_file(f)]
+ if non_ignored:
+ lines.append(f" - Shared files: `{'`, `'.join(non_ignored[:5])}`")
+ lines.append("")
+
+
+def format_low_risk_section(low_risk: list[tuple], current_pr: int, lines: list[str]):
+ """Format the low risk section."""
+ if not low_risk:
+ return
+
+ lines.append("### 🟢 Low Risk — File Overlap Only\n")
+ lines.append("These PRs touch the same files but different sections (click to expand)
\n")
+
+ for o, _ in low_risk:
+ other = o.pr_b if o.pr_a.number == current_pr else o.pr_a
+ non_ignored = [f for f in o.overlapping_files if not should_ignore_file(f)]
+ if non_ignored:
+ format_pr_entry(other, lines)
+ if o.line_overlaps:
+ for file_path, ranges in o.line_overlaps.items():
+ range_strs = [f"L{r[0]}-{r[1]}" if r[0] != r[1] else f"L{r[0]}" for r in ranges]
+ lines.append(f" - `{file_path}`: {', '.join(range_strs)}")
+ else:
+ lines.append(f" - Shared files: `{'`, `'.join(non_ignored[:5])}`")
+
+ lines.append("\nThese PRs touch the same files but different sections (click to expand)
\n")
- for o, _ in low_risk:
- other = o.pr_b if o.pr_a.number == current_pr else o.pr_a
- non_ignored = [f for f in o.overlapping_files if not should_ignore_file(f)]
- if non_ignored: # Only show if there are non-ignored files
- updated = format_relative_time(other.updated_at)
- updated_str = f" · updated {updated}" if updated else ""
- lines.append(f"- **#{other.number}** ({other.author}{updated_str}): [{other.title}]({other.url})")
- if o.line_overlaps:
- for file_path, ranges in o.line_overlaps.items():
- range_strs = [f"L{r[0]}-{r[1]}" if r[0] != r[1] else f"L{r[0]}" for r in ranges]
- lines.append(f" - `{file_path}`: {', '.join(range_strs)}")
- else:
- lines.append(f" - Shared files: `{'`, `'.join(non_ignored[:5])}`")
- lines.append("\n