307 lines
10 KiB
Python
307 lines
10 KiB
Python
"""Local branches/ worktree hygiene scan for the web UI (#432)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from typing import Any, Callable
|
|
|
|
from issue_lock_provenance import ISSUE_LOCK_FILE
|
|
from merged_cleanup_reconcile import (
|
|
branch_worktree_folder,
|
|
read_issue_lock,
|
|
read_local_worktree_state,
|
|
)
|
|
|
|
_REVIEW_WORKTREE_RE = re.compile(
|
|
r"branches/(?:review-pr\d+|merge-simulation-pr\d+|review-[\w-]+)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
CLASSIFICATIONS = frozenset({
|
|
"active-pr",
|
|
"active-issue",
|
|
"dirty",
|
|
"stale-clean",
|
|
"detached-review",
|
|
"unsafe-unknown",
|
|
"orphan",
|
|
})
|
|
|
|
CLEANUP_PROMPT = (
|
|
"Task: clean up branch/worktree for PR #<pr> / issue #<n> after merge. "
|
|
"Confirm the merge on remote master before any deletion; never "
|
|
"force-remove a dirty worktree. Full steps: "
|
|
"skills/llm-project-workflow/templates/worktree-cleanup.md"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WorktreeEntry:
|
|
rel_path: str
|
|
folder_name: str
|
|
classification: str
|
|
branch: str | None
|
|
head_sha: str | None
|
|
dirty_tracked: int
|
|
dirty_untracked: bool
|
|
detached: bool
|
|
registered_worktree: bool
|
|
notes: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HygieneSnapshot:
|
|
repo_root: str
|
|
entries: tuple[WorktreeEntry, ...]
|
|
anomalies: tuple[str, ...]
|
|
cleanup_prompt: str
|
|
scan_error: str | None = None
|
|
|
|
|
|
def _repo_root() -> str:
|
|
override = (os.environ.get("WEBUI_REPO_ROOT") or "").strip()
|
|
if override:
|
|
return os.path.realpath(override)
|
|
return os.path.realpath(os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
|
|
def _relative_branches_path(project_root: str, path: str) -> str:
|
|
root = os.path.normpath(project_root)
|
|
normalized = os.path.normpath(path)
|
|
if normalized.startswith(root + os.sep):
|
|
return normalized[len(root) + 1 :].replace("\\", "/")
|
|
return normalized.replace("\\", "/")
|
|
|
|
|
|
def parse_worktree_list_porcelain(porcelain: str) -> list[dict[str, Any]]:
|
|
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
|
|
if current:
|
|
entries.append(current)
|
|
return entries
|
|
|
|
|
|
def list_branches_directories(project_root: str) -> list[str]:
|
|
branches_root = os.path.join(project_root, "branches")
|
|
if not os.path.isdir(branches_root):
|
|
return []
|
|
names: list[str] = []
|
|
for entry in sorted(os.listdir(branches_root)):
|
|
if entry.startswith("."):
|
|
continue
|
|
full = os.path.join(branches_root, entry)
|
|
if os.path.isdir(full):
|
|
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_entry(
|
|
*,
|
|
rel_path: str,
|
|
worktree_record: dict[str, Any] | None,
|
|
state: dict[str, Any],
|
|
open_pr_branches: set[str],
|
|
active_lock_branch: str | None,
|
|
) -> tuple[str, str]:
|
|
record = worktree_record or {}
|
|
branch_name = (record.get("branch") or state.get("current_branch") or "").strip()
|
|
folder_name = rel_path.split("/", 1)[-1] if "/" in rel_path else rel_path
|
|
|
|
if branch_name in open_pr_branches:
|
|
return "active-pr", "Head branch matches an open PR"
|
|
if active_lock_branch and branch_name == active_lock_branch:
|
|
return "active-issue", "Matches active issue lock branch"
|
|
|
|
dirty_tracked = len(state.get("dirty_files") or [])
|
|
dirty_untracked = bool(state.get("dirty_untracked"))
|
|
if dirty_tracked or dirty_untracked:
|
|
return "dirty", "Local tracked or untracked changes present"
|
|
|
|
if not record and not state.get("exists"):
|
|
return "orphan", "Directory missing on disk"
|
|
|
|
if not record:
|
|
return "orphan", "Directory exists but is not a registered git worktree"
|
|
|
|
if record.get("detached") and REVIEW_WORKTREE_RE.search(rel_path):
|
|
return "detached-review", "Detached HEAD in reviewer/simulation worktree"
|
|
|
|
if state.get("exists") and state.get("clean"):
|
|
return "stale-clean", "Registered worktree with clean working tree"
|
|
|
|
return "unsafe-unknown", "Could not classify safely — inspect manually before cleanup"
|
|
|
|
|
|
def _missing_preserved_anomalies(
|
|
project_root: str,
|
|
lock: dict[str, Any] | None,
|
|
entries_by_path: dict[str, WorktreeEntry],
|
|
) -> tuple[str, ...]:
|
|
anomalies: list[str] = []
|
|
if not lock:
|
|
return tuple(anomalies)
|
|
|
|
branch = (lock.get("branch_name") or "").strip()
|
|
issue_number = lock.get("issue_number")
|
|
worktree_path = (lock.get("worktree_path") or "").strip()
|
|
expected_rel = _relative_branches_path(project_root, worktree_path) if worktree_path else ""
|
|
if worktree_path and not os.path.isdir(worktree_path):
|
|
anomalies.append(
|
|
f"Missing preserved worktree for issue #{issue_number}: "
|
|
f"lock path {worktree_path!r} not found on disk (#404)"
|
|
)
|
|
elif expected_rel.startswith("branches/") and expected_rel not in entries_by_path:
|
|
anomalies.append(
|
|
f"Missing preserved worktree folder {expected_rel} for locked branch {branch!r}"
|
|
)
|
|
elif branch:
|
|
folder = f"branches/{branch_worktree_folder(branch)}"
|
|
entry = entries_by_path.get(folder)
|
|
if entry and entry.classification in {"orphan", "unsafe-unknown"}:
|
|
anomalies.append(
|
|
f"Locked branch {branch!r} maps to {folder} but classification is "
|
|
f"{entry.classification}"
|
|
)
|
|
return tuple(anomalies)
|
|
|
|
|
|
def load_hygiene_snapshot(
|
|
*,
|
|
project_root: str | None = None,
|
|
worktree_porcelain: str | None = None,
|
|
open_pr_branches: set[str] | None = None,
|
|
issue_lock_path: str | None = None,
|
|
run_git: Callable[[list[str]], subprocess.CompletedProcess[str]] | None = None,
|
|
) -> HygieneSnapshot:
|
|
root = project_root or _repo_root()
|
|
lock_path = issue_lock_path if issue_lock_path is not None else ISSUE_LOCK_FILE
|
|
lock = read_issue_lock(lock_path)
|
|
active_lock_branch = (lock.get("branch_name") or "").strip() if lock else None
|
|
|
|
git_run = run_git or (
|
|
lambda args: subprocess.run(args, capture_output=True, text=True, check=False)
|
|
)
|
|
|
|
scan_error: str | None = None
|
|
porcelain = worktree_porcelain
|
|
if porcelain is None:
|
|
result = git_run(["git", "-C", root, "worktree", "list", "--porcelain"])
|
|
if result.returncode != 0:
|
|
scan_error = (result.stderr or result.stdout or "git worktree list failed").strip()
|
|
porcelain = ""
|
|
else:
|
|
porcelain = result.stdout or ""
|
|
|
|
worktrees = parse_worktree_list_porcelain(porcelain)
|
|
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
|
|
|
|
open_heads = set(open_pr_branches or ())
|
|
if not open_heads:
|
|
try:
|
|
from webui.queue_loader import load_queue_snapshot
|
|
|
|
snapshot = load_queue_snapshot()
|
|
open_heads = {
|
|
item.extra.get("branch", "")
|
|
for item in snapshot.prs
|
|
if item.extra.get("branch")
|
|
}
|
|
except Exception:
|
|
open_heads = set()
|
|
|
|
entries: list[WorktreeEntry] = []
|
|
for rel_path in list_branches_directories(root):
|
|
abs_path = os.path.join(root, rel_path)
|
|
record = worktree_by_rel.get(rel_path)
|
|
state = read_local_worktree_state(abs_path)
|
|
if state.get("exists"):
|
|
status = git_run(["git", "-C", abs_path, "status", "--porcelain"])
|
|
state = {
|
|
**state,
|
|
"dirty_untracked": _untracked_dirty(status.stdout or ""),
|
|
}
|
|
classification, note = classify_entry(
|
|
rel_path=rel_path,
|
|
worktree_record=record,
|
|
state=state,
|
|
open_pr_branches=open_heads,
|
|
active_lock_branch=active_lock_branch,
|
|
)
|
|
entries.append(
|
|
WorktreeEntry(
|
|
rel_path=rel_path,
|
|
folder_name=rel_path.split("/", 1)[-1],
|
|
classification=classification,
|
|
branch=(record or {}).get("branch") or state.get("current_branch"),
|
|
head_sha=(record or {}).get("head_sha") or state.get("head_sha"),
|
|
dirty_tracked=len(state.get("dirty_files") or []),
|
|
dirty_untracked=bool(state.get("dirty_untracked")),
|
|
detached=bool((record or {}).get("detached")),
|
|
registered_worktree=record is not None,
|
|
notes=note,
|
|
)
|
|
)
|
|
|
|
entries_by_path = {entry.rel_path: entry for entry in entries}
|
|
anomalies = _missing_preserved_anomalies(root, lock, entries_by_path)
|
|
|
|
return HygieneSnapshot(
|
|
repo_root=root,
|
|
entries=tuple(entries),
|
|
anomalies=anomalies,
|
|
cleanup_prompt=CLEANUP_PROMPT,
|
|
scan_error=scan_error,
|
|
)
|
|
|
|
|
|
def snapshot_to_dict(snapshot: HygieneSnapshot) -> dict[str, Any]:
|
|
return {
|
|
"repo_root": snapshot.repo_root,
|
|
"count": len(snapshot.entries),
|
|
"cleanup_prompt": snapshot.cleanup_prompt,
|
|
"anomalies": list(snapshot.anomalies),
|
|
"scan_error": snapshot.scan_error,
|
|
"entries": [
|
|
{
|
|
"rel_path": entry.rel_path,
|
|
"folder_name": entry.folder_name,
|
|
"classification": entry.classification,
|
|
"branch": entry.branch,
|
|
"head_sha": entry.head_sha,
|
|
"dirty_tracked": entry.dirty_tracked,
|
|
"dirty_untracked": entry.dirty_untracked,
|
|
"detached": entry.detached,
|
|
"registered_worktree": entry.registered_worktree,
|
|
"notes": entry.notes,
|
|
}
|
|
for entry in snapshot.entries
|
|
],
|
|
} |