From 4cc31a42f93e06ce3fe62a9c72ceeff027cdb1e5 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 00:16:42 -0400 Subject: [PATCH] feat: add merged PR cleanup reconciliation report (#269) Implement gitea_reconcile_merged_cleanups with dry-run reporting for merged PR remote branches and local branches/ worktrees, plus confirmation-gated execute mode. Safety gates cover protected branches, open PR references, active issue locks, dirty worktrees, and delete_branch capability. Closes #269 Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 136 ++++++++++ merged_cleanup_reconcile.py | 333 +++++++++++++++++++++++++ task_capability_map.py | 4 + tests/test_mcp_server.py | 55 ++++ tests/test_merged_cleanup_reconcile.py | 159 ++++++++++++ 5 files changed, 687 insertions(+) create mode 100644 merged_cleanup_reconcile.py create mode 100644 tests/test_merged_cleanup_reconcile.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index dd1480c..021b609 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -374,6 +374,7 @@ import role_namespace_gate # noqa: E402 import task_capability_map # noqa: E402 import review_proofs # noqa: E402 import issue_lock_worktree # noqa: E402 +import merged_cleanup_reconcile # noqa: E402 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue, @@ -2859,6 +2860,141 @@ def gitea_delete_branch( return {"success": True, "message": f"Remote branch '{branch}' deleted."} +def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> bool: + import urllib.parse + + encoded = urllib.parse.quote(branch, safe="") + url = f"{repo_api_url(h, o, r)}/branches/{encoded}" + try: + api_request("GET", url, auth) + return True + except Exception as exc: + message = str(exc).lower() + if "404" in message or "not found" in message: + return False + raise + + +@mcp.tool() +def gitea_reconcile_merged_cleanups( + dry_run: bool = True, + execute_confirmed: bool = False, + limit: int = 50, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Reconcile merged PRs by reporting and optionally cleaning remote branches and local worktrees. + + Args: + dry_run: Defaults to True. When True, only builds the reconciliation report. + execute_confirmed: Must be True when dry_run=False. + limit: Max number of closed PRs to inspect. + remote: Known Gitea instance ('dadeschools' or 'prgs'). + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + dict with reconciliation report and any executed cleanup actions. + """ + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "performed": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + if not dry_run and not execute_confirmed: + raise ValueError( + "execute_confirmed must be True when dry_run=False (fail closed)" + ) + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + closed_prs = api_get_all(f"{base}/pulls?state=closed", auth, limit=limit) + open_prs = api_get_all(f"{base}/pulls?state=open", auth) + + merged_closed: list[dict] = [] + remote_branch_exists: dict[str, bool] = {} + head_on_master: dict[int, bool | None] = {} + master_ref = f"{remote}/master" if remote in REMOTES else "origin/master" + + for pr in closed_prs: + if not (pr.get("merged") or pr.get("merged_at")): + continue + merged_closed.append(pr) + head_ref = (pr.get("head") or {}).get("ref") or "" + head_sha = (pr.get("head") or {}).get("sha") + if head_ref and head_ref not in remote_branch_exists: + remote_branch_exists[head_ref] = _remote_branch_exists( + h, o, r, auth, head_ref + ) + head_on_master[int(pr["number"])] = merged_cleanup_reconcile.is_head_ancestor_of_ref( + PROJECT_ROOT, head_sha, master_ref + ) + + delete_capability_allowed = not _profile_operation_gate("gitea.branch.delete") + report = merged_cleanup_reconcile.build_reconciliation_report( + project_root=PROJECT_ROOT, + closed_prs=merged_closed, + open_prs=open_prs, + remote_branch_exists=remote_branch_exists, + head_on_master=head_on_master, + delete_capability_allowed=delete_capability_allowed, + ) + + if dry_run: + report["dry_run"] = True + report["executed"] = False + return {"success": True, "performed": False, **report} + + verify_preflight_purity(remote) + actions: list[dict] = [] + for entry in report.get("entries") or []: + head_branch = entry.get("head_branch") or "" + remote_assessment = entry.get("remote_branch") or {} + local_assessment = entry.get("local_worktree") or {} + + if remote_assessment.get("safe_to_delete_remote"): + import urllib.parse + + encoded = urllib.parse.quote(head_branch, safe="") + url = f"{base}/branches/{encoded}" + with _audited( + "delete_branch", + host=h, + remote=remote, + org=o, + repo=r, + target_branch=head_branch, + request_metadata={"branch": head_branch, "source": "reconcile_merged_cleanups"}, + ): + api_request("DELETE", url, auth) + actions.append( + { + "action": "delete_remote_branch", + "branch": head_branch, + "success": True, + } + ) + + if local_assessment.get("safe_to_remove_worktree"): + result = merged_cleanup_reconcile.remove_local_worktree( + PROJECT_ROOT, head_branch + ) + actions.append({"action": "remove_local_worktree", **result}) + + report["dry_run"] = False + report["executed"] = True + report["actions"] = actions + return {"success": True, "performed": True, **report} + + @mcp.tool() def gitea_close_issue( issue_number: int, diff --git a/merged_cleanup_reconcile.py b/merged_cleanup_reconcile.py new file mode 100644 index 0000000..60204f6 --- /dev/null +++ b/merged_cleanup_reconcile.py @@ -0,0 +1,333 @@ +"""Merged PR branch and worktree cleanup reconciliation (#269). + +Builds dry-run reports for merged pull requests and optionally executes +remote branch deletion and local worktree removal when every safety gate passes. +""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +from typing import Any + +from reviewer_worktree import parse_dirty_tracked_files + +PROTECTED_BRANCHES = frozenset({"master", "main", "dev"}) +ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json") +CLOSES_FIXES_RE = re.compile(r"\b(?:closes|fixes)\s+#(\d+)\b", re.IGNORECASE) + + +def extract_linked_issue(title: str, body: str) -> int | None: + """Return the first Closes/Fixes issue number from PR metadata.""" + for text in (title or "", body or ""): + match = CLOSES_FIXES_RE.search(text) + if match: + return int(match.group(1)) + return None + + +def branch_worktree_folder(branch: str) -> str: + return (branch or "").replace("/", "-") + + +def resolve_worktree_path(project_root: str, branch: str) -> str: + return os.path.join(project_root, "branches", branch_worktree_folder(branch)) + + +def read_issue_lock(path: str | None = None) -> dict[str, Any] | None: + lock_path = (path or ISSUE_LOCK_FILE).strip() + if not lock_path or not os.path.exists(lock_path): + return None + try: + with open(lock_path, encoding="utf-8") as handle: + data = json.load(handle) + except (OSError, json.JSONDecodeError): + return None + return data if isinstance(data, dict) else None + + +def has_active_issue_lock(branch: str, lock_path: str | None = None) -> bool: + lock = read_issue_lock(lock_path) + if not lock: + return False + return (lock.get("branch_name") or "").strip() == (branch or "").strip() + + +def collect_open_pr_heads(open_prs: list[dict[str, Any]]) -> set[str]: + heads: set[str] = set() + for pr in open_prs or []: + head = pr.get("head") + if isinstance(head, dict): + ref = head.get("ref") + else: + ref = head + if ref: + heads.add(str(ref)) + return heads + + +def read_local_worktree_state(worktree_path: str) -> dict[str, Any]: + path = (worktree_path or "").strip() + if not path or not os.path.isdir(path): + return { + "exists": False, + "clean": None, + "current_branch": None, + "head_sha": None, + "dirty_files": [], + } + + branch_res = subprocess.run( + ["git", "-C", path, "branch", "--show-current"], + capture_output=True, + text=True, + check=False, + ) + current_branch = (branch_res.stdout or "").strip() or None + + status_res = subprocess.run( + ["git", "-C", path, "status", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + dirty_files = parse_dirty_tracked_files(status_res.stdout or "") + + head_res = subprocess.run( + ["git", "-C", path, "rev-parse", "HEAD"], + capture_output=True, + text=True, + check=False, + ) + head_sha = (head_res.stdout or "").strip() if head_res.returncode == 0 else None + + return { + "exists": True, + "clean": not dirty_files, + "current_branch": current_branch, + "head_sha": head_sha, + "dirty_files": dirty_files, + } + + +def is_head_ancestor_of_ref(project_root: str, head_sha: str | None, base_ref: str) -> bool | None: + if not head_sha: + return None + res = subprocess.run( + ["git", "-C", project_root, "merge-base", "--is-ancestor", head_sha, base_ref], + capture_output=True, + text=True, + check=False, + ) + if res.returncode == 0: + return True + if res.returncode == 1: + return False + return None + + +def assess_remote_branch_cleanup( + *, + pr_number: int, + head_branch: str, + merged: bool, + remote_branch_exists: bool, + open_pr_heads: set[str], + protected_branches: frozenset[str] | None = None, + head_on_master: bool | None, + delete_capability_allowed: bool, + active_lock: bool, +) -> dict[str, Any]: + protected = protected_branches or PROTECTED_BRANCHES + reasons: list[str] = [] + if not merged: + reasons.append("PR is not merged") + if not remote_branch_exists: + reasons.append("remote branch already absent") + if head_branch in protected: + reasons.append(f"branch '{head_branch}' is protected") + if head_branch in open_pr_heads: + reasons.append("an open PR still references this head branch") + if active_lock: + reasons.append("active issue lock references this branch") + if head_on_master is False: + reasons.append("PR head is not an ancestor of master") + if not delete_capability_allowed: + reasons.append("delete_branch capability is not allowed in the active profile") + + safe = merged and remote_branch_exists and not reasons + return { + "pr_number": pr_number, + "head_branch": head_branch, + "remote_branch_exists": remote_branch_exists, + "safe_to_delete_remote": safe, + "block_reasons": reasons, + "recommended_action": "delete_remote_branch" if safe else "keep_remote_branch", + } + + +def assess_local_worktree_cleanup( + *, + pr_number: int, + head_branch: str, + merged: bool, + worktree_state: dict[str, Any], + active_lock: bool, +) -> dict[str, Any]: + reasons: list[str] = [] + exists = bool(worktree_state.get("exists")) + if not merged: + reasons.append("PR is not merged") + if not exists: + reasons.append("local worktree not present") + if exists and worktree_state.get("clean") is False: + dirty = worktree_state.get("dirty_files") or [] + reasons.append( + "local worktree has tracked edits" + + (f" ({', '.join(dirty)})" if dirty else "") + ) + if exists: + current_branch = worktree_state.get("current_branch") + if current_branch and current_branch != head_branch: + reasons.append( + f"worktree branch '{current_branch}' does not match PR head '{head_branch}'" + ) + if active_lock: + reasons.append("active issue lock references this branch") + + safe = merged and exists and not reasons + return { + "pr_number": pr_number, + "head_branch": head_branch, + "worktree_path": worktree_state.get("worktree_path"), + "worktree_exists": exists, + "worktree_clean": worktree_state.get("clean"), + "safe_to_remove_worktree": safe, + "block_reasons": reasons, + "recommended_action": "remove_local_worktree" if safe else "keep_local_worktree", + } + + +def build_pr_cleanup_entry( + *, + pr: dict[str, Any], + project_root: str, + open_pr_heads: set[str], + remote_branch_exists: bool, + head_on_master: bool | None, + delete_capability_allowed: bool, + issue_lock_path: str | None = None, + protected_branches: frozenset[str] | None = None, +) -> dict[str, Any]: + pr_number = int(pr["number"]) + head_branch = pr.get("head") or "" + if isinstance(head_branch, dict): + head_branch = head_branch.get("ref") or "" + title = pr.get("title") or "" + body = pr.get("body") or "" + merged = bool(pr.get("merged_at")) + worktree_path = resolve_worktree_path(project_root, head_branch) + worktree_state = read_local_worktree_state(worktree_path) + worktree_state["worktree_path"] = worktree_path + active_lock = has_active_issue_lock(head_branch, issue_lock_path) + + remote = assess_remote_branch_cleanup( + pr_number=pr_number, + head_branch=head_branch, + merged=merged, + remote_branch_exists=remote_branch_exists, + open_pr_heads=open_pr_heads, + protected_branches=protected_branches, + head_on_master=head_on_master, + delete_capability_allowed=delete_capability_allowed, + active_lock=active_lock, + ) + local = assess_local_worktree_cleanup( + pr_number=pr_number, + head_branch=head_branch, + merged=merged, + worktree_state=worktree_state, + active_lock=active_lock, + ) + return { + "pr_number": pr_number, + "issue_number": extract_linked_issue(title, body), + "title": title, + "head_branch": head_branch, + "merge_commit_sha": pr.get("merge_commit_sha"), + "merged_at": pr.get("merged_at"), + "merged": merged, + "remote_branch": remote, + "local_worktree": local, + } + + +def build_reconciliation_report( + *, + project_root: str, + closed_prs: list[dict[str, Any]], + open_prs: list[dict[str, Any]], + remote_branch_exists: dict[str, bool], + head_on_master: dict[int, bool | None], + delete_capability_allowed: bool, + issue_lock_path: str | None = None, + protected_branches: frozenset[str] | None = None, +) -> dict[str, Any]: + open_heads = collect_open_pr_heads(open_prs) + entries: list[dict[str, Any]] = [] + for pr in closed_prs or []: + if not pr.get("merged_at"): + continue + pr_number = int(pr["number"]) + head = pr.get("head") or "" + if isinstance(head, dict): + head = head.get("ref") or "" + entries.append( + build_pr_cleanup_entry( + pr=pr, + project_root=project_root, + open_pr_heads=open_heads, + remote_branch_exists=bool(remote_branch_exists.get(head)), + head_on_master=head_on_master.get(pr_number), + delete_capability_allowed=delete_capability_allowed, + issue_lock_path=issue_lock_path, + protected_branches=protected_branches, + ) + ) + return { + "project_root": os.path.realpath(project_root), + "merged_pr_count": len(entries), + "entries": entries, + "dry_run": True, + "executed": False, + } + + +def remove_local_worktree(project_root: str, branch: str) -> dict[str, Any]: + worktree_path = resolve_worktree_path(project_root, branch) + if not os.path.isdir(worktree_path): + return { + "success": False, + "performed": False, + "message": f"worktree not found: {worktree_path}", + } + res = subprocess.run( + ["git", "-C", project_root, "worktree", "remove", worktree_path], + capture_output=True, + text=True, + check=False, + ) + if res.returncode != 0: + return { + "success": False, + "performed": False, + "message": (res.stderr or res.stdout or "worktree remove failed").strip(), + } + return { + "success": True, + "performed": True, + "message": f"removed worktree {worktree_path}", + "worktree_path": worktree_path, + } \ No newline at end of file diff --git a/task_capability_map.py b/task_capability_map.py index fd3c344..7607170 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -80,6 +80,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.branch.delete", "role": "author", }, + "reconcile_merged_cleanups": { + "permission": "gitea.read", + "role": "author", + }, } # Issue-mutating MCP tools and their resolver task keys. diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 133e515..6b2827f 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -24,6 +24,7 @@ from mcp_server import ( # noqa: E402 gitea_merge_pr, gitea_review_pr, gitea_delete_branch, + gitea_reconcile_merged_cleanups, gitea_edit_pr, gitea_get_file, gitea_commit_files, @@ -990,6 +991,60 @@ class TestDeleteBranch(unittest.TestCase): self.assertIn("feat%2Fbranch", url) +# --------------------------------------------------------------------------- +# Reconcile merged cleanups (#269) +# --------------------------------------------------------------------------- +READ_ENV = { + "GITEA_ALLOWED_OPERATIONS": "gitea.read", +} + + +class TestReconcileMergedCleanups(unittest.TestCase): + + @patch("mcp_server.api_get_all") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_dry_run_builds_report_without_mutations(self, _auth, mock_api_get_all): + mock_api_get_all.side_effect = [ + [ + { + "number": 50, + "title": "merged feature", + "body": "Closes #50", + "merged": True, + "merged_at": "2026-07-06T12:00:00Z", + "merge_commit_sha": "deadbeef", + "head": {"ref": "feat/issue-50-example", "sha": "cafebabe"}, + } + ], + [], + ] + with patch.dict(os.environ, READ_ENV, clear=True): + with patch( + "mcp_server._remote_branch_exists", + return_value=True, + ): + with patch( + "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref", + return_value=True, + ): + result = gitea_reconcile_merged_cleanups( + dry_run=True, + remote="prgs", + limit=10, + ) + self.assertTrue(result["success"]) + self.assertTrue(result["dry_run"]) + self.assertFalse(result["executed"]) + self.assertEqual(result["merged_pr_count"], 1) + self.assertEqual(result["entries"][0]["issue_number"], 50) + + def test_execute_requires_confirmation(self): + with patch.dict(os.environ, READ_ENV, clear=True): + with self.assertRaises(ValueError) as ctx: + gitea_reconcile_merged_cleanups(dry_run=False, execute_confirmed=False) + self.assertIn("execute_confirmed", str(ctx.exception)) + + # --------------------------------------------------------------------------- # Edit PR # --------------------------------------------------------------------------- diff --git a/tests/test_merged_cleanup_reconcile.py b/tests/test_merged_cleanup_reconcile.py new file mode 100644 index 0000000..779df64 --- /dev/null +++ b/tests/test_merged_cleanup_reconcile.py @@ -0,0 +1,159 @@ +"""Tests for merged PR cleanup reconciliation (#269).""" + +import os +import sys +import tempfile +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import merged_cleanup_reconcile as mcr # noqa: E402 + + +class TestMergedCleanupAssessment(unittest.TestCase): + def test_extract_linked_issue_from_closes(self): + issue = mcr.extract_linked_issue( + "feat: cleanup (Closes #269)", + "No other refs", + ) + self.assertEqual(issue, 269) + + def test_merged_remote_branch_safe_when_gates_pass(self): + result = mcr.assess_remote_branch_cleanup( + pr_number=42, + head_branch="feat/issue-269-merged-pr-cleanup-reconcile", + merged=True, + remote_branch_exists=True, + open_pr_heads=set(), + head_on_master=True, + delete_capability_allowed=True, + active_lock=False, + ) + self.assertTrue(result["safe_to_delete_remote"]) + self.assertEqual(result["block_reasons"], []) + + def test_open_pr_blocks_remote_delete(self): + result = mcr.assess_remote_branch_cleanup( + pr_number=42, + head_branch="feat/issue-9-example", + merged=True, + remote_branch_exists=True, + open_pr_heads={"feat/issue-9-example"}, + head_on_master=True, + delete_capability_allowed=True, + active_lock=False, + ) + self.assertFalse(result["safe_to_delete_remote"]) + self.assertIn("open PR", result["block_reasons"][0]) + + def test_protected_branch_blocks_remote_delete(self): + result = mcr.assess_remote_branch_cleanup( + pr_number=1, + head_branch="master", + merged=True, + remote_branch_exists=True, + open_pr_heads=set(), + head_on_master=True, + delete_capability_allowed=True, + active_lock=False, + ) + self.assertFalse(result["safe_to_delete_remote"]) + self.assertIn("protected", result["block_reasons"][0]) + + def test_active_lock_blocks_remote_and_worktree_cleanup(self): + remote = mcr.assess_remote_branch_cleanup( + pr_number=42, + head_branch="feat/issue-9-example", + merged=True, + remote_branch_exists=True, + open_pr_heads=set(), + head_on_master=True, + delete_capability_allowed=True, + active_lock=True, + ) + local = mcr.assess_local_worktree_cleanup( + pr_number=42, + head_branch="feat/issue-9-example", + merged=True, + worktree_state={ + "exists": True, + "clean": True, + "current_branch": "feat/issue-9-example", + "worktree_path": "/repo/branches/feat-issue-9-example", + }, + active_lock=True, + ) + self.assertFalse(remote["safe_to_delete_remote"]) + self.assertFalse(local["safe_to_remove_worktree"]) + self.assertIn("active issue lock", remote["block_reasons"][0]) + + def test_dirty_worktree_blocks_local_cleanup(self): + result = mcr.assess_local_worktree_cleanup( + pr_number=42, + head_branch="feat/issue-9-example", + merged=True, + worktree_state={ + "exists": True, + "clean": False, + "dirty_files": ["gitea_mcp_server.py"], + "current_branch": "feat/issue-9-example", + "worktree_path": "/repo/branches/feat-issue-9-example", + }, + active_lock=False, + ) + self.assertFalse(result["safe_to_remove_worktree"]) + self.assertIn("tracked edits", result["block_reasons"][0]) + + +class TestMergedCleanupReport(unittest.TestCase): + def test_build_report_skips_unmerged_closed_prs(self): + with tempfile.TemporaryDirectory() as tmp: + report = mcr.build_reconciliation_report( + project_root=tmp, + closed_prs=[ + { + "number": 10, + "title": "closed not merged", + "body": "Closes #10", + "head": {"ref": "feat/issue-10-x"}, + "merged_at": None, + }, + { + "number": 11, + "title": "merged", + "body": "Closes #11", + "head": {"ref": "feat/issue-11-x"}, + "merged_at": "2026-07-06T12:00:00Z", + "merge_commit_sha": "abc123", + }, + ], + open_prs=[], + remote_branch_exists={"feat/issue-11-x": True}, + head_on_master={11: True}, + delete_capability_allowed=False, + ) + self.assertEqual(report["merged_pr_count"], 1) + entry = report["entries"][0] + self.assertEqual(entry["pr_number"], 11) + self.assertEqual(entry["issue_number"], 11) + self.assertFalse(entry["remote_branch"]["safe_to_delete_remote"]) + self.assertIn("delete_branch capability", entry["remote_branch"]["block_reasons"][0]) + + def test_has_active_issue_lock_reads_lock_file(self): + with tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False) as handle: + handle.write('{"branch_name": "feat/issue-9-example"}') + lock_path = handle.name + try: + self.assertTrue( + mcr.has_active_issue_lock("feat/issue-9-example", lock_path=lock_path) + ) + self.assertFalse( + mcr.has_active_issue_lock("feat/other-branch", lock_path=lock_path) + ) + finally: + os.remove(lock_path) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file