Gitea-Tools/worktree_cleanup_audit.py

"""Session-owned worktree cleanup audit, TTL enforcement, and integrity (#401, #404).

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/``.
* ``capture_cleanup_snapshot`` / ``reconcile_cleanup_audit`` — before/after
  reconciliation for bulk cleanup audits (#404).

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

from merged_cleanup_reconcile import (
    branch_worktree_folder,
    is_head_ancestor_of_ref,
    read_local_worktree_state,
)
from reviewer_worktree import parse_dirty_tracked_files, REVIEW_WORKTREE_RE

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

# Merged-PR linkage outcomes for issue worktrees (#858). Only ``LINKAGE_MERGED``
# is ownership proof; every other outcome leaves the worktree protected.
LINKAGE_MERGED = "merged_pr"
LINKAGE_OPEN = "open_pr"
LINKAGE_NONE = "no_owning_pr"
LINKAGE_AMBIGUOUS = "ambiguous"
LINKAGE_UNKNOWN = "unknown"

_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 build_pr_index(prs: list[dict[str, Any]] | None) -> dict[str, list[dict[str, Any]]]:
    """Index PR records by head branch for deterministic worktree linkage (#858).

    Accepts Gitea PR payloads (``head`` as a dict) and pre-flattened records
    (``head_branch``/``head_sha``). Records without a usable head branch or
    number are dropped rather than guessed at, so a branch is only ever linked
    to a PR the caller actually proved.
    """
    index: dict[str, list[dict[str, Any]]] = {}
    for pr in prs or []:
        head = pr.get("head")
        if isinstance(head, dict):
            head_branch = head.get("ref")
            head_sha = head.get("sha")
        else:
            head_branch = pr.get("head_branch") or (head if isinstance(head, str) else None)
            head_sha = pr.get("head_sha")
        number = pr.get("number")
        if not head_branch or number is None:
            continue
        try:
            pr_number = int(number)
        except (TypeError, ValueError):
            continue
        index.setdefault(str(head_branch).strip(), []).append(
            {
                "pr_number": pr_number,
                "head_branch": str(head_branch).strip(),
                "head_sha": head_sha,
                "merged": bool(pr.get("merged") or pr.get("merged_at")),
                "state": pr.get("state"),
            }
        )
    return index


def resolve_owning_pr(
    *,
    branch: str | None,
    pr_index: dict[str, list[dict[str, Any]]] | None,
) -> dict[str, Any]:
    """Resolve the single PR that owns ``branch``, failing closed when unclear.

    Ownership is only ``LINKAGE_MERGED`` when exactly one PR claims the branch
    and that PR is merged. Several distinct PRs on one branch is a competing
    claim (``LINKAGE_AMBIGUOUS``), and a still-open owner is reported as
    ``LINKAGE_OPEN`` — both keep the worktree protected while still exposing
    the PR number the audit resolved.
    """
    if pr_index is None:
        return {
            "status": LINKAGE_UNKNOWN,
            "pr_number": None,
            "candidate_pr_numbers": [],
            "reasons": ["live PR state was not supplied; ownership unproven"],
        }
    branch_name = (branch or "").strip()
    if not branch_name:
        return {
            "status": LINKAGE_UNKNOWN,
            "pr_number": None,
            "candidate_pr_numbers": [],
            "reasons": ["worktree has no attached branch; ownership unproven"],
        }

    candidates = list(pr_index.get(branch_name) or [])
    numbers = sorted({c["pr_number"] for c in candidates})
    if not candidates:
        return {
            "status": LINKAGE_NONE,
            "pr_number": None,
            "candidate_pr_numbers": [],
            "reasons": [f"no PR claims branch '{branch_name}'"],
        }
    if len(numbers) > 1:
        return {
            "status": LINKAGE_AMBIGUOUS,
            "pr_number": None,
            "candidate_pr_numbers": numbers,
            "reasons": [
                f"branch '{branch_name}' is claimed by competing PRs {numbers}; "
                "ownership is ambiguous"
            ],
        }

    owner = candidates[0]
    pr_number = owner["pr_number"]
    if owner.get("head_branch") != branch_name:
        return {
            "status": LINKAGE_UNKNOWN,
            "pr_number": pr_number,
            "candidate_pr_numbers": numbers,
            "reasons": [
                f"PR #{pr_number} head branch '{owner.get('head_branch')}' does not "
                f"match worktree branch '{branch_name}'"
            ],
        }
    if not owner.get("merged"):
        return {
            "status": LINKAGE_OPEN,
            "pr_number": pr_number,
            "candidate_pr_numbers": numbers,
            "pr_head_sha": owner.get("head_sha"),
            "reasons": [f"owning PR #{pr_number} is not merged"],
        }
    return {
        "status": LINKAGE_MERGED,
        "pr_number": pr_number,
        "candidate_pr_numbers": numbers,
        "pr_head_sha": owner.get("head_sha"),
        "reasons": [],
    }


def assess_merged_pr_worktree_cleanup(
    *,
    linkage: dict[str, Any] | None,
    head_sha: str | None,
    head_in_master: bool | None,
    is_dirty: bool,
    has_open_pr: bool,
    has_active_lease: bool,
    has_active_issue_lock: bool,
    is_protected: bool,
    has_live_session: bool = False,
) -> dict[str, Any]:
    """Decide whether a merged issue worktree satisfies the full cleanup policy.

    Every condition must be independently proven: conclusive merged-PR
    ownership, agreement between the worktree branch and the PR head branch,
    containment of the worktree head in authoritative master (which is what
    proves no unmerged commits remain), absence of any open/competing PR,
    lease, issue lock, or live session, a clean tree, and a worktree that is
    not the protected control checkout. Anything unknown blocks.
    """
    link = linkage or {
        "status": LINKAGE_UNKNOWN,
        "pr_number": None,
        "reasons": ["no linkage assessment supplied"],
    }
    status = link.get("status")
    reasons: list[str] = []

    if status != LINKAGE_MERGED:
        reasons.extend(
            link.get("reasons") or ["owning PR could not be conclusively identified"]
        )
    if is_protected:
        reasons.append("worktree is protected or the stable control checkout")
    if is_dirty:
        reasons.append("worktree has uncommitted changes")
    if has_open_pr:
        reasons.append("worktree branch has an open PR")
    if has_active_lease:
        reasons.append("worktree has an active lease")
    if has_active_issue_lock:
        reasons.append("an active issue lock references this branch")
    if has_live_session:
        reasons.append("a live process or session is using this worktree")
    if not head_sha:
        reasons.append("worktree head sha is unknown")
    if head_in_master is None:
        reasons.append("containment of the worktree head in master is unknown")
    elif not head_in_master:
        reasons.append(
            "worktree head is not contained in authoritative master "
            "(unmerged commits remain)"
        )

    proven = not reasons
    return {
        "linkage_status": status,
        "pr_number": link.get("pr_number"),
        "pr_head_sha": link.get("pr_head_sha"),
        "head_in_master": head_in_master,
        "proven": proven,
        "block_reasons": reasons,
    }


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,
    merged_pr_cleanup: dict[str, Any] | None = None,
    has_live_session: bool = False,
) -> 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 has_live_session:
        return CLASS_ACTIVE_ISSUE_WORK  # a live session still owns this tree
    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
    if workflow_type == WORKFLOW_ISSUE_WORK:
        # #858: an issue worktree becomes removable only on authoritative
        # merged-PR evidence satisfying the whole cleanup policy. Age alone
        # never proves the branch landed, so TTL cannot qualify one by itself
        # — otherwise a worktree holding unmerged commits would be reclaimed.
        if (merged_pr_cleanup or {}).get("proven"):
            return CLASS_CLEAN_STALE_REMOVABLE
        return CLASS_ACTIVE_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 head_contained_in_ref(
    project_root: str, head_sha: str | None, ref: str | None
) -> bool | None:
    """Return True when ``head_sha`` is already contained in ``ref``.

    Shares :mod:`merged_cleanup_reconcile`'s ancestry check so the audit and
    the PR-scoped reconciler agree on what "already landed" means (#858).
    Returns None when containment cannot be determined, which fails closed.
    """
    if not head_sha or not ref:
        return None
    return is_head_ancestor_of_ref(project_root, head_sha, ref)


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,
    pr_index: dict[str, list[dict[str, Any]]] | None = None,
    leased_issue_numbers: set[int] | None = None,
    live_session_paths: set[str] | None = None,
    master_ref: str | None = None,
) -> 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.

    ``pr_index`` (see :func:`build_pr_index`) supplies the authoritative PR
    ownership used to link issue worktrees to their merged PR (#858).
    ``master_ref`` is the ref a worktree head must be contained in before it
    can be considered landed. Both are optional and their absence only ever
    fails closed: without them no issue worktree becomes removable.
    """
    open_pr_branches = open_pr_branches or set()
    leased_branches = leased_branches or set()
    active_issue_branches = active_issue_branches or set()
    leased_issue_numbers = leased_issue_numbers or set()
    live_session_paths = {
        os.path.abspath(p) for p in (live_session_paths or set()) if p
    }

    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"))
        head_sha = entry.get("head")
        linkage = resolve_owning_pr(branch=branch, pr_index=pr_index)
        metadata = build_worktree_metadata(
            path=path,
            branch=branch,
            head_sha=head_sha,
            pr_number=linkage.get("pr_number"),
        )
        has_open_pr = bool(branch) and branch in open_pr_branches
        # A lease on issue N protects that issue's own work worktree. It must
        # not incidentally protect a baseline/review scratch tree that merely
        # carries the same issue marker in its name, which would change the
        # classification of worktrees this policy does not own.
        has_active_lease = (bool(branch) and branch in leased_branches) or (
            metadata["workflow_type"] == WORKFLOW_ISSUE_WORK
            and metadata.get("issue_number") is not None
            and metadata["issue_number"] in leased_issue_numbers
        )
        has_active_lock = bool(branch) and branch in active_issue_branches
        has_live_session = bool(path) and os.path.abspath(path) in live_session_paths
        head_in_master = (
            head_contained_in_ref(project_root, head_sha, master_ref)
            if master_ref
            else None
        )
        merged_pr_cleanup = assess_merged_pr_worktree_cleanup(
            linkage=linkage,
            head_sha=head_sha,
            head_in_master=head_in_master,
            is_dirty=is_dirty,
            has_open_pr=has_open_pr,
            has_active_lease=has_active_lease,
            has_active_issue_lock=has_active_lock,
            is_protected=is_protected,
            has_live_session=has_live_session,
        )
        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,
            merged_pr_cleanup=merged_pr_cleanup,
            has_live_session=has_live_session,
        )
        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,
                "has_live_session": has_live_session,
                "is_protected": is_protected,
                "merged_pr_linkage": linkage,
                "merged_pr_cleanup": merged_pr_cleanup,
                "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),
    }
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",
})





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}