Gitea-Tools/master_parity_gate.py

"""Master-parity staleness gate (#420).

The Gitea MCP server loads its capability-gate code and workflow logic into
memory when the process starts. When ``master`` advances -- for example a newly
merged security gate such as the branch-delete capability gate (#408/#410) --
the running process keeps executing the *old* code until it is restarted. A
stale server can therefore still perform a mutation that the updated codebase
would forbid.

Runtime profile/config data is already read live from disk on every call
(``gitea_config.load_config`` re-reads the JSON file each time), so profile
``allowed_operations`` changes take effect immediately without a restart. The
gap this module closes is *code* parity: it captures the server process's
source-tree commit at startup and detects, at mutation time, when the on-disk
``master`` HEAD has advanced past it. Detected staleness fails closed with a
restart-required recovery report, while read-only operations stay allowed so a
stale server can still be inspected.

The core assessment is pure -- callers inject the observed HEAD SHAs -- so the
logic is fully unit-testable without a git checkout.
"""

from __future__ import annotations

import os
import subprocess
import time

# Live-remote head cache: the parity gate runs on every mutation and every
# runtime-context read, so the ``git ls-remote`` result is cached briefly to
# avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
_REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
_REMOTE_HEAD_TTL = 60.0

# When True, ``read_remote_master_head`` never performs ``git ls-remote`` unless
# ``GITEA_TEST_LIVE_REMOTE_HEAD`` is set. Conftest enables this suite-wide so
# feature worktrees (whose HEAD differs from live master) cannot flip legacy
# runtime-context assertions to live_stale, and so unit tests never depend on
# a live network (PR #788 F1/F2 / issue #610). Module-level (not env-only) so
# ``patch.dict(os.environ, …, clear=True)`` cannot re-enable the probe.
_HERMETIC_TEST_MODE: bool = False


def _clear_remote_head_cache() -> None:
    """Reset the live-remote head cache (test isolation / forced refresh)."""
    _REMOTE_HEAD_CACHE.clear()


def set_hermetic_test_mode(enabled: bool) -> None:
    """Enable or disable suite-wide hermetic live-remote reads (tests only)."""
    global _HERMETIC_TEST_MODE
    _HERMETIC_TEST_MODE = bool(enabled)
    _clear_remote_head_cache()


def hermetic_test_mode() -> bool:
    """Return whether hermetic live-remote reads are active."""
    return bool(_HERMETIC_TEST_MODE)


# Environment escape hatches (ops + tests):
#   GITEA_MCP_DISABLE_PARITY_GATE  -> disable enforcement entirely (fail open).
#   GITEA_TEST_CURRENT_HEAD        -> force the "current" HEAD read, for tests.
ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
# GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
# GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE -> opt a single test into a real ls-remote
# even when hermetic mode is on (rare; prefer ENV_TEST_LIVE_REMOTE_HEAD).
ENV_TEST_ALLOW_LIVE_REMOTE_PROBE = "GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE"


def read_git_head(root: str) -> str | None:
    """Return the current ``HEAD`` commit SHA of *root*, or ``None``.

    ``None`` means the SHA could not be determined (not a git checkout, git
    unavailable, or an error). A test override via ``GITEA_TEST_CURRENT_HEAD``
    takes precedence so the gate can be exercised deterministically.
    """
    forced = os.environ.get(ENV_TEST_CURRENT_HEAD)
    if forced is not None:
        return forced.strip() or None
    if not root:
        return None
    try:
        res = subprocess.run(
            ["git", "-C", root, "rev-parse", "HEAD"],
            capture_output=True,
            text=True,
            check=False,
        )
    except Exception:
        return None
    if res.returncode != 0:
        return None
    return (res.stdout or "").strip() or None


def read_remote_master_head(
    root: str,
    remote: str = "origin",
    branch: str = "master",
    ttl: float = _REMOTE_HEAD_TTL,
) -> str | None:
    """Return the live remote ``branch`` commit SHA, or ``None`` (#610).

    Resolves the *live* target commit via ``git ls-remote`` so parity can tell
    a daemon that is behind the live remote master apart from one whose local
    checkout simply hasn't been pulled. ``None`` means the live head could not
    be resolved (offline, no such remote, git unavailable, error) -- callers
    must treat unknown live state as *not mutation-safe* while never blocking
    read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
    precedence so the wiring can be exercised deterministically and offline.

    The result is cached for *ttl* seconds per (root, remote, branch) so the
    gate does not run a network probe on every mutation/read (``ttl=0`` forces
    a live probe). Both hits and ``None`` misses are cached to bound offline
    latency; the env override bypasses the cache and the subprocess entirely.

    Under suite hermetic mode (``set_hermetic_test_mode(True)``, set by
    conftest) a missing override returns ``None`` without network I/O so
    feature-worktree test runs cannot observe live_stale against real master
    (PR #788 F1) and unit tests stay offline (F2). Opt out with an explicit
    ``GITEA_TEST_LIVE_REMOTE_HEAD`` pin or ``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``.
    """
    forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
    if forced is not None:
        return forced.strip() or None
    if _HERMETIC_TEST_MODE and not (
        os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or ""
    ).strip():
        # Hermetic default: live head unknown. live_stale stays False;
        # mutation_safe is False when live is unknown (documented #610 note).
        return None
    # Defense in depth: even without the module flag, never probe while pytest
    # is running unless the test opted into a real probe or set an override.
    if (os.environ.get("PYTEST_CURRENT_TEST") or "").strip() and not (
        os.environ.get(ENV_TEST_ALLOW_LIVE_REMOTE_PROBE) or ""
    ).strip():
        return None
    if not root:
        return None
    key = (root, remote, branch)
    now = time.monotonic()
    if ttl > 0:
        cached = _REMOTE_HEAD_CACHE.get(key)
        if cached is not None and (now - cached[0]) < ttl:
            return cached[1]
    sha: str | None = None
    try:
        res = subprocess.run(
            ["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
            capture_output=True,
            text=True,
            check=False,
            timeout=5,
        )
        if res.returncode == 0:
            lines = (res.stdout or "").strip().splitlines()
            if lines:
                sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
    except Exception:
        sha = None
    _REMOTE_HEAD_CACHE[key] = (now, sha)
    return sha


def capture_startup_parity(root: str, head: str | None = None) -> dict:
    """Capture the process source-tree baseline once at server startup.

    *head* may be injected (tests); otherwise it is read from *root*. The result
    is an opaque baseline handed back to :func:`assess_master_parity`.
    """
    startup_head = head if head is not None else read_git_head(root)
    return {"root": root, "startup_head": startup_head}


def _short(sha: str | None) -> str:
    return sha[:12] if sha else "unknown"


def assess_master_parity(
    startup: dict | None,
    current_head: str | None,
    live_remote_head: str | None = None,
) -> dict:
    """Compare the startup baseline against the current on-disk ``HEAD``.

    Pure: all HEADs are supplied by the caller. Returns a structured result:

    - ``in_parity``    -- server code matches the on-disk master (or parity
      could not be determined, which is not treated as stale).
    - ``stale``        -- the on-disk master has definitively advanced past the
      running process.
    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
    - ``determinable`` -- whether both local HEADs were known well enough to
      compare.
    - ``startup_head`` / ``current_head`` / ``reasons``.

    #610 adds live-remote awareness so a daemon that is stale relative to the
    *live* remote master cannot report a mutation-safe result even when the
    local checkout HEAD still matches the daemon's startup commit:

    - ``daemon_start_head`` -- the commit the running process started at
      (alias of ``startup_head``, named for clarity in reports).
    - ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
    - ``live_remote_head`` -- the live remote target commit, or ``None`` when it
      could not be fetched.
    - ``live_known`` -- whether the live remote target was resolved.
    - ``live_stale`` -- the live remote master has advanced past the running
      process (daemon is behind live master) even if local parity is green.
    - ``mutation_safe`` -- the daemon code, local checkout, and live remote
      target all agree; the only state in which a mutation may rely on parity.
    """
    startup_head = (startup or {}).get("startup_head")
    reasons: list[str] = []

    if startup_head is None:
        reasons.append(
            "startup commit was not captured; code parity cannot be enforced")
        return _result(True, False, False, startup_head, current_head,
                       live_remote_head, False, reasons)

    if current_head is None:
        reasons.append(
            "current workspace HEAD could not be read; code parity cannot be "
            "enforced")
        return _result(True, False, False, startup_head, current_head,
                       live_remote_head, False, reasons)

    local_in_parity = startup_head == current_head
    local_stale = not local_in_parity
    if local_stale:
        reasons.append(
            f"MCP server started at commit {_short(startup_head)} but the "
            f"workspace master is now {_short(current_head)}; restart the "
            f"server to load the current capability gates")

    live_known = live_remote_head is not None
    live_stale = live_known and live_remote_head != startup_head
    if live_stale:
        reasons.append(
            f"live remote master is {_short(live_remote_head)} but the MCP "
            f"server started at {_short(startup_head)}; the daemon is stale "
            f"relative to live master -- restart/reconnect before mutating")

    return _result(
        local_in_parity, local_stale, True, startup_head, current_head,
        live_remote_head, live_stale, reasons)


def _result(in_parity, stale, determinable, startup_head, current_head,
            live_remote_head, live_stale, reasons):
    live_known = live_remote_head is not None
    mutation_safe = (
        determinable and in_parity and live_known and not live_stale)
    return {
        "in_parity": in_parity,
        "stale": stale,
        "restart_required": stale or live_stale,
        "determinable": determinable,
        "startup_head": startup_head,
        "current_head": current_head,
        # #610 distinguished signals:
        "daemon_start_head": startup_head,
        "local_head": current_head,
        "live_remote_head": live_remote_head,
        "live_known": live_known,
        "live_stale": live_stale,
        "mutation_safe": mutation_safe,
        "reasons": list(reasons),
    }


def gate_disabled() -> bool:
    """Whether the parity gate is disabled by env escape hatch."""
    return bool((os.environ.get(ENV_DISABLE) or "").strip())


def parity_block_reasons(assessment: dict) -> list[str]:
    """Block reasons for a mutation gate (empty when the mutation may proceed).

    A disabled gate or an in-parity / non-determinable assessment yields no
    reasons. A definitively stale server blocks, and (#610) a daemon that is
    stale relative to the *live* remote master blocks even when the local
    checkout HEAD still matches the daemon's startup commit.
    """
    if gate_disabled():
        return []
    if assessment.get("stale") or assessment.get("live_stale"):
        return list(assessment.get("reasons") or
                    ["server code is stale relative to master (fail closed)"])
    return []


def parity_report(assessment: dict) -> dict:
    """Structured stale-server report for permission-block payloads."""
    return {
        "kind": "server_stale",
        "restart_required": True,
        "startup_head": assessment.get("startup_head"),
        "current_head": assessment.get("current_head"),
        # #610: name the live remote target so the report distinguishes a
        # local-code stale from a daemon-behind-live-master stale.
        "live_remote_head": assessment.get("live_remote_head"),
        "live_stale": bool(assessment.get("live_stale")),
        "reasons": list(assessment.get("reasons") or []),
        "recovery": [
            "The running MCP server is executing code older than the current "
            "master and may not enforce newly merged capability gates.",
            "Restart the Gitea MCP server so it reloads master's capability "
            "gates and execution profiles before retrying the mutation.",
        ],
    }


def parity_resolver_disagreement(
    assessment: dict,
    resolver_restart_required: bool,
) -> dict | None:
    """Typed blocker when the resolver requires restart but parity looks green.

    The capability resolver (``gitea_resolve_task_capability``) detects stale
    runtime authoritatively for mutation safety (#610). When it requires a
    restart, local-only parity must never override it: this returns a typed,
    fail-closed blocker that names the resolver as authoritative. Returns
    ``None`` when the resolver does not require a restart.
    """
    if not resolver_restart_required:
        return None
    parity_optimistic = bool(assessment.get("in_parity")) and not (
        assessment.get("stale") or assessment.get("live_stale"))
    return {
        "kind": "parity_resolver_disagreement",
        "restart_required": True,
        "resolver_authoritative": True,
        "parity_optimistic": parity_optimistic,
        "daemon_start_head": assessment.get("daemon_start_head"),
        "local_head": assessment.get("local_head"),
        "live_remote_head": assessment.get("live_remote_head"),
        "reasons": [
            "The capability resolver requires a restart/reconnect (stale "
            "runtime) but master-parity reported local code as in-parity. "
            "The resolver is authoritative for mutation safety; do not mutate "
            "on local parity alone. Restart/reconnect the Gitea MCP server "
            "and re-verify before mutating.",
        ],
        "recovery": [
            "Trust the resolver: treat this session as stale.",
            "Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
            "current master and live target state, then re-run preflight.",
        ],
    }


def format_parity(assessment: dict) -> str:
    """One-line human summary for logs / runtime context."""
    if assessment.get("stale"):
        return (f"STALE: started {_short(assessment.get('startup_head'))}, "
                f"master now {_short(assessment.get('current_head'))} "
                f"(restart required)")
    if not assessment.get("determinable"):
        return "parity indeterminate (baseline or current HEAD unknown)"
    return f"in parity at {_short(assessment.get('current_head'))}"


# ---------------------------------------------------------------------------
# Target-repository parity (#739 F3)
#
# Everything above measures ONE dimension: the Gitea-Tools server's own
# implementation commit, comparing the SHA this process was loaded from against
# the SHA now on disk at PROJECT_ROOT. That is deliberate and is left untouched
# — it is what proves the in-memory capability gates are current.
#
# It is not, however, a statement about the repository a cross-repository
# namespace actually mutates. The assessment below is a separate, separately
# labelled dimension for the configured canonical target repository. It never
# feeds the mutation gate and never changes startup_head/current_head.
# ---------------------------------------------------------------------------

# Legacy hardcoded target tracking ref. Retained for callers that still pass an
# explicit ref, but no longer the default: assuming a remote named ``origin``
# read an unrelated (often orphaned) remote-tracking ref in any checkout whose
# remote is named something else, and reported the target stale against a commit
# from a remote that may no longer even be configured (#983).
DEFAULT_TARGET_TRACKING_REF = "refs/remotes/origin/master"


def _git_capture(root: str, *args: str) -> str | None:
    """Run a read-only git command in *root*; ``None`` on any failure.

    Deliberately does not honour ``GITEA_TEST_CURRENT_HEAD``: that override
    exists to pin the *server's* HEAD, and applying it here would make a target
    repository silently report the server's forced SHA.
    """
    if not root:
        return None
    try:
        res = subprocess.run(
            ["git", "-C", root, *args],
            capture_output=True,
            text=True,
            check=False,
        )
    except Exception:
        return None
    if res.returncode != 0:
        return None
    return (res.stdout or "").strip() or None


def assess_target_repository_parity(
    *,
    canonical_root: str | None,
    source: str | None,
    tracking_ref: str | None = None,
) -> dict:
    """Assess the configured cross-repository target checkout.

    Reports the target's canonical root, repository identity, checked-out
    commit, and last-known remote master commit, plus whether the checkout is
    behind that ref. No network call is made: the remote side is read from the
    existing remote-tracking ref, so a target that has never been fetched is
    reported as indeterminate rather than guessed at.

    An unconfigured namespace is ``configured=False`` and never ``stale`` — the
    single-repository default has no second dimension to be stale about. A
    configured root that cannot be read is ``determinable=False`` with reasons.

    *tracking_ref* defaults to None, meaning **derive the target from the
    repository itself** through the same resolver the mutation guard uses, so
    gating and reporting can never disagree about which ref is authoritative
    (#983). Passing an explicit ref preserves the previous behaviour verbatim.
    """
    result = {
        "configured": bool(canonical_root),
        "canonical_repository_root": None,
        "source": source,
        "repository_slug": None,
        "checkout_head": None,
        "tracking_ref": tracking_ref,
        "base_remote": None,
        "base_branch": None,
        "tracking_ref_source": "explicit_tracking_ref" if tracking_ref else None,
        "remote_tracking_head": None,
        "determinable": False,
        "stale": False,
        "reason_code": None,
        "reasons": [],
    }
    if not canonical_root:
        result["determinable"] = True
        return result

    result["canonical_repository_root"] = canonical_root
    if not os.path.isdir(canonical_root):
        result["reasons"].append(
            f"configured canonical repository root '{canonical_root}' "
            f"does not exist or is not a directory"
        )
        return result

    toplevel = _git_capture(canonical_root, "rev-parse", "--show-toplevel")
    if not toplevel:
        result["reasons"].append(
            f"configured canonical repository root '{canonical_root}' "
            f"is not a git checkout"
        )
        return result

    head = _git_capture(canonical_root, "rev-parse", "HEAD")
    if not head:
        result["reasons"].append(
            f"target repository HEAD could not be read at '{canonical_root}'"
        )
        return result
    result["checkout_head"] = head
    result["determinable"] = True

    # Local import keeps this module dependency-light for its startup role.
    import canonical_repository_root as _crr

    # #983: identity comes from whichever remote actually proves it, not from a
    # remote assumed to be named 'origin'. In a checkout whose only remote is
    # 'prgs', the old lookup failed outright and reported the identity as
    # underivable while a leftover refs/remotes/origin/master still resolved.
    #
    # #983 B2: this must be the *ambiguity-aware* resolver. The first-wins
    # `resolve_identity_remote` picks whichever remote probes first, so on a
    # target where distinct remotes claim different repositories the report
    # confidently named one of them while the mutation gate refused the same
    # target — gating and reporting evaluating different repositories, which is
    # precisely the divergence this issue exists to end.
    identity = _crr.assess_identity_remote(canonical_root)
    result["repository_slug"] = identity["slug"]
    if not identity["slug"]:
        result["reason_code"] = identity["reason_code"]
        result["reasons"].extend(
            identity["reasons"]
            or ["target repository identity could not be derived from its git remote"]
        )
    if identity["ambiguous"]:
        # An ambiguous target has no single authoritative base, so reporting one
        # would be a guess. Fail closed here exactly as the gate does.
        return result

    # Identity is otherwise resolved independently of the base ref: a target that
    # has never been fetched still has a provable repository identity, and
    # reporting it as unidentifiable would lose real information over an
    # unrelated missing ref.
    if not tracking_ref:
        # No remote argument. The identity remote resolved just above was
        # *inferred here*, and feeding it back in would tell the resolver a
        # caller had explicitly disambiguated the repository, suppressing its
        # ambiguity gate (#983 B2). Only an operator-supplied remote may do that,
        # and this call site has none.
        base = _crr.resolve_target_base_ref(canonical_root)
        if not base.get("proven"):
            result["reason_code"] = base.get("reason_code")
            result["reasons"].extend(base.get("reasons") or [])
            return result
        tracking_ref = base["tracking_ref"]
        result["tracking_ref"] = tracking_ref
        result["tracking_ref_source"] = base.get("source")
        result["base_remote"] = base.get("remote")
        result["base_branch"] = base.get("branch")

    tracking_head = _git_capture(canonical_root, "rev-parse", tracking_ref)
    if not tracking_head:
        result["reasons"].append(
            f"remote-tracking ref '{tracking_ref}' is unknown in the target "
            f"checkout; target staleness is indeterminate (no fetch is "
            f"performed by this assessment)"
        )
        return result
    result["remote_tracking_head"] = tracking_head
    result["stale"] = tracking_head != head
    return result