Merge branch 'master' into feat/issue-610-live-remote-parity

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

@@ -12366,10 +12366,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 _current_runtime_mode_report(refresh: bool = False) -> dict:
@@ -16310,15 +16339,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:
@@ -16367,6 +16415,13 @@ 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"],
@@ -16383,7 +16438,7 @@ def gitea_assess_master_parity(
             source=canonical_source,
         ),
     }
-    if parity["stale"] and enforced:
+    if parity["restart_required"] and enforced:
         out["report"] = master_parity_gate.parity_report(parity)
     return out
 

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"):

runtime_recovery_guard.py

@@ -49,7 +49,7 @@ from __future__ import annotations
 
 import os
 import re
-from typing import Any, Iterable, Iterator
+from typing import Any, Iterable
 
 # Single source of truth for both the redactor and the gated-mutation set: the
 # #671 guard already owns them, so the two contamination models can never drift
@@ -86,26 +86,8 @@ REMEDIATION = (
 
 # Split a compound command line into simple commands on shell separators so
 # ``ps aux | grep mcp_server`` is analysed segment by segment and its harmless
-# inspection half never reaches the kill classifier. The background separator
-# ``&`` is a separator too: without it ``sleep 1 & pkill -f mcp_server.py`` was
-# a single segment whose command position held ``sleep``, so the kill was never
-# classified (#787).
-#
-# Splitting is *quote-aware*, and a regex alternation cannot express that, so
-# the scan below replaces the earlier ``_SEGMENT_SPLIT_RE`` pattern. A separator
-# only separates where it is syntactically active: outside single and double
-# quotes, and not backslash-escaped. Without that, adding ``&`` made every
-# benign mention of the canonical kill string classify as a real kill — a commit
-# message quoting ``sleep 1 & pkill -f mcp_server.py``, an ``echo`` of the same
-# sentence, a ``grep`` for it — and a false contamination marker fails review,
-# merge, close and completion mutations closed until a reconciler clears it (PR
-# #789 review finding F1). Quote-awareness is not specific to ``&``: it also
-# retires the same false-positive class that ``;`` and ``|`` carried before #787.
-_SEPARATOR_CHARS = frozenset("|&;\n")
-
-#: Two-character logical separators, consumed whole so ``&&`` and ``||`` are
-#: never split into single characters leaving a stray operator behind.
-_LOGICAL_SEPARATORS = ("&&", "||")
+# inspection half never reaches the kill classifier.
+_SEGMENT_SPLIT_RE = re.compile(r"(?:\|\||&&|\||;|\n)")
 
 _KILL_VERBS = frozenset({"kill", "pkill", "killall"})
 
@@ -151,128 +133,8 @@ def _clean(value: str | None) -> str:
     return (value or "").strip()
 
 
-def _iter_active(text: str) -> Iterator[tuple[int, str]]:
-    """Yield ``(index, char)`` for every *syntactically active* character.
-
-    Active means outside single and double quotes and not backslash-escaped —
-    the positions where a shell metacharacter actually carries its meaning.
-    Quoted runs, the quote characters themselves, and escaped characters are
-    skipped, so a separator written inside a commit message or a ``grep``
-    pattern is literal text rather than syntax. A backslash escapes nothing
-    inside single quotes, matching POSIX.
-
-    An unterminated quote swallows the rest of the line, exactly as it does for
-    the shell — which would reject such a command as a syntax error rather than
-    run its tail, so nothing executable hides behind it.
-    """
-    quote: str | None = None
-    index = 0
-    end = len(text)
-    while index < end:
-        char = text[index]
-        if quote == "'":
-            if char == "'":
-                quote = None
-            index += 1
-        elif quote == '"':
-            if char == "\\" and index + 1 < end:
-                index += 2
-            else:
-                if char == '"':
-                    quote = None
-                index += 1
-        elif char == "\\" and index + 1 < end:
-            index += 2
-        elif char in ("'", '"'):
-            quote = char
-            index += 1
-        else:
-            yield index, char
-            index += 1
-
-
-def _is_redirection(command: str, index: int, active: frozenset[int]) -> bool:
-    """Is the ``&``/``|`` at *index* part of a redirection, not a separator?
-
-    ``2>&1`` and ``>&2`` put the character immediately after a redirection
-    operator, and ``&>log`` immediately before one; in neither position does it
-    separate commands. Without this, ``a 2>&1`` split into ``['a 2>', '1']``
-    (PR #789 review finding F3).
-    """
-    previous = command[index - 1] if index else ""
-    if previous in ("<", ">") and (index - 1) in active:
-        return True
-    return (
-        command[index] == "&"
-        and command[index + 1:index + 2] == ">"
-        and (index + 1) in active
-    )
-
-
-def _closes_leading_paren(body: str) -> bool:
-    """Does *body* end with the active ``)`` matching a stripped leading ``(``?"""
-    if not body.endswith(")"):
-        return False
-    depth = 0
-    for index, char in _iter_active(body):
-        if char == "(":
-            depth += 1
-        elif char == ")":
-            if depth == 0:
-                return index == len(body) - 1
-            depth -= 1
-    return False
-
-
-def _strip_subshell(segment: str) -> str:
-    """Remove subshell wrappers so ``(pkill -f mcp_server.py)`` is classified.
-
-    The parentheses are shell syntax, not part of the simple command, so a
-    wrapped kill otherwise put ``(pkill`` in command position and never
-    reached the kill classifier (#787). Nested wrappers are unwrapped too.
-
-    A trailing ``)`` is removed only when it closes a leading ``(`` this call
-    stripped. Removing one unconditionally mangled balanced command
-    substitution — ``kill $(pgrep -f myapp)`` became ``kill $(pgrep -f myapp``
-    (PR #789 review finding F3). An unmatched leading ``(`` is still dropped on
-    its own, because splitting a wrapped compound orphans the opening half.
-    """
-    stripped = segment.strip()
-    while stripped.startswith("("):
-        body = stripped[1:].strip()
-        if _closes_leading_paren(body):
-            body = body[:-1].strip()
-        stripped = body
-    return stripped
-
-
 def _split_segments(command: str) -> list[str]:
-    """Split *command* into simple commands on syntactically active separators."""
-    active = frozenset(index for index, _ in _iter_active(command))
-    segments: list[str] = []
-    start = 0
-    index = 0
-    end = len(command)
-    while index < end:
-        char = command[index]
-        if (
-            char not in _SEPARATOR_CHARS
-            or index not in active
-            or (char in "&|" and _is_redirection(command, index, active))
-        ):
-            index += 1
-            continue
-        width = (
-            2
-            if command[index:index + 2] in _LOGICAL_SEPARATORS
-            and (index + 1) in active
-            else 1
-        )
-        segments.append(command[start:index])
-        index += width
-        start = index
-    segments.append(command[start:])
-    return [seg for seg in (_strip_subshell(seg) for seg in segments) if seg]
+    return [seg for seg in _SEGMENT_SPLIT_RE.split(command) if seg.strip()]
 
 
 def is_sanctioned_recovery(text: str | None) -> bool:

tests/test_issue_630_runtime_recovery_guard.py

@@ -91,132 +91,6 @@ def test_compound_command_detects_the_kill_half():
     assert result["contamination"] is True
 
 
-# ── #787: background separator and subshell forms reach the classifier ───────
-
-def test_background_separator_kill_is_contamination():
-    result = guard.classify_recovery_command("sleep 1 & pkill -f mcp_server.py")
-    assert result["process_kill"] is True
-    assert result["contamination"] is True
-    assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
-    assert result["ambiguous"] is False
-
-
-def test_subshell_wrapped_kill_is_contamination():
-    result = guard.classify_recovery_command("(pkill -f mcp_server.py)")
-    assert result["process_kill"] is True
-    assert result["contamination"] is True
-    assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
-    assert result["ambiguous"] is False
-
-
-def test_further_background_and_subshell_forms_are_contamination():
-    for command in (
-        "pkill -f mcp_server.py &",
-        "( sudo pkill -f mcp_server.py )",
-        "((pkill -f gitea_mcp_server))",
-        "sleep 1 & killall mcp_server",
-        "(ps aux | grep mcp_server) & pkill -f mcp_server.py",
-    ):
-        result = guard.classify_recovery_command(command)
-        assert result["contamination"] is True, command
-        assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL, command
-
-
-def test_logical_operators_are_not_split_into_single_characters():
-    # ``&&``/``||`` must still be consumed whole by the separator scan.
-    assert guard._split_segments("a && b || c") == ["a", "b", "c"]
-    assert guard._split_segments("a & b") == ["a", "b"]
-    assert guard._split_segments("(a)") == ["a"]
-    assert guard._split_segments("a; b\nc | d") == ["a", "b", "c", "d"]
-
-
-# ── #789 F1: separators only separate outside quoted or escaped text ─────────
-
-# The three commands the PR #789 review measured as regressions at head
-# 6b58f04: each merely *mentions* the canonical kill string inside quotes.
-F1_QUOTED_COMMANDS = (
-    'git commit -m "block sleep 1 & pkill -f mcp_server.py as recovery"',
-    'echo "docs: sleep 1 & pkill -f mcp_server.py is now detected"',
-    'grep -rn "sleep 1 & pkill -f mcp_server.py" docs/',
-)
-
-
-def test_quoted_ampersand_examples_from_review_f1_are_not_kills():
-    for command in F1_QUOTED_COMMANDS:
-        result = guard.classify_recovery_command(command)
-        assert result["process_kill"] is False, command
-        assert result["contamination"] is False, command
-        assert result["reason_class"] is None, command
-
-
-def test_ampersand_inside_double_quotes_is_not_a_separator():
-    assert guard._split_segments('echo "a & b"') == ['echo "a & b"']
-    result = guard.classify_recovery_command(
-        'echo "restart it: sleep 1 & pkill -f mcp_server.py"'
-    )
-    assert result["process_kill"] is False
-    assert result["contamination"] is False
-
-
-def test_ampersand_inside_single_quotes_is_not_a_separator():
-    assert guard._split_segments("echo 'a & b'") == ["echo 'a & b'"]
-    result = guard.classify_recovery_command(
-        "git commit -m 'sleep 1 & pkill -f mcp_server.py stays quoted'"
-    )
-    assert result["process_kill"] is False
-    assert result["contamination"] is False
-
-
-def test_backslash_escaped_ampersand_is_not_a_separator():
-    command = r"echo a \& pkill -f mcp_server.py"
-    assert guard._split_segments(command) == [command]
-    result = guard.classify_recovery_command(command)
-    assert result["process_kill"] is False
-    assert result["contamination"] is False
-
-
-def test_backslash_does_not_escape_inside_single_quotes():
-    # POSIX: a backslash is literal inside single quotes, so the closing quote
-    # still closes and the following ``&`` is a genuinely active separator.
-    command = r"echo 'a\' & pkill -f mcp_server.py"
-    assert guard._split_segments(command) == [r"echo 'a\'", "pkill -f mcp_server.py"]
-    assert guard.classify_recovery_command(command)["contamination"] is True
-
-
-def test_quote_awareness_also_retires_the_pre_existing_semicolon_and_pipe_cases():
-    # ``;`` and ``|`` misclassified quoted text before #787 as well. The fix is
-    # the quote-unawareness, not the ``&`` instance the issue happens to name.
-    for command in (
-        'git commit -m "fix; pkill -f mcp_server.py"',
-        'git commit -m "fix | pkill -f mcp_server.py"',
-    ):
-        result = guard.classify_recovery_command(command)
-        assert result["process_kill"] is False, command
-        assert result["contamination"] is False, command
-
-
-# ── #789 F3: subshell stripping and redirection stay syntactically honest ────
-
-def test_command_substitution_is_not_mangled_by_subshell_stripping():
-    # Only a wrapper this call opened may be unwrapped; a ``)`` closing ``$(``
-    # must survive intact.
-    assert guard._strip_subshell("kill $(pgrep -f myapp)") == "kill $(pgrep -f myapp)"
-    result = guard.classify_recovery_command("kill $(pgrep -f myapp)")
-    assert result["contamination"] is False
-    assert result["ambiguous"] is True
-
-
-def test_redirection_is_not_treated_as_a_background_separator():
-    assert guard._split_segments("a 2>&1") == ["a 2>&1"]
-    assert guard._split_segments("a &> log") == ["a &> log"]
-    assert guard._split_segments("pkill -f mcp_server.py 2>&1") == [
-        "pkill -f mcp_server.py 2>&1"
-    ]
-    result = guard.classify_recovery_command("pkill -f mcp_server.py 2>&1")
-    assert result["contamination"] is True
-    assert result["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
-
-
 # ── no false positives ───────────────────────────────────────────────────────
 
 def test_read_only_inspection_is_not_a_kill():
@@ -238,22 +112,6 @@ def test_unrelated_pkill_target_is_not_contamination():
     assert result["ambiguous"] is False
 
 
-def test_user_scoped_pkill_of_unrelated_app_is_not_contamination():
-    # ``-u`` consumes ``mcpuser``; the surviving operand names no daemon (#787).
-    result = guard.classify_recovery_command("pkill -u mcpuser -f myapp")
-    assert result["process_kill"] is True
-    assert result["contamination"] is False
-    assert result["ambiguous"] is False
-
-
-def test_commit_message_quoting_the_kill_string_is_not_a_kill():
-    result = guard.classify_recovery_command(
-        'git commit -m "block pkill -f mcp_server.py as workflow recovery"'
-    )
-    assert result["process_kill"] is False
-    assert result["contamination"] is False
-
-
 def test_bare_kill_of_unknown_pid_is_ambiguous_not_contamination():
     result = guard.classify_recovery_command("kill 31337")
     assert result["contamination"] is False
@@ -475,43 +333,6 @@ def test_record_tool_marks_manual_daemon_kill():
     assert "mcp_server.py" in loaded["command_summary"]
 
 
-def test_record_tool_marks_background_separator_kill():
-    _clear_marker()
-    res = srv.gitea_record_daemon_process_kill_attempt(
-        command="sleep 1 & pkill -f mcp_server.py", remote="prgs"
-    )
-    assert res["contaminated"] is True
-    assert res["marked"] is True
-    assert res["marker"]["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
-    loaded = srv._load_runtime_recovery_marker("prgs")
-    assert loaded is not None
-    assert "mcp_server.py" in loaded["command_summary"]
-
-
-def test_record_tool_marks_subshell_wrapped_kill():
-    _clear_marker()
-    res = srv.gitea_record_daemon_process_kill_attempt(
-        command="(pkill -f mcp_server.py)", remote="prgs"
-    )
-    assert res["contaminated"] is True
-    assert res["marked"] is True
-    assert res["marker"]["reason_class"] == guard.REASON_MANUAL_DAEMON_KILL
-    loaded = srv._load_runtime_recovery_marker("prgs")
-    assert loaded is not None
-    assert "mcp_server.py" in loaded["command_summary"]
-
-
-def test_record_tool_does_not_mark_a_quoted_mention_of_the_kill_string():
-    # The marker is what fails review/merge/close closed and only a reconciler
-    # may clear it, so a quoted mention must never create one (PR #789 F1).
-    for command in F1_QUOTED_COMMANDS:
-        _clear_marker()
-        res = srv.gitea_record_daemon_process_kill_attempt(command=command, remote="prgs")
-        assert res["contaminated"] is False, command
-        assert res["marked"] is False, command
-        assert srv._load_runtime_recovery_marker("prgs") is None, command
-
-
 def test_record_tool_marks_broad_sweep():
     _clear_marker()
     res = srv.gitea_record_daemon_process_kill_attempt(

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()