feat: add worktree hygiene dashboard to web UI (Closes #432) #453

Merged
sysadmin merged 5 commits from feat/issue-432-worktree-hygiene into master 2026-07-09 08:24:35 -05:00
6 changed files with 582 additions and 16 deletions
+13 -1
View File
@@ -47,6 +47,9 @@ Optional environment variables:
| `/api/runtime` | JSON runtime health export |
| `/audit` | Report audit paste + validator preview (#431) |
| `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) |
| `/worktrees` | Worktree hygiene dashboard (#432) |
| `/api/worktrees` | JSON worktree scan with classifications and anomalies |
| `/leases` | Stub — lease visibility (#433) |
| `/worktrees` | Stub — hygiene dashboard (#432) |
| `/leases` | Lease and collision visibility (#433) |
| `/api/leases` | JSON lease/collision export |
@@ -94,6 +97,15 @@ credentials. The UI surfaces pagination proof (returned count, pages fetched,
If credentials are missing or the fetch fails, the page shows an explicit error
instead of an empty queue (fail closed).
## Worktree hygiene (#432)
`/worktrees` scans local `branches/` directories and registered git worktrees.
Each entry is classified (`active-pr`, `active-issue`, `dirty`, `stale-clean`,
`detached-review`, `unsafe-unknown`, `orphan`). Missing preserved worktrees
referenced by the issue lock file are flagged as anomalies (#404). The page
includes a copy/paste canonical cleanup prompt only — no deletion actions.
Override scan root with `WEBUI_REPO_ROOT` (defaults to repository root).
## Lease visibility (#433)
`/leases` surfaces read-only lease and collision state: local issue lock file,
@@ -114,6 +126,6 @@ no tokens or MCP restart actions are exposed.
## Tests
```bash
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_audit.py -q
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_audit.py tests/test_webui_worktree_hygiene.py -q
pytest tests/test_webui_skeleton.py tests/test_webui_project_registry.py tests/test_webui_prompt_library.py tests/test_webui_queue_dashboard.py tests/test_webui_lease_visibility.py tests/test_webui_runtime_health.py -q
```
+9 -11
View File
@@ -34,13 +34,6 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertIn("Runtime health", response.text)
def test_route_stubs_render(self):
for path in ("/worktrees",):
with self.subTest(path=path):
response = self.client.get(path)
self.assertEqual(response.status_code, 200)
self.assertIn("child issue", response.text.lower())
def test_audit_is_implemented(self):
response = self.client.get("/audit")
self.assertEqual(response.status_code, 200)
@@ -56,14 +49,17 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertEqual(response.status_code, 200)
self.assertIn("Gitea-Tools", response.text)
def test_worktrees_is_implemented(self):
response = self.client.get("/worktrees")
self.assertEqual(response.status_code, 200)
self.assertIn("Worktree hygiene", response.text)
def test_leases_is_implemented(self):
response = self.client.get("/leases")
self.assertEqual(response.status_code, 200)
self.assertIn("Leases", response.text)
self.assertIn("Collision warnings", response.text)
def test_extra_stub_routes(self):
self.assertEqual(self.client.get("/worktrees").status_code, 200)
def test_post_is_rejected(self):
response = self.client.post("/health")
@@ -76,10 +72,12 @@ class TestWebuiSkeleton(unittest.TestCase):
self.assertIn("Live queue", response.text)
def test_nav_links_on_all_pages(self):
for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
paths = ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
hrefs = ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees", "/leases")
for path in paths:
with self.subTest(path=path):
text = self.client.get(path).text
for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/leases"):
for href in hrefs:
self.assertIn(f'href="{href}"', text)
+114
View File
@@ -0,0 +1,114 @@
"""Tests for web UI worktree hygiene dashboard (#432)."""
import json
import sys
import tempfile
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from starlette.testclient import TestClient
from webui.app import create_app
from webui.worktree_scanner import (
classify_entry,
load_hygiene_snapshot,
parse_worktree_list_porcelain,
snapshot_to_dict,
)
class TestWorktreeScanner(unittest.TestCase):
def test_parse_worktree_porcelain(self):
porcelain = (
"worktree /repo/branches/feat-issue-1-foo\n"
"HEAD abcdef0123456789\n"
"branch refs/heads/feat/issue-1-foo\n"
"\n"
)
entries = parse_worktree_list_porcelain(porcelain)
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0]["branch"], "feat/issue-1-foo")
def test_classify_active_pr(self):
classification, _ = classify_entry(
rel_path="branches/feat-issue-99-demo",
worktree_record={"branch": "feat/issue-99-demo"},
state={"exists": True, "clean": True, "dirty_files": []},
open_pr_branches={"feat/issue-99-demo"},
active_lock_branch=None,
)
self.assertEqual(classification, "active-pr")
def test_classify_dirty(self):
classification, _ = classify_entry(
rel_path="branches/feat-issue-2-bar",
worktree_record={"branch": "feat/issue-2-bar"},
state={"exists": True, "clean": False, "dirty_files": ["webui/app.py"]},
open_pr_branches=set(),
active_lock_branch=None,
)
self.assertEqual(classification, "dirty")
def test_missing_lock_worktree_anomaly(self):
with tempfile.TemporaryDirectory() as tmp:
branches = Path(tmp) / "branches" / "other-worktree"
branches.mkdir(parents=True)
lock_path = Path(tmp) / "lock.json"
lock_path.write_text(
json.dumps({
"issue_number": 432,
"branch_name": "feat/issue-432-worktree-hygiene",
"worktree_path": str(Path(tmp) / "branches" / "missing-tree"),
}),
encoding="utf-8",
)
snapshot = load_hygiene_snapshot(
project_root=tmp,
worktree_porcelain="",
open_pr_branches=set(),
issue_lock_path=str(lock_path),
)
self.assertTrue(snapshot.anomalies)
self.assertIn("Missing preserved worktree", snapshot.anomalies[0])
def test_snapshot_to_dict(self):
with tempfile.TemporaryDirectory() as tmp:
(Path(tmp) / "branches").mkdir()
snapshot = load_hygiene_snapshot(
project_root=tmp,
worktree_porcelain="",
open_pr_branches=set(),
issue_lock_path=str(Path(tmp) / "no-lock.json"),
)
data = snapshot_to_dict(snapshot)
self.assertIn("entries", data)
self.assertIn("cleanup_prompt", data)
class TestWorktreeRoutes(unittest.TestCase):
def setUp(self):
self.client = TestClient(create_app())
def test_worktrees_page_renders(self):
response = self.client.get("/worktrees")
self.assertEqual(response.status_code, 200)
self.assertIn("Worktree hygiene", response.text)
self.assertIn("Canonical cleanup prompt", response.text)
self.assertIn("Copy cleanup prompt", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_api_worktrees_json(self):
response = self.client.get("/api/worktrees")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertIn("entries", data)
self.assertIn("cleanup_prompt", data)
self.assertIn("anomalies", data)
def test_nav_includes_worktrees(self):
self.assertIn('href="/worktrees"', self.client.get("/").text)
if __name__ == "__main__":
unittest.main()
+9 -4
View File
@@ -22,6 +22,8 @@ from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_sn
from webui.lease_views import render_leases_page
from webui.queue_loader import load_queue_snapshot, snapshot_to_dict as queue_snapshot_to_dict
from webui.queue_views import render_queue_page
from webui.worktree_scanner import load_hygiene_snapshot, snapshot_to_dict as worktree_snapshot_to_dict
from webui.worktree_views import render_worktrees_page
from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runtime_snapshot_to_dict
from webui.runtime_views import render_runtime_page
@@ -181,10 +183,12 @@ async def api_audit(request: Request) -> JSONResponse:
async def worktrees(_request: Request) -> HTMLResponse:
return _stub_page(
"Worktrees",
"Worktree hygiene will summarize branches/ session folders and cleanup risk.",
)
snapshot = load_hygiene_snapshot()
return HTMLResponse(render_worktrees_page(snapshot))
async def api_worktrees(_request: Request) -> JSONResponse:
return JSONResponse(worktree_snapshot_to_dict(load_hygiene_snapshot()))
async def leases(_request: Request) -> HTMLResponse:
@@ -228,6 +232,7 @@ def create_app() -> Starlette:
Route("/audit", audit, methods=["GET", "POST"]),
Route("/api/audit", api_audit, methods=["GET", "POST"]),
Route("/worktrees", worktrees, methods=["GET"]),
Route("/api/worktrees", api_worktrees, methods=["GET"]),
Route("/leases", leases, methods=["GET"]),
Route("/api/leases", api_leases, methods=["GET"]),
],
+307
View File
@@ -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 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
],
}
+130
View File
@@ -0,0 +1,130 @@
"""HTML views for branches/ worktree hygiene dashboard (#432)."""
from __future__ import annotations
import html
from webui.layout import render_page
from webui.worktree_scanner import HygieneSnapshot, WorktreeEntry
def _escape(text: str) -> str:
return html.escape(text, quote=True)
def _badge(classification: str) -> str:
return (
f'<span class="badge badge-{_escape(classification)}">'
f"{_escape(classification)}</span>"
)
def _entry_row(entry: WorktreeEntry) -> str:
branch = entry.branch or ""
head = entry.head_sha[:12] if entry.head_sha else ""
dirty = (
f"{entry.dirty_tracked} tracked"
+ (" + untracked" if entry.dirty_untracked else "")
if entry.dirty_tracked or entry.dirty_untracked
else "clean"
)
return (
"<tr>"
f"<td><code>{_escape(entry.rel_path)}</code></td>"
f"<td>{_badge(entry.classification)}</td>"
f"<td><code>{_escape(branch)}</code></td>"
f"<td><code>{_escape(head)}</code></td>"
f"<td>{_escape(dirty)}</td>"
f"<td>{'yes' if entry.detached else 'no'}</td>"
f"<td class='muted'>{_escape(entry.notes)}</td>"
"</tr>"
)
WORKTREE_PAGE_SCRIPT = """
<script>
document.querySelectorAll('.copy-cleanup-btn').forEach((btn) => {
btn.addEventListener('click', async () => {
const node = document.getElementById('cleanup-prompt');
if (!node) return;
try {
await navigator.clipboard.writeText(node.textContent || '');
const prior = btn.textContent;
btn.textContent = 'Copied';
setTimeout(() => { btn.textContent = prior; }, 1200);
} catch (_err) {
btn.textContent = 'Copy failed';
}
});
});
</script>
"""
WORKTREE_PAGE_STYLES = """
<style>
table.worktrees td:nth-child(7) { font-size: 0.85rem; }
.anomaly {
margin: 1rem 0;
padding: 0.75rem 0.9rem;
border: 1px solid #7a3b3b;
border-radius: 8px;
background: rgba(122, 59, 59, 0.15);
color: #f0c4c4;
}
.badge-active-pr { color: #9ec8f0; border-color: #3d5f7a; }
.badge-active-issue { color: #8fd19e; border-color: #3d6b4a; }
.badge-dirty { color: #e0c27a; border-color: #6b5730; }
.badge-stale-clean { color: #c9b8e8; border-color: #5a4a78; }
.badge-detached-review { color: #9ec8f0; border-color: #3d5f7a; }
.badge-unsafe-unknown { color: #f0a8a8; border-color: #7a3b3b; }
.badge-orphan { color: #f0a8a8; border-color: #7a3b3b; }
</style>
"""
def render_worktrees_page(snapshot: HygieneSnapshot) -> str:
error_block = ""
if snapshot.scan_error:
error_block = (
'<div class="stub"><p><strong>Partial scan:</strong> '
f"{_escape(snapshot.scan_error)}</p></div>"
)
anomaly_block = ""
if snapshot.anomalies:
items = "".join(f"<li>{_escape(text)}</li>" for text in snapshot.anomalies)
anomaly_block = (
'<section class="anomaly"><h3>Missing preserved worktree anomalies</h3>'
f"<ul>{items}</ul></section>"
)
if snapshot.entries:
rows = "".join(_entry_row(entry) for entry in snapshot.entries)
table = (
"<table class='registry worktrees'>"
"<thead><tr>"
"<th>Path</th><th>Class</th><th>Branch</th><th>HEAD</th>"
"<th>Dirty</th><th>Detached</th><th>Notes</th>"
"</tr></thead>"
f"<tbody>{rows}</tbody></table>"
)
else:
table = "<p class='muted'>No directories under <code>branches/</code>.</p>"
body = (
"<h2>Worktree hygiene</h2>"
"<p>Read-only scan of local <code>branches/</code> worktrees. "
"No deletion actions — copy the canonical cleanup prompt when merge is confirmed.</p>"
f"{error_block}"
f"{anomaly_block}"
f"<p class='meta'>Repo root: <code>{_escape(snapshot.repo_root)}</code> · "
f"entries: {len(snapshot.entries)}</p>"
f"{table}"
"<h3>Canonical cleanup prompt</h3>"
f'<pre class="prompt-text" id="cleanup-prompt">{_escape(snapshot.cleanup_prompt)}</pre>'
'<button type="button" class="copy-btn copy-cleanup-btn">Copy cleanup prompt</button>'
"<p><a href=\"/api/worktrees\">JSON API</a></p>"
f"{WORKTREE_PAGE_STYLES}"
f"{WORKTREE_PAGE_SCRIPT}"
)
return render_page(title="Worktrees", body_html=body)