Gitea-Tools/root_checkout_guard.py

"""Root checkout guard (#475).

The project root checkout is the stable control checkout on master/prgs/master.
Author/reviewer/merge flows must fail closed when the control checkout is
contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of
prgs/master). Isolated ``branches/...`` worktrees remain allowed.
"""

from __future__ import annotations

import os
import subprocess

from author_mutation_worktree import is_path_under_branches
from reviewer_worktree import parse_dirty_tracked_files

REMEDIATION = (
    "Root checkout is not on master. Preserve state, switch root back to master, "
    "and use scripts/worktree-review or the sanctioned issue worktree flow."
)

BASE_BRANCHES = frozenset({"master", "main", "dev"})
REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master")


def resolve_remote_master_sha(
    canonical_repo_root: str,
    *,
    remote_refs: tuple[str, ...] | None = None,
) -> str | None:
    """Return the commit SHA for the tracking master ref when available."""
    root = (canonical_repo_root or "").strip()
    if not root:
        return None
    for ref in remote_refs or REMOTE_MASTER_REFS:
        res = subprocess.run(
            ["git", "-C", root, "rev-parse", "--verify", ref],
            capture_output=True,
            text=True,
            check=False,
        )
        if res.returncode == 0:
            sha = (res.stdout or "").strip()
            if sha:
                return sha
    return None


resolve_tracking_master_sha = resolve_remote_master_sha


def assess_root_checkout_guard(
    *,
    workspace_path: str,
    canonical_repo_root: str,
    current_branch: str | None,
    head_sha: str | None,
    porcelain_status: str,
    remote_master_sha: str | None,
    resolved_role: str | None = None,
    actual_role: str | None = None,
) -> dict:
    """Fail closed when the control checkout is not clean master/prgs/master.

    ``resolved_role`` is the preflight-resolved *task* role and ``actual_role``
    is the *active profile* role (#540). The reconciler exemption honours either
    signal so a ``comment_issue`` preflight (which stamps the task role as
    ``author``) cannot strip a genuine reconciler of its exemption. An actual
    author profile classifies as ``author`` in both signals, so author blocking
    on a contaminated control checkout is preserved.

    Merger *strictness* (a merger must not be auto-exempted by working from a
    ``branches/`` worktree) stays keyed on the resolved task role: merge
    operations resolve their own task role, and widening the merger test with
    ``actual_role`` would wrongly subject a merger operating from its clean
    workspace under a non-merge task to full control-checkout checks.
    """
    reasons: list[str] = []
    root = os.path.realpath(canonical_repo_root)
    workspace = os.path.realpath(workspace_path)
    branch = (current_branch or "").strip()
    dirty_files = parse_dirty_tracked_files(porcelain_status)

    if resolved_role == "reconciler" or actual_role == "reconciler":
        return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)

    if resolved_role != "merger" and is_path_under_branches(workspace, root):
        return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)

    if dirty_files:
        reasons.append(
            "control checkout has tracked local edits before role work "
            f"(dirty files: {', '.join(dirty_files)})"
        )

    if not branch:
        reasons.append("control checkout is detached HEAD; expected branch 'master'")
    elif branch not in BASE_BRANCHES:
        reasons.append(
            f"control checkout branch '{branch}' is not a stable base branch "
            f"({'/'.join(sorted(BASE_BRANCHES))})"
        )

    if remote_master_sha and head_sha and head_sha != remote_master_sha:
        reasons.append(
            "control checkout HEAD does not match prgs/master "
            f"(HEAD {head_sha[:12]}, prgs/master {remote_master_sha[:12]})"
        )

    proven = not reasons
    return _assessment(proven, reasons, root, workspace, branch or None, head_sha, dirty_files)


def format_root_checkout_guard_error(assessment: dict) -> str:
    """Single RuntimeError message for MCP preflight gates."""
    root = assessment.get("canonical_repo_root") or "(unknown)"
    workspace = assessment.get("workspace_path") or "(unknown)"
    reasons = "; ".join(assessment.get("reasons") or ["unknown root checkout violation"])
    return (
        f"Root checkout guard (#475): {reasons}. "
        f"canonical repository root: {root}; workspace: {workspace}. "
        f"{REMEDIATION}"
    )


def _assessment(
    proven: bool,
    reasons: list[str],
    canonical_repo_root: str,
    workspace_path: str,
    current_branch: str | None,
    head_sha: str | None,
    dirty_files: list[str],
) -> dict:
    return {
        "proven": proven,
        "block": not proven,
        "reasons": reasons,
        "canonical_repo_root": canonical_repo_root,
        "workspace_path": workspace_path,
        "current_branch": current_branch,
        "head_sha": head_sha,
        "dirty_files": dirty_files,
        "remediation": REMEDIATION,
    }


assess_root_checkout = assess_root_checkout_guard