Gitea-Tools/reviewer_mutation_workspace.py

"""Reviewer mutation workspace binding (#503).

Reviewer mutation tools must validate the same branches/ worktree that
read-only preflight and runtime_context inspect — not only the MCP server
process root.
"""

from __future__ import annotations

import os

import author_mutation_worktree as amw

ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV


def resolve_reviewer_mutation_workspace(
    *,
    worktree_path: str | None,
    worktree: str | None,
    process_project_root: str,
    active_worktree_env: str | None = None,
    session_lease_worktree: str | None = None,
) -> str:
    """Resolve the workspace inspected before reviewer mutations."""
    for candidate in (
        worktree_path,
        worktree,
        active_worktree_env,
        session_lease_worktree,
    ):
        text = (candidate or "").strip()
        if text:
            return os.path.realpath(os.path.abspath(text))
    return os.path.realpath(process_project_root)


def assess_metadata_only_worktree_binding(
    *,
    declared_worktree_path: str | None,
    mutation_workspace: str,
    process_project_root: str,
) -> dict:
    """Fail closed when declared worktree_path would not redirect mutations."""
    declared = (declared_worktree_path or "").strip()
    process_root = os.path.realpath(process_project_root)
    mutation_root = os.path.realpath(mutation_workspace)
    if not declared:
        return {"block": False, "reasons": [], "metadata_only": False}

    declared_root = os.path.realpath(os.path.abspath(declared))
    if declared_root == mutation_root:
        return {"block": False, "reasons": [], "metadata_only": False}

    if declared_root != process_root and mutation_root == process_root:
        return {
            "block": True,
            "metadata_only": True,
            "reasons": [
                "worktree_path is metadata-only for reviewer mutations: preflight "
                f"inspected '{declared_root}' but mutation tools would still "
                f"validate MCP server process root '{process_root}'"
            ],
            "declared_worktree_path": declared_root,
            "mutation_workspace": mutation_root,
            "process_project_root": process_root,
        }

    return {"block": False, "reasons": [], "metadata_only": False}


def format_stale_mcp_workspace_binding_error(
    *,
    process_project_root: str,
    mutation_workspace: str,
    declared_worktree_path: str | None = None,
    metadata_only: bool = False,
) -> str:
    """Canonical handoff when reviewer mutations bind to the wrong workspace."""
    process_root = os.path.realpath(process_project_root)
    mutation_root = os.path.realpath(mutation_workspace)
    declared = (declared_worktree_path or "").strip()
    declared_root = (
        os.path.realpath(os.path.abspath(declared)) if declared else None
    )

    if metadata_only and declared_root:
        lead = (
            "Reviewer mutation workspace mismatch (#503): worktree_path affected "
            f"read-only preflight only ('{declared_root}') while mutation tools "
            f"still inspect MCP server process root '{process_root}'."
        )
    elif mutation_root == process_root:
        lead = (
            "Reviewer mutation workspace mismatch (#503): mutation tools are "
            f"bound to the stable control checkout '{process_root}' instead of "
            "an active branches/ review worktree."
        )
    else:
        lead = (
            "Reviewer mutation workspace mismatch (#503): mutation workspace "
            f"'{mutation_root}' does not match the declared active task worktree."
        )

    fixes = (
        "Fix: relaunch or reconnect the MCP server from the branches/ review "
        "worktree, set GITEA_ACTIVE_WORKTREE to that path, pass worktree_path "
        "to reviewer mutation tools, or acquire a reviewer PR lease from the "
        "same branches/ worktree so mutations inherit the validated workspace."
    )
    return f"{lead} {fixes}"


def assess_reviewer_mutation_workspace(
    *,
    worktree_path: str | None,
    worktree: str | None,
    process_project_root: str,
    active_worktree_env: str | None = None,
    session_lease_worktree: str | None = None,
    current_branch: str | None = None,
) -> dict:
    """Evaluate reviewer mutation workspace binding before preflight/mutation."""
    mutation_workspace = resolve_reviewer_mutation_workspace(
        worktree_path=worktree_path,
        worktree=worktree,
        process_project_root=process_project_root,
        active_worktree_env=active_worktree_env,
        session_lease_worktree=session_lease_worktree,
    )
    process_root = os.path.realpath(process_project_root)
    metadata = assess_metadata_only_worktree_binding(
        declared_worktree_path=worktree_path,
        mutation_workspace=mutation_workspace,
        process_project_root=process_root,
    )
    reasons = list(metadata.get("reasons") or [])
    canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
    branches = amw.assess_author_mutation_worktree(
        workspace_path=mutation_workspace,
        project_root=canonical_root,
        current_branch=current_branch,
    )
    if branches["block"]:
        reasons.extend(branches["reasons"])

    block = bool(reasons)
    return {
        "block": block,
        "reasons": reasons,
        "mutation_workspace": mutation_workspace,
        "process_project_root": process_root,
        "canonical_repo_root": canonical_root,
        "metadata_only": metadata.get("metadata_only", False),
        "declared_worktree_path": metadata.get("declared_worktree_path"),
        "under_branches": branches.get("under_branches", False),
    }