from __future__ import annotations import argparse import json import os import shlex 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 from urllib.parse import urlparse 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 git_lines(repo: Path, args: list[str]) -> list[str]: result = run_git(repo, args) if result.code != 0: raise CheckError(result.combined or f"Git command failed: {' '.join(args)}") return [line.strip() for line in result.stdout.splitlines() if line.strip()] def list_branches(repo: Path) -> dict[str, Any]: current = run_git(repo, ["branch", "--show-current"]).stdout.strip() local = git_lines(repo, ["branch", "--format=%(refname:short)"]) remote = [ branch for branch in git_lines(repo, ["branch", "--remotes", "--format=%(refname:short)"]) if "/" in branch and not branch.endswith("/HEAD") ] return { "repo": str(repo), "current": current, "local": local, "remote": remote, "all": sorted(dict.fromkeys([*local, *remote])), } def fetch_updates(repo: Path) -> dict[str, Any]: result = run_git(repo, ["fetch", "--all", "--prune"], timeout=300) if result.code != 0: raise CheckError(f"Fetch failed.\n{result.combined}") return { "ok": True, "summary": "Fetched all remotes and pruned stale remote-tracking branches.", "gitOutput": result.combined, } def upstream_status(repo: Path, ref: str) -> dict[str, Any]: upstream = run_git(repo, ["rev-parse", "--abbrev-ref", f"{ref}@{{upstream}}"]) if upstream.code != 0 or not upstream.stdout.strip(): return { "ref": ref, "upstream": "", "ahead": None, "behind": None, "summary": f"{ref} has no configured upstream.", "state": "unknown", } upstream_ref = upstream.stdout.strip() counts = run_git(repo, ["rev-list", "--left-right", "--count", f"{ref}...{upstream_ref}"]) if counts.code != 0: return { "ref": ref, "upstream": upstream_ref, "ahead": None, "behind": None, "summary": f"Could not compare {ref} with {upstream_ref}.", "state": "unknown", } left, right = (counts.stdout.split() + ["0", "0"])[:2] ahead = int(left) behind = int(right) if ahead == 0 and behind == 0: summary = f"{ref} is up to date with {upstream_ref}." state = "ok" elif behind: summary = f"{ref} is {behind} commit{'s' if behind != 1 else ''} behind {upstream_ref}." state = "behind" else: summary = f"{ref} is {ahead} commit{'s' if ahead != 1 else ''} ahead of {upstream_ref}." state = "ahead" return { "ref": ref, "upstream": upstream_ref, "ahead": ahead, "behind": behind, "summary": summary, "state": state, } def compare_freshness(repo: Path, target: str, source: str) -> list[dict[str, Any]]: return [upstream_status(repo, target), upstream_status(repo, source)] def resolve_repo_file(repo: Path, relative_path: str) -> Path: if not relative_path or not str(relative_path).strip(): raise CheckError("Choose a file path.") candidate = (repo / relative_path).resolve() try: candidate.relative_to(repo) except ValueError as exc: raise CheckError("File path must stay inside the selected repository.") from exc return candidate def reveal_path(payload: dict[str, Any]) -> dict[str, Any]: repo = repo_root(str(payload.get("repoPath", ""))) target = resolve_repo_file(repo, str(payload.get("path", ""))) reveal_target = target if target.exists() else target.parent if sys.platform.startswith("win"): subprocess.Popen(["explorer", "/select,", str(reveal_target)]) else: opener = "open" if sys.platform == "darwin" else "xdg-open" subprocess.Popen([opener, str(reveal_target.parent if reveal_target.is_file() else reveal_target)]) return {"ok": True, "path": str(target)} def open_in_editor(payload: dict[str, Any]) -> dict[str, Any]: repo = repo_root(str(payload.get("repoPath", ""))) target = resolve_repo_file(repo, str(payload.get("path", ""))) editor_command = str(payload.get("editorCommand", "")).strip() if editor_command: parts = shlex.split(editor_command, posix=not sys.platform.startswith("win")) if not parts: raise CheckError("Editor command is empty.") command = [part.replace("{path}", str(target)) for part in parts] if not any("{path}" in part for part in parts): command.append(str(target)) subprocess.Popen(command, cwd=str(repo)) elif sys.platform.startswith("win"): os.startfile(str(target)) # type: ignore[attr-defined] else: opener = "open" if sys.platform == "darwin" else "xdg-open" subprocess.Popen([opener, str(target)]) return {"ok": True, "path": str(target)} 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.") fetch_result = None if bool(payload.get("fetchBeforeCheck")): fetch_result = fetch_updates(repo) target_sha = verify_commit(repo, target, "target") source_sha = verify_commit(repo, source, "source") freshness = compare_freshness(repo, target, 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], "fetch": fetch_result, "freshness": freshness, "checks": checks, } def generate_report(data: dict[str, Any]) -> str: checks = data.get("checks", []) has_conflict = any(not check.get("ok") for check in checks) lines = [ "# Conflict Catcher Report", "", f"- Repository: `{data.get('repo', '')}`", f"- Target/base: `{data.get('target', '')}` (`{data.get('targetSha', '')}`)", f"- Source/incoming: `{data.get('source', '')}` (`{data.get('sourceSha', '')}`)", f"- Overall: {'conflicts need review' if has_conflict else 'clean'}", "", "## Freshness", "", ] for item in data.get("freshness", []): lines.append(f"- {item.get('summary', '')}") lines.extend(["", "## Checks", ""]) for check in checks: lines.append(f"### {str(check.get('operation', '')).title()}") lines.append("") lines.append(f"- Status: {'clean' if check.get('ok') else 'conflicts'}") lines.append(f"- Summary: {check.get('summary', '')}") conflicts = check.get("conflicts", []) if conflicts: lines.extend(["", "| Status | Meaning | Path |", "| --- | --- | --- |"]) for conflict in conflicts: lines.append( f"| {conflict.get('status', '')} | {conflict.get('meaning', '')} | `{conflict.get('path', '')}` |" ) else: lines.append("- Conflict files: none") lines.append("") return "\n".join(lines).strip() + "\n" INDEX_HTML = r""" Conflict Catcher

Check merge and rebase risk before your real branch gets involved.

Run clean-room Git dry runs, review exactly which files would conflict, then go back to your normal workflow with the answer in hand.

No checkoutYour active working tree stays untouched.
No commitDry runs abort and clean up after themselves.
Both pathsTest merge, rebase, or both in one pass.

Preflight setup

Point at a repo and name the branch direction.

Ready

Use target<source. You can also fill the separate fields below if that is clearer.

Check type

Results

Waiting for a repository and branch pair.

Idle

Ready when your branch pair is.

Enter a repository 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: path = urlparse(self.path).path if 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: path = urlparse(self.path).path handlers = { "/api/check": check_conflicts, "/api/branches": lambda payload: list_branches(repo_root(str(payload.get("repoPath", "")))), "/api/report": lambda payload: {"markdown": generate_report(payload)}, "/api/reveal": reveal_path, "/api/open-editor": open_in_editor, } handler = handlers.get(path) if handler is None: 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, handler(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.")