sysadminandClaude Opus 4.8
cedf451a2b
Stale worktrees accumulate under branches/ when runs stop early, race a
sibling session, hit a validation failure, or lose cwd state, making later
workflow decisions harder and riskier.
Changes:
- worktree_cleanup_audit.py — ownership metadata, safety-first classifier
(active_open_pr, active_issue_work, dirty_local_worktree,
clean_stale_removable, detached_review_leftover, unsafe_unknown), TTL
expiry, per-worktree removal proof, and success-completion cleanup plan.
Only clean_stale_removable and detached_review_leftover are removable;
dirty/PR/leased/protected/unknown worktrees are never auto-deleted.
- gitea_audit_worktree_cleanup — read-only MCP tool that classifies every
branches/ entry, fetches live open-PR heads (fails closed if unavailable),
and returns counts, removable candidates, and git worktree list proof.
- work-issue.md 22A + review-merge-pr.md 28 — cleanup/TTL workflow rules.
- tests/test_worktree_cleanup_audit.py — 30 cases covering all nine
acceptance scenarios plus inference, metadata, TTL, and report accuracy.
Validation:
/Users/jasonwalker/Development/Gitea-Tools/venv/bin/python \
-m unittest tests.test_worktree_cleanup_audit tests.test_merged_cleanup_reconcile -q
38 passed.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
384 lines
14 KiB
Python
384 lines
14 KiB
Python
"""Tests for session-owned worktree cleanup audit and TTL enforcement (#401)."""
|
|
|
|
import sys
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import worktree_cleanup_audit as wca # noqa: E402
|
|
|
|
|
|
NOW = datetime(2026, 7, 7, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def _iso(dt):
|
|
return dt.isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
class TestWorkflowTypeInference(unittest.TestCase):
|
|
def test_review_pr_path(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type("branches/review-pr376", "review-pr376"),
|
|
wca.WORKFLOW_REVIEW,
|
|
)
|
|
|
|
def test_baseline_path(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type("branches/baseline-master-issue-401"),
|
|
wca.WORKFLOW_BASELINE,
|
|
)
|
|
|
|
def test_merge_simulation_path(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type("branches/merge-sim-pr380"),
|
|
wca.WORKFLOW_MERGE_SIMULATION,
|
|
)
|
|
|
|
def test_issue_work_branch(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type(
|
|
"branches/issue-401-worktree", "feat/issue-401-worktree"
|
|
),
|
|
wca.WORKFLOW_ISSUE_WORK,
|
|
)
|
|
|
|
def test_conflict_fix_path(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type("branches/conflict-fix-pr376"),
|
|
wca.WORKFLOW_CONFLICT_FIX,
|
|
)
|
|
|
|
def test_unknown_path(self):
|
|
self.assertEqual(
|
|
wca.infer_workflow_type("branches/scratchpad"), wca.WORKFLOW_UNKNOWN
|
|
)
|
|
|
|
|
|
class TestMetadata(unittest.TestCase):
|
|
def test_metadata_fields_present(self):
|
|
meta = wca.build_worktree_metadata(
|
|
path="branches/issue-401-worktree",
|
|
branch="feat/issue-401-worktree",
|
|
head_sha="abc123",
|
|
creator="jcwalker3",
|
|
profile="prgs-author",
|
|
created_at=_iso(NOW),
|
|
last_used_at=_iso(NOW),
|
|
)
|
|
for field in (
|
|
"path",
|
|
"workflow_type",
|
|
"issue_number",
|
|
"pr_number",
|
|
"branch",
|
|
"head_sha",
|
|
"creator",
|
|
"profile",
|
|
"created_at",
|
|
"last_used_at",
|
|
"cleanup_eligibility",
|
|
):
|
|
self.assertIn(field, meta)
|
|
self.assertEqual(meta["issue_number"], 401)
|
|
self.assertEqual(meta["workflow_type"], wca.WORKFLOW_ISSUE_WORK)
|
|
|
|
def test_review_metadata_auto_removable_flag(self):
|
|
meta = wca.build_worktree_metadata(path="branches/review-pr42")
|
|
self.assertTrue(meta["auto_remove_on_success"])
|
|
|
|
|
|
class TestTTL(unittest.TestCase):
|
|
def test_expired(self):
|
|
old = _iso(NOW - timedelta(hours=48))
|
|
self.assertTrue(wca.is_ttl_expired(last_used_at=old, now=NOW, ttl_hours=24))
|
|
|
|
def test_not_expired(self):
|
|
recent = _iso(NOW - timedelta(hours=1))
|
|
self.assertFalse(
|
|
wca.is_ttl_expired(last_used_at=recent, now=NOW, ttl_hours=24)
|
|
)
|
|
|
|
def test_unknown_timestamp_fails_safe(self):
|
|
self.assertFalse(wca.is_ttl_expired(last_used_at=None, now=NOW))
|
|
self.assertFalse(
|
|
wca.is_ttl_expired(last_used_at="not-a-date", now=NOW)
|
|
)
|
|
|
|
|
|
class TestClassification(unittest.TestCase):
|
|
def test_successful_review_cleanup(self):
|
|
# Scenario 1: clean review worktree -> removable.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_REVIEW, is_dirty=False
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
|
|
self.assertTrue(wca.is_removable(cls))
|
|
|
|
def test_dirty_worktree_preserved(self):
|
|
# Scenario 3: dirty worktree is never removable.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_REVIEW, is_dirty=True
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_DIRTY_LOCAL)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_active_pr_worktree_preserved(self):
|
|
# Scenario 4: open PR wins over an otherwise-removable review worktree.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_REVIEW,
|
|
is_dirty=False,
|
|
has_open_pr=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_ACTIVE_OPEN_PR)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_stale_clean_issue_worktree_removable(self):
|
|
# Scenario 5: clean issue worktree, TTL expired, no lock -> removable.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=False,
|
|
ttl_expired=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
|
|
self.assertTrue(wca.is_removable(cls))
|
|
|
|
def test_fresh_clean_issue_worktree_preserved(self):
|
|
# Clean issue worktree not yet TTL-expired stays active.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=False,
|
|
ttl_expired=False,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_detached_review_worktree_classified(self):
|
|
# Scenario 6: detached review worktree -> detached_review_leftover.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_REVIEW,
|
|
is_dirty=False,
|
|
is_detached=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_DETACHED_REVIEW_LEFTOVER)
|
|
self.assertTrue(wca.is_removable(cls))
|
|
|
|
def test_ttl_expired_but_dirty_preserved(self):
|
|
# Scenario 7: dirty wins over TTL expiry.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=True,
|
|
ttl_expired=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_DIRTY_LOCAL)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_lease_protected_worktree_preserved(self):
|
|
# Scenario 8: active lease is never removable.
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_REVIEW,
|
|
is_dirty=False,
|
|
has_active_lease=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_active_issue_lock_preserved(self):
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=False,
|
|
has_active_issue_lock=True,
|
|
ttl_expired=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_protected_base_worktree_never_removable(self):
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_ISSUE_WORK,
|
|
is_dirty=False,
|
|
is_protected=True,
|
|
ttl_expired=True,
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_UNSAFE_UNKNOWN)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
def test_unknown_workflow_type_unsafe(self):
|
|
cls = wca.classify_worktree(
|
|
workflow_type=wca.WORKFLOW_UNKNOWN, is_dirty=False, ttl_expired=True
|
|
)
|
|
self.assertEqual(cls, wca.CLASS_UNSAFE_UNKNOWN)
|
|
self.assertFalse(wca.is_removable(cls))
|
|
|
|
|
|
class TestRemovalDecision(unittest.TestCase):
|
|
def test_safe_removal_proof(self):
|
|
decision = wca.assess_worktree_removal(
|
|
path="branches/review-pr42",
|
|
branch="review-pr42",
|
|
head_sha="abc123",
|
|
is_dirty=False,
|
|
has_open_pr=False,
|
|
has_active_lease=False,
|
|
classification=wca.CLASS_CLEAN_STALE_REMOVABLE,
|
|
)
|
|
self.assertTrue(decision["safe_to_remove"])
|
|
self.assertEqual(decision["block_reasons"], [])
|
|
self.assertTrue(decision["clean"])
|
|
self.assertTrue(decision["no_active_pr"])
|
|
self.assertTrue(decision["no_active_lease"])
|
|
|
|
def test_dirty_blocks_removal(self):
|
|
decision = wca.assess_worktree_removal(
|
|
path="branches/review-pr42",
|
|
branch="review-pr42",
|
|
head_sha="abc123",
|
|
is_dirty=True,
|
|
has_open_pr=False,
|
|
has_active_lease=False,
|
|
classification=wca.CLASS_DIRTY_LOCAL,
|
|
)
|
|
self.assertFalse(decision["safe_to_remove"])
|
|
self.assertIn("worktree has uncommitted changes", decision["block_reasons"])
|
|
|
|
def test_open_pr_and_lease_block_removal(self):
|
|
decision = wca.assess_worktree_removal(
|
|
path="branches/review-pr42",
|
|
branch="review-pr42",
|
|
head_sha="abc123",
|
|
is_dirty=False,
|
|
has_open_pr=True,
|
|
has_active_lease=True,
|
|
classification=wca.CLASS_ACTIVE_OPEN_PR,
|
|
)
|
|
self.assertFalse(decision["safe_to_remove"])
|
|
self.assertIn("worktree branch has an open PR", decision["block_reasons"])
|
|
self.assertIn("worktree has an active lease", decision["block_reasons"])
|
|
|
|
|
|
class TestSuccessCleanupPlan(unittest.TestCase):
|
|
def test_review_worktree_removed_at_success(self):
|
|
# Scenario 1 end-to-end: clean review worktree removed at success.
|
|
meta = wca.build_worktree_metadata(path="branches/review-pr42")
|
|
plan = wca.plan_success_cleanup(
|
|
metadata=meta, is_dirty=False, has_open_pr=False, has_active_lease=False
|
|
)
|
|
self.assertTrue(plan["remove"])
|
|
|
|
def test_failed_review_leaves_worktree_reported(self):
|
|
# Scenario 2: dirty review worktree preserved and reported at failure.
|
|
meta = wca.build_worktree_metadata(path="branches/review-pr42")
|
|
plan = wca.plan_success_cleanup(
|
|
metadata=meta, is_dirty=True, has_open_pr=False, has_active_lease=False
|
|
)
|
|
self.assertFalse(plan["remove"])
|
|
self.assertIn("uncommitted", plan["reason"])
|
|
report = wca.cleanup_failure_report(meta["path"], plan["reason"])
|
|
self.assertFalse(report["removed"])
|
|
self.assertEqual(report["path"], "branches/review-pr42")
|
|
|
|
def test_issue_worktree_preserved_by_policy(self):
|
|
meta = wca.build_worktree_metadata(
|
|
path="branches/issue-401-worktree", branch="feat/issue-401-worktree"
|
|
)
|
|
plan = wca.plan_success_cleanup(
|
|
metadata=meta, is_dirty=False, has_open_pr=False, has_active_lease=False
|
|
)
|
|
self.assertFalse(plan["remove"])
|
|
self.assertIn("preserved by policy", plan["reason"])
|
|
|
|
|
|
class TestPorcelainParser(unittest.TestCase):
|
|
def test_parse_branch_and_detached(self):
|
|
text = (
|
|
"worktree /repo\n"
|
|
"HEAD 1111111111111111111111111111111111111111\n"
|
|
"branch refs/heads/master\n"
|
|
"\n"
|
|
"worktree /repo/branches/review-pr42\n"
|
|
"HEAD 2222222222222222222222222222222222222222\n"
|
|
"detached\n"
|
|
)
|
|
entries = wca.parse_worktree_porcelain(text)
|
|
self.assertEqual(len(entries), 2)
|
|
self.assertEqual(entries[0]["branch"], "master")
|
|
self.assertFalse(entries[0]["detached"])
|
|
self.assertIsNone(entries[1]["branch"])
|
|
self.assertTrue(entries[1]["detached"])
|
|
|
|
|
|
class TestAuditReportAccuracy(unittest.TestCase):
|
|
"""Scenario 9: cleanup report accuracy over a mixed branches/ directory."""
|
|
|
|
PORCELAIN = (
|
|
"worktree /repo\n"
|
|
"HEAD 1111111111111111111111111111111111111111\n"
|
|
"branch refs/heads/master\n"
|
|
"\n"
|
|
"worktree /repo/branches/review-pr42\n"
|
|
"HEAD 2222222222222222222222222222222222222222\n"
|
|
"branch refs/heads/review-pr42\n"
|
|
"\n"
|
|
"worktree /repo/branches/issue-400-open-pr\n"
|
|
"HEAD 3333333333333333333333333333333333333333\n"
|
|
"branch refs/heads/feat/issue-400-open-pr\n"
|
|
"\n"
|
|
"worktree /repo/branches/issue-401-dirty\n"
|
|
"HEAD 4444444444444444444444444444444444444444\n"
|
|
"branch refs/heads/feat/issue-401-dirty\n"
|
|
)
|
|
|
|
def _fake_dirty(self, path):
|
|
if path.endswith("issue-401-dirty"):
|
|
return {"exists": True, "dirty": True, "dirty_files": [" M x.py"]}
|
|
return {"exists": True, "dirty": False, "dirty_files": []}
|
|
|
|
def test_mixed_directory_classified(self):
|
|
with patch.object(
|
|
wca,
|
|
"list_worktrees",
|
|
return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
|
|
), patch.object(
|
|
wca, "read_worktree_dirty", side_effect=self._fake_dirty
|
|
), patch.object(
|
|
wca, "git_worktree_list", return_value="(mocked)"
|
|
):
|
|
report = wca.audit_branches_directory(
|
|
"/repo",
|
|
open_pr_branches={"feat/issue-400-open-pr"},
|
|
)
|
|
|
|
by_path = {wt["path"]: wt for wt in report["worktrees"]}
|
|
# main checkout on master -> protected -> unsafe/unknown, not removable
|
|
self.assertEqual(
|
|
by_path["/repo"]["classification"], wca.CLASS_UNSAFE_UNKNOWN
|
|
)
|
|
# clean review worktree -> removable
|
|
self.assertEqual(
|
|
by_path["/repo/branches/review-pr42"]["classification"],
|
|
wca.CLASS_CLEAN_STALE_REMOVABLE,
|
|
)
|
|
# open PR branch -> preserved
|
|
self.assertEqual(
|
|
by_path["/repo/branches/issue-400-open-pr"]["classification"],
|
|
wca.CLASS_ACTIVE_OPEN_PR,
|
|
)
|
|
# dirty worktree -> preserved
|
|
self.assertEqual(
|
|
by_path["/repo/branches/issue-401-dirty"]["classification"],
|
|
wca.CLASS_DIRTY_LOCAL,
|
|
)
|
|
# exactly one removable candidate (the clean review worktree)
|
|
self.assertEqual(report["removable_count"], 1)
|
|
self.assertEqual(
|
|
report["removable_candidates"][0]["path"],
|
|
"/repo/branches/review-pr42",
|
|
)
|
|
self.assertEqual(report["total"], 4)
|
|
self.assertEqual(report["git_worktree_list"], "(mocked)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|