Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat: add worktree cleanup audit integrity (Closes #404)

Browse Source 
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)
This commit is contained in:
sysadmin
2026-07-07 11:50:29 -04:00
parent 1a1e679246
commit e4c0a51e5a
5 changed files with 834 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
final_report_validator.py
+22
View File
@@ -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,
gitea_mcp_server.py
+54
View File
@@ -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,
skills/llm-project-workflow/templates/worktree-cleanup.md
+40
View File
@@ -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.
tests/test_worktree_cleanup_audit.py
+164
View File
@@ -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()
worktree_cleanup_audit.py
+554
View File
@@ -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/<name>`` 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}