Gitea-Tools/reviewer_validation_cwd_proof.py

"""Explicit worktree and cwd proof for PR review validation (#398)."""

from __future__ import annotations

import re
from typing import Any

_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)

_PWD_RE = re.compile(
    r"(?:^|\n)\s*(?:pwd|working\s+directory|cwd)\s*:\s*(\S+)",
    re.IGNORECASE,
)
_HEAD_RE = re.compile(
    r"(?:git\s+rev-parse\s+head|observed\s+head\s+sha)\s*:\s*([0-9a-f]{7,40})",
    re.IGNORECASE | re.MULTILINE,
)
_EXPECTED_HEAD_RE = re.compile(
    r"(?:expected\s+(?:pr\s+)?head\s+sha|candidate\s+head\s+sha|pinned\s+head)\s*:\s*([0-9a-f]{7,40})",
    re.IGNORECASE,
)
_STATUS_RE = re.compile(
    r"git\s+status\s+(?:--short\s+--branch|--short|-sb)\s*:\s*(.+)",
    re.IGNORECASE,
)
_VALIDATION_CMD_RE = re.compile(
    r"validation\s+command\s*:\s*(.+)",
    re.IGNORECASE,
)
_GIT_C_CMD_RE = re.compile(r"git\s+-C\s+\S+", re.IGNORECASE)
_CD_CMD_RE = re.compile(r"(?:^|&&\s*)cd\s+\S+", re.IGNORECASE)
_BASELINE_CWD_RE = re.compile(
    r"baseline\s+(?:worktree|working\s+directory|cwd)\s*:\s*(\S+)",
    re.IGNORECASE,
)
_BASELINE_SHA_RE = re.compile(
    r"baseline\s+(?:target\s+)?sha\s*:\s*([0-9a-f]{7,40})",
    re.IGNORECASE,
)
_BASELINE_CMD_RE = re.compile(
    r"baseline\s+validation\s+command\s*:\s*(.+)",
    re.IGNORECASE,
)


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


def _path_under_branches(path: str, project_root: str | None = None) -> bool:
    normalized = _normalize_path(path)
    if not normalized:
        return False
    if "/branches/" in f"{normalized}/":
        return True
    if normalized.endswith("/branches"):
        return True
    if project_root:
        root = _normalize_path(project_root)
        if normalized.startswith(f"{root}/"):
            rel = normalized[len(root) + 1 :]
            return rel == "branches" or rel.startswith("branches/")
    return False


def _expand_sha(sha: str) -> str:
    return (sha or "").strip().lower()


def _sha_matches(expected: str, observed: str) -> bool:
    exp = _expand_sha(expected)
    obs = _expand_sha(observed)
    if not exp or not obs:
        return False
    if len(exp) == 40 and len(obs) == 40:
        return exp == obs
    return obs.startswith(exp) or exp.startswith(obs)


def _command_has_explicit_cwd(command: str, cwd: str) -> bool:
    text = (command or "").strip()
    if not text:
        return False
    if _GIT_C_CMD_RE.search(text):
        return True
    if _CD_CMD_RE.search(text):
        return True
    if cwd and cwd in text:
        return True
    return False


def assess_validation_cwd_proof_report(
    report_text: str,
    *,
    validation_session: dict | None = None,
    project_root: str | None = None,
) -> dict[str, Any]:
    """Require cwd/HEAD proof before reviewer validation claims (#398)."""
    text = report_text or ""
    session = dict(validation_session or {})
    reasons: list[str] = []
    violations: list[str] = []

    claims_validation = bool(
        session.get("validation_ran")
        or _VALIDATION_CMD_RE.search(text)
        or session.get("command")
    )
    if not claims_validation:
        return {
            "proven": True,
            "block": False,
            "claims_validation": False,
            "reasons": [],
            "violations": [],
            "safe_next_action": "proceed",
        }

    expected_head = (
        session.get("expected_head_sha")
        or session.get("candidate_head_sha")
        or ""
    ).strip()
    if not expected_head:
        match = _EXPECTED_HEAD_RE.search(text)
        expected_head = (match.group(1) if match else "").strip()

    observed_head = (session.get("observed_head_sha") or "").strip()
    if not observed_head:
        match = _HEAD_RE.search(text)
        observed_head = (match.group(1) if match else "").strip()

    cwd = (
        session.get("working_directory")
        or session.get("cwd")
        or session.get("pwd")
        or ""
    ).strip()
    if not cwd:
        match = _PWD_RE.search(text)
        cwd = (match.group(1) if match else "").strip().rstrip(",.;")

    command = (session.get("command") or "").strip()
    if not command:
        match = _VALIDATION_CMD_RE.search(text)
        command = (match.group(1) if match else "").strip().rstrip(".;")

    if not cwd:
        reasons.append(
            "validation claimed without pwd/working-directory proof (#398)"
        )
    elif not _path_under_branches(cwd, project_root):
        violations.append(
            f"validation cwd {cwd!r} is not under branches/ (#398)"
        )
        reasons.append(
            "reviewer validation must run from a branches/ worktree, "
            "not the main checkout (#398)"
        )

    if not observed_head:
        reasons.append(
            "validation claimed without git rev-parse HEAD / observed HEAD SHA "
            "proof (#398)"
        )
    elif expected_head and not _sha_matches(expected_head, observed_head):
        violations.append(
            f"observed HEAD {observed_head} does not match expected "
            f"PR head {expected_head} (#398)"
        )
        reasons.append("validation HEAD SHA must match pinned PR head (#398)")

    if not _STATUS_RE.search(text) and session.get("git_status") is None:
        reasons.append(
            "validation claimed without git status --short --branch proof (#398)"
        )

    if command and cwd and not _command_has_explicit_cwd(command, cwd):
        if session.get("tool_working_directory") is not True:
            reasons.append(
                "validation command must use git -C <worktree>, "
                "cd <worktree> && ..., or tool-provided cwd metadata (#398)"
            )

    baseline_ran = bool(
        session.get("baseline_validation_ran")
        or _BASELINE_CMD_RE.search(text)
    )
    if baseline_ran:
        baseline_cwd = (session.get("baseline_worktree_path") or "").strip()
        if not baseline_cwd:
            match = _BASELINE_CWD_RE.search(text)
            baseline_cwd = (match.group(1) if match else "").strip().rstrip(",.;")
        if not baseline_cwd or not _path_under_branches(baseline_cwd, project_root):
            reasons.append(
                "baseline validation claimed without baseline worktree cwd "
                "under branches/ (#398)"
            )
        baseline_sha = (session.get("baseline_target_sha") or "").strip()
        if not baseline_sha:
            match = _BASELINE_SHA_RE.search(text)
            baseline_sha = (match.group(1) if match else "").strip()
        if not baseline_sha:
            reasons.append(
                "baseline validation claimed without baseline target SHA (#398)"
            )
        baseline_cmd = (session.get("baseline_command") or "").strip()
        if not baseline_cmd:
            match = _BASELINE_CMD_RE.search(text)
            baseline_cmd = (match.group(1) if match else "").strip()
        if not baseline_cmd:
            reasons.append(
                "baseline validation claimed without exact baseline command (#398)"
            )

    proven = not reasons and not violations
    return {
        "proven": proven,
        "block": bool(violations) or not proven,
        "claims_validation": True,
        "expected_head_sha": expected_head or None,
        "observed_head_sha": observed_head or None,
        "working_directory": cwd or None,
        "reasons": reasons,
        "violations": violations,
        "safe_next_action": (
            "before validation record pwd, git rev-parse HEAD, git status, "
            "expected PR head SHA; run commands with git -C or cd in the same line"
            if not proven
            else "proceed"
        ),
    }