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


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

# 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"


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.
    """
    forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
    if forced is not None:
        return forced.strip() or 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'))}"