Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat: add merged PR cleanup reconciliation report (#269)

Browse Source 
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) <[email protected]>
This commit is contained in:
sysadmin
2026-07-07 00:16:42 -04:00
co-authored by Claude Opus 4.8
parent d6f4f936e3
commit 4cc31a42f9
5 changed files with 687 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+136
View File
@@ -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,
merged_cleanup_reconcile.py
+333
View File
@@ -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,
}
task_capability_map.py
+4
View File
@@ -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.
tests/test_mcp_server.py
+55
View File
@@ -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
# ---------------------------------------------------------------------------
tests/test_merged_cleanup_reconcile.py
+159
View File
@@ -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()