gitea_mcp_server.py

@@ -4737,6 +4737,34 @@ def gitea_reconcile_merged_cleanups(
         )
 
     delete_capability_allowed = not _profile_operation_gate("gitea.branch.delete")
+
+    # #534: discover reviewer scratch trees and active leases for those PRs.
+    scratch_candidates = merged_cleanup_reconcile.discover_reviewer_scratch_worktrees(
+        PROJECT_ROOT
+    )
+    active_reviewer_leases: dict[int, bool] = {}
+    pr_states: dict[int, dict] = {}
+    for scratch in scratch_candidates:
+        pr_num = int(scratch["pr_number"])
+        if pr_num in active_reviewer_leases:
+            continue
+        try:
+            comments = api_get_all(f"{base}/issues/{pr_num}/comments", auth)
+        except Exception:
+            comments = []
+        lease = reviewer_pr_lease.find_active_reviewer_lease(
+            comments, pr_number=pr_num
+        )
+        active_reviewer_leases[pr_num] = bool(lease)
+        try:
+            pr_live = api_request("GET", f"{base}/pulls/{pr_num}", auth)
+        except Exception:
+            pr_live = {}
+        pr_states[pr_num] = {
+            "merged": bool((pr_live or {}).get("merged") or (pr_live or {}).get("merged_at")),
+            "closed": (pr_live or {}).get("state") == "closed",
+        }
+
     report = merged_cleanup_reconcile.build_reconciliation_report(
         project_root=PROJECT_ROOT,
         closed_prs=merged_closed,
@@ -4744,6 +4772,8 @@ def gitea_reconcile_merged_cleanups(
         remote_branch_exists=remote_branch_exists,
         head_on_master=head_on_master,
         delete_capability_allowed=delete_capability_allowed,
+        active_reviewer_leases=active_reviewer_leases,
+        pr_states=pr_states,
     )
 
     if dry_run:
@@ -4787,6 +4817,14 @@ def gitea_reconcile_merged_cleanups(
             )
             actions.append({"action": "remove_local_worktree", **result})
 
+    for scratch in report.get("reviewer_scratch_entries") or []:
+        if not scratch.get("safe_to_remove_worktree"):
+            continue
+        result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree(
+            PROJECT_ROOT, scratch.get("worktree_path") or ""
+        )
+        actions.append({"action": "remove_reviewer_scratch_worktree", **result})
+
     report["dry_run"] = False
     report["executed"] = True
     report["actions"] = actions

merged_cleanup_reconcile.py

@@ -17,6 +17,11 @@ import issue_lock_store
 
 PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
 CLOSES_FIXES_RE = re.compile(r"\b(?:closes|fixes)\s+#(\d+)\b", re.IGNORECASE)
+# Reviewer scratch folders: review-pr<N> or review-pr<N>-submit (#534).
+REVIEWER_SCRATCH_NAME_RE = re.compile(
+    r"^review-pr(?P<pr>\d+)(?:-submit)?$",
+    re.IGNORECASE,
+)
 
 
 def extract_linked_issue(title: str, body: str) -> int | None:
@@ -36,8 +41,8 @@ def resolve_worktree_path(project_root: str, branch: str) -> str:
     return os.path.join(project_root, "branches", branch_worktree_folder(branch))
 
 
-def discover_local_worktrees(project_root: str) -> dict[str, str]:
-    """Return a mapping from branch name to absolute worktree path (#528)."""
+def _git_worktree_porcelain(project_root: str) -> list[dict[str, str | None]]:
+    """Parse ``git worktree list --porcelain`` into structured records."""
     res = subprocess.run(
         ["git", "-C", project_root, "worktree", "list", "--porcelain"],
         capture_output=True,
@@ -45,25 +50,139 @@ def discover_local_worktrees(project_root: str) -> dict[str, str]:
         check=False,
     )
     if res.returncode != 0:
-        return {}
+        return []
 
-    mapping: dict[str, str] = {}
-    current_worktree = None
+    records: list[dict[str, str | None]] = []
+    current: dict[str, str | None] = {}
     for line in res.stdout.splitlines():
         line = line.strip()
         if not line:
-            current_worktree = None
+            if current.get("worktree"):
+                records.append(current)
+            current = {}
             continue
         if line.startswith("worktree "):
-            current_worktree = line[9:].strip()
-        elif line.startswith("branch ") and current_worktree:
+            if current.get("worktree"):
+                records.append(current)
+            current = {"worktree": line[9:].strip(), "branch": None, "head": None}
+        elif line.startswith("branch ") and current:
             ref = line[7:].strip()
-            if ref.startswith("refs/heads/"):
-                branch_name = ref[11:]
-                mapping[branch_name] = current_worktree
+            current["branch"] = (
+                ref[11:] if ref.startswith("refs/heads/") else ref
+            )
+        elif line.startswith("HEAD ") and current:
+            current["head"] = line[5:].strip()
+        elif line == "detached" and current:
+            current["branch"] = None
+    if current.get("worktree"):
+        records.append(current)
+    return records
+
+
+def discover_local_worktrees(project_root: str) -> dict[str, str]:
+    """Return a mapping from branch name to absolute worktree path (#528)."""
+    mapping: dict[str, str] = {}
+    for record in _git_worktree_porcelain(project_root):
+        branch_name = record.get("branch")
+        path = record.get("worktree")
+        if branch_name and path:
+            mapping[str(branch_name)] = str(path)
     return mapping
 
 
+def parse_reviewer_scratch_folder(folder_name: str) -> int | None:
+    """Return PR number for a reviewer scratch folder name, else None (#534)."""
+    match = REVIEWER_SCRATCH_NAME_RE.match((folder_name or "").strip())
+    if not match:
+        return None
+    return int(match.group("pr"))
+
+
+def discover_reviewer_scratch_worktrees(project_root: str) -> list[dict[str, Any]]:
+    """Discover detached reviewer scratch worktrees under ``branches/`` (#534).
+
+    Matches folder names ``review-pr<N>`` and ``review-pr<N>-submit`` only.
+    These are never treated as author worktree paths.
+    """
+    root = os.path.realpath(project_root)
+    branches_root = os.path.realpath(os.path.join(root, "branches"))
+    found: list[dict[str, Any]] = []
+    for record in _git_worktree_porcelain(project_root):
+        path = record.get("worktree")
+        if not path:
+            continue
+        real = os.path.realpath(path)
+        parent = os.path.dirname(real)
+        if parent != branches_root:
+            continue
+        folder = os.path.basename(real)
+        pr_number = parse_reviewer_scratch_folder(folder)
+        if pr_number is None:
+            continue
+        found.append(
+            {
+                "pr_number": pr_number,
+                "worktree_path": real,
+                "folder_name": folder,
+                "current_branch": record.get("branch"),
+                "head_sha": record.get("head"),
+                "worktree_kind": "reviewer_scratch",
+            }
+        )
+    return found
+
+
+def assess_reviewer_scratch_cleanup(
+    *,
+    pr_number: int,
+    worktree_path: str,
+    folder_name: str,
+    pr_merged: bool,
+    pr_closed: bool,
+    worktree_state: dict[str, Any],
+    active_reviewer_lease: bool,
+) -> dict[str, Any]:
+    """Assess whether a reviewer scratch worktree is safe to remove (#534)."""
+    reasons: list[str] = []
+    exists = bool(worktree_state.get("exists"))
+    if not (pr_merged or pr_closed):
+        reasons.append("associated PR is still open")
+    if not exists:
+        reasons.append("reviewer scratch worktree not present")
+    if exists and worktree_state.get("clean") is False:
+        dirty = worktree_state.get("dirty_files") or []
+        reasons.append(
+            "reviewer scratch worktree has tracked edits"
+            + (f" ({', '.join(dirty)})" if dirty else "")
+        )
+    if active_reviewer_lease:
+        reasons.append("active reviewer lease still requires this worktree")
+    current_branch = worktree_state.get("current_branch")
+    if current_branch:
+        # Scratch trees are expected detached; a named branch is ambiguous ownership.
+        reasons.append(
+            f"reviewer scratch worktree is on branch '{current_branch}' "
+            "(expected detached HEAD for review-pr scratch trees)"
+        )
+
+    safe = exists and not reasons
+    return {
+        "pr_number": pr_number,
+        "worktree_path": worktree_path,
+        "folder_name": folder_name,
+        "worktree_kind": "reviewer_scratch",
+        "worktree_exists": exists,
+        "worktree_clean": worktree_state.get("clean"),
+        "safe_to_remove_worktree": safe,
+        "block_reasons": reasons,
+        "recommended_action": (
+            "remove_reviewer_scratch_worktree" if safe else "keep_reviewer_scratch_worktree"
+        ),
+        # Never treat as author worktree cleanup.
+        "author_worktree_cleanup": False,
+    }
+
+
 def read_issue_lock(path: str | None = None) -> dict[str, Any] | None:
     if path:
         return issue_lock_store.read_lock_file(path.strip())
@@ -310,6 +429,8 @@ def build_reconciliation_report(
     delete_capability_allowed: bool,
     issue_lock_path: str | None = None,
     protected_branches: frozenset[str] | None = None,
+    active_reviewer_leases: dict[int, bool] | None = None,
+    pr_states: dict[int, dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
     open_heads = collect_open_pr_heads(open_prs)
     worktree_map = discover_local_worktrees(project_root)
@@ -334,10 +455,51 @@ def build_reconciliation_report(
                 worktree_map=worktree_map,
             )
         )
+
+    # #534: reviewer scratch worktrees are reported separately from author cleanup.
+    lease_map = active_reviewer_leases or {}
+    state_map = pr_states or {}
+    open_pr_numbers = {
+        int(pr["number"]) for pr in (open_prs or []) if pr.get("number") is not None
+    }
+    closed_by_number = {
+        int(pr["number"]): pr for pr in (closed_prs or []) if pr.get("number") is not None
+    }
+    scratch_entries: list[dict[str, Any]] = []
+    for scratch in discover_reviewer_scratch_worktrees(project_root):
+        pr_number = int(scratch["pr_number"])
+        state_info = state_map.get(pr_number) or {}
+        closed_pr = closed_by_number.get(pr_number)
+        if closed_pr is not None:
+            pr_merged = bool(closed_pr.get("merged_at") or closed_pr.get("merged"))
+            pr_closed = True
+        elif pr_number in open_pr_numbers:
+            pr_merged = False
+            pr_closed = False
+        else:
+            pr_merged = bool(state_info.get("merged"))
+            pr_closed = bool(state_info.get("closed") or state_info.get("merged"))
+        worktree_path = scratch["worktree_path"]
+        worktree_state = read_local_worktree_state(worktree_path)
+        worktree_state["worktree_path"] = worktree_path
+        # Prefer live branch from state (git) over porcelain branch field.
+        assessment = assess_reviewer_scratch_cleanup(
+            pr_number=pr_number,
+            worktree_path=worktree_path,
+            folder_name=scratch["folder_name"],
+            pr_merged=pr_merged,
+            pr_closed=pr_closed,
+            worktree_state=worktree_state,
+            active_reviewer_lease=bool(lease_map.get(pr_number)),
+        )
+        scratch_entries.append(assessment)
+
     return {
         "project_root": os.path.realpath(project_root),
         "merged_pr_count": len(entries),
         "entries": entries,
+        "reviewer_scratch_entries": scratch_entries,
+        "reviewer_scratch_count": len(scratch_entries),
         "dry_run": True,
         "executed": False,
     }
@@ -370,4 +532,65 @@ def remove_local_worktree(project_root: str, branch: str) -> dict[str, Any]:
         "performed": True,
         "message": f"removed worktree {worktree_path}",
         "worktree_path": worktree_path,
+    }
+
+
+def remove_reviewer_scratch_worktree(
+    project_root: str,
+    worktree_path: str,
+) -> dict[str, Any]:
+    """Remove a reviewer scratch worktree by absolute path (#534).
+
+    Fail closed unless the path is a direct ``branches/review-pr*`` child of
+    *project_root*. Never routes through author branch-name derivation.
+    """
+    root = os.path.realpath(project_root)
+    branches_root = os.path.realpath(os.path.join(root, "branches"))
+    real = os.path.realpath((worktree_path or "").strip())
+    folder = os.path.basename(real)
+    if os.path.dirname(real) != branches_root:
+        return {
+            "success": False,
+            "performed": False,
+            "worktree_kind": "reviewer_scratch",
+            "message": (
+                "reviewer scratch path must be a direct child of "
+                f"{branches_root}; got {real}"
+            ),
+        }
+    if parse_reviewer_scratch_folder(folder) is None:
+        return {
+            "success": False,
+            "performed": False,
+            "worktree_kind": "reviewer_scratch",
+            "message": f"path is not a reviewer scratch folder: {folder}",
+        }
+    if not os.path.isdir(real):
+        return {
+            "success": False,
+            "performed": False,
+            "worktree_kind": "reviewer_scratch",
+            "message": f"worktree not found: {real}",
+        }
+    res = subprocess.run(
+        ["git", "-C", project_root, "worktree", "remove", real],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if res.returncode != 0:
+        return {
+            "success": False,
+            "performed": False,
+            "worktree_kind": "reviewer_scratch",
+            "message": (res.stderr or res.stdout or "worktree remove failed").strip(),
+            "worktree_path": real,
+        }
+    return {
+        "success": True,
+        "performed": True,
+        "worktree_kind": "reviewer_scratch",
+        "message": f"removed reviewer scratch worktree {real}",
+        "worktree_path": real,
+        "author_worktree_cleanup": False,
     }

tests/test_merged_cleanup_reconcile.py

@@ -238,5 +238,206 @@ class TestMergedCleanupReport(unittest.TestCase):
         )
 
 
+class TestReviewerScratchCleanup(unittest.TestCase):
+    """#534: authorized cleanup path for reviewer scratch worktrees."""
+
+    def test_parse_reviewer_scratch_folder(self):
+        self.assertEqual(mcr.parse_reviewer_scratch_folder("review-pr512-submit"), 512)
+        self.assertEqual(mcr.parse_reviewer_scratch_folder("review-pr99"), 99)
+        self.assertIsNone(mcr.parse_reviewer_scratch_folder("feat-issue-11-x"))
+        self.assertIsNone(mcr.parse_reviewer_scratch_folder("review-feat-issue-11"))
+        self.assertIsNone(mcr.parse_reviewer_scratch_folder("review-pr512-extra-tail"))
+
+    def test_assess_safe_clean_detached_after_merge(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=True,
+            pr_closed=True,
+            worktree_state={
+                "exists": True,
+                "clean": True,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": [],
+            },
+            active_reviewer_lease=False,
+        )
+        self.assertTrue(assessment["safe_to_remove_worktree"])
+        self.assertFalse(assessment["author_worktree_cleanup"])
+        self.assertEqual(
+            assessment["recommended_action"], "remove_reviewer_scratch_worktree"
+        )
+        self.assertEqual(assessment["worktree_kind"], "reviewer_scratch")
+
+    def test_assess_blocks_dirty_scratch(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=True,
+            pr_closed=True,
+            worktree_state={
+                "exists": True,
+                "clean": False,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": ["foo.py"],
+            },
+            active_reviewer_lease=False,
+        )
+        self.assertFalse(assessment["safe_to_remove_worktree"])
+        self.assertTrue(
+            any("tracked edits" in r for r in assessment["block_reasons"])
+        )
+
+    def test_assess_blocks_active_lease(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=True,
+            pr_closed=True,
+            worktree_state={
+                "exists": True,
+                "clean": True,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": [],
+            },
+            active_reviewer_lease=True,
+        )
+        self.assertFalse(assessment["safe_to_remove_worktree"])
+        self.assertTrue(
+            any("active reviewer lease" in r for r in assessment["block_reasons"])
+        )
+
+    def test_assess_blocks_open_pr(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=False,
+            pr_closed=False,
+            worktree_state={
+                "exists": True,
+                "clean": True,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": [],
+            },
+            active_reviewer_lease=False,
+        )
+        self.assertFalse(assessment["safe_to_remove_worktree"])
+        self.assertTrue(any("still open" in r for r in assessment["block_reasons"]))
+
+    @patch("subprocess.run")
+    @patch("merged_cleanup_reconcile.read_local_worktree_state")
+    def test_report_lists_scratch_separately_from_author(self, mock_state, mock_run):
+        mock_output = (
+            "worktree /repo/Gitea-Tools\n"
+            "bare\n\n"
+            "worktree /repo/Gitea-Tools/branches/feat-issue-11-x\n"
+            "branch refs/heads/feat/issue-11-x\n"
+            "HEAD deadbeef\n\n"
+            "worktree /repo/Gitea-Tools/branches/review-pr512-submit\n"
+            "HEAD abcdef0\n"
+            "detached\n\n"
+        )
+        mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
+
+        def _state(path):
+            if "review-pr512" in path:
+                return {
+                    "exists": True,
+                    "clean": True,
+                    "current_branch": None,
+                    "head_sha": "abcdef0",
+                    "dirty_files": [],
+                }
+            return {
+                "exists": True,
+                "clean": True,
+                "current_branch": "feat/issue-11-x",
+                "head_sha": "deadbeef",
+                "dirty_files": [],
+            }
+
+        mock_state.side_effect = _state
+        report = mcr.build_reconciliation_report(
+            project_root="/repo/Gitea-Tools",
+            closed_prs=[
+                {
+                    "number": 11,
+                    "title": "merged author",
+                    "body": "Closes #11",
+                    "head": {"ref": "feat/issue-11-x"},
+                    "merged_at": "2026-07-06T12:00:00Z",
+                },
+                {
+                    "number": 512,
+                    "title": "merged reviewed",
+                    "body": "Closes #512",
+                    "head": {"ref": "feat/issue-512-x"},
+                    "merged_at": "2026-07-06T12:00:00Z",
+                },
+            ],
+            open_prs=[],
+            remote_branch_exists={
+                "feat/issue-11-x": False,
+                "feat/issue-512-x": False,
+            },
+            head_on_master={11: True, 512: True},
+            delete_capability_allowed=True,
+            active_reviewer_leases={512: False},
+        )
+        self.assertEqual(report["merged_pr_count"], 2)
+        self.assertEqual(report["reviewer_scratch_count"], 1)
+        scratch = report["reviewer_scratch_entries"][0]
+        self.assertEqual(scratch["pr_number"], 512)
+        self.assertEqual(scratch["worktree_kind"], "reviewer_scratch")
+        self.assertFalse(scratch["author_worktree_cleanup"])
+        self.assertTrue(scratch["safe_to_remove_worktree"])
+        # Author entries must not swallow the scratch path as local_worktree.
+        author_paths = {
+            (e.get("local_worktree") or {}).get("worktree_path")
+            for e in report["entries"]
+        }
+        self.assertNotIn(
+            "/repo/Gitea-Tools/branches/review-pr512-submit", author_paths
+        )
+
+    @patch("subprocess.run")
+    def test_remove_reviewer_scratch_worktree_path_guard(self, mock_run):
+        mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+        with patch("os.path.isdir", return_value=True):
+            bad = mcr.remove_reviewer_scratch_worktree(
+                "/repo/Gitea-Tools",
+                "/repo/Gitea-Tools/branches/feat-issue-11-x",
+            )
+            self.assertFalse(bad["performed"])
+            good = mcr.remove_reviewer_scratch_worktree(
+                "/repo/Gitea-Tools",
+                "/repo/Gitea-Tools/branches/review-pr512-submit",
+            )
+        self.assertTrue(good["success"])
+        self.assertTrue(good["performed"])
+        self.assertFalse(good["author_worktree_cleanup"])
+        mock_run.assert_any_call(
+            [
+                "git",
+                "-C",
+                "/repo/Gitea-Tools",
+                "worktree",
+                "remove",
+                "/repo/Gitea-Tools/branches/review-pr512-submit",
+            ],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+
 if __name__ == "__main__":
     unittest.main()