feat: add final-report audit paste tool to web UI (Closes #431)
Expose /audit and /api/audit for local validator previews using final_report_validator and review schema checks. Operators can paste reports, auto-detect task kind, and get structured findings with suggested next prompts and issue-comment drafts. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -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 "",
|
||||
}
|
||||
Reference in New Issue
Block a user