Gitea-Tools/webui/audit_validator.py

"""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 "",
    }