Gitea-Tools/reviewer_merge_simulation.py

"""Merge-simulation worktree mutation verifier for reviewer reports (#317)."""

from __future__ import annotations

import re
from typing import Any

_WORKTREE_INDEX_FIELD_RE = re.compile(
    r"^\s*worktree/index mutations\s*:\s*(.+?)\s*$",
    re.IGNORECASE | re.MULTILINE,
)
_READONLY_DIAG_RE = re.compile(
    r"read[- ]only diagnostics\s*:\s*(.+)$",
    re.IGNORECASE | re.MULTILINE,
)
_WORKTREE_PATH_RE = re.compile(
    r"(?:worktree path|diagnostic worktree|simulation worktree)\s*:",
    re.IGNORECASE,
)
_PRE_CLEAN_RE = re.compile(
    r"(?:pre[- ]simulation|before simulation|clean before).*(?:clean|status)",
    re.IGNORECASE,
)
_MERGE_RESULT_RE = re.compile(
    r"(?:merge result|simulation result|conflict status|merge conflict)",
    re.IGNORECASE,
)
_ABORT_RE = re.compile(
    r"(?:merge --abort|abort/reset command|abort command|git merge --abort)",
    re.IGNORECASE,
)
_POST_CLEAN_RE = re.compile(
    r"(?:post[- ]abort|after abort|clean after).*(?:clean|status)",
    re.IGNORECASE,
)


def _command_text(entry: Any) -> str:
    if isinstance(entry, dict):
        return str(entry.get("command") or "").strip()
    return str(entry or "").strip()


def _git_subcommand_line(command: str) -> str | None:
    tokens = command.split()
    if not tokens or tokens[0] != "git":
        return None
    return " ".join(tokens[1:])


def merge_simulation_commands(command_log: list | None) -> list[str]:
    """Return logged commands that perform local merge simulation (#317)."""
    out: list[str] = []
    for entry in command_log or []:
        command = _command_text(entry)
        rest = _git_subcommand_line(command)
        if rest is None:
            continue
        lower = command.lower()
        if rest.startswith("merge"):
            if "--no-commit" in lower or (
                "--no-ff" in lower and "merge" in lower and "--commit" not in lower
            ):
                out.append(command)
            elif "--abort" in lower:
                out.append(command)
    return out


def assess_merge_simulation_report(
    report_text: str,
    *,
    command_log: list | None = None,
) -> dict[str, Any]:
    """Reject merge simulation classified as read-only; require worktree proof (#317)."""
    text = report_text or ""
    lower = text.lower()
    reasons: list[str] = []
    simulation_commands = merge_simulation_commands(command_log)
    has_abort = any("--abort" in c.lower() for c in simulation_commands)
    has_simulation = any(
        "--no-commit" in c.lower() or (
            "--no-ff" in c.lower() and "--abort" not in c.lower()
        )
        for c in simulation_commands
    )

    if not simulation_commands:
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "simulation_commands": [],
            "safe_next_action": "proceed",
        }

    field = _WORKTREE_INDEX_FIELD_RE.search(text)
    value = (field.group(1).strip().lower() if field else "") or ""
    if not value or value in {"none", "n/a", "no"}:
        reasons.append(
            "merge simulation commands ran but report lacks "
            "'Worktree/index mutations' entry"
        )
    elif "merge" not in value and "simulation" not in value:
        if not any(cmd.lower() in lower for cmd in simulation_commands):
            reasons.append(
                "Worktree/index mutations must document merge simulation commands"
            )

    readonly = _READONLY_DIAG_RE.search(text)
    if readonly:
        readonly_value = readonly.group(1)
        for command in simulation_commands:
            if command.lower() in readonly_value.lower():
                reasons.append(
                    "merge simulation may not be listed under read-only diagnostics"
                )
        if has_simulation and "merge simulation" in readonly_value.lower():
            reasons.append(
                "merge simulation may not be classified as read-only diagnostics"
            )

    if has_simulation:
        if not _WORKTREE_PATH_RE.search(text):
            reasons.append("merge simulation report missing worktree path")
        if not _PRE_CLEAN_RE.search(text):
            reasons.append("merge simulation report missing pre-simulation clean status")
        if not _MERGE_RESULT_RE.search(text):
            reasons.append("merge simulation report missing merge result/conflict status")
    if has_abort or has_simulation:
        if has_abort and not _ABORT_RE.search(text):
            reasons.append("merge simulation report missing abort/reset command proof")
        if has_abort and not _POST_CLEAN_RE.search(text):
            reasons.append("merge simulation report missing post-abort clean status")

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "simulation_commands": simulation_commands,
        "safe_next_action": (
            "classify merge simulation under Worktree/index mutations with full "
            "pre/merge/abort/post-clean proof; never list as read-only diagnostics"
            if reasons
            else "proceed"
        ),
    }