Merge remote-tracking branch 'prgs/master' into feat/issue-434-gated-actions
# Conflicts: # docs/webui-local-dev.md # tests/test_webui_skeleton.py # webui/app.py
This commit is contained in:
+59
-9
@@ -14,16 +14,23 @@ from webui.project_registry import find_project, load_registry, registry_to_dict
|
||||
from webui.project_views import render_project_detail, render_projects_list
|
||||
from webui.prompt_library import find_prompt, library_to_dict
|
||||
from webui.prompt_views import render_prompt_detail, render_prompts_page
|
||||
from final_report_validator import FINAL_REPORT_TASK_KINDS
|
||||
|
||||
from webui.gated_actions import attempt_action, load_action_registry, preview_action
|
||||
from webui.gated_action_views import render_actions_page
|
||||
from webui.audit_validator import audit_report, audit_to_dict
|
||||
from webui.audit_views import render_audit_page
|
||||
from webui.lease_loader import load_lease_snapshot, snapshot_to_dict as lease_snapshot_to_dict
|
||||
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
|
||||
|
||||
_READ_ONLY_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
|
||||
_AUDIT_MUTATION_PATHS = frozenset({"/audit", "/api/audit"})
|
||||
|
||||
|
||||
def _stub_page(title: str, description: str) -> HTMLResponse:
|
||||
@@ -135,18 +142,56 @@ async def api_runtime(_request: Request) -> JSONResponse:
|
||||
return JSONResponse(runtime_snapshot_to_dict(load_runtime_snapshot()))
|
||||
|
||||
|
||||
async def audit(_request: Request) -> HTMLResponse:
|
||||
return _stub_page(
|
||||
"Audit",
|
||||
"Report audit will accept pasted final reports and run validator previews.",
|
||||
async def _parse_audit_form(request: Request) -> tuple[str, str | None]:
|
||||
if request.method == "GET":
|
||||
return "", None
|
||||
content_type = (request.headers.get("content-type") or "").lower()
|
||||
if "application/json" in content_type:
|
||||
payload = await request.json()
|
||||
if not isinstance(payload, dict):
|
||||
return "", None
|
||||
report_text = str(payload.get("report_text") or "")
|
||||
task_kind = payload.get("task_kind")
|
||||
return report_text, str(task_kind) if task_kind else None
|
||||
form = await request.form()
|
||||
report_text = str(form.get("report_text") or "")
|
||||
task_kind = form.get("task_kind")
|
||||
return report_text, str(task_kind) if task_kind else None
|
||||
|
||||
|
||||
async def audit(request: Request) -> HTMLResponse:
|
||||
report_text, task_kind = await _parse_audit_form(request)
|
||||
result = None
|
||||
if request.method == "POST" and report_text.strip():
|
||||
result = audit_report(report_text, task_kind=task_kind)
|
||||
return HTMLResponse(
|
||||
render_audit_page(
|
||||
report_text=report_text,
|
||||
task_kind=task_kind,
|
||||
result=result,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
async def api_audit(request: Request) -> JSONResponse:
|
||||
if request.method == "GET":
|
||||
return JSONResponse({
|
||||
"usage": "POST JSON with report_text and optional task_kind",
|
||||
"task_kinds": sorted(FINAL_REPORT_TASK_KINDS),
|
||||
})
|
||||
report_text, task_kind = await _parse_audit_form(request)
|
||||
if not report_text.strip():
|
||||
return JSONResponse({"error": "report_text required"}, status_code=400)
|
||||
return JSONResponse(audit_to_dict(audit_report(report_text, task_kind=task_kind)))
|
||||
|
||||
|
||||
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:
|
||||
@@ -194,6 +239,9 @@ async def api_action_attempt(request: Request) -> JSONResponse:
|
||||
|
||||
|
||||
async def method_not_allowed(request: Request, _exc: Exception) -> Response:
|
||||
path = request.url.path
|
||||
if path in _AUDIT_MUTATION_PATHS and request.method == "POST":
|
||||
return JSONResponse({"error": "not_found"}, status_code=404)
|
||||
if request.method not in _READ_ONLY_METHODS:
|
||||
return JSONResponse(
|
||||
{"error": "read-only-mvp", "detail": f"{request.method} not permitted"},
|
||||
@@ -219,8 +267,10 @@ def create_app() -> Starlette:
|
||||
Route("/api/prompts", api_prompts, methods=["GET"]),
|
||||
Route("/runtime", runtime, methods=["GET"]),
|
||||
Route("/api/runtime", api_runtime, methods=["GET"]),
|
||||
Route("/audit", audit, methods=["GET"]),
|
||||
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("/actions", actions, methods=["GET"]),
|
||||
Route("/api/actions", api_actions, methods=["GET"]),
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
"""Final-report audit preview for the web UI (#431)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from final_report_validator import FINAL_REPORT_TASK_KINDS, assess_final_report_validator
|
||||
|
||||
_TASK_KIND_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
|
||||
("review_pr", re.compile(r"\b(?:review\s+pr|task\s*:\s*review|review-merge-pr)\b", re.I)),
|
||||
(
|
||||
"reconcile_already_landed",
|
||||
re.compile(
|
||||
r"\b(?:reconcile\s+already[- ]landed|already[- ]landed\s+pr|"
|
||||
r"reconcile-landed-pr)\b",
|
||||
re.I,
|
||||
),
|
||||
),
|
||||
("work_issue", re.compile(r"\b(?:work[- ]issue|task\s*:\s*work|selected\s+issue\s*:)\b", re.I)),
|
||||
("issue_filing", re.compile(r"\b(?:issue\s+filing|create\s+issue|task\s*:\s*file)\b", re.I)),
|
||||
("issue_selection", re.compile(r"\b(?:issue\s+selection|select\s+next\s+issue)\b", re.I)),
|
||||
("inventory", re.compile(r"\b(?:issue\s+inventory|queue\s+inventory)\b", re.I)),
|
||||
)
|
||||
|
||||
_SUGGESTED_PROMPTS: dict[str, str] = {
|
||||
"review_pr": (
|
||||
"Review the next eligible open PR in this project. Merge it only if every "
|
||||
"gate passes. Post a final report matching review-merge-pr schema."
|
||||
),
|
||||
"work_issue": (
|
||||
"Find the next eligible issue in this project, work on it only if all gates "
|
||||
"pass, and create a PR when complete."
|
||||
),
|
||||
"reconcile_already_landed": (
|
||||
"Reconcile the oldest eligible already-landed open PR. Post read-only audit "
|
||||
"first; authorize cleanup only when required."
|
||||
),
|
||||
"issue_filing": "File a new Gitea issue using the canonical issue-filing workflow.",
|
||||
"issue_selection": "Build a complete open-issue inventory and select the next eligible issue.",
|
||||
"inventory": "Produce a proof-backed inventory of open PRs or issues without mutating state.",
|
||||
}
|
||||
|
||||
|
||||
def infer_task_kind(report_text: str, explicit: str | None = None) -> str:
|
||||
"""Infer workflow task kind from explicit selection or report content."""
|
||||
if explicit:
|
||||
normalized = explicit.strip().lower().replace(" ", "_")
|
||||
if normalized in FINAL_REPORT_TASK_KINDS:
|
||||
return normalized
|
||||
text = report_text or ""
|
||||
for kind, pattern in _TASK_KIND_PATTERNS:
|
||||
if pattern.search(text):
|
||||
return kind
|
||||
if re.search(r"\brole\s*:\s*reviewer\b", text, re.I):
|
||||
return "review_pr"
|
||||
if re.search(r"\brole\s*:\s*author\b", text, re.I):
|
||||
return "work_issue"
|
||||
return "review_pr"
|
||||
|
||||
|
||||
def _merge_findings(*result_sets: dict[str, Any]) -> list[dict[str, str]]:
|
||||
seen: set[tuple[str, str, str]] = set()
|
||||
merged: list[dict[str, str]] = []
|
||||
for result in result_sets:
|
||||
for finding in result.get("findings") or []:
|
||||
key = (
|
||||
str(finding.get("rule_id", "")),
|
||||
str(finding.get("field", "")),
|
||||
str(finding.get("reason", "")),
|
||||
)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
merged.append(finding)
|
||||
return merged
|
||||
|
||||
|
||||
def _aggregate_result(task_kind: str, findings: list[dict[str, str]], base: dict[str, Any]) -> dict[str, Any]:
|
||||
blocked = any(f.get("severity") == "block" for f in findings)
|
||||
downgraded = any(f.get("severity") == "downgrade" for f in findings)
|
||||
warning = any(f.get("severity") == "warning" for f in findings)
|
||||
if blocked:
|
||||
grade = "blocked"
|
||||
elif downgraded:
|
||||
grade = "downgraded"
|
||||
elif warning:
|
||||
grade = "warning"
|
||||
else:
|
||||
grade = base.get("grade") or "A"
|
||||
safe_next_action = base.get("safe_next_action") or "proceed"
|
||||
for finding in findings:
|
||||
if finding.get("severity") in {"block", "downgrade"}:
|
||||
safe_next_action = finding.get("safe_next_action") or safe_next_action
|
||||
break
|
||||
return {
|
||||
"task_kind": task_kind,
|
||||
"inferred_task_kind": task_kind,
|
||||
"grade": grade,
|
||||
"blocked": blocked or bool(base.get("blocked")),
|
||||
"downgraded": downgraded or bool(base.get("downgraded")),
|
||||
"findings": findings,
|
||||
"reasons": [f"{f.get('rule_id')}: {f.get('reason')}" for f in findings],
|
||||
"safe_next_action": safe_next_action,
|
||||
"complete": grade == "A" and not findings,
|
||||
"suggested_next_prompt": _SUGGESTED_PROMPTS.get(task_kind, ""),
|
||||
"suggested_issue_comment": _build_issue_comment(findings, safe_next_action),
|
||||
}
|
||||
|
||||
|
||||
def _build_issue_comment(findings: list[dict[str, str]], safe_next_action: str) -> str:
|
||||
if not findings:
|
||||
return ""
|
||||
lines = [
|
||||
"## Report audit preview (local validator)",
|
||||
"",
|
||||
"Automated findings from pasted final report:",
|
||||
"",
|
||||
]
|
||||
for finding in findings[:12]:
|
||||
severity = finding.get("severity", "info")
|
||||
rule_id = finding.get("rule_id", "unknown")
|
||||
reason = finding.get("reason", "")
|
||||
lines.append(f"- **{severity}** `{rule_id}` — {reason}")
|
||||
if len(findings) > 12:
|
||||
lines.append(f"- …and {len(findings) - 12} more finding(s)")
|
||||
lines.extend(["", f"Safe next action: {safe_next_action}", ""])
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def audit_report(report_text: str, *, task_kind: str | None = None) -> dict[str, Any]:
|
||||
"""Run composable final-report validation on pasted text."""
|
||||
text = (report_text or "").strip()
|
||||
if not text:
|
||||
return {
|
||||
"task_kind": task_kind or "review_pr",
|
||||
"inferred_task_kind": task_kind or "review_pr",
|
||||
"grade": "blocked",
|
||||
"blocked": True,
|
||||
"downgraded": False,
|
||||
"findings": [],
|
||||
"reasons": ["empty report text"],
|
||||
"safe_next_action": "paste a non-empty final report",
|
||||
"complete": False,
|
||||
"suggested_next_prompt": "",
|
||||
"suggested_issue_comment": "",
|
||||
}
|
||||
|
||||
resolved_kind = infer_task_kind(text, task_kind)
|
||||
base = assess_final_report_validator(text, resolved_kind)
|
||||
findings = list(base.get("findings") or [])
|
||||
|
||||
if resolved_kind == "review_pr":
|
||||
from review_final_report_schema import assess_review_final_report_schema
|
||||
|
||||
schema = assess_review_final_report_schema(text)
|
||||
findings = _merge_findings(base, schema)
|
||||
|
||||
return _aggregate_result(resolved_kind, findings, base)
|
||||
|
||||
|
||||
def audit_to_dict(result: dict[str, Any]) -> dict[str, Any]:
|
||||
"""JSON-safe export for /api/audit."""
|
||||
return {
|
||||
"task_kind": result.get("task_kind"),
|
||||
"inferred_task_kind": result.get("inferred_task_kind"),
|
||||
"grade": result.get("grade"),
|
||||
"blocked": result.get("blocked"),
|
||||
"downgraded": result.get("downgraded"),
|
||||
"complete": result.get("complete"),
|
||||
"safe_next_action": result.get("safe_next_action"),
|
||||
"findings": result.get("findings") or [],
|
||||
"reasons": result.get("reasons") or [],
|
||||
"suggested_next_prompt": result.get("suggested_next_prompt") or "",
|
||||
"suggested_issue_comment": result.get("suggested_issue_comment") or "",
|
||||
}
|
||||
@@ -0,0 +1,139 @@
|
||||
"""HTML views for final-report audit paste (#431)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
|
||||
from final_report_validator import FINAL_REPORT_TASK_KINDS
|
||||
from webui.audit_validator import audit_report
|
||||
from webui.layout import render_page
|
||||
|
||||
_TASK_KIND_OPTIONS = sorted(FINAL_REPORT_TASK_KINDS)
|
||||
|
||||
|
||||
def _escape(text: str) -> str:
|
||||
return html.escape(text, quote=True)
|
||||
|
||||
|
||||
def _render_findings(result: dict) -> str:
|
||||
findings = result.get("findings") or []
|
||||
if not findings:
|
||||
return '<p class="muted">No validator findings — report structure looks complete for the selected task kind.</p>'
|
||||
rows = []
|
||||
for finding in findings:
|
||||
rows.append(
|
||||
"<tr>"
|
||||
f"<td><code>{_escape(str(finding.get('rule_id', '')))}</code></td>"
|
||||
f"<td>{_escape(str(finding.get('severity', '')))}</td>"
|
||||
f"<td>{_escape(str(finding.get('field', '')))}</td>"
|
||||
f"<td>{_escape(str(finding.get('reason', '')))}</td>"
|
||||
f"<td>{_escape(str(finding.get('safe_next_action', '')))}</td>"
|
||||
"</tr>"
|
||||
)
|
||||
return (
|
||||
'<table class="detail audit-findings">'
|
||||
"<thead><tr><th>Rule</th><th>Severity</th><th>Field</th>"
|
||||
"<th>Reason</th><th>Safe next action</th></tr></thead>"
|
||||
f"<tbody>{''.join(rows)}</tbody></table>"
|
||||
)
|
||||
|
||||
|
||||
def _render_task_kind_select(selected: str | None) -> str:
|
||||
options = ['<option value="">Auto-detect from report</option>']
|
||||
for kind in _TASK_KIND_OPTIONS:
|
||||
sel = ' selected' if selected == kind else ""
|
||||
options.append(f'<option value="{_escape(kind)}"{sel}>{_escape(kind)}</option>')
|
||||
return f'<select id="task_kind" name="task_kind">{"".join(options)}</select>'
|
||||
|
||||
|
||||
def render_audit_page(
|
||||
*,
|
||||
report_text: str = "",
|
||||
task_kind: str | None = None,
|
||||
result: dict | None = None,
|
||||
) -> str:
|
||||
result_html = ""
|
||||
if result is not None:
|
||||
grade = _escape(str(result.get("grade", "")))
|
||||
inferred = _escape(str(result.get("inferred_task_kind", "")))
|
||||
safe_next = _escape(str(result.get("safe_next_action", "")))
|
||||
suggested_prompt = result.get("suggested_next_prompt") or ""
|
||||
suggested_comment = result.get("suggested_issue_comment") or ""
|
||||
result_html = (
|
||||
'<section class="audit-result">'
|
||||
f"<h3>Validator preview</h3>"
|
||||
f'<p class="meta">Task kind: <code>{inferred}</code> · '
|
||||
f'Grade: <strong>{grade}</strong> · '
|
||||
f"Blocked: <code>{result.get('blocked')}</code> · "
|
||||
f"Downgraded: <code>{result.get('downgraded')}</code></p>"
|
||||
f"<p><strong>Safe next action:</strong> {safe_next}</p>"
|
||||
f"{_render_findings(result)}"
|
||||
)
|
||||
if suggested_prompt:
|
||||
result_html += (
|
||||
"<h4>Suggested next prompt</h4>"
|
||||
f'<pre class="prompt-text">{_escape(suggested_prompt)}</pre>'
|
||||
)
|
||||
if suggested_comment:
|
||||
result_html += (
|
||||
"<h4>Suggested issue/PR comment</h4>"
|
||||
f'<pre class="prompt-text">{_escape(suggested_comment)}</pre>'
|
||||
)
|
||||
result_html += "</section>"
|
||||
|
||||
body = (
|
||||
"<h2>Report audit</h2>"
|
||||
"<p>Paste an LLM final report for local validator preview. "
|
||||
"Uses <code>final_report_validator</code> and review schema checks — "
|
||||
"does not post to Gitea or store reports server-side.</p>"
|
||||
'<form method="post" action="/audit" class="audit-form">'
|
||||
"<label for=\"task_kind\">Task kind</label>"
|
||||
f"{_render_task_kind_select(task_kind)}"
|
||||
'<label for="report_text">Final report</label>'
|
||||
f'<textarea id="report_text" name="report_text" rows="18" '
|
||||
f'placeholder="Paste final report markdown here…">{_escape(report_text)}</textarea>'
|
||||
'<button type="submit" class="copy-btn">Run validator preview</button>'
|
||||
"</form>"
|
||||
f"{result_html}"
|
||||
"<p><a href=\"/api/audit\">JSON API</a> (POST <code>report_text</code>, "
|
||||
"optional <code>task_kind</code>)</p>"
|
||||
)
|
||||
return render_page(title="Audit", body_html=body, extra_head=AUDIT_PAGE_STYLES)
|
||||
|
||||
|
||||
AUDIT_PAGE_STYLES = """
|
||||
<style>
|
||||
.audit-form label {
|
||||
display: block;
|
||||
margin: 0.75rem 0 0.35rem;
|
||||
color: var(--muted);
|
||||
font-size: 0.9rem;
|
||||
}
|
||||
.audit-form select,
|
||||
.audit-form textarea {
|
||||
width: 100%;
|
||||
font: inherit;
|
||||
color: var(--text);
|
||||
background: var(--surface);
|
||||
border: 1px solid var(--border);
|
||||
border-radius: 6px;
|
||||
padding: 0.55rem 0.65rem;
|
||||
}
|
||||
.audit-form textarea {
|
||||
font-family: ui-monospace, monospace;
|
||||
font-size: 0.85rem;
|
||||
line-height: 1.45;
|
||||
resize: vertical;
|
||||
}
|
||||
.audit-result {
|
||||
margin-top: 1.5rem;
|
||||
padding-top: 1rem;
|
||||
border-top: 1px solid var(--border);
|
||||
}
|
||||
table.audit-findings td:nth-child(4),
|
||||
table.audit-findings td:nth-child(5) {
|
||||
font-size: 0.85rem;
|
||||
color: var(--muted);
|
||||
}
|
||||
</style>
|
||||
"""
|
||||
@@ -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
|
||||
],
|
||||
}
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user