"""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}