feat: add worktree hygiene dashboard to web UI (Closes #432)
Adds read-only /worktrees and /api/worktrees routes that scan branches/ directories, classify worktree state, flag missing preserved worktrees from the issue lock, and surface the canonical cleanup prompt without deletion. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,307 @@
|
||||
"""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 merged_cleanup_reconcile import (
|
||||
ISSUE_LOCK_FILE,
|
||||
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
|
||||
],
|
||||
}
|
||||
Reference in New Issue
Block a user