feat: make master-parity live-remote aware so a stale daemon fails closed (Closes #610)

Master-parity previously compared only the daemon's startup commit against the
local on-disk HEAD. When the checkout was not pulled, parity reported green even
though the live remote master had advanced, so a stale daemon could claim a
mutation-safe result while running outdated capability gates (observed during
PR #592 recovery, where the resolver correctly required restart but parity said
in_parity=true).

Changes:
- master_parity_gate.assess_master_parity() gains an optional live_remote_head
  and reports the three commits distinctly (daemon_start_head, local_head,
  live_remote_head) plus live_known / live_stale / mutation_safe. A result is
  mutation_safe only when daemon, local checkout, and live remote all agree.
- parity_block_reasons() now blocks mutations on live-staleness too; read-only
  operations remain unblocked (non-goal: never block diagnostics offline).
- New parity_resolver_disagreement(): typed fail-closed blocker naming the
  capability resolver as authoritative when it requires restart but parity
  looks locally green.
- New read_remote_master_head(): best-effort `git ls-remote` for the live
  target, cached with a 60s TTL (bounded offline latency, no network probe per
  gate call); env override GITEA_TEST_LIVE_REMOTE_HEAD keeps tests hermetic.
- Server: _current_master_parity() reads the live remote head;
  gitea_assess_master_parity and runtime_context surface the distinguished
  SHAs, mutation_safe, and resolver-authoritative guidance.

Tests: 13 new cases (live-remote parity, live-stale blocking + typed blocker,
remote-head reader + TTL cache, server wiring). Full suite: 2423 passed; the 8
remaining failures (test_config TestAuthIntegration, test_credentials
TestGetCredentials) are baseline-proven keychain/env failures identical on the
unmodified base.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

gitea_mcp_server.py

@@ -6609,10 +6609,39 @@ def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
     return False
 
 
+def _git_default_remote_name(root: str) -> str:
+    """First configured git remote name for *root*, defaulting to 'origin'.
+
+    Used to resolve the live remote master target for parity (#610). Best
+    effort: any failure falls back to 'origin' so callers never raise.
+    """
+    try:
+        res = subprocess.run(
+            ["git", "-C", root, "remote"],
+            capture_output=True, text=True, check=False,
+        )
+    except Exception:
+        return "origin"
+    if res.returncode != 0:
+        return "origin"
+    names = [n.strip() for n in (res.stdout or "").splitlines() if n.strip()]
+    return names[0] if names else "origin"
+
+
 def _current_master_parity() -> dict:
-    """Assess this process's code against the on-disk master HEAD (#420)."""
+    """Assess this process's code against local and live remote master (#420/#610).
+
+    Compares the daemon's startup commit, the on-disk checkout HEAD, and the
+    live remote master target. A stale daemon relative to live master fails
+    closed for mutations even when the local checkout HEAD still matches the
+    startup commit. The live-remote read is best effort: an unresolved live
+    head leaves read-only diagnostics unblocked but is never mutation-safe.
+    """
     current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
-    return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
+    live_head = master_parity_gate.read_remote_master_head(
+        PROJECT_ROOT, remote=_git_default_remote_name(PROJECT_ROOT))
+    return master_parity_gate.assess_master_parity(
+        _STARTUP_PARITY, current_head, live_remote_head=live_head)
 
 
 def _master_parity_block(op: str) -> list[str]:
@@ -8942,15 +8971,34 @@ def gitea_get_runtime_context(
         "restart_required": parity["restart_required"],
         "startup_head": parity["startup_head"],
         "current_head": parity["current_head"],
+        # #610 distinguished mutation-safety signals:
+        "daemon_start_head": parity["daemon_start_head"],
+        "local_head": parity["local_head"],
+        "live_remote_head": parity["live_remote_head"],
+        "live_known": parity["live_known"],
+        "live_stale": parity["live_stale"],
+        "mutation_safe": parity["mutation_safe"],
         "summary": master_parity_gate.format_parity(parity),
         "mutation_gate_enforced": not master_parity_gate.gate_disabled(),
+        # #610: the capability resolver is authoritative for mutation safety;
+        # local parity alone must never authorize a mutation.
+        "resolver_authoritative_for_mutation_safety": True,
     }
-    if parity["stale"] and not master_parity_gate.gate_disabled():
-        safe_next_action = (
-            "Server code is stale relative to master; restart the Gitea MCP "
-            "server to load current capability gates before mutating. "
-            f"({master_parity_gate.format_parity(parity)})"
-        )
+    if parity["restart_required"] and not master_parity_gate.gate_disabled():
+        if parity["live_stale"]:
+            safe_next_action = (
+                "Daemon is stale relative to LIVE remote master "
+                f"(started {parity['startup_head'][:12] if parity['startup_head'] else 'unknown'}, "
+                f"live master {parity['live_remote_head'][:12] if parity['live_remote_head'] else 'unknown'}); "
+                "restart/reconnect the Gitea MCP server before mutating. The "
+                "capability resolver is authoritative for mutation safety."
+            )
+        else:
+            safe_next_action = (
+                "Server code is stale relative to master; restart the Gitea MCP "
+                "server to load current capability gates before mutating. "
+                f"({master_parity_gate.format_parity(parity)})"
+            )
         result["safe_next_action"] = safe_next_action
 
     if reveal and h:
@@ -8989,12 +9037,19 @@ def gitea_assess_master_parity(
         "determinable": parity["determinable"],
         "startup_head": parity["startup_head"],
         "current_head": parity["current_head"],
+        # #610 distinguished mutation-safety signals:
+        "daemon_start_head": parity["daemon_start_head"],
+        "local_head": parity["local_head"],
+        "live_remote_head": parity["live_remote_head"],
+        "live_known": parity["live_known"],
+        "live_stale": parity["live_stale"],
+        "mutation_safe": parity["mutation_safe"],
         "mutation_gate_enforced": enforced,
         "summary": master_parity_gate.format_parity(parity),
         "reasons": parity["reasons"],
         "process_root": PROJECT_ROOT,
     }
-    if parity["stale"] and enforced:
+    if parity["restart_required"] and enforced:
         out["report"] = master_parity_gate.parity_report(parity)
     return out
 def gitea_record_pre_review_command(

master_parity_gate.py

@@ -24,12 +24,26 @@ 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:
@@ -58,6 +72,57 @@ def read_git_head(root: str) -> str | 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.
 
@@ -72,18 +137,38 @@ def _short(sha: str | None) -> str:
     return sha[:12] if sha else "unknown"
 
 
-def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
+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: both HEADs are supplied by the caller. Returns a structured result:
+    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`` -- alias of ``stale``; the recovery action.
-    - ``determinable`` -- whether both HEADs were known well enough to compare.
+    - ``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] = []
@@ -91,32 +176,56 @@ def assess_master_parity(startup: dict | None, current_head: str | None) -> dict
     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, reasons)
+        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, reasons)
+        return _result(True, False, False, startup_head, current_head,
+                       live_remote_head, False, reasons)
 
-    if startup_head == current_head:
-        return _result(True, False, True, startup_head, current_head, 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")
 
-    reasons.append(
-        f"MCP server started at commit {_short(startup_head)} but the workspace "
-        f"master is now {_short(current_head)}; restart the server to load the "
-        f"current capability gates")
-    return _result(False, True, True, startup_head, current_head, reasons)
+    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, 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,
+        "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),
     }
 
@@ -130,11 +239,13 @@ 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; only a definitively stale server blocks.
+    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"):
+    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 []
@@ -147,6 +258,10 @@ def parity_report(assessment: dict) -> dict:
         "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 "
@@ -157,6 +272,45 @@ def parity_report(assessment: dict) -> dict:
     }
 
 
+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"):

tests/test_master_parity_gate.py

@@ -78,6 +78,95 @@ class TestBlockReasonsAndReport(unittest.TestCase):
         self.assertTrue(report["recovery"])
 
 
+class TestLiveRemoteParity(unittest.TestCase):
+    """#610: parity must account for the live remote master, not just local.
+
+    The daemon can be stale relative to the live remote target while the local
+    checkout HEAD still matches the daemon's startup commit, so local parity
+    reports green even though a mutation would run against outdated code.
+    """
+
+    SHA_C = "c" * 40
+
+    def test_distinguishes_three_shas(self):
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
+        self.assertEqual(res["daemon_start_head"], SHA_A)
+        self.assertEqual(res["local_head"], SHA_A)
+        self.assertEqual(res["live_remote_head"], SHA_B)
+
+    def test_mutation_safe_only_when_all_three_match(self):
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
+        self.assertTrue(res["mutation_safe"])
+        self.assertTrue(res["live_known"])
+        self.assertFalse(res["live_stale"])
+
+    def test_live_stale_when_remote_advanced_past_daemon(self):
+        # Local checkout still matches the daemon start (local parity green),
+        # but the live remote master has advanced -> daemon is live-stale.
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
+        self.assertTrue(res["in_parity"])  # local parity still green
+        self.assertTrue(res["live_stale"])
+        self.assertFalse(res["mutation_safe"])
+        self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
+
+    def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
+        # Non-goal: unfetchable live remote must not be treated as stale for
+        # read-only, but a mutation-safe claim fails closed.
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=None)
+        self.assertFalse(res["live_known"])
+        self.assertFalse(res["mutation_safe"])
+        self.assertFalse(res["live_stale"])
+        self.assertTrue(res["in_parity"])
+
+    def test_default_live_remote_preserves_legacy_shape(self):
+        # Callers that do not supply a live head keep the pre-#610 behavior:
+        # in-parity, not live-stale, no live-derived block.
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
+        self.assertFalse(res["live_stale"])
+        self.assertEqual(mp.parity_block_reasons(res), [])
+
+
+class TestLiveStaleBlockAndReport(unittest.TestCase):
+    """#610: live-staleness must block mutations and surface a typed blocker."""
+
+    def test_live_stale_produces_block_reasons(self):
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
+        self.assertTrue(mp.parity_block_reasons(res))
+
+    def test_disable_env_suppresses_live_stale_block(self):
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
+        with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
+            self.assertEqual(mp.parity_block_reasons(res), [])
+
+    def test_resolver_disagreement_returns_typed_blocker(self):
+        # Parity says local-green, resolver says restart required -> disagreement
+        # is a typed, fail-closed blocker naming the resolver as authoritative.
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
+        blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
+        self.assertIsNotNone(blocker)
+        self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
+        self.assertTrue(blocker["restart_required"])
+        self.assertTrue(blocker["resolver_authoritative"])
+
+    def test_no_disagreement_when_resolver_agrees(self):
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
+        self.assertIsNone(
+            mp.parity_resolver_disagreement(res, resolver_restart_required=False))
+
+    def test_live_stale_report_names_live_remote(self):
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
+        report = mp.parity_report(res)
+        self.assertEqual(report["live_remote_head"], SHA_B)
+        self.assertTrue(report["restart_required"])
+
+
 class TestReadGitHead(unittest.TestCase):
     def test_test_override_takes_precedence(self):
         with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -95,6 +184,78 @@ class TestReadGitHead(unittest.TestCase):
             self.assertIsNone(mp.read_git_head(""))
 
 
+class TestReadRemoteMasterHead(unittest.TestCase):
+    """#610: live remote master head reader (env-overridable, fails to None)."""
+
+    def test_test_override_takes_precedence(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
+            self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
+
+    def test_blank_override_is_none(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: "   "}):
+            self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
+
+    def test_unfetchable_remote_is_none(self):
+        # No override; a bogus root/remote must fail closed to None, never raise.
+        env = {k: v for k, v in os.environ.items()
+               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
+        with patch.dict(os.environ, env, clear=True):
+            self.assertIsNone(
+                mp.read_remote_master_head("/nonexistent", remote="nope"))
+
+
+class TestRemoteHeadCache(unittest.TestCase):
+    """#610: live remote reads are cached with a TTL to stay off the network.
+
+    The parity gate runs on every mutation and every runtime-context read, so an
+    unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
+    """
+
+    def setUp(self):
+        mp._clear_remote_head_cache()
+        env = {k: v for k, v in os.environ.items()
+               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
+        self._env = patch.dict(os.environ, env, clear=True)
+        self._env.start()
+        self.addCleanup(self._env.stop)
+        self.addCleanup(mp._clear_remote_head_cache)
+
+    def _fake_run(self, sha):
+        class _R:
+            returncode = 0
+            stdout = f"{sha}\trefs/heads/master\n"
+        calls = {"n": 0}
+
+        def run(*args, **kwargs):
+            calls["n"] += 1
+            return _R()
+        return run, calls
+
+    def test_second_call_within_ttl_uses_cache(self):
+        run, calls = self._fake_run(SHA_B)
+        with patch.object(mp.subprocess, "run", run):
+            a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
+            b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
+        self.assertEqual(a, SHA_B)
+        self.assertEqual(b, SHA_B)
+        self.assertEqual(calls["n"], 1)
+
+    def test_zero_ttl_bypasses_cache(self):
+        run, calls = self._fake_run(SHA_B)
+        with patch.object(mp.subprocess, "run", run):
+            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
+            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
+        self.assertEqual(calls["n"], 2)
+
+    def test_env_override_never_touches_subprocess(self):
+        run, calls = self._fake_run(SHA_B)
+        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
+            with patch.object(mp.subprocess, "run", run):
+                self.assertEqual(
+                    mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
+        self.assertEqual(calls["n"], 0)
+
+
 class TestServerWiring(unittest.TestCase):
     """Integration with the gate choke point in the server namespace."""
 
@@ -105,6 +266,13 @@ class TestServerWiring(unittest.TestCase):
         self._saved = self.srv._STARTUP_PARITY
         self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
                                     "startup_head": SHA_A}
+        # Keep the live-remote read hermetic (no real ls-remote network call):
+        # default the live master to the daemon start so parity is fully green
+        # unless a test overrides the live head explicitly (#610).
+        self._live_patch = patch.dict(
+            os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
+        self._live_patch.start()
+        self.addCleanup(self._live_patch.stop)
 
     def tearDown(self):
         self.srv._STARTUP_PARITY = self._saved
@@ -147,6 +315,36 @@ class TestServerWiring(unittest.TestCase):
             self.assertTrue(out["in_parity"])
             self.assertNotIn("report", out)
 
+    # --- #610: live-remote wiring -------------------------------------------
+
+    def test_live_stale_blocks_mutation_though_local_green(self):
+        # Local checkout matches the daemon start (local parity green) but the
+        # live remote master has advanced -> mutations must fail closed.
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
+                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
+            self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
+            self.assertTrue(
+                self.srv._master_parity_block("gitea.pr.create"))
+
+    def test_assess_tool_exposes_three_distinct_shas(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
+                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
+            out = self.srv.gitea_assess_master_parity(remote="prgs")
+            self.assertEqual(out["daemon_start_head"], SHA_A)
+            self.assertEqual(out["local_head"], SHA_A)
+            self.assertEqual(out["live_remote_head"], SHA_B)
+            self.assertTrue(out["live_stale"])
+            self.assertFalse(out["mutation_safe"])
+            self.assertIn("report", out)
+
+    def test_assess_tool_mutation_safe_when_all_three_match(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
+                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
+            out = self.srv.gitea_assess_master_parity(remote="prgs")
+            self.assertTrue(out["mutation_safe"])
+            self.assertFalse(out["live_stale"])
+            self.assertNotIn("report", out)
+
 
 if __name__ == "__main__":
     unittest.main()