From cedf451a2ba635d42ebe7d35391abe11ceb5daf8 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 11:56:44 -0400 Subject: [PATCH] feat: add session-owned worktree cleanup audit and TTL enforcement (Closes #401) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- gitea_mcp_server.py | 85 ++++ .../workflows/review-merge-pr.md | 10 + .../workflows/work-issue.md | 33 ++ tests/test_worktree_cleanup_audit.py | 383 ++++++++++++++ worktree_cleanup_audit.py | 480 ++++++++++++++++++ 5 files changed, 991 insertions(+) create mode 100644 tests/test_worktree_cleanup_audit.py create mode 100644 worktree_cleanup_audit.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 9ea9050..d82542b 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -547,6 +547,7 @@ import reconciliation_workflow # noqa: E402 import review_merge_state_machine # noqa: E402 import pr_work_lease # noqa: E402 import native_mcp_preference # noqa: E402 +import worktree_cleanup_audit # noqa: E402 # Keyed issue-lock storage (#443): per remote/org/repo/issue files under @@ -4270,6 +4271,90 @@ def gitea_scan_already_landed_open_prs( } +@mcp.tool() +def gitea_audit_worktree_cleanup( + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + ttl_hours: float = worktree_cleanup_audit.DEFAULT_TTL_HOURS, +) -> dict: + """Read-only: classify every session-owned worktree under ``branches/`` (#401). + + Audits the local ``branches/`` directory, classifying each worktree as + active open PR, active issue work, dirty local, clean stale removable, + detached review leftover, or unsafe/unknown. Open PR branch heads are + fetched live so a worktree tied to an open PR is never marked removable; + the active issue-lock branch is read from the local lock file and treated + as active work. Deletes nothing and mutates no Gitea state. + + Fails closed if the live open-PR list cannot be fetched: without it, + removability cannot be proven, so no candidates are returned. + + Args: + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + ttl_hours: Age (hours) after which a clean issue/conflict-fix + worktree becomes stale-removable (default from + GITEA_WORKTREE_TTL_HOURS). + + Returns: + dict with per-worktree classifications, counts, removable + candidates, and the ``git worktree list`` verification proof. + """ + 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"), + } + + try: + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + open_prs = api_get_all(f"{repo_api_url(h, o, r)}/pulls?state=open", auth) + except Exception as exc: + return { + "success": False, + "performed": False, + "open_pr_state_verified": False, + "reasons": [ + "could not fetch live open PRs; removability unverified " + f"(fail closed): {_redact(str(exc))}" + ], + } + + open_pr_branches = { + str((pr.get("head") or {}).get("ref")) + for pr in open_prs + if (pr.get("head") or {}).get("ref") + } + + active_issue_branches: set[str] = set() + lock = merged_cleanup_reconcile.read_issue_lock(ISSUE_LOCK_FILE) + if lock and lock.get("branch_name"): + active_issue_branches.add(str(lock["branch_name"]).strip()) + + report = worktree_cleanup_audit.audit_branches_directory( + PROJECT_ROOT, + open_pr_branches=open_pr_branches, + active_issue_branches=active_issue_branches, + now=datetime.now(timezone.utc), + ttl_hours=ttl_hours, + ) + return { + "success": True, + "performed": False, + "open_pr_state_verified": True, + "task_mode": "work-issue", + **report, + } + + @mcp.tool() def gitea_reconcile_already_landed_pr( pr_number: int, diff --git a/skills/llm-project-workflow/workflows/review-merge-pr.md b/skills/llm-project-workflow/workflows/review-merge-pr.md index aca0bfc..e187863 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -849,6 +849,16 @@ Confirm: Clean only the session-owned `branches/` review worktree if the project workflow explicitly allows cleanup. +Review, baseline, and merge-simulation worktrees created during this run are +transient and are removed automatically at successful completion once they are +clean, carry no open PR, and hold no active lease (#401). Use +`gitea_audit_worktree_cleanup` (read-only) to classify `branches/` entries; only +`clean_stale_removable` and `detached_review_leftover` may be removed, one-by-one, +after `git worktree list` proof plus per-worktree proof of path, branch/HEAD, +clean/dirty status, no active PR/lease, and the removal result. Dirty, +active-PR, active-issue, and leased worktrees are never deleted automatically; a +failed removal must be reported with the leftover path and reason. + Do not delete or mutate unrelated branches/worktrees. Do not touch the main checkout except to update the stable branch after merge if explicitly allowed by the workflow. diff --git a/skills/llm-project-workflow/workflows/work-issue.md b/skills/llm-project-workflow/workflows/work-issue.md index ceef51c..2fb2818 100644 --- a/skills/llm-project-workflow/workflows/work-issue.md +++ b/skills/llm-project-workflow/workflows/work-issue.md @@ -691,6 +691,39 @@ Do not update the main checkout unless the canonical workflow explicitly allows Any cleanup is a mutation and must be reported. +### 22A. Session-owned worktree cleanup and TTL (#401) + +Every session-owned worktree created under `branches/` has ownership metadata: +path, workflow type, issue number, PR number, branch/head SHA, creator +identity/profile, created timestamp, last-used timestamp, and cleanup +eligibility. + +Cleanup is classification-driven. `gitea_audit_worktree_cleanup` (read-only) +classifies every `branches/` entry as exactly one of: + +* `active_open_pr` — branch has an open PR; never auto-removed. +* `active_issue_work` — active claim/lease or fresh issue worktree; never + auto-removed. +* `dirty_local_worktree` — uncommitted changes; never auto-removed. +* `clean_stale_removable` — clean, no PR, no lease; removable. +* `detached_review_leftover` — clean detached review/baseline/merge-simulation + worktree; removable. +* `unsafe_unknown` — protected base checkout or unknown workflow type; never + auto-removed. + +Only `clean_stale_removable` and `detached_review_leftover` may be removed, and +only one-by-one after `git worktree list` proof plus per-worktree proof of: +worktree path, branch/HEAD, clean/dirty status, no active PR/lease, and the +removal result. Review, baseline, and merge-simulation worktrees are removed +automatically at successful workflow completion; issue/conflict-fix worktrees +are removed only after their TTL (`GITEA_WORKTREE_TTL_HOURS`, default 24h) +expires and no lock/lease is held. + +Dirty, active-PR, active-issue, and leased worktrees are never deleted +automatically. If a removal fails, the final report must list the leftover +worktree path and the reason. Include the `git worktree list` output as final +cleanup verification. + ## 23. Recovery handoff rules If blocked, produce a recovery handoff with: diff --git a/tests/test_worktree_cleanup_audit.py b/tests/test_worktree_cleanup_audit.py new file mode 100644 index 0000000..638355e --- /dev/null +++ b/tests/test_worktree_cleanup_audit.py @@ -0,0 +1,383 @@ +"""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() diff --git a/worktree_cleanup_audit.py b/worktree_cleanup_audit.py new file mode 100644 index 0000000..9c5f743 --- /dev/null +++ b/worktree_cleanup_audit.py @@ -0,0 +1,480 @@ +"""Session-owned worktree cleanup audit and TTL enforcement (#401). + +LLM workflows create many session-owned worktrees under ``branches/`` +(review, baseline, merge-simulation, issue, and conflict-fix worktrees). +When a run stops early, races a sibling session, hits a validation failure, +or loses shell/cwd state, those worktrees are left behind and later workflow +decisions get harder and riskier. + +This module provides: + +* ``build_worktree_metadata`` — ownership/purpose metadata for a worktree. +* ``classify_worktree`` — safety-first classification into the audit + vocabulary (active open PR, active issue work, dirty, clean stale + removable, detached review leftover, unsafe/unknown). +* ``assess_worktree_removal`` — a per-worktree removal decision with an + explicit proof and block reasons. +* git-shelling helpers (``list_worktrees``, ``read_worktree_dirty``, + ``git_worktree_list``, ``remove_worktree``) and ``audit_branches_directory`` + that classify every entry under ``branches/``. + +Pure assessment functions take explicit state so they are unit-testable +without a filesystem or network. Only the thin git helpers shell out, and +removal is only ever executed after ``assess_worktree_removal`` proves the +worktree is safe to delete. +""" + +from __future__ import annotations + +import os +import re +import subprocess +from datetime import datetime, timezone +from typing import Any + +PROTECTED_BRANCHES = frozenset({"master", "main", "dev"}) +DEFAULT_TTL_HOURS = float(os.environ.get("GITEA_WORKTREE_TTL_HOURS", "24") or 24) + +# Workflow types that can create session-owned worktrees. +WORKFLOW_REVIEW = "review" +WORKFLOW_BASELINE = "baseline" +WORKFLOW_MERGE_SIMULATION = "merge_simulation" +WORKFLOW_ISSUE_WORK = "issue_work" +WORKFLOW_CONFLICT_FIX = "conflict_fix" +WORKFLOW_UNKNOWN = "unknown" + +# Workflow types whose worktrees are transient and removed automatically at +# successful workflow completion (acceptance criterion 2). +AUTO_REMOVE_ON_SUCCESS = frozenset( + {WORKFLOW_REVIEW, WORKFLOW_BASELINE, WORKFLOW_MERGE_SIMULATION} +) + +# Cleanup-audit classification vocabulary (acceptance criterion 4). +CLASS_ACTIVE_OPEN_PR = "active_open_pr" +CLASS_ACTIVE_ISSUE_WORK = "active_issue_work" +CLASS_DIRTY_LOCAL = "dirty_local_worktree" +CLASS_CLEAN_STALE_REMOVABLE = "clean_stale_removable" +CLASS_DETACHED_REVIEW_LEFTOVER = "detached_review_leftover" +CLASS_UNSAFE_UNKNOWN = "unsafe_unknown" + +# Only these two classifications may ever be removed automatically. +REMOVABLE_CLASSES = frozenset( + {CLASS_CLEAN_STALE_REMOVABLE, CLASS_DETACHED_REVIEW_LEFTOVER} +) + +_ISSUE_REF_RE = re.compile(r"issue-(\d+)", re.IGNORECASE) +_ISSUE_BRANCH_PREFIXES = ("feat/", "fix/", "docs/", "chore/") + + +def infer_workflow_type(path: str | None, branch: str | None = None) -> str: + """Infer the creating workflow type from a worktree path or branch name.""" + text = f"{path or ''} {branch or ''}".lower() + if "baseline" in text: + return WORKFLOW_BASELINE + if "merge-sim" in text or "merge_sim" in text or "mergesim" in text: + return WORKFLOW_MERGE_SIMULATION + if "review" in text or "review-pr" in text: + return WORKFLOW_REVIEW + if "conflict" in text: + return WORKFLOW_CONFLICT_FIX + if "issue-" in text or (branch or "").startswith(_ISSUE_BRANCH_PREFIXES): + return WORKFLOW_ISSUE_WORK + return WORKFLOW_UNKNOWN + + +def _extract_issue_number(path: str | None, branch: str | None) -> int | None: + match = _ISSUE_REF_RE.search(f"{path or ''} {branch or ''}") + return int(match.group(1)) if match else None + + +def build_worktree_metadata( + *, + path: str, + branch: str | None = None, + head_sha: str | None = None, + workflow_type: str | None = None, + issue_number: int | None = None, + pr_number: int | None = None, + creator: str | None = None, + profile: str | None = None, + created_at: str | None = None, + last_used_at: str | None = None, + cleanup_eligibility: str | None = None, +) -> dict[str, Any]: + """Return ownership/purpose metadata for a session-owned worktree. + + Covers acceptance criterion 1: path, workflow type, issue/PR number, + branch/head SHA, creator identity/profile, created and last-used + timestamps, and cleanup eligibility. + """ + wt = workflow_type or infer_workflow_type(path, branch) + issue = issue_number + if issue is None: + issue = _extract_issue_number(path, branch) + return { + "path": path, + "workflow_type": wt, + "issue_number": issue, + "pr_number": pr_number, + "branch": branch, + "head_sha": head_sha, + "creator": creator, + "profile": profile, + "created_at": created_at, + "last_used_at": last_used_at, + "auto_remove_on_success": wt in AUTO_REMOVE_ON_SUCCESS, + "cleanup_eligibility": cleanup_eligibility, + } + + +def _parse_timestamp(value: str | None) -> datetime | None: + if not value: + return None + text = str(value).strip() + if not text: + return None + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed + + +def is_ttl_expired( + *, + last_used_at: str | None, + now: datetime | str | None, + ttl_hours: float = DEFAULT_TTL_HOURS, +) -> bool: + """Return True only when the age is known and exceeds ``ttl_hours``. + + Unknown or unparseable timestamps fail safe (not expired) so a worktree + is never treated as removable merely because its age is unknown. + """ + last = _parse_timestamp(last_used_at) + now_dt = now if isinstance(now, datetime) else _parse_timestamp(now) + if last is None or now_dt is None: + return False + if now_dt.tzinfo is None: + now_dt = now_dt.replace(tzinfo=timezone.utc) + return (now_dt - last).total_seconds() > ttl_hours * 3600.0 + + +def classify_worktree( + *, + workflow_type: str, + is_dirty: bool, + has_open_pr: bool = False, + has_active_lease: bool = False, + has_active_issue_lock: bool = False, + is_detached: bool = False, + branch_gone: bool = False, + ttl_expired: bool = False, + is_protected: bool = False, + metadata_known: bool = True, +) -> str: + """Classify a worktree, safety-first: any preservation signal wins. + + Dirty, open-PR, leased, active-lock, protected, and unknown states are + all non-removable and are checked before any removable classification, + so nothing removable can shadow a preservation signal (criteria 6-8). + """ + if is_protected: + # The main checkout / a protected base branch is never removable. + return CLASS_UNSAFE_UNKNOWN + if is_dirty: + return CLASS_DIRTY_LOCAL # never auto-deleted (criterion 6) + if has_open_pr: + return CLASS_ACTIVE_OPEN_PR # never auto-deleted (criterion 7) + if has_active_lease: + return CLASS_ACTIVE_ISSUE_WORK # never auto-deleted (criterion 8) + if has_active_issue_lock: + return CLASS_ACTIVE_ISSUE_WORK + if not metadata_known or workflow_type == WORKFLOW_UNKNOWN: + return CLASS_UNSAFE_UNKNOWN # never auto-deleted without proof + + # Clean, no PR, no lease, no lock, known workflow type. + if workflow_type in AUTO_REMOVE_ON_SUCCESS: + if is_detached or branch_gone: + return CLASS_DETACHED_REVIEW_LEFTOVER + return CLASS_CLEAN_STALE_REMOVABLE + # issue_work / conflict_fix: only removable once the TTL has expired. + if ttl_expired: + return CLASS_CLEAN_STALE_REMOVABLE + return CLASS_ACTIVE_ISSUE_WORK + + +def is_removable(classification: str) -> bool: + """Return True only for the two auto-removable classifications.""" + return classification in REMOVABLE_CLASSES + + +def assess_worktree_removal( + *, + path: str, + branch: str | None, + head_sha: str | None, + is_dirty: bool, + has_open_pr: bool, + has_active_lease: bool, + classification: str, +) -> dict[str, Any]: + """Return a per-worktree removal decision with proof (criterion 9). + + A worktree is only safe to remove when it is clean, has no active PR, + has no active lease, and its classification is auto-removable. + """ + block_reasons: list[str] = [] + if is_dirty: + block_reasons.append("worktree has uncommitted changes") + if has_open_pr: + block_reasons.append("worktree branch has an open PR") + if has_active_lease: + block_reasons.append("worktree has an active lease") + if not is_removable(classification): + block_reasons.append( + f"classification '{classification}' is not auto-removable" + ) + return { + "path": path, + "branch": branch, + "head_sha": head_sha, + "classification": classification, + "clean": not is_dirty, + "no_active_pr": not has_open_pr, + "no_active_lease": not has_active_lease, + "safe_to_remove": not block_reasons, + "block_reasons": block_reasons, + } + + +def plan_success_cleanup( + *, + metadata: dict[str, Any], + is_dirty: bool, + has_open_pr: bool, + has_active_lease: bool, +) -> dict[str, Any]: + """Decide whether a just-completed worktree is removed at success. + + Review/baseline/merge-simulation worktrees are removed automatically at + successful completion (criterion 2); everything else is preserved and + reported. Dirty/PR/leased worktrees are always preserved (criteria 6-8). + """ + workflow_type = metadata.get("workflow_type", WORKFLOW_UNKNOWN) + if not metadata.get("auto_remove_on_success"): + return { + "remove": False, + "reason": f"workflow type '{workflow_type}' is preserved by policy", + } + classification = classify_worktree( + workflow_type=workflow_type, + is_dirty=is_dirty, + has_open_pr=has_open_pr, + has_active_lease=has_active_lease, + ) + decision = assess_worktree_removal( + path=metadata.get("path", ""), + branch=metadata.get("branch"), + head_sha=metadata.get("head_sha"), + is_dirty=is_dirty, + has_open_pr=has_open_pr, + has_active_lease=has_active_lease, + classification=classification, + ) + return { + "remove": decision["safe_to_remove"], + "reason": "clean transient worktree removable at success completion" + if decision["safe_to_remove"] + else "; ".join(decision["block_reasons"]), + "classification": classification, + "decision": decision, + } + + +def cleanup_failure_report(path: str, reason: str) -> dict[str, Any]: + """Structured leftover-worktree record for the final report (criterion 3).""" + return {"path": path, "removed": False, "reason": reason} + + +# -------------------------------------------------------------------------- +# git-shelling helpers (only these touch the filesystem) +# -------------------------------------------------------------------------- + + +def parse_worktree_porcelain(text: str) -> list[dict[str, Any]]: + """Parse ``git worktree list --porcelain`` output into entries.""" + entries: list[dict[str, Any]] = [] + current: dict[str, Any] = {} + for raw in (text or "").splitlines(): + line = raw.rstrip("\n") + if not line: + if current: + entries.append(current) + current = {} + continue + if line.startswith("worktree "): + if current: + entries.append(current) + current = { + "path": line[len("worktree ") :].strip(), + "head": None, + "branch": None, + "detached": False, + "bare": False, + } + elif line.startswith("HEAD "): + current["head"] = line[len("HEAD ") :].strip() + elif line.startswith("branch "): + ref = line[len("branch ") :].strip() + current["branch"] = ref.replace("refs/heads/", "", 1) + elif line == "detached": + current["detached"] = True + elif line == "bare": + current["bare"] = True + if current: + entries.append(current) + return entries + + +def list_worktrees(project_root: str) -> list[dict[str, Any]]: + """Return parsed ``git worktree list`` entries for ``project_root``.""" + result = subprocess.run( + ["git", "-C", project_root, "worktree", "list", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + if result.returncode != 0: + return [] + return parse_worktree_porcelain(result.stdout) + + +def git_worktree_list(project_root: str) -> str: + """Return plain ``git worktree list`` output for final verification (criterion 10).""" + result = subprocess.run( + ["git", "-C", project_root, "worktree", "list"], + capture_output=True, + text=True, + check=False, + ) + return (result.stdout or "").strip() + + +def read_worktree_dirty(path: str) -> dict[str, Any]: + """Return dirty state for a worktree path via ``git status --porcelain``.""" + if not path or not os.path.isdir(path): + return {"exists": False, "dirty": None, "dirty_files": []} + result = subprocess.run( + ["git", "-C", path, "status", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + files = [ln for ln in (result.stdout or "").splitlines() if ln.strip()] + return {"exists": True, "dirty": bool(files), "dirty_files": files} + + +def remove_worktree(project_root: str, path: str) -> dict[str, Any]: + """Remove a single worktree via ``git worktree remove`` (no ``--force``).""" + result = subprocess.run( + ["git", "-C", project_root, "worktree", "remove", path], + capture_output=True, + text=True, + check=False, + ) + ok = result.returncode == 0 + return { + "path": path, + "removed": ok, + "reason": None if ok else (result.stderr or "git worktree remove failed").strip(), + } + + +def _is_under_branches(project_root: str, path: str) -> bool: + branches_root = os.path.join(os.path.abspath(project_root), "branches") + return os.path.abspath(path or "").startswith(branches_root + os.sep) + + +def audit_branches_directory( + project_root: str, + *, + open_pr_branches: set[str] | None = None, + leased_branches: set[str] | None = None, + active_issue_branches: set[str] | None = None, + now: datetime | str | None = None, + ttl_hours: float = DEFAULT_TTL_HOURS, +) -> dict[str, Any]: + """Classify every session-owned worktree under ``branches/``. + + Read-only: shells out to git for discovery and dirty state, then applies + the pure classifier. Returns per-worktree classifications, counts, the + list of removable candidates, and the ``git worktree list`` proof. + """ + open_pr_branches = open_pr_branches or set() + leased_branches = leased_branches or set() + active_issue_branches = active_issue_branches or set() + + worktrees: list[dict[str, Any]] = [] + for entry in list_worktrees(project_root): + path = entry.get("path") or "" + branch = entry.get("branch") + is_protected = (branch in PROTECTED_BRANCHES) or not _is_under_branches( + project_root, path + ) + dirty_state = read_worktree_dirty(path) + is_dirty = bool(dirty_state.get("dirty")) + metadata = build_worktree_metadata( + path=path, branch=branch, head_sha=entry.get("head") + ) + has_open_pr = bool(branch) and branch in open_pr_branches + has_active_lease = bool(branch) and branch in leased_branches + has_active_lock = bool(branch) and branch in active_issue_branches + ttl_expired = is_ttl_expired( + last_used_at=metadata.get("last_used_at"), now=now, ttl_hours=ttl_hours + ) + classification = classify_worktree( + workflow_type=metadata["workflow_type"], + is_dirty=is_dirty, + has_open_pr=has_open_pr, + has_active_lease=has_active_lease, + has_active_issue_lock=has_active_lock, + is_detached=bool(entry.get("detached")), + branch_gone=branch is None and not entry.get("detached"), + ttl_expired=ttl_expired, + is_protected=is_protected, + ) + metadata["cleanup_eligibility"] = classification + worktrees.append( + { + **metadata, + "detached": bool(entry.get("detached")), + "dirty": is_dirty, + "dirty_files": dirty_state.get("dirty_files", []), + "has_open_pr": has_open_pr, + "has_active_lease": has_active_lease, + "has_active_issue_lock": has_active_lock, + "is_protected": is_protected, + "classification": classification, + "removable": is_removable(classification), + } + ) + + counts: dict[str, int] = {} + for wt in worktrees: + counts[wt["classification"]] = counts.get(wt["classification"], 0) + 1 + removable = [wt for wt in worktrees if wt["removable"]] + + return { + "project_root": project_root, + "worktrees": worktrees, + "counts": counts, + "removable_candidates": removable, + "removable_count": len(removable), + "total": len(worktrees), + "git_worktree_list": git_worktree_list(project_root), + } -- 2.43.7