Gitea-Tools/reviewer_worktree_ownership.py

"""Reviewer worktree ownership and safe-reuse proof verifier (#312)."""

from __future__ import annotations

import re
from typing import Any

_SESSION_OWNED_RE = re.compile(
    r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)",
    re.IGNORECASE,
)
_WORKTREE_PATH_RE = re.compile(
    r"(?:review worktree path|worktree path|session-owned worktree)\s*:\s*(\S+)",
    re.IGNORECASE,
)
_BRANCHES_PATH_RE = re.compile(r"/branches/|\bbranches/", re.IGNORECASE)
_MAIN_CHECKOUT_RE = re.compile(
    r"main checkout|not (?:the )?main checkout|outside branches/",
    re.IGNORECASE,
)
_SAFE_REUSE_RE = re.compile(r"safe[- ]reuse proof", re.IGNORECASE)
_REUSE_POLICY_RE = re.compile(
    r"(?:project policy|policy) (?:allows|allowing) (?:reuse|reset)",
    re.IGNORECASE,
)
_CLEAN_TRACKED_RE = re.compile(
    r"(?:clean tracked state|tracked state\s*:\s*clean|no uncommitted tracked)",
    re.IGNORECASE,
)
_CLEAN_UNTRACKED_RE = re.compile(
    r"(?:clean untracked state|untracked state\s*:\s*clean|no untracked files)",
    re.IGNORECASE,
)
_NOT_OTHER_SESSION_RE = re.compile(
    r"not owned by (?:another|other) (?:active )?(?:task|session)",
    re.IGNORECASE,
)
_BRANCH_BEFORE_RESET_RE = re.compile(
    r"(?:branch/head before reset|head before reset|branch before reset)\s*:\s*(\S+)",
    re.IGNORECASE,
)
_RESET_TARGET_RE = re.compile(
    r"(?:reset target sha|reset target)\s*:\s*([0-9a-f]{7,40})",
    re.IGNORECASE,
)
_DESTRUCTIVE_CMD_RE = re.compile(
    r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
    r"(?:reset\s+--hard|clean(?:\s+-[A-Za-z]+)*|checkout\s+(?!HEAD\s+--)|switch\s+)",
    re.IGNORECASE,
)
_WORKSPACE_NONE_RE = re.compile(r"workspace mutations\s*:\s*none", re.IGNORECASE)
_WORKTREE_MUTATION_RE = re.compile(
    r"(?:worktree(?:/index)? mutations|destructive reset)\s*:\s*(?!none\b).+",
    re.IGNORECASE,
)
_RESET_IN_REPORT_RE = re.compile(r"reset\s+--hard", re.IGNORECASE)


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


def _destructive_commands(command_log: list | None) -> list[str]:
    return [
        cmd
        for cmd in (_command_text(entry) for entry in (command_log or []))
        if cmd and _DESTRUCTIVE_CMD_RE.search(cmd)
    ]


def _extract_worktree_path(text: str, session: dict) -> str:
    path = (session.get("worktree_path") or "").strip()
    if path:
        return path
    match = _WORKTREE_PATH_RE.search(text or "")
    return match.group(1).strip() if match else ""


def _is_session_owned_path(path: str) -> bool:
    normalized = (path or "").replace("\\", "/")
    return bool(_SESSION_OWNED_RE.search(normalized))


def _safe_reuse_fields_present(text: str) -> list[str]:
    missing: list[str] = []
    if not _WORKTREE_PATH_RE.search(text):
        missing.append("exact worktree path")
    if not _BRANCHES_PATH_RE.search(text):
        missing.append("worktree inside branches/")
    if not _MAIN_CHECKOUT_RE.search(text):
        missing.append("worktree is not the main checkout")
    if not _NOT_OTHER_SESSION_RE.search(text):
        missing.append("worktree not owned by another active session")
    if not _CLEAN_TRACKED_RE.search(text):
        missing.append("clean tracked state")
    if not _CLEAN_UNTRACKED_RE.search(text):
        missing.append("clean untracked state")
    if not _BRANCH_BEFORE_RESET_RE.search(text):
        missing.append("branch/head before reset")
    if not _RESET_TARGET_RE.search(text):
        missing.append("reset target SHA")
    if not _REUSE_POLICY_RE.search(text):
        missing.append("explicit project policy allowing reuse/reset")
    return missing


def assess_worktree_ownership_report(
    report_text: str,
    *,
    ownership_session: dict | None = None,
    command_log: list | None = None,
) -> dict[str, Any]:
    """Prove reviewer worktree ownership before reset or validation (#312)."""
    text = report_text or ""
    session = dict(ownership_session or {})
    reasons: list[str] = []

    worktree_path = _extract_worktree_path(text, session)
    destructive = _destructive_commands(command_log or session.get("command_log"))
    if not destructive and session.get("destructive_reset"):
        destructive = ["git reset --hard (session)"]

    session_owned = bool(
        session.get("session_owned")
        or (worktree_path and _is_session_owned_path(worktree_path))
    )
    safe_reuse = bool(session.get("safe_reuse") or _SAFE_REUSE_RE.search(text))
    main_checkout = bool(session.get("main_checkout"))
    dirty_tracked = session.get("dirty_tracked")
    dirty_untracked = session.get("dirty_untracked")
    other_session_owned = bool(session.get("other_session_owned"))

    if worktree_path or destructive or session.get("validation_ran"):
        if not worktree_path:
            reasons.append("report missing exact review worktree path")
        elif main_checkout or "branches/" not in worktree_path.replace("\\", "/"):
            reasons.append("review worktree must be inside branches/, not the main checkout")
        elif not _BRANCHES_PATH_RE.search(worktree_path.replace("\\", "/")):
            reasons.append("review worktree path must be under branches/")

    if other_session_owned and not safe_reuse:
        reasons.append(
            "reused worktree appears owned by another active task/session; "
            "safe-reuse proof required"
        )

    if dirty_tracked:
        reasons.append(
            "worktree has uncommitted tracked changes before reset or validation"
        )
    if dirty_untracked and safe_reuse:
        reasons.append("safe-reuse proof requires clean untracked state")

    if safe_reuse and not session_owned:
        reasons.extend(
            f"safe-reuse proof missing {field}"
            for field in _safe_reuse_fields_present(text)
        )

    if destructive:
        if not session_owned and not safe_reuse:
            reasons.append(
                "destructive worktree commands forbidden without session-owned "
                "worktree or safe-reuse proof"
            )
        if _WORKSPACE_NONE_RE.search(text) and (
            _RESET_IN_REPORT_RE.search(text) or any("reset" in c.lower() for c in destructive)
        ):
            reasons.append(
                "final report must not claim 'Workspace mutations: none' when "
                "git reset --hard occurred"
            )
        if not _WORKTREE_MUTATION_RE.search(text) and not _RESET_IN_REPORT_RE.search(text):
            reasons.append(
                "destructive reset must be reported under Worktree/index mutations "
                "or destructive reset operations"
            )

    if (
        worktree_path
        and not session_owned
        and not safe_reuse
        and (destructive or session.get("validation_ran"))
        and not _is_session_owned_path(worktree_path)
    ):
        reasons.append(
            "reused branch-named worktree requires safe-reuse proof before reset or validation"
        )

    proven = not reasons
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "worktree_path": worktree_path or None,
        "session_owned": session_owned,
        "safe_reuse": safe_reuse,
        "destructive_commands": destructive,
        "safe_next_action": (
            "use a fresh session-owned review worktree under branches/review-pr<N>-* "
            "or document full safe-reuse proof before destructive reset"
            if reasons
            else "proceed"
        ),
    }