Sweep a Pruned Worktree Entry's Branch in the Same Run in pr_review.py

scripts/pr_review.py

@@ -7051,11 +7051,11 @@ def run(self) -> int:
         for w in trees:
             if w.path == self.primary and w.branch:
                 self.permanent.add(w.branch)
-        held = {w.branch for w in trees if w.branch}
         for w in trees:
             if not w.prunable:
                 self.operating |= cleanup_operation_names(w.path)
-        self.prune([w for w in linked if w.prunable and not w.locked])
+        pruned = self.prune([w for w in linked if w.prunable and not w.locked])
+        held = {w.branch for w in trees if w.branch and w.path not in pruned}
         for w in linked:
             if not w.prunable or w.locked:
                 self.linked(w)
@@ -7070,8 +7070,12 @@ def run(self) -> int:
         self.base()
         return 2 if self.problems else 0
 
-    def prune(self, stale: list[Worktree]) -> None:
-        """Prune the entries whose directory is gone and whose record is old enough."""
+    def prune(self, stale: list[Worktree]) -> set[str]:
+        """Prune the entries whose directory is gone and whose record is old enough.
+
+        Returns the paths pruned, or under a report the paths a prune would remove, so their
+        branches are no longer held and are judged as loose branches in the same run.
+        """
         old = []
         for w in stale:
             hours = self.entry_hours(w)
@@ -7080,24 +7084,25 @@ def prune(self, stale: list[Worktree]) -> None:
             else:
                 old.append(w)
         if not old:
-            return
+            return set()
         if not self.apply:
             for w in old:
                 print(f"would-remove prunable worktree entry {w.path}")
-            return
+            return {w.path for w in old}
         expire = f"{int(self.min_age * 3600)}.seconds.ago"
         try:
             cleanup_git(self.primary, "worktree", "prune", "--expire", expire)
             left = {w.path for w in cleanup_worktrees(self.primary)}
         except CleanupUnread as exc:
             self.failed("prunable worktree entries", exc)
-            return
+            return set()
         for w in old:
             if w.path in left:
                 self.problems += 1
                 print(f"unread prunable worktree entry {w.path} is still listed after the prune")
             else:
                 print(f"removed prunable worktree entry {w.path}")
+        return {w.path for w in old if w.path not in left}
 
     def entry_hours(self, w: Worktree) -> float:
         """The idle hours of a missing worktree's record, from the `index` file prune reads.

tests/test_pr_review.py

@@ -12718,8 +12718,46 @@ def test_a_branch_without_a_reflog_is_kept_as_age_unreadable(self) -> None:
         self.assertIn("keep age-unreadable branch loose-done", self.out.getvalue())
         self.assertIn("loose-done", self.branches())
 
+    def gone_on_branch(self, branch: str, entry_hours: float = 48) -> str:
+        """A finished, idle branch whose worktree directory is gone, its record `entry_hours` old."""
+        gone = os.path.join(self.root, "wt", branch)
+        run_git(self.primary, "worktree", "add", "-b", branch, gone, "origin/develop")
+        admin = Path(gone, ".git").read_text(encoding="utf-8").split(":", 1)[1].strip()
+        shutil.rmtree(gone)
+        old = time.time() - entry_hours * 3600
+        os.utime(os.path.join(admin, "index"), (old, old))
+        self.age_branch(branch, 48)
+        return gone
+
+    def test_a_pruned_entry_s_branch_is_swept_in_the_same_run(self) -> None:
+        gone = self.gone_on_branch("gone-done")
+        self.advance_origin()
+        self.wire_sweep(remote_tip=None)
+        self.assertEqual(0, self.sweep("--apply"))
+        self.assertNotIn(gone, self.listed())
+        self.assertNotIn("gone-done", self.branches())
+
+    def test_a_dry_run_reports_a_prunable_entry_s_branch(self) -> None:
+        self.gone_on_branch("gone-done")
+        self.advance_origin()
+        self.wire_sweep(remote_tip=None)
+        self.assertEqual(0, self.sweep())
+        self.assertIn("gone-done", self.branches())
+        self.assertIn("would-remove branch gone-done", self.out.getvalue())
+
+    def test_a_too_recent_prunable_entry_still_holds_its_branch(self) -> None:
+        gone = self.gone_on_branch("gone-recent", entry_hours=0)
+        self.advance_origin()
+        self.wire_sweep(remote_tip=None)
+        self.assertEqual(0, self.sweep())
+        self.assertNotIn("would-remove branch gone-recent", self.out.getvalue())
+        self.assertEqual(0, self.sweep("--apply"))
+        self.assertIn(gone, self.listed())
+        self.assertIn("gone-recent", self.branches())
+
     def test_a_failed_prune_is_counted_and_the_other_items_still_run(self) -> None:
         made = self.scenario()
+        self.gone_on_branch("gone-failed")
         tip = run_git(self.primary, "rev-parse", CLEANUP_BRANCH)
         self.wire_sweep(remote_tip=tip)
         real = pr_review.cleanup_git
@@ -12732,6 +12770,8 @@ def failing(root: str, *args: str, **kw: Any) -> subprocess.CompletedProcess[str
         with mock.patch.object(pr_review, "cleanup_git", side_effect=failing):
             self.assertEqual(2, self.sweep("--apply"))
         self.assertIn("unread prunable worktree entries", self.out.getvalue())
+        self.assertNotIn("branch gone-failed", self.out.getvalue())
+        self.assertIn("gone-failed", self.branches())
         self.assertNotIn(made["merged"], self.listed())
 
     def test_a_stale_remote_ref_for_a_deleted_branch_is_not_containment(self) -> None: