Gitea-Tools/root_checkout_guard.py

"""Root checkout guard (#475).

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

The tracking ref is *derived per repository* (#983) rather than assumed to be
``prgs/master``: a namespace bound to another repository — say remote ``MDCPS``
on branch ``dev`` — is gated against ``refs/remotes/MDCPS/dev``. Derivation is
delegated to :mod:`canonical_repository_root`, the authoritative
repository/context resolver, so gating and parity reporting share one resolved
target instead of maintaining two disagreeing hardcoded defaults.
"""

from __future__ import annotations

import os
import subprocess

import canonical_repository_root
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"})

# Legacy PRGS-specific probe order. Retained only for callers that pass an
# explicit ``remote_refs`` override; it is no longer the silent default, because
# inheriting it in a non-PRGS checkout compared that checkout against a ref it
# can never have (#983).
REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master")


def _derive_probe_refs(root: str, explicit_remote: str | None) -> dict:
    """Derive the ordered tracking refs to probe for *root*.

    Returns the derivation payload from :mod:`canonical_repository_root` plus a
    ``refs`` tuple, which is empty when the target is not provable.
    """
    derived = canonical_repository_root.resolve_target_base_ref(
        root, explicit_remote=explicit_remote
    )
    return {
        "refs": tuple(derived.get("tracking_refs") or ()),
        "remote": derived.get("remote"),
        "branch": derived.get("branch"),
        "source": derived.get("source"),
        "proven": bool(derived.get("proven")),
        "reason_code": derived.get("reason_code"),
        "reasons": list(derived.get("reasons") or []),
        "cached_remote_head_branch": derived.get("cached_remote_head_branch"),
        "cached_remote_head_conflicts": bool(derived.get("cached_remote_head_conflicts")),
    }


def resolve_remote_master_sha(
    canonical_repo_root: str,
    *,
    remote_refs: tuple[str, ...] | None = None,
    explicit_remote: str | None = None,
) -> str | None:
    """Return the commit SHA for the tracking integration ref when available.

    This remains the single place that turns a ref into a SHA. Returns None when
    the target cannot be derived or resolved — exactly what this function already
    returned when ``rev-parse`` failed. Callers that must fail closed on missing
    evidence (the #749/#757 bootstrap path) surface that None as *missing
    evidence*, never as "no constraint".

    *explicit_remote* is a caller-supplied disambiguation and is named that way
    deliberately: an internally inferred remote handed back in would suppress the
    resolver's ambiguity gate (#983 B2). No production caller supplies it.
    """
    root = (canonical_repo_root or "").strip()
    if not root:
        return None
    if remote_refs:
        probe: tuple[str, ...] = tuple(remote_refs)
    else:
        derived = _derive_probe_refs(root, explicit_remote)
        if not derived["proven"]:
            return None
        probe = derived["refs"]
    for ref in probe:
        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


def resolve_remote_master_ref_state(
    canonical_repo_root: str,
    *,
    remote_refs: tuple[str, ...] | None = None,
    explicit_remote: str | None = None,
) -> dict:
    """Resolve the tracking integration ref together with its commit SHA.

    Returns ``sha``, the ``ref`` it came from, the derived ``remote`` /
    ``branch``, a machine-checkable ``reason_code``, and ``reasons``. ``sha`` is
    None whenever the target cannot be resolved — never a fallback to some other
    repository's commit.

    The SHA itself is obtained through :func:`resolve_remote_master_sha` rather
    than by re-probing here, so one public function stays authoritative for
    ref-to-SHA resolution. An explicit *remote_refs* keeps the historical
    behaviour exactly: those refs are probed in order and no derivation happens.
    """
    root = (canonical_repo_root or "").strip()
    state: dict = {
        "sha": None,
        "ref": None,
        "remote": None,
        "branch": None,
        "source": None,
        "reason_code": None,
        "reasons": [],
        "cached_remote_head_branch": None,
        "cached_remote_head_conflicts": False,
    }
    if not root:
        state["reasons"].append("no canonical repository root supplied (fail closed)")
        return state

    if remote_refs:
        probe: tuple[str, ...] = tuple(remote_refs)
        state["source"] = "explicit_remote_refs"
    else:
        derived = _derive_probe_refs(root, explicit_remote)
        state["remote"] = derived["remote"]
        state["branch"] = derived["branch"]
        state["source"] = derived["source"]
        state["cached_remote_head_branch"] = derived["cached_remote_head_branch"]
        state["cached_remote_head_conflicts"] = derived["cached_remote_head_conflicts"]
        if not derived["proven"]:
            state["reason_code"] = derived["reason_code"]
            state["reasons"] = derived["reasons"]
            return state
        probe = derived["refs"]

    sha = resolve_remote_master_sha(root, remote_refs=probe, explicit_remote=explicit_remote)
    if not sha:
        state["reasons"].append(
            "tracking integration ref "
            f"{' / '.join(probe) if probe else '(none derived)'} does not resolve in "
            f"'{root}' (fail closed)"
        )
        return state

    state["sha"] = sha
    # Name the ref that actually carries this commit. When the SHA comes from a
    # test double no probe will match, so fall back to the first derived ref,
    # which is the one the guard is conceptually comparing against.
    for ref in probe:
        res = subprocess.run(
            ["git", "-C", root, "rev-parse", "--verify", ref],
            capture_output=True,
            text=True,
            check=False,
        )
        if res.returncode == 0 and (res.stdout or "").strip() == sha:
            state["ref"] = ref
            break
    else:
        state["ref"] = probe[0] if probe else None
    return state


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,
    remote_master_ref: str | None = None,
) -> dict:
    """Fail closed when the control checkout is not clean on its integration ref.

    ``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:
        # #983: name the ref that was actually compared. Reporting a literal
        # 'prgs/master' in a checkout gated against refs/remotes/MDCPS/dev sends
        # the operator to inspect a ref that repository does not have.
        ref_label = (remote_master_ref or "").strip() or "the tracking integration ref"
        reasons.append(
            f"control checkout HEAD does not match {ref_label} "
            f"(HEAD {head_sha[:12]}, {ref_label} {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