From e4c0a51e5a0a1cb46cf2438b9b5689e8dfd9d4e2 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 11:50:28 -0400 Subject: [PATCH] feat: add worktree cleanup audit integrity (Closes #404) Reconcile before/after branches/ snapshots so preserved worktrees cannot disappear without removal logs or explained state transitions. - worktree_cleanup_audit.py: snapshot capture, disposition reconciliation - gitea_capture_branches_worktree_snapshot, gitea_assess_worktree_cleanup_integrity - Final-report rule author.worktree_cleanup_audit_proof - worktree-cleanup.md bulk audit section - tests/test_worktree_cleanup_audit.py (11 cases) --- final_report_validator.py | 22 + gitea_mcp_server.py | 54 ++ .../templates/worktree-cleanup.md | 40 ++ tests/test_worktree_cleanup_audit.py | 164 ++++++ worktree_cleanup_audit.py | 554 ++++++++++++++++++ 5 files changed, 834 insertions(+) create mode 100644 tests/test_worktree_cleanup_audit.py create mode 100644 worktree_cleanup_audit.py diff --git a/final_report_validator.py b/final_report_validator.py index 3fbac0f..46ec941 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -490,6 +490,27 @@ def _rule_reviewer_validation_failure_history( ] +def _rule_worktree_cleanup_audit_proof(report_text: str) -> list[dict[str, str]]: + from worktree_cleanup_audit import assess_cleanup_audit_final_report + + text = report_text or "" + if "cleanup audit" not in text.lower() and "reconciliation table" not in text.lower(): + return [] + result = assess_cleanup_audit_final_report(text) + if result.get("proven"): + return [] + return _findings_from_reasons( + "author.worktree_cleanup_audit_proof", + result.get("reasons") or [], + field="Worktree cleanup audit", + severity="block", + safe_next_action=( + "include reconciliation table counts, disposition rows, and " + "final git worktree list proof" + ), + ) + + def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]: text = report_text or "" if not _BARE_PYTEST_RE.search(text): @@ -930,6 +951,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_shared_controller_handoff, _rule_shared_email_disclosure, _rule_reviewer_vague_mutations_none, + _rule_worktree_cleanup_audit_proof, ], "issue_filing": [ _rule_shared_controller_handoff, diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 0bc82e4..a1ea72e 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -504,6 +504,7 @@ import already_landed_reconcile # noqa: E402 import author_mutation_worktree # noqa: E402 import issue_claim_heartbeat # noqa: E402 import merged_cleanup_reconcile # noqa: E402 +import worktree_cleanup_audit # noqa: E402 import reconciler_profile # noqa: E402 import reconciliation_workflow # noqa: E402 import review_merge_state_machine # noqa: E402 @@ -3509,6 +3510,59 @@ def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> boo raise +@mcp.tool() +def gitea_capture_branches_worktree_snapshot( + open_pr_branches: list[str] | None = None, + active_lock_branch: str | None = None, + leased_paths: list[str] | None = None, + worktree_path: str | None = None, +) -> dict: + """Read-only: capture ``branches/`` and worktree audit snapshot (#404).""" + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + root = PROJECT_ROOT + if worktree_path: + root = os.path.realpath(os.path.abspath(worktree_path)) + git_root = _get_git_root(root) + if git_root: + root = git_root + snapshot = worktree_cleanup_audit.capture_branches_worktree_snapshot( + root, + open_pr_branches=open_pr_branches, + active_lock_branch=active_lock_branch, + leased_paths=leased_paths, + ) + return {"success": True, "snapshot": snapshot} + + +@mcp.tool() +def gitea_assess_worktree_cleanup_integrity( + before_snapshot: dict, + after_snapshot: dict, + removals: list[dict] | None = None, + explained_missing: dict[str, str] | None = None, +) -> dict: + """Read-only: reconcile cleanup before/after snapshots (#404).""" + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "integrity_passed": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + return worktree_cleanup_audit.assess_worktree_cleanup_integrity( + before=before_snapshot, + after=after_snapshot, + removals=removals, + explained_missing=explained_missing, + ) + + @mcp.tool() def gitea_reconcile_merged_cleanups( dry_run: bool = True, diff --git a/skills/llm-project-workflow/templates/worktree-cleanup.md b/skills/llm-project-workflow/templates/worktree-cleanup.md index a9ffa95..4884525 100644 --- a/skills/llm-project-workflow/templates/worktree-cleanup.md +++ b/skills/llm-project-workflow/templates/worktree-cleanup.md @@ -26,3 +26,43 @@ Steps: Handoff: merge confirmed, issue closed, branch+worktree removed, checkout clean. ``` + +## Branches cleanup audit integrity (#404) + +Any bulk or multi-path cleanup under `branches/` must capture auditable before/after +identity for every initial directory and registered worktree. Use +`worktree_cleanup_audit.capture_cleanup_snapshot` before and after cleanup, record +every intentional removal in a removal log (path, method, order, timestamp, +pre-removal proof), then run `reconcile_cleanup_audit` and +`assess_cleanup_audit_integrity`. + +The cleanup report must include a reconciliation table: + +* initial count +* removed count +* preserved count +* missing-unexplained count +* final count + +Fail closed when: + +* a preserved (active PR, dirty, claim/lease, or unsafe) worktree disappears without + a removal log entry or explicit explanation +* the removal log omits a removed clean-stale path +* final counts do not reconcile with initial minus removed + +If another session removes or mutates a worktree during cleanup, record the path +under explained missing entries — never treat silent disappearance as success. + +## Bulk `branches/` cleanup audit (#404) + +Before removing multiple session-owned worktrees: + +1. Call `gitea_capture_branches_worktree_snapshot` and record the before snapshot. +2. Remove only paths classified as `clean_stale_removable` with explicit per-path proof. +3. Log every removal with path, method, and timestamp/order. +4. Capture an after snapshot with the same tool. +5. Call `gitea_assess_worktree_cleanup_integrity` with before, after, and the removal log. +6. Fail closed when any protected path (active PR, dirty, claim/lease) disappears + without an explained state transition. +7. Final report must include the reconciliation table and `git worktree list` proof. diff --git a/tests/test_worktree_cleanup_audit.py b/tests/test_worktree_cleanup_audit.py new file mode 100644 index 0000000..3fccfee --- /dev/null +++ b/tests/test_worktree_cleanup_audit.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python3 +"""Regression tests for worktree cleanup audit integrity (#404).""" + +from __future__ import annotations + +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from worktree_cleanup_audit import ( # noqa: E402 + assess_cleanup_audit_final_report, + assess_cleanup_audit_integrity, + assess_worktree_cleanup_integrity, + classify_branches_entry, + parse_worktree_list_porcelain, + reconcile_cleanup_audit, +) + + +def _entry( + path: str, + *, + classification: str, + registered: bool = True, + preserve: bool | None = None, +) -> dict: + preserve_flag = preserve if preserve is not None else classification in { + "active_open_pr", + "active_issue_work", + "dirty_local_worktree", + "unsafe_unknown", + } + return { + "path": path, + "classification": classification, + "preserve": preserve_flag, + "registered_worktree": registered, + "worktree_state": {"exists": True, "clean": classification == "clean_stale_removable"}, + "worktree_record": {"branch": "feat/x"} if registered else None, + } + + +def _snapshot(entries: list[dict]) -> dict: + return {"entries": entries} + + +class TestParseWorktreePorcelain(unittest.TestCase): + def test_parses_multiple_worktrees(self): + text = "\n".join([ + "worktree /proj/branches/foo", + "HEAD abcdef0123456789abcdef0123456789abcdef0", + "branch refs/heads/feat/foo", + "", + "worktree /proj", + "HEAD 1111111111111111111111111111111111111111", + "branch refs/heads/master", + ]) + parsed = parse_worktree_list_porcelain(text) + self.assertEqual(len(parsed), 2) + self.assertEqual(parsed[0]["branch"], "feat/foo") + + +class TestClassifyEntry(unittest.TestCase): + def test_active_pr_classification(self): + result = classify_branches_entry( + rel_path="branches/feat-issue-1-x", + worktree_record={"branch": "feat/issue-1-x"}, + worktree_state={"exists": True, "clean": True, "dirty_files": []}, + open_pr_branches={"feat/issue-1-x"}, + ) + self.assertEqual(result, "active_open_pr") + + def test_dirty_classification(self): + result = classify_branches_entry( + rel_path="branches/dirty-one", + worktree_record={"branch": "feat/dirty-one"}, + worktree_state={"exists": True, "dirty_files": ["a.py"]}, + open_pr_branches=set(), + ) + self.assertEqual(result, "dirty_local_worktree") + + +class TestCleanupIntegrity(unittest.TestCase): + def test_preserved_worktree_remains(self): + path = "branches/keep-me" + before = _snapshot([_entry(path, classification="clean_stale_removable", preserve=False)]) + after = _snapshot([_entry(path, classification="clean_stale_removable", preserve=False)]) + result = assess_worktree_cleanup_integrity(before=before, after=after) + self.assertTrue(result["integrity_passed"]) + + def test_intentional_removal_passes(self): + path = "branches/remove-me" + before = _snapshot([_entry(path, classification="clean_stale_removable", preserve=False)]) + after = _snapshot([]) + result = assess_worktree_cleanup_integrity( + before=before, + after=after, + removals=[{ + "path": path, + "method": "git worktree remove", + "pre_removal_proof": "clean status", + }], + ) + self.assertTrue(result["integrity_passed"]) + + def test_dirty_worktree_disappears_fails(self): + path = "branches/dirty-wt" + before = _snapshot([_entry(path, classification="dirty_local_worktree")]) + after = _snapshot([]) + recon = reconcile_cleanup_audit(before, after) + result = assess_cleanup_audit_integrity(recon) + self.assertFalse(result["proven"]) + + def test_active_pr_worktree_disappears_fails(self): + path = "branches/review-pr99" + before = _snapshot([_entry(path, classification="active_open_pr")]) + after = _snapshot([]) + result = assess_worktree_cleanup_integrity(before=before, after=after) + self.assertFalse(result["integrity_passed"]) + + def test_explained_missing_allowed(self): + path = "branches/review-pr382" + before = _snapshot([ + _entry(path, classification="detached_review_leftover", preserve=False), + ]) + after = _snapshot([]) + result = assess_worktree_cleanup_integrity( + before=before, + after=after, + explained_missing={path: "removed concurrently by sibling session"}, + ) + self.assertTrue(result["integrity_passed"]) + + def test_removal_log_omits_clean_stale_fails(self): + path = "branches/a" + before = _snapshot([_entry(path, classification="clean_stale_removable", preserve=False)]) + after = _snapshot([]) + recon = reconcile_cleanup_audit(before, after, removal_log=[]) + self.assertFalse(recon["removal_log_complete"]) + + +class TestCleanupReportProof(unittest.TestCase): + def test_complete_report_passes(self): + report = "\n".join([ + "Cleanup audit reconciliation table:", + "Initial count: 10", + "Removed count: 3", + "Preserved count: 7", + "Missing-unexplained count: 0", + "Final count: 7", + "Final verification: git worktree list proof attached", + ]) + result = assess_cleanup_audit_final_report(report) + self.assertTrue(result["proven"]) + + def test_incomplete_report_fails(self): + result = assess_cleanup_audit_final_report("removed some worktrees") + self.assertFalse(result["proven"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/worktree_cleanup_audit.py b/worktree_cleanup_audit.py new file mode 100644 index 0000000..f60dfe7 --- /dev/null +++ b/worktree_cleanup_audit.py @@ -0,0 +1,554 @@ +"""Worktree cleanup audit integrity and reconciliation (#404). + +Captures before/after snapshots of ``branches/`` directories and registered +worktrees, reconciles every initial path to exactly one disposition, and fails +closed when preserved worktrees disappear without an explicit removal record. +""" + +from __future__ import annotations + +import os +import re +from typing import Any + +from merged_cleanup_reconcile import branch_worktree_folder, read_local_worktree_state +from reviewer_worktree import parse_dirty_tracked_files + +CLASSIFICATIONS = frozenset({ + "active_open_pr", + "active_issue_work", + "dirty_local_worktree", + "clean_stale_removable", + "detached_review_leftover", + "orphan_directory", + "unsafe_unknown", +}) + +PRESERVE_CLASSIFICATIONS = frozenset({ + "active_open_pr", + "active_issue_work", + "dirty_local_worktree", + "unsafe_unknown", +}) + +DISPOSITIONS = frozenset({ + "removed_intentionally", + "preserved_exists", + "preserved_missing_explained", + "not_registered_worktree", + "unsafe_unknown", +}) + +REVIEW_WORKTREE_RE = re.compile( + r"branches/(?:review-pr\d+|merge-simulation-pr\d+|review-[\w-]+)", + re.IGNORECASE, +) + + +def normalize_path(path: str) -> str: + return os.path.normpath((path or "").strip()) + + +def relative_branches_path(project_root: str, path: str) -> str: + root = normalize_path(project_root) + normalized = normalize_path(path) + if normalized.startswith(root + os.sep): + return normalized[len(root) + 1 :] + return normalized.replace("\\", "/") + + +def parse_worktree_list_porcelain(porcelain: str) -> list[dict[str, Any]]: + """Parse ``git worktree list --porcelain`` into worktree records.""" + entries: list[dict[str, Any]] = [] + current: dict[str, Any] = {} + for raw in (porcelain or "").splitlines(): + line = raw.strip() + if not line: + if current: + entries.append(current) + current = {} + continue + if line.startswith("worktree "): + if current: + entries.append(current) + current = {"path": line.split(" ", 1)[1].strip()} + elif line.startswith("HEAD "): + current["head_sha"] = line.split(" ", 1)[1].strip() + elif line.startswith("branch "): + current["branch"] = line.split(" ", 1)[1].strip().removeprefix("refs/heads/") + elif line == "detached": + current["detached"] = True + elif line == "bare": + current["bare"] = True + if current: + entries.append(current) + return entries + + +def list_branches_directories(project_root: str, dir_names: list[str] | None = None) -> list[str]: + """Return relative ``branches/`` paths for first-level directories.""" + branches_root = os.path.join(project_root, "branches") + if dir_names is not None: + return sorted( + f"branches/{name}" + for name in dir_names + if name and not name.startswith(".") + ) + if not os.path.isdir(branches_root): + return [] + names: list[str] = [] + for entry in sorted(os.listdir(branches_root)): + full = os.path.join(branches_root, entry) + if entry.startswith(".") or not os.path.isdir(full): + continue + names.append(f"branches/{entry}") + return names + + +def _untracked_dirty(porcelain: str) -> bool: + return any(line.startswith("??") for line in (porcelain or "").splitlines()) + + +def classify_branches_entry( + *, + rel_path: str, + worktree_record: dict[str, Any] | None, + worktree_state: dict[str, Any] | None, + open_pr_branches: set[str] | None = None, + active_lock_branches: set[str] | None = None, + active_issue_branches: set[str] | None = None, +) -> str: + """Classify a ``branches/`` directory for cleanup policy.""" + open_pr_branches = open_pr_branches or set() + active_lock_branches = active_lock_branches or set() + active_issue_branches = active_issue_branches or set() + state = worktree_state or {} + record = worktree_record or {} + + branch_name = (record.get("branch") or "").strip() + folder_name = rel_path.split("/", 1)[-1] if "/" in rel_path else rel_path + inferred_branch = folder_name.replace("-", "/") if "/" not in folder_name else folder_name + + candidate_branches = {b for b in (branch_name, inferred_branch) if b} + open_folder_names = {branch_worktree_folder(b) for b in open_pr_branches} + if folder_name in open_folder_names or any( + b in open_pr_branches for b in candidate_branches + ): + return "active_open_pr" + if any(b in active_lock_branches or b in active_issue_branches for b in candidate_branches): + return "active_issue_work" + + dirty_tracked = bool(state.get("dirty_files")) + dirty_untracked = bool(state.get("dirty_untracked")) + if dirty_tracked or dirty_untracked: + return "dirty_local_worktree" + + if not record: + return "orphan_directory" + + if record.get("detached") and REVIEW_WORKTREE_RE.search(rel_path): + return "detached_review_leftover" + + if state.get("exists") and state.get("clean"): + return "clean_stale_removable" + + return "unsafe_unknown" + + +def capture_cleanup_snapshot( + project_root: str, + *, + branch_dirs: list[str] | None = None, + worktree_porcelain: str | None = None, + open_pr_branches: set[str] | None = None, + active_lock_branches: set[str] | None = None, + active_issue_branches: set[str] | None = None, + issue_lock_path: str | None = None, +) -> dict[str, Any]: + """Capture audit snapshot for ``branches/`` dirs and registered worktrees.""" + root = normalize_path(project_root) + rel_dirs = list_branches_directories(root, branch_dirs) + worktrees = parse_worktree_list_porcelain(worktree_porcelain or "") + worktree_by_rel: dict[str, dict[str, Any]] = {} + for wt in worktrees: + rel = relative_branches_path(root, wt.get("path") or "") + if rel.startswith("branches/"): + worktree_by_rel[rel] = wt + + lock_branches = set(active_lock_branches or []) + lock = None + if issue_lock_path: + from merged_cleanup_reconcile import read_issue_lock + + lock = read_issue_lock(issue_lock_path) + if lock and lock.get("branch_name"): + lock_branches.add(str(lock["branch_name"])) + + entries: list[dict[str, Any]] = [] + for rel_path in rel_dirs: + abs_path = os.path.join(root, rel_path) + wt_record = worktree_by_rel.get(rel_path) + state = read_local_worktree_state(abs_path) if os.path.isdir(abs_path) else { + "exists": False, + "clean": None, + "dirty_files": [], + } + if state.get("exists"): + status_res = state.get("porcelain_status") + if status_res is None and os.path.isdir(abs_path): + import subprocess + + proc = subprocess.run( + ["git", "-C", abs_path, "status", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + status_res = proc.stdout or "" + state["dirty_untracked"] = _untracked_dirty(status_res or "") + state["dirty_files"] = state.get("dirty_files") or parse_dirty_tracked_files( + status_res or "" + ) + state["clean"] = not state["dirty_files"] and not state["dirty_untracked"] + + classification = classify_branches_entry( + rel_path=rel_path, + worktree_record=wt_record, + worktree_state=state, + open_pr_branches=open_pr_branches, + active_lock_branches=lock_branches, + active_issue_branches=active_issue_branches, + ) + entries.append( + { + "path": rel_path, + "absolute_path": abs_path, + "classification": classification, + "registered_worktree": bool(wt_record), + "worktree_record": wt_record or None, + "worktree_state": state, + "preserve": classification in PRESERVE_CLASSIFICATIONS, + } + ) + + return { + "project_root": root, + "branch_directory_count": len(rel_dirs), + "registered_branches_worktree_count": len(worktree_by_rel), + "entries": entries, + "worktrees": worktrees, + } + + +def _index_snapshot_entries(snapshot: dict[str, Any]) -> dict[str, dict[str, Any]]: + return {entry["path"]: entry for entry in snapshot.get("entries") or []} + + +def _removal_paths(removal_log: list[dict[str, Any]] | None) -> dict[str, dict[str, Any]]: + indexed: dict[str, dict[str, Any]] = {} + for item in removal_log or []: + rel = (item.get("path") or "").strip().replace("\\", "/") + if rel: + indexed[rel] = item + return indexed + + +def reconcile_cleanup_audit( + before: dict[str, Any], + after: dict[str, Any], + removal_log: list[dict[str, Any]] | None = None, + *, + explained_missing: dict[str, str] | None = None, + concurrent_mutations: list[str] | None = None, +) -> dict[str, Any]: + """Reconcile before/after snapshots to exactly one disposition per path.""" + before_index = _index_snapshot_entries(before) + after_index = _index_snapshot_entries(after) + removals = _removal_paths(removal_log) + explained = { + (k or "").strip().replace("\\", "/"): (v or "").strip() + for k, v in (explained_missing or {}).items() + } + concurrent = { + (p or "").strip().replace("\\", "/") + for p in (concurrent_mutations or []) + } + + rows: list[dict[str, Any]] = [] + for path, before_entry in sorted(before_index.items()): + after_entry = after_index.get(path) + after_exists = bool(after_entry and after_entry.get("worktree_state", {}).get("exists")) + classification = before_entry.get("classification") or "unsafe_unknown" + preserve = bool(before_entry.get("preserve")) or classification in PRESERVE_CLASSIFICATIONS + removal = removals.get(path) + explanation = explained.get(path, "") + + if removal: + disposition = "removed_intentionally" + reasons = [] + elif after_exists: + disposition = "preserved_exists" + reasons = [] + elif explanation: + disposition = "preserved_missing_explained" + reasons = [explanation] + elif not before_entry.get("registered_worktree"): + disposition = "not_registered_worktree" + reasons = ["directory was not a registered git worktree at audit start"] + elif path in concurrent: + disposition = "preserved_missing_explained" + reasons = ["removed or mutated by another session during cleanup"] + elif preserve: + disposition = "unsafe_unknown" + reasons = [ + f"preserved classification '{classification}' disappeared without " + "removal log or explanation" + ] + else: + disposition = "unsafe_unknown" + reasons = [ + "clean/removable path disappeared without removal log entry" + ] + + rows.append( + { + "path": path, + "classification": classification, + "preserve": preserve, + "disposition": disposition, + "removed_intentionally": disposition == "removed_intentionally", + "removal_record": removal, + "after_exists": after_exists, + "reasons": reasons, + } + ) + + for path, removal in removals.items(): + if path not in before_index: + rows.append( + { + "path": path, + "classification": "unsafe_unknown", + "preserve": False, + "disposition": "unsafe_unknown", + "removed_intentionally": True, + "removal_record": removal, + "after_exists": path in after_index, + "reasons": ["removal log references path absent from before snapshot"], + } + ) + + counts = { + "initial_count": len(before_index), + "removed_count": sum(1 for r in rows if r["disposition"] == "removed_intentionally"), + "preserved_count": sum(1 for r in rows if r["disposition"] == "preserved_exists"), + "missing_unexplained_count": sum( + 1 + for r in rows + if r["disposition"] == "unsafe_unknown" + and r.get("preserve") + and not r.get("after_exists") + ), + "missing_explained_count": sum( + 1 for r in rows if r["disposition"] == "preserved_missing_explained" + ), + "final_count": len(after_index), + "orphan_directory_count": sum( + 1 for r in rows if r["disposition"] == "not_registered_worktree" + ), + } + expected_final = ( + counts["initial_count"] + - counts["removed_count"] + - counts["missing_explained_count"] + ) + counts["count_reconciles"] = counts["final_count"] == expected_final + + return { + "rows": rows, + "counts": counts, + "removal_log_complete": _removal_log_complete(before_index, after_index, removals), + } + + +def _removal_log_complete( + before_index: dict[str, dict[str, Any]], + after_index: dict[str, dict[str, Any]], + removals: dict[str, dict[str, Any]], +) -> bool: + for path, before_entry in before_index.items(): + if path in after_index: + continue + classification = before_entry.get("classification") or "" + if classification == "clean_stale_removable" and path not in removals: + return False + return True + + +def assess_cleanup_audit_integrity(reconciliation: dict[str, Any]) -> dict[str, Any]: + """Fail closed when preserved worktrees vanish or counts do not reconcile.""" + reasons: list[str] = [] + counts = reconciliation.get("counts") or {} + + if counts.get("missing_unexplained_count"): + reasons.append( + f"{counts['missing_unexplained_count']} preserved worktree(s) missing " + "without explanation" + ) + + for row in reconciliation.get("rows") or []: + if not row.get("preserve") or row.get("after_exists"): + continue + if row.get("disposition") == "removed_intentionally": + continue + message = ( + f"preserved worktree {row.get('path')} missing " + f"({row.get('disposition')})" + ) + if message not in reasons: + reasons.append(message) + for item in row.get("reasons") or []: + if item not in reasons: + reasons.append(item) + + if not counts.get("count_reconciles"): + reasons.append( + "final directory count does not reconcile with initial minus removed " + f"(initial={counts.get('initial_count')}, removed={counts.get('removed_count')}, " + f"final={counts.get('final_count')})" + ) + + if reconciliation.get("removal_log_complete") is False: + reasons.append("removal log omits one or more removed clean-stale worktrees") + + for row in reconciliation.get("rows") or []: + removal = row.get("removal_record") or {} + if row.get("disposition") == "removed_intentionally": + if not removal.get("method"): + reasons.append(f"removal log for {row.get('path')} missing method") + if not removal.get("pre_removal_proof"): + reasons.append(f"removal log for {row.get('path')} missing pre-removal proof") + + block = bool(reasons) + return { + "block": block, + "proven": not block, + "reasons": reasons, + "counts": counts, + "safe_next_action": ( + "capture before/after snapshots, record every removal with proof, and " + "explain any preserved path that disappears" + if block + else "proceed" + ), + } + + +def build_cleanup_reconciliation_table(reconciliation: dict[str, Any]) -> dict[str, Any]: + """Return the operator-facing reconciliation summary table.""" + counts = dict(reconciliation.get("counts") or {}) + return { + "initial_count": counts.get("initial_count", 0), + "removed_count": counts.get("removed_count", 0), + "preserved_count": counts.get("preserved_count", 0), + "missing_unexplained_count": counts.get("missing_unexplained_count", 0), + "missing_explained_count": counts.get("missing_explained_count", 0), + "final_count": counts.get("final_count", 0), + "count_reconciles": counts.get("count_reconciles", False), + } + + +def _read_worktree_porcelain(project_root: str) -> str: + import subprocess + + try: + res = subprocess.run( + ["git", "-C", project_root, "worktree", "list", "--porcelain"], + capture_output=True, + text=True, + check=False, + ) + except OSError: + return "" + return res.stdout if res.returncode == 0 else "" + + +def capture_branches_worktree_snapshot( + project_root: str, + *, + open_pr_branches: list[str] | None = None, + active_lock_branch: str | None = None, + leased_paths: list[str] | None = None, + issue_lock_path: str | None = None, +) -> dict[str, Any]: + """MCP-facing snapshot capture for live ``branches/`` cleanup audits.""" + root = normalize_path(project_root) + lock_branches = set() + if active_lock_branch: + lock_branches.add(active_lock_branch) + issue_branches = set() + for path in leased_paths or []: + rel = relative_branches_path(root, path) + if rel.startswith("branches/"): + issue_branches.add(rel.split("/", 1)[-1].replace("-", "/")) + return capture_cleanup_snapshot( + root, + worktree_porcelain=_read_worktree_porcelain(root), + open_pr_branches=set(open_pr_branches or []), + active_lock_branches=lock_branches, + active_issue_branches=issue_branches, + issue_lock_path=issue_lock_path, + ) + + +def assess_worktree_cleanup_integrity( + *, + before: dict[str, Any], + after: dict[str, Any], + removals: list[dict[str, Any]] | None = None, + explained_missing: dict[str, str] | None = None, +) -> dict[str, Any]: + """MCP-facing integrity assessment over before/after cleanup snapshots.""" + reconciliation = reconcile_cleanup_audit( + before, + after, + removals, + explained_missing=explained_missing, + ) + integrity = assess_cleanup_audit_integrity(reconciliation) + return { + **integrity, + "integrity_passed": integrity.get("proven", False), + "reconciliation": reconciliation, + "reconciliation_table": build_cleanup_reconciliation_table(reconciliation), + "rows": reconciliation.get("rows") or [], + } + + +_RECON_INITIAL_RE = re.compile(r"initial count\s*:\s*(\d+)", re.I) +_RECON_REMOVED_RE = re.compile(r"removed count\s*:\s*(\d+)", re.I) +_RECON_PRESERVED_RE = re.compile(r"preserved count\s*:\s*(\d+)", re.I) +_RECON_MISSING_RE = re.compile(r"missing-unexplained count\s*:\s*(\d+)", re.I) +_RECON_FINAL_RE = re.compile(r"final count\s*:\s*(\d+)", re.I) +_WORKTREE_LIST_RE = re.compile(r"git worktree list|worktree list proof", re.I) + + +def assess_cleanup_audit_final_report(report_text: str) -> dict[str, Any]: + """Validate cleanup final report includes reconciliation proof (#404).""" + text = report_text or "" + reasons: list[str] = [] + for pattern in ( + _RECON_INITIAL_RE, + _RECON_REMOVED_RE, + _RECON_PRESERVED_RE, + _RECON_MISSING_RE, + _RECON_FINAL_RE, + ): + if not pattern.search(text): + reasons.append( + f"cleanup report missing field matching /{pattern.pattern}/" + ) + if not _WORKTREE_LIST_RE.search(text): + reasons.append("final verification missing git worktree list proof") + proven = not reasons + return {"proven": proven, "block": not proven, "reasons": reasons} \ No newline at end of file