diff --git a/docs/webui-local-dev.md b/docs/webui-local-dev.md index 23d4c56..6053f7f 100644 --- a/docs/webui-local-dev.md +++ b/docs/webui-local-dev.md @@ -45,13 +45,23 @@ Optional environment variables: | `/api/prompts` | JSON prompt export with workflow hashes | | `/runtime` | MCP runtime health and stale detection (#430) | | `/api/runtime` | JSON runtime health export | -| `/audit` | Stub — report audit paste (#431) | +| `/audit` | Report audit paste + validator preview (#431) | +| `/api/audit` | JSON validator preview (POST `report_text`, optional `task_kind`) | | `/worktrees` | Stub — hygiene dashboard (#432) | | `/leases` | Lease and collision visibility (#433) | | `/api/leases` | JSON lease/collision export | -All routes are GET-only. POST/PUT/PATCH/DELETE return `405` with -`read-only-mvp`. +Most routes are GET-only. POST/PUT/PATCH/DELETE return `405` with +`read-only-mvp`, except `/audit` and `/api/audit` which accept POST for +local validator preview only (no Gitea mutations, no server-side storage). + +## Report audit (#431) + +Paste an LLM final report at `/audit` or POST JSON to `/api/audit`. The UI +reuses `final_report_validator` and review schema checks to surface missing +proof fields, wrong validation vocabulary, mutation contradictions, and a +suggested next prompt or issue-comment draft. Task kind can be auto-detected +or selected explicitly. ## Project registry (#427) @@ -104,5 +114,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_lease_visibility.py tests/test_webui_runtime_health.py -q ``` \ No newline at end of file diff --git a/tests/test_webui_audit.py b/tests/test_webui_audit.py new file mode 100644 index 0000000..51f8cd9 --- /dev/null +++ b/tests/test_webui_audit.py @@ -0,0 +1,94 @@ +"""Tests for web UI report audit paste (#431).""" +import sys +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.audit_validator import audit_report, infer_task_kind + + +_MINIMAL_REVIEW_REPORT = """## Controller Handoff + +- Task: review PR #1 +- Repo: Scaled-Tech-Consulting/Gitea-Tools +- Role: reviewer +- Identity: sysadmin / prgs-reviewer +- Review decision: approve +""" + + +class TestAuditValidator(unittest.TestCase): + def test_infer_task_kind_from_review_report(self): + self.assertEqual(infer_task_kind(_MINIMAL_REVIEW_REPORT), "review_pr") + + def test_infer_task_kind_explicit_override(self): + self.assertEqual( + infer_task_kind(_MINIMAL_REVIEW_REPORT, explicit="work_issue"), + "work_issue", + ) + + def test_empty_report_blocked(self): + result = audit_report("") + self.assertTrue(result["blocked"]) + self.assertIn("empty", result["reasons"][0].lower()) + + def test_minimal_report_produces_findings(self): + result = audit_report(_MINIMAL_REVIEW_REPORT) + self.assertIn(result["grade"], {"blocked", "downgraded", "warning", "A"}) + self.assertTrue(result["findings"] or result["blocked"] or result["downgraded"]) + + +class TestAuditRoutes(unittest.TestCase): + def setUp(self): + self.client = TestClient(create_app()) + + def test_audit_page_renders_form(self): + response = self.client.get("/audit") + self.assertEqual(response.status_code, 200) + self.assertIn("Report audit", response.text) + self.assertIn('name="report_text"', response.text) + self.assertIn("Run validator preview", response.text) + self.assertNotIn("child issue", response.text.lower()) + + def test_audit_post_runs_validator(self): + response = self.client.post( + "/audit", + data={"report_text": _MINIMAL_REVIEW_REPORT, "task_kind": "review_pr"}, + ) + self.assertEqual(response.status_code, 200) + self.assertIn("Validator preview", response.text) + self.assertIn("Safe next action", response.text) + + def test_api_audit_post_json(self): + response = self.client.post( + "/api/audit", + json={"report_text": _MINIMAL_REVIEW_REPORT, "task_kind": "review_pr"}, + ) + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertEqual(data["task_kind"], "review_pr") + self.assertIn("grade", data) + self.assertIn("findings", data) + self.assertIn("suggested_next_prompt", data) + + def test_api_audit_rejects_empty_body(self): + response = self.client.post("/api/audit", json={"report_text": ""}) + self.assertEqual(response.status_code, 400) + + def test_audit_post_allowed_other_post_still_blocked(self): + self.assertEqual(self.client.post("/health").status_code, 405) + self.assertEqual( + self.client.post( + "/audit", + data={"report_text": _MINIMAL_REVIEW_REPORT}, + ).status_code, + 200, + ) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_webui_skeleton.py b/tests/test_webui_skeleton.py index e5b71b6..8e72b40 100644 --- a/tests/test_webui_skeleton.py +++ b/tests/test_webui_skeleton.py @@ -35,12 +35,17 @@ class TestWebuiSkeleton(unittest.TestCase): self.assertIn("Runtime health", response.text) def test_route_stubs_render(self): - for path in ("/audit",): + 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) + self.assertIn("Report audit", response.text) + def test_prompts_is_implemented(self): response = self.client.get("/prompts") self.assertEqual(response.status_code, 200) diff --git a/webui/app.py b/webui/app.py index fb1aab2..2fe5018 100644 --- a/webui/app.py +++ b/webui/app.py @@ -14,6 +14,10 @@ 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.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 @@ -22,6 +26,7 @@ from webui.runtime_health import load_runtime_snapshot, snapshot_to_dict as runt 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: @@ -132,13 +137,49 @@ 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", @@ -156,6 +197,9 @@ async def api_leases(_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"}, @@ -181,7 +225,8 @@ 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("/leases", leases, methods=["GET"]), Route("/api/leases", api_leases, methods=["GET"]), diff --git a/webui/audit_validator.py b/webui/audit_validator.py new file mode 100644 index 0000000..d19e7ff --- /dev/null +++ b/webui/audit_validator.py @@ -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 "", + } \ No newline at end of file diff --git a/webui/audit_views.py b/webui/audit_views.py new file mode 100644 index 0000000..1316eea --- /dev/null +++ b/webui/audit_views.py @@ -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 '

No validator findings — report structure looks complete for the selected task kind.

' + rows = [] + for finding in findings: + rows.append( + "" + f"{_escape(str(finding.get('rule_id', '')))}" + f"{_escape(str(finding.get('severity', '')))}" + f"{_escape(str(finding.get('field', '')))}" + f"{_escape(str(finding.get('reason', '')))}" + f"{_escape(str(finding.get('safe_next_action', '')))}" + "" + ) + return ( + '' + "" + "" + f"{''.join(rows)}
RuleSeverityFieldReasonSafe next action
" + ) + + +def _render_task_kind_select(selected: str | None) -> str: + options = [''] + for kind in _TASK_KIND_OPTIONS: + sel = ' selected' if selected == kind else "" + options.append(f'') + return f'' + + +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 = ( + '
' + f"

Validator preview

" + f'

Task kind: {inferred} · ' + f'Grade: {grade} · ' + f"Blocked: {result.get('blocked')} · " + f"Downgraded: {result.get('downgraded')}

" + f"

Safe next action: {safe_next}

" + f"{_render_findings(result)}" + ) + if suggested_prompt: + result_html += ( + "

Suggested next prompt

" + f'
{_escape(suggested_prompt)}
' + ) + if suggested_comment: + result_html += ( + "

Suggested issue/PR comment

" + f'
{_escape(suggested_comment)}
' + ) + result_html += "
" + + body = ( + "

Report audit

" + "

Paste an LLM final report for local validator preview. " + "Uses final_report_validator and review schema checks — " + "does not post to Gitea or store reports server-side.

" + '
' + "" + f"{_render_task_kind_select(task_kind)}" + '' + f'' + '' + "
" + f"{result_html}" + "

JSON API (POST report_text, " + "optional task_kind)

" + ) + return render_page(title="Audit", body_html=body, extra_head=AUDIT_PAGE_STYLES) + + +AUDIT_PAGE_STYLES = """ + +""" \ No newline at end of file