diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index 0a5b935..1873305 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -45,7 +45,8 @@ Optional environment variables: | `/api/prompts` | JSON prompt export with workflow hashes | | `/runtime` | Stub — MCP runtime health (#430) | | `/audit` | Stub — report audit paste (#431) | -| `/worktrees` | Stub — hygiene dashboard (#432) | +| `/worktrees` | Worktree hygiene dashboard (#432) | +| `/api/worktrees` | JSON worktree scan with classifications and anomalies | | `/leases` | Stub — lease visibility (#433) | All routes are GET-only. POST/PUT/PATCH/DELETE return `405` with @@ -82,8 +83,18 @@ 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). + ## 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 -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_worktree_hygiene.py -q ``` \ No newline at end of file diff --git a/tests/test_webui_skeleton.py b/tests/test_webui_skeleton.py index 5dd23a1..bd290e8 100644 --- a/tests/test_webui_skeleton.py +++ b/tests/test_webui_skeleton.py @@ -46,10 +46,13 @@ 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_extra_stub_routes(self): - for path in ("/worktrees", "/leases"): - with self.subTest(path=path): - self.assertEqual(self.client.get(path).status_code, 200) + self.assertEqual(self.client.get("/leases").status_code, 200) def test_post_is_rejected(self): response = self.client.post("/health") @@ -62,10 +65,10 @@ 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"): + for path in ("/", "/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees"): with self.subTest(path=path): text = self.client.get(path).text - for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit"): + for href in ("/queue", "/projects", "/prompts", "/runtime", "/audit", "/worktrees"): self.assertIn(f'href="{href}"', text) diff --git a/tests/test_webui_worktree_hygiene.py b/tests/test_webui_worktree_hygiene.py new file mode 100644 index 0000000..ee62596 --- /dev/null +++ b/tests/test_webui_worktree_hygiene.py @@ -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() \ No newline at end of file diff --git a/webui/app.py b/webui/app.py index a97d1ad..3dccded 100644 --- a/webui/app.py +++ b/webui/app.py @@ -16,6 +16,8 @@ from webui.prompt_library import find_prompt, library_to_dict from webui.prompt_views import render_prompt_detail, render_prompts_page from webui.queue_loader import load_queue_snapshot, 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 _READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"}) @@ -134,10 +136,12 @@ async def audit(_request: Request) -> HTMLResponse: 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: @@ -174,6 +178,7 @@ def create_app() -> Starlette: Route("/runtime", runtime, methods=["GET"]), Route("/audit", audit, methods=["GET"]), Route("/worktrees", worktrees, methods=["GET"]), + Route("/api/worktrees", api_worktrees, methods=["GET"]), Route("/leases", leases, methods=["GET"]), ], exception_handlers={405: method_not_allowed}, diff --git a/webui/worktree_scanner.py b/webui/worktree_scanner.py new file mode 100644 index 0000000..0916d9c --- /dev/null +++ b/webui/worktree_scanner.py @@ -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 # / issue # 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 + ], + } \ No newline at end of file diff --git a/webui/worktree_views.py b/webui/worktree_views.py new file mode 100644 index 0000000..77bd23a --- /dev/null +++ b/webui/worktree_views.py @@ -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'' + f"{_escape(classification)}" + ) + + +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 ( + "" + f"{_escape(entry.rel_path)}" + f"{_badge(entry.classification)}" + f"{_escape(branch)}" + f"{_escape(head)}" + f"{_escape(dirty)}" + f"{'yes' if entry.detached else 'no'}" + f"{_escape(entry.notes)}" + "" + ) + + +WORKTREE_PAGE_SCRIPT = """ + +""" + +WORKTREE_PAGE_STYLES = """ + +""" + + +def render_worktrees_page(snapshot: HygieneSnapshot) -> str: + error_block = "" + if snapshot.scan_error: + error_block = ( + '

Partial scan: ' + f"{_escape(snapshot.scan_error)}

" + ) + + anomaly_block = "" + if snapshot.anomalies: + items = "".join(f"
  • {_escape(text)}
    • " for text in snapshot.anomalies) + anomaly_block = ( + '

    Missing preserved worktree anomalies

    ' + f"
      {items}
    " + ) + + if snapshot.entries: + rows = "".join(_entry_row(entry) for entry in snapshot.entries) + table = ( + "" + "" + "" + "" + "" + f"{rows}
    PathClassBranchHEADDirtyDetachedNotes
    " + ) + else: + table = "

    No directories under branches/.

    " + + body = ( + "

    Worktree hygiene

    " + "

    Read-only scan of local branches/ worktrees. " + "No deletion actions — copy the canonical cleanup prompt when merge is confirmed.

    " + f"{error_block}" + f"{anomaly_block}" + f"

    Repo root: {_escape(snapshot.repo_root)} · " + f"entries: {len(snapshot.entries)}

    " + f"{table}" + "

    Canonical cleanup prompt

    " + f'
    {_escape(snapshot.cleanup_prompt)}
    ' + '' + "

    JSON API

    " + f"{WORKTREE_PAGE_STYLES}" + f"{WORKTREE_PAGE_SCRIPT}" + ) + return render_page(title="Worktrees", body_html=body) \ No newline at end of file