from __future__ import annotations import argparse import json import os import shutil import subprocess import sys import tempfile import threading import webbrowser from dataclasses import dataclass from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any APP_TITLE = "Conflict Catcher" DEFAULT_HOST = "127.0.0.1" DEFAULT_PORT = 8765 class CheckError(Exception): pass @dataclass class GitResult: code: int stdout: str stderr: str @property def combined(self) -> str: return "\n".join(part for part in (self.stdout, self.stderr) if part).strip() def run_git(repo: Path, args: list[str], *, cwd: Path | None = None, timeout: int = 120) -> GitResult: command = ["git", *args] try: completed = subprocess.run( command, cwd=str(cwd or repo), text=True, capture_output=True, timeout=timeout, check=False, ) except FileNotFoundError as exc: raise CheckError("Git was not found on PATH. Install Git and try again.") from exc except subprocess.TimeoutExpired as exc: raise CheckError(f"Git command timed out: {' '.join(command)}") from exc return GitResult(completed.returncode, completed.stdout.strip(), completed.stderr.strip()) def repo_root(path: str) -> Path: if not path or not path.strip(): raise CheckError("Choose a Git repository path.") candidate = Path(path).expanduser().resolve() if not candidate.exists(): raise CheckError(f"Path does not exist: {candidate}") result = run_git(candidate, ["rev-parse", "--show-toplevel"]) if result.code != 0: raise CheckError(f"Not a Git repository: {candidate}") return Path(result.stdout).resolve() def parse_branch_spec(spec: str, target: str, source: str) -> tuple[str, str]: spec = (spec or "").strip() target = (target or "").strip() source = (source or "").strip() if spec: if "<" not in spec: raise CheckError("Branch spec must look like main str: result = run_git(repo, ["rev-parse", "--verify", f"{ref}^{{commit}}"]) if result.code != 0: raise CheckError(f"Could not find {label} branch/ref: {ref}") return result.stdout def list_conflicts(worktree: Path) -> list[dict[str, str]]: result = run_git(worktree, ["diff", "--name-status", "--diff-filter=U"], cwd=worktree) conflicts: list[dict[str, str]] = [] if result.code != 0 or not result.stdout: return conflicts status_names = { "AA": "both added", "AU": "added by us", "UA": "added by them", "DD": "both deleted", "DU": "deleted by us", "UD": "deleted by them", "UU": "both modified", "U": "unmerged", } for line in result.stdout.splitlines(): parts = line.split("\t", 1) if len(parts) == 2: conflicts.append( { "status": parts[0], "meaning": status_names.get(parts[0], "unmerged"), "path": parts[1], } ) return conflicts def make_worktree(repo: Path, ref: str) -> Path: temp_parent = Path(tempfile.mkdtemp(prefix="conflict-catcher-")) worktree = temp_parent / "worktree" result = run_git(repo, ["worktree", "add", "--detach", str(worktree), ref], timeout=180) if result.code != 0: shutil.rmtree(temp_parent, ignore_errors=True) raise CheckError(f"Could not create a temporary worktree for {ref}.\n{result.combined}") return worktree def cleanup_worktree(repo: Path, worktree: Path) -> None: run_git(repo, ["worktree", "remove", "--force", str(worktree)], timeout=180) shutil.rmtree(worktree.parent, ignore_errors=True) run_git(repo, ["worktree", "prune"], timeout=180) def dry_run_merge(repo: Path, target: str, source: str) -> dict[str, Any]: worktree = make_worktree(repo, target) try: result = run_git( worktree, ["merge", "--no-commit", "--no-ff", source], cwd=worktree, timeout=300, ) conflicts = list_conflicts(worktree) ok = result.code == 0 and not conflicts summary = ( f"{source} can be merged into {target} without conflicts." if ok else f"{source} would conflict when merged into {target}." ) return { "ok": ok, "operation": "merge", "summary": summary, "conflicts": conflicts, "gitOutput": result.combined, } finally: run_git(worktree, ["merge", "--abort"], cwd=worktree) cleanup_worktree(repo, worktree) def dry_run_rebase(repo: Path, target: str, source: str) -> dict[str, Any]: worktree = make_worktree(repo, source) try: result = run_git( worktree, ["rebase", "--no-verify", target], cwd=worktree, timeout=300, ) conflicts = list_conflicts(worktree) ok = result.code == 0 and not conflicts summary = ( f"{source} can be rebased onto {target} without conflicts." if ok else f"{source} would conflict when rebased onto {target}." ) return { "ok": ok, "operation": "rebase", "summary": summary, "conflicts": conflicts, "gitOutput": result.combined, } finally: run_git(worktree, ["rebase", "--abort"], cwd=worktree) cleanup_worktree(repo, worktree) def check_conflicts(payload: dict[str, Any]) -> dict[str, Any]: repo = repo_root(str(payload.get("repoPath", ""))) target, source = parse_branch_spec( str(payload.get("branchSpec", "")), str(payload.get("targetBranch", "")), str(payload.get("sourceBranch", "")), ) operation = str(payload.get("operation", "both")).lower() if operation not in {"merge", "rebase", "both"}: raise CheckError("Operation must be merge, rebase, or both.") target_sha = verify_commit(repo, target, "target") source_sha = verify_commit(repo, source, "source") checks: list[dict[str, Any]] = [] if operation in {"merge", "both"}: checks.append(dry_run_merge(repo, target, source)) if operation in {"rebase", "both"}: checks.append(dry_run_rebase(repo, target, source)) return { "repo": str(repo), "target": target, "source": source, "targetSha": target_sha[:12], "sourceSha": source_sha[:12], "checks": checks, } INDEX_HTML = r""" Conflict Catcher

Conflict Catcher

Dry-run branch merge and rebase checks before you touch the real working tree.

Git preflight

Results

Waiting for a repository and branch pair.

Idle

Enter a repo path and a branch pair such as main<develop.

""" class ConflictCatcherHandler(BaseHTTPRequestHandler): server_version = "ConflictCatcher/1.0" def log_message(self, format: str, *args: Any) -> None: sys.stderr.write("%s - %s\n" % (self.address_string(), format % args)) def send_json(self, status: int, payload: dict[str, Any]) -> None: encoded = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(encoded))) self.end_headers() self.wfile.write(encoded) def do_GET(self) -> None: if self.path not in {"/", "/index.html"}: self.send_error(404) return encoded = INDEX_HTML.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(encoded))) self.end_headers() self.wfile.write(encoded) def do_POST(self) -> None: if self.path != "/api/check": self.send_error(404) return try: length = int(self.headers.get("Content-Length", "0")) payload = json.loads(self.rfile.read(length).decode("utf-8")) self.send_json(200, check_conflicts(payload)) except CheckError as exc: self.send_json(400, {"error": str(exc)}) except json.JSONDecodeError: self.send_json(400, {"error": "Request body must be valid JSON."}) except Exception as exc: self.send_json(500, {"error": f"Unexpected error: {exc}"}) def run_server(host: str, port: int, open_browser: bool) -> None: server = ThreadingHTTPServer((host, port), ConflictCatcherHandler) url = f"http://{host}:{server.server_port}" if open_browser: threading.Timer(0.4, lambda: webbrowser.open(url)).start() print(f"{APP_TITLE} running at {url}") print("Press Ctrl+C to stop.") server.serve_forever() def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Check Git branch conflicts before merging or rebasing.") parser.add_argument("--host", default=os.environ.get("CONFLICT_CATCHER_HOST", DEFAULT_HOST)) parser.add_argument("--port", type=int, default=int(os.environ.get("CONFLICT_CATCHER_PORT", DEFAULT_PORT))) parser.add_argument("--no-browser", action="store_true", help="Start the server without opening a browser.") return parser if __name__ == "__main__": args = build_parser().parse_args() try: run_server(args.host, args.port, not args.no_browser) except KeyboardInterrupt: print("\nStopped.")