Gitea-Tools/reviewer_validation_worktree_edits.py

"""PR validation worktree read-only edit verifier (#315)."""

from __future__ import annotations

import re
from typing import Any

_VALIDATION_WORKTREE_RE = re.compile(
    r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)",
    re.IGNORECASE,
)
_DIAGNOSTIC_WORKTREE_RE = re.compile(
    r"branches/(?:diagnostic[\w/-]*|scratch[\w/-]*)",
    re.IGNORECASE,
)
_FILE_EDITS_NONE_RE = re.compile(
    r"file edits by reviewer\s*:\s*none\b",
    re.IGNORECASE,
)
_FILE_EDITS_FIELD_RE = re.compile(
    r"file edits by reviewer\s*:\s*(.+)",
    re.IGNORECASE,
)
_VALIDATION_WORKTREE_PATH_RE = re.compile(
    r"(?:validation worktree path|review worktree path)\s*:\s*(\S+)",
    re.IGNORECASE,
)
_DIAGNOSTIC_WORKTREE_PATH_RE = re.compile(
    r"(?:diagnostic (?:scratch )?worktree path|diagnostic worktree)\s*:\s*(\S+)",
    re.IGNORECASE,
)
_DIAGNOSTIC_LABEL_RE = re.compile(
    r"diagnostic local experiment|not pr-head validation|diagnostic-only",
    re.IGNORECASE,
)
_OFFICIAL_VALIDATION_RE = re.compile(
    r"(?:official (?:pr-head )?validation|pr-head validation result)",
    re.IGNORECASE,
)
_POST_EDIT_OFFICIAL_RE = re.compile(
    r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit)",
    re.IGNORECASE,
)

_PERFORMED_EDIT_KINDS = frozenset({"file_edit", "edit", "write", "edited", "created", "wrote"})


def _normalize_path(path: str) -> str:
    return (path or "").replace("\\", "/").strip()


def _is_validation_worktree(path: str) -> bool:
    return bool(_VALIDATION_WORKTREE_RE.search(_normalize_path(path)))


def _is_diagnostic_worktree(path: str) -> bool:
    normalized = _normalize_path(path)
    return bool(
        _DIAGNOSTIC_WORKTREE_RE.search(normalized)
        or "diagnostic" in normalized.lower()
    )


def _performed_edits(action_log: list[dict] | None) -> list[dict[str, str]]:
    edits: list[dict[str, str]] = []
    for entry in action_log or []:
        if entry.get("gated_rejected") or entry.get("performed") is False:
            continue
        kind = (entry.get("kind") or entry.get("action") or "file_edit").strip().lower()
        if kind not in _PERFORMED_EDIT_KINDS:
            continue
        path = (entry.get("path") or "").strip()
        if not path:
            continue
        worktree = _normalize_path(str(entry.get("worktree_path") or entry.get("cwd") or ""))
        edits.append({"path": path, "worktree_path": worktree})
    return edits


def _extract_validation_path(text: str, session: dict) -> str:
    path = _normalize_path(str(session.get("validation_worktree_path") or ""))
    if path:
        return path
    match = _VALIDATION_WORKTREE_PATH_RE.search(text or "")
    return _normalize_path(match.group(1)) if match else ""


def _extract_diagnostic_path(text: str, session: dict) -> str:
    path = _normalize_path(str(session.get("diagnostic_worktree_path") or ""))
    if path:
        return path
    match = _DIAGNOSTIC_WORKTREE_PATH_RE.search(text or "")
    return _normalize_path(match.group(1)) if match else ""


def assess_validation_worktree_edit_report(
    report_text: str,
    *,
    validation_session: dict | None = None,
    action_log: list[dict] | None = None,
) -> dict[str, Any]:
    """Block edits in PR validation worktrees; require diagnostic separation (#315)."""
    text = report_text or ""
    session = dict(validation_session or {})
    reasons: list[str] = []

    validation_path = _extract_validation_path(text, session)
    diagnostic_path = _extract_diagnostic_path(text, session)
    validation_edits = list(session.get("validation_worktree_edits") or [])
    diagnostic_edits = list(session.get("diagnostic_edits") or [])
    edits = _performed_edits(action_log)

    if not validation_edits:
        for entry in edits:
            wt = entry.get("worktree_path") or validation_path
            if wt and _is_validation_worktree(wt) and not _is_diagnostic_worktree(wt):
                validation_edits.append(entry["path"])

    if not diagnostic_edits:
        for entry in edits:
            wt = entry.get("worktree_path") or ""
            if wt and _is_diagnostic_worktree(wt):
                diagnostic_edits.append(entry["path"])
            elif entry["path"] and diagnostic_path and not validation_edits:
                if _is_diagnostic_worktree(diagnostic_path):
                    diagnostic_edits.append(entry["path"])

    claimed_none = bool(_FILE_EDITS_NONE_RE.search(text))
    any_edits = bool(validation_edits or diagnostic_edits or edits)

    if validation_edits:
        reasons.append(
            "reviewer must not edit files in the PR validation worktree; "
            f"observed edits: {', '.join(validation_edits)}"
        )

    if diagnostic_edits or session.get("diagnostic_experiment"):
        if not diagnostic_path and not _DIAGNOSTIC_WORKTREE_PATH_RE.search(text):
            reasons.append(
                "diagnostic experiments require a separate diagnostic scratch worktree path"
            )
        if not _DIAGNOSTIC_LABEL_RE.search(text):
            reasons.append(
                "diagnostic experiments must be labeled "
                "'Diagnostic local experiment — not PR-head validation'"
            )
        if not _FILE_EDITS_FIELD_RE.search(text) or claimed_none:
            reasons.append(
                "diagnostic edits must be reported under 'File edits by reviewer'"
            )

    if any_edits and claimed_none:
        reasons.append(
            "report claims 'File edits by reviewer: none' but reviewer file edits occurred"
        )

    if session.get("official_validation_after_edit") or _POST_EDIT_OFFICIAL_RE.search(text):
        if validation_edits or (validation_path and diagnostic_edits):
            reasons.append(
                "post-edit test results cannot be reported as official PR-head validation"
            )

    if validation_edits and _OFFICIAL_VALIDATION_RE.search(text):
        if not _DIAGNOSTIC_LABEL_RE.search(text):
            reasons.append(
                "validation worktree was edited; official PR-head validation is no longer valid"
            )

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "validation_worktree_path": validation_path or None,
        "diagnostic_worktree_path": diagnostic_path or None,
        "validation_worktree_edits": validation_edits,
        "diagnostic_edits": diagnostic_edits,
        "safe_next_action": (
            "keep PR validation worktrees read-only; use a separate diagnostic "
            "scratch worktree and report all reviewer file edits"
            if reasons
            else "proceed"
        ),
    }