"""Merged-PR awareness for the worktree cleanup audit (#858). Before #858 an ``issue_work`` worktree could never leave ``active_issue_work``: the audit had no PR linkage at all (``pr_number`` was structurally ``None``) and its only route to ``clean_stale_removable`` was a TTL derived from a ``last_used_at`` that nothing ever populated. A merged, clean, unprotected worktree was therefore reported as active work forever, disagreeing with the PR-scoped reconciler. These tests use fabricated temporary repositories and synthetic PR records only. Nothing here removes a worktree or deletes a branch. """ import os import subprocess 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 import worktree_cleanup_audit as wca # noqa: E402 MERGED_BRANCH = "feat/issue-777-timeline" MERGED_PATH = "/repo/branches/issue-777-timeline" HEAD_SHA = "a" * 40 def _pr(number, branch, *, merged=True, sha=HEAD_SHA, state=None): """Synthetic Gitea PR payload.""" return { "number": number, "head": {"ref": branch, "sha": sha}, "merged_at": "2026-07-24T01:00:00Z" if merged else None, "state": state or ("closed" if merged else "open"), } def _porcelain(*entries): out = [] for path, branch, sha in entries: out.append(f"worktree {path}") out.append(f"HEAD {sha}") if branch is None: out.append("detached") else: out.append(f"branch refs/heads/{branch}") out.append("") return "\n".join(out) class _AuditHarness(unittest.TestCase): """Runs audit_branches_directory over a fabricated worktree listing.""" PORCELAIN = _porcelain( ("/repo", "master", "f" * 40), (MERGED_PATH, MERGED_BRANCH, HEAD_SHA), ) def run_audit(self, *, dirty_paths=(), contained=True, **kwargs): def fake_dirty(path): if path in dirty_paths: return {"exists": True, "dirty": True, "dirty_files": [" M x.py"]} return {"exists": True, "dirty": False, "dirty_files": []} with patch.object( wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(self.PORCELAIN), ), patch.object( wca, "read_worktree_dirty", side_effect=fake_dirty ), patch.object( wca, "git_worktree_list", return_value="(mocked)" ), patch.object( wca, "is_head_ancestor_of_ref", return_value=contained ): report = wca.audit_branches_directory("/repo", **kwargs) return {wt["path"]: wt for wt in report["worktrees"]}, report def merged_audit(self, **kwargs): kwargs.setdefault("pr_index", wca.build_pr_index([_pr(849, MERGED_BRANCH)])) kwargs.setdefault("master_ref", "prgs/master") return self.run_audit(**kwargs) class TestMergedWorktreeBecomesRemovable(_AuditHarness): def test_clean_merged_issue_worktree_is_linked_and_removable(self): by_path, report = self.merged_audit() entry = by_path[MERGED_PATH] self.assertEqual(entry["classification"], wca.CLASS_CLEAN_STALE_REMOVABLE) self.assertTrue(entry["removable"]) self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_MERGED) self.assertEqual(entry["merged_pr_cleanup"]["block_reasons"], []) self.assertIn(MERGED_PATH, [c["path"] for c in report["removable_candidates"]]) def test_pr_number_populated_from_authoritative_linkage(self): by_path, _ = self.merged_audit() self.assertEqual(by_path[MERGED_PATH]["pr_number"], 849) def test_regression_without_pr_evidence_stays_active_issue_work(self): """The pre-#858 behaviour, still correct when no PR state is supplied.""" by_path, _ = self.run_audit() entry = by_path[MERGED_PATH] self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) self.assertFalse(entry["removable"]) self.assertIsNone(entry["pr_number"]) class TestProtectiveSignalsSurvive(_AuditHarness): def test_open_pr_worktree_is_not_removable(self): index = wca.build_pr_index([_pr(900, MERGED_BRANCH, merged=False)]) by_path, _ = self.run_audit( pr_index=index, master_ref="prgs/master", open_pr_branches={MERGED_BRANCH}, ) entry = by_path[MERGED_PATH] self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_OPEN_PR) self.assertFalse(entry["removable"]) # linkage still reports the owning PR, it just is not merge proof self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_OPEN) self.assertEqual(entry["pr_number"], 900) def test_dirty_tracked_worktree_is_not_removable(self): by_path, _ = self.merged_audit(dirty_paths=(MERGED_PATH,)) entry = by_path[MERGED_PATH] self.assertEqual(entry["classification"], wca.CLASS_DIRTY_LOCAL) self.assertFalse(entry["removable"]) self.assertIn( "worktree has uncommitted changes", entry["merged_pr_cleanup"]["block_reasons"], ) def test_untracked_only_worktree_is_not_removable(self): """``git status --porcelain`` reports untracked files as dirty too.""" def untracked(path): if path == MERGED_PATH: return {"exists": True, "dirty": True, "dirty_files": ["?? scratch.txt"]} return {"exists": True, "dirty": False, "dirty_files": []} with patch.object( wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(self.PORCELAIN), ), patch.object( wca, "read_worktree_dirty", side_effect=untracked ), patch.object( wca, "git_worktree_list", return_value="(mocked)" ), patch.object( wca, "is_head_ancestor_of_ref", return_value=True ): report = wca.audit_branches_directory( "/repo", pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]), master_ref="prgs/master", ) entry = {wt["path"]: wt for wt in report["worktrees"]}[MERGED_PATH] self.assertEqual(entry["classification"], wca.CLASS_DIRTY_LOCAL) self.assertFalse(entry["removable"]) def test_active_lease_by_issue_number_is_protective(self): by_path, _ = self.merged_audit(leased_issue_numbers={777}) entry = by_path[MERGED_PATH] self.assertTrue(entry["has_active_lease"]) self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) self.assertFalse(entry["removable"]) def test_active_lease_by_branch_is_protective(self): by_path, _ = self.merged_audit(leased_branches={MERGED_BRANCH}) entry = by_path[MERGED_PATH] self.assertTrue(entry["has_active_lease"]) self.assertFalse(entry["removable"]) def test_active_issue_lock_is_protective(self): by_path, _ = self.merged_audit(active_issue_branches={MERGED_BRANCH}) entry = by_path[MERGED_PATH] self.assertTrue(entry["has_active_issue_lock"]) self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) self.assertFalse(entry["removable"]) def test_live_session_worktree_is_protective(self): by_path, _ = self.merged_audit(live_session_paths={MERGED_PATH}) entry = by_path[MERGED_PATH] self.assertTrue(entry["has_live_session"]) self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) self.assertFalse(entry["removable"]) def test_head_not_contained_in_master_is_not_removable(self): by_path, _ = self.merged_audit(contained=False) entry = by_path[MERGED_PATH] self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) self.assertFalse(entry["removable"]) self.assertIn( "worktree head is not contained in authoritative master " "(unmerged commits remain)", entry["merged_pr_cleanup"]["block_reasons"], ) def test_unknown_containment_fails_closed(self): by_path, _ = self.merged_audit(contained=None) entry = by_path[MERGED_PATH] self.assertFalse(entry["removable"]) self.assertIn( "containment of the worktree head in master is unknown", entry["merged_pr_cleanup"]["block_reasons"], ) def test_missing_master_ref_fails_closed(self): by_path, _ = self.run_audit( pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]) ) self.assertFalse(by_path[MERGED_PATH]["removable"]) def test_unmerged_owning_pr_is_not_removable(self): index = wca.build_pr_index([_pr(901, MERGED_BRANCH, merged=False)]) by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master") entry = by_path[MERGED_PATH] self.assertFalse(entry["removable"]) self.assertIn( "owning PR #901 is not merged", entry["merged_pr_cleanup"]["block_reasons"], ) def test_control_checkout_is_never_removable(self): by_path, _ = self.merged_audit() control = by_path["/repo"] self.assertTrue(control["is_protected"]) self.assertEqual(control["classification"], wca.CLASS_UNSAFE_UNKNOWN) self.assertFalse(control["removable"]) def test_control_checkout_not_removable_even_if_linked_and_merged(self): """A merged PR on the control checkout must not unlock removal.""" porcelain = _porcelain(("/repo", MERGED_BRANCH, HEAD_SHA)) with patch.object( wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(porcelain) ), patch.object( wca, "read_worktree_dirty", return_value={"exists": True, "dirty": False, "dirty_files": []}, ), patch.object( wca, "git_worktree_list", return_value="(mocked)" ), patch.object( wca, "is_head_ancestor_of_ref", return_value=True ): report = wca.audit_branches_directory( "/repo", pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]), master_ref="prgs/master", ) entry = report["worktrees"][0] self.assertEqual(entry["classification"], wca.CLASS_UNSAFE_UNKNOWN) self.assertFalse(entry["removable"]) class TestAmbiguousLinkageFailsClosed(_AuditHarness): def test_competing_prs_on_one_branch_fail_closed(self): index = wca.build_pr_index( [_pr(849, MERGED_BRANCH), _pr(860, MERGED_BRANCH)] ) by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master") entry = by_path[MERGED_PATH] self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_AMBIGUOUS) self.assertIsNone(entry["pr_number"]) self.assertEqual(entry["classification"], wca.CLASS_ACTIVE_ISSUE_WORK) self.assertFalse(entry["removable"]) def test_merged_plus_open_pr_on_one_branch_fails_closed(self): index = wca.build_pr_index( [_pr(849, MERGED_BRANCH), _pr(861, MERGED_BRANCH, merged=False)] ) by_path, _ = self.run_audit(pr_index=index, master_ref="prgs/master") entry = by_path[MERGED_PATH] self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_AMBIGUOUS) self.assertFalse(entry["removable"]) def test_no_owning_pr_fails_closed(self): by_path, _ = self.run_audit( pr_index=wca.build_pr_index([_pr(849, "feat/other-branch")]), master_ref="prgs/master", ) entry = by_path[MERGED_PATH] self.assertEqual(entry["merged_pr_linkage"]["status"], wca.LINKAGE_NONE) self.assertFalse(entry["removable"]) def test_malformed_pr_records_are_dropped_not_guessed(self): index = wca.build_pr_index( [ {"number": None, "head": {"ref": MERGED_BRANCH}}, {"number": 5, "head": {}}, {"number": "not-an-int", "head": {"ref": MERGED_BRANCH}}, ] ) self.assertEqual(index, {}) self.assertEqual( wca.resolve_owning_pr(branch=MERGED_BRANCH, pr_index=index)["status"], wca.LINKAGE_NONE, ) def test_detached_worktree_has_no_branch_linkage(self): self.assertEqual( wca.resolve_owning_pr(branch=None, pr_index={})["status"], wca.LINKAGE_UNKNOWN, ) class TestUnrelatedClassificationsUnchanged(unittest.TestCase): """Non-issue_work worktrees keep their pre-#858 classifications.""" PORCELAIN = _porcelain( ("/repo", "master", "f" * 40), ("/repo/branches/review-pr42", "review-pr42", "2" * 40), ("/repo/branches/baseline-master-x", "baseline-master-x", "3" * 40), ("/repo/branches/conflict-fix-pr50", "conflict-fix-pr50", "4" * 40), ("/repo/branches/review-pr99", None, "5" * 40), ) def _audit(self, **kwargs): with patch.object( wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(self.PORCELAIN), ), patch.object( wca, "read_worktree_dirty", return_value={"exists": True, "dirty": False, "dirty_files": []}, ), patch.object( wca, "git_worktree_list", return_value="(mocked)" ), patch.object( wca, "is_head_ancestor_of_ref", return_value=True ): report = wca.audit_branches_directory("/repo", **kwargs) return {wt["path"]: wt for wt in report["worktrees"]} def test_classifications_identical_with_and_without_pr_evidence(self): without = self._audit() with_evidence = self._audit( pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]), master_ref="prgs/master", ) self.assertEqual( {p: e["classification"] for p, e in without.items()}, {p: e["classification"] for p, e in with_evidence.items()}, ) def test_lease_on_issue_does_not_capture_similarly_named_scratch_trees(self): """A lease on issue 777 protects issue work, not baseline/review trees.""" porcelain = _porcelain( ("/repo/branches/baseline-master-issue-777", "baseline-issue-777", "7" * 40), ("/repo/branches/issue-777-timeline", MERGED_BRANCH, HEAD_SHA), ) with patch.object( wca, "list_worktrees", return_value=wca.parse_worktree_porcelain(porcelain) ), patch.object( wca, "read_worktree_dirty", return_value={"exists": True, "dirty": False, "dirty_files": []}, ), patch.object( wca, "git_worktree_list", return_value="(mocked)" ), patch.object( wca, "is_head_ancestor_of_ref", return_value=True ): report = wca.audit_branches_directory( "/repo", pr_index=wca.build_pr_index([_pr(849, MERGED_BRANCH)]), master_ref="prgs/master", leased_issue_numbers={777}, ) by_path = {wt["path"]: wt for wt in report["worktrees"]} baseline = by_path["/repo/branches/baseline-master-issue-777"] self.assertFalse(baseline["has_active_lease"]) self.assertEqual(baseline["classification"], wca.CLASS_CLEAN_STALE_REMOVABLE) issue_work = by_path["/repo/branches/issue-777-timeline"] self.assertTrue(issue_work["has_active_lease"]) self.assertFalse(issue_work["removable"]) def test_review_and_baseline_still_removable(self): by_path = self._audit( pr_index=wca.build_pr_index([]), master_ref="prgs/master" ) self.assertEqual( by_path["/repo/branches/review-pr42"]["classification"], wca.CLASS_CLEAN_STALE_REMOVABLE, ) self.assertEqual( by_path["/repo/branches/baseline-master-x"]["classification"], wca.CLASS_CLEAN_STALE_REMOVABLE, ) self.assertEqual( by_path["/repo/branches/review-pr99"]["classification"], wca.CLASS_DETACHED_REVIEW_LEFTOVER, ) def test_conflict_fix_ttl_behaviour_unchanged(self): """conflict_fix still needs only TTL expiry; #858 did not touch it.""" self.assertEqual( wca.classify_worktree( workflow_type=wca.WORKFLOW_CONFLICT_FIX, is_dirty=False, ttl_expired=True, ), wca.CLASS_CLEAN_STALE_REMOVABLE, ) self.assertEqual( wca.classify_worktree( workflow_type=wca.WORKFLOW_CONFLICT_FIX, is_dirty=False, ttl_expired=False, ), wca.CLASS_ACTIVE_ISSUE_WORK, ) def test_issue_work_ttl_alone_no_longer_grants_removal(self): """Age is not landing proof: TTL alone must not reclaim issue work.""" self.assertEqual( wca.classify_worktree( workflow_type=wca.WORKFLOW_ISSUE_WORK, is_dirty=False, ttl_expired=True, ), wca.CLASS_ACTIVE_ISSUE_WORK, ) class TestAssessorPerformsNoDeletion(_AuditHarness): def test_audit_never_removes_a_worktree(self): with patch.object(wca, "remove_worktree") as removal: self.merged_audit() removal.assert_not_called() def test_audit_shells_out_to_no_destructive_git_command(self): seen = [] real_run = subprocess.run def recording_run(cmd, *args, **kwargs): seen.append(cmd) return real_run(["true"], *args, **kwargs) with patch.object(subprocess, "run", side_effect=recording_run): wca.audit_branches_directory("/nonexistent-repo-for-audit") joined = [" ".join(c) if isinstance(c, list) else str(c) for c in seen] for cmd in joined: self.assertNotIn("worktree remove", cmd) self.assertNotIn("branch -D", cmd) self.assertNotIn("push", cmd) class TestAgreementWithPrScopedReconciler(unittest.TestCase): """The audit and merged_cleanup_reconcile must agree on identical input. Uses a real throwaway git repository so containment is computed by git rather than asserted. Nothing outside the temporary directory is touched. """ def _git(self, *args): subprocess.run( ["git", "-C", self.root, *args], check=True, capture_output=True, text=True, ) def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.root = os.path.realpath(self._tmp.name) self._git("init", "-b", "master", ".") self._git("config", "user.email", "test@example.invalid") self._git("config", "user.name", "Test") with open(os.path.join(self.root, "seed.txt"), "w") as fh: fh.write("seed\n") self._git("add", "seed.txt") self._git("commit", "-m", "seed") self.branch = "feat/issue-777-timeline" self._git("checkout", "-b", self.branch) with open(os.path.join(self.root, "feature.txt"), "w") as fh: fh.write("feature\n") self._git("add", "feature.txt") self._git("commit", "-m", "feature") self.head_sha = subprocess.run( ["git", "-C", self.root, "rev-parse", "HEAD"], capture_output=True, text=True, check=True, ).stdout.strip() self._git("checkout", "master") self._git("merge", "--no-ff", "-m", "merge feature", self.branch) self.worktree = os.path.join(self.root, "branches", "issue-777-timeline") self._git("worktree", "add", self.worktree, self.branch) def tearDown(self): self._tmp.cleanup() def _pr_index(self): return wca.build_pr_index( [ { "number": 849, "head": {"ref": self.branch, "sha": self.head_sha}, "merged_at": "2026-07-24T01:00:00Z", } ] ) def _audit_entry(self): report = wca.audit_branches_directory( self.root, pr_index=self._pr_index(), master_ref="master" ) return next(wt for wt in report["worktrees"] if wt["path"] == self.worktree) def _reconciler_entry(self): return mcr.assess_local_worktree_cleanup( pr_number=849, head_branch=self.branch, merged=True, worktree_state=mcr.resolve_cleanup_worktree_state( project_root=self.root, head_branch=self.branch, issue_number=777, pr_head_sha=self.head_sha, target_ref="master", ), active_lock=False, ) def test_both_assessors_agree_the_worktree_is_safe(self): audit_entry = self._audit_entry() reconciler = self._reconciler_entry() self.assertTrue(reconciler["safe_to_remove_worktree"], reconciler) self.assertTrue(audit_entry["removable"], audit_entry) self.assertEqual(audit_entry["pr_number"], reconciler["pr_number"]) self.assertEqual(audit_entry["merged_pr_cleanup"]["block_reasons"], []) self.assertEqual(reconciler["block_reasons"], []) def test_both_assessors_agree_a_dirty_worktree_is_unsafe(self): with open(os.path.join(self.worktree, "feature.txt"), "a") as fh: fh.write("local edit\n") audit_entry = self._audit_entry() reconciler = self._reconciler_entry() self.assertFalse(audit_entry["removable"]) self.assertFalse(reconciler["safe_to_remove_worktree"]) def test_worktree_still_present_after_audit(self): self._audit_entry() self.assertTrue(os.path.isdir(self.worktree)) if __name__ == "__main__": unittest.main()