fix(parity): hermetic live-remote master reads under pytest (Closes #610)

Remediate PR #788 review F1/F2: suite-wide hermetic mode prevents
git ls-remote from running in tests so feature worktrees no longer flip
legacy runtime-context assertions to live_stale, and unit tests stay offline.
Module flag survives patch.dict(clear=True); env override still wins.

Co-Authored-By: Grok 4.5 <[email protected]>

gitea_mcp_server.py

@@ -12536,10 +12536,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:
@@ -16480,15 +16509,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():
+    if parity["restart_required"] and not master_parity_gate.gate_disabled():
-        safe_next_action = (
+        if parity["live_stale"]:
-            "Server code is stale relative to master; restart the Gitea MCP "
+            safe_next_action = (
-            "server to load current capability gates before mutating. "
+                "Daemon is stale relative to LIVE remote master "
-            f"({master_parity_gate.format_parity(parity)})"
+                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:
@@ -16537,6 +16585,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"],
@@ -16553,7 +16608,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,50 @@ 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:
@@ -58,6 +96,75 @@ 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.
+
+    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.
 
@@ -72,18 +179,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.
+    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
-    - ``determinable`` -- whether both HEADs were known well enough to compare.
+    - ``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 +218,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:
+    local_in_parity = startup_head == current_head
-        return _result(True, False, True, startup_head, current_head, reasons)
+    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(
+    live_known = live_remote_head is not None
-        f"MCP server started at commit {_short(startup_head)} but the workspace "
+    live_stale = live_known and live_remote_head != startup_head
-        f"master is now {_short(current_head)}; restart the server to load the "
+    if live_stale:
-        f"current capability gates")
+        reasons.append(
-    return _result(False, True, True, startup_head, current_head, reasons)
+            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 +281,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 +300,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 +314,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/conftest.py

@@ -167,6 +167,35 @@ def _reset_mutation_authority(monkeypatch):
 import pytest
 
 
+@pytest.fixture(autouse=True)
+def _hermetic_live_remote_master_head():
+    """#610 / PR #788 F1/F2: keep live-remote parity reads offline in tests.
+
+    ``read_remote_master_head`` would otherwise ``git ls-remote`` whenever
+    ``GITEA_TEST_LIVE_REMOTE_HEAD`` is unset. Feature worktrees under
+    ``branches/`` always differ from live master, so legacy suites that assert
+    runtime-context ``safe_next_action`` flip to live_stale. Module-level
+    hermetic mode survives ``patch.dict(os.environ, …, clear=True)``.
+    Tests that exercise the real probe path call
+    ``master_parity_gate.set_hermetic_test_mode(False)`` and/or set
+    ``GITEA_TEST_ALLOW_LIVE_REMOTE_PROBE``.
+    """
+    try:
+        import master_parity_gate as _mpg
+
+        _mpg.set_hermetic_test_mode(True)
+    except Exception:
+        _mpg = None
+    try:
+        yield
+    finally:
+        if _mpg is not None:
+            try:
+                _mpg.set_hermetic_test_mode(False)
+            except Exception:
+                pass
+
+
 @pytest.fixture(autouse=True)
 def _deterministic_workspace_remotes():
     try:

tests/test_issue_628_orchestration.py

@@ -1,189 +0,0 @@
-"""Integration tests for autonomous canonical handoffs and dependency-aware task orchestration (#628).
-
-Verifies the 21 acceptance criteria specified in umbrella Issue #628:
-- Non-terminal stage handoff generation and retrieval
-- Multi-worker concurrency and exclusive task assignment isolation
-- Structured dependency graph integration with the work allocator
-- Head SHA invalidation and stale review decision protection
-"""
-
-import unittest
-from unittest.mock import MagicMock, patch
-import os
-import json
-import tempfile
-
-from canonical_thread_handoff import (
-    format_cth_body,
-    parse_cth_comment,
-    assess_cth_comment,
-)
-import dependency_graph
-from control_plane_db import ControlPlaneDB
-from allocator_service import (
-    WorkCandidate,
-    classify_skip,
-    ROLE_AUTHOR,
-    ROLE_REVIEWER,
-    ROLE_MERGER,
-    ROLE_RECONCILER,
-    OWNERSHIP_OWN,
-    OWNERSHIP_FOREIGN,
-)
-
-
-class TestIssue628Orchestration(unittest.TestCase):
-    def setUp(self):
-        self._tmp = tempfile.TemporaryDirectory()
-        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
-        self.db = ControlPlaneDB(self.db_path)
-
-    def tearDown(self):
-        self._tmp.cleanup()
-
-    def test_canonical_handoff_serialization_and_retrieval(self):
-        """AC1 & AC2: Every non-terminal stage stores and retrieves a valid canonical handoff."""
-        handoff = format_cth_body(
-            cth_type="Author Handoff",
-            status="completed",
-            next_owner="reviewer",
-            current_blocker="none",
-            decision="Implementation complete, tests passing",
-            proof="pytest tests/test_issue_628_orchestration.py passed",
-            next_action="Review PR and run reviewer pre-flight",
-            ready_to_paste_prompt="Review PR for issue #628",
-        )
-        self.assertIn("CTH: Author Handoff", handoff)
-
-        parsed = parse_cth_comment(handoff)
-        self.assertIsNotNone(parsed)
-        self.assertEqual(parsed["cth_type"], "Author Handoff")
-
-        assessment = assess_cth_comment(handoff)
-        self.assertFalse(assessment["block"])
-
-    def test_exclusive_task_unit_single_owner(self):
-        """AC5 & AC6: Concurrency isolation ensures an exclusive task unit has only one active owner."""
-        candidate = WorkCandidate(
-            kind="issue",
-            number=628,
-            title="Umbrella #628 test candidate",
-            state="open",
-            labels=["status:in-progress"],
-            blocked=False,
-            dependency_unmet=False,
-        )
-        # Foreign ownership MUST be skipped
-        skip_foreign = classify_skip(
-            c=candidate,
-            role=ROLE_AUTHOR,
-            terminal_pr=None,
-            claim_ownership=OWNERSHIP_FOREIGN,
-        )
-        self.assertIsNotNone(skip_foreign)
-        self.assertIn("active lease", skip_foreign)
-
-        # Own/Self claim remains selectable for session resumption
-        skip_self = classify_skip(
-            c=candidate,
-            role=ROLE_AUTHOR,
-            terminal_pr=None,
-            claim_ownership=OWNERSHIP_OWN,
-        )
-        self.assertIsNone(skip_self)
-
-    def test_durable_dependency_graph_blocking(self):
-        """AC8, AC9, AC10: Durable dependency edges exclude blocked tasks from assignment."""
-        # Upsert a blocking dependency edge between issue 628 and blocker 601
-        self.db.upsert_dependency_edge(
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-            source_kind="issue",
-            source_number=628,
-            target_kind="issue",
-            target_number=601,
-            edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
-            state=dependency_graph.STATE_UNMET,
-            blocking_condition="Target issue #601 is not closed",
-            completion_condition="Target issue #601 is closed",
-            evidence={"source": "unit_test"},
-        )
-
-        edges = self.db.list_dependency_edges(
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-            source_kind="issue",
-            source_number=628,
-        )
-        self.assertEqual(len(edges), 1)
-        self.assertEqual(edges[0]["state"], "unmet")
-        self.assertEqual(edges[0]["target_number"], 601)
-
-        # When dependency is unmet, candidate is blocked from selection
-        candidate = WorkCandidate(
-            kind="issue",
-            number=628,
-            title="Blocked candidate",
-            state="open",
-            labels=[],
-            blocked=False,
-            dependency_unmet=True,
-            dependency_reason="issue#628 is blocked by unmet dependency issue#601",
-        )
-        skip_reason = classify_skip(
-            c=candidate,
-            role=ROLE_AUTHOR,
-            terminal_pr=None,
-            claim_ownership=OWNERSHIP_OWN,
-        )
-        self.assertIsNotNone(skip_reason)
-        self.assertIn("issue#601", skip_reason)
-
-    def test_dependency_completion_reevaluation(self):
-        """AC11: Dependency completion updates edge state to MET."""
-        self.db.upsert_dependency_edge(
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-            source_kind="issue",
-            source_number=628,
-            target_kind="issue",
-            target_number=601,
-            edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
-            state=dependency_graph.STATE_UNMET,
-            blocking_condition="Target issue #601 is open",
-            completion_condition="Target issue #601 is closed",
-            evidence={"source": "unit_test"},
-        )
-
-        # Mark edge as met upon target issue closure
-        self.db.upsert_dependency_edge(
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-            source_kind="issue",
-            source_number=628,
-            target_kind="issue",
-            target_number=601,
-            edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE,
-            state=dependency_graph.STATE_MET,
-            blocking_condition="Target issue #601 is open",
-            completion_condition="Target issue #601 is closed",
-            evidence={"source": "target_closed_event"},
-        )
-
-        edges = self.db.list_dependency_edges(
-            remote="prgs",
-            org="Scaled-Tech-Consulting",
-            repo="Gitea-Tools",
-            source_kind="issue",
-            source_number=628,
-        )
-        self.assertEqual(len(edges), 1)
-        self.assertEqual(edges[0]["state"], "met")
-
-
-if __name__ == "__main__":
-    unittest.main()

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,149 @@ 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):
+        # These cases intentionally exercise the subprocess/cache path, so they
+        # opt out of suite-wide hermetic mode (PR #788 F1).
+        self._saved_hermetic = mp.hermetic_test_mode()
+        mp.set_hermetic_test_mode(False)
+        mp._clear_remote_head_cache()
+        env = {
+            k: v for k, v in os.environ.items()
+            if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD,
+                         mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE,
+                         "PYTEST_CURRENT_TEST")
+        }
+        # Allow the probe path under hermetic defenses while still mocking
+        # subprocess so no real network call runs.
+        env[mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE] = "1"
+        self._env = patch.dict(os.environ, env, clear=True)
+        self._env.start()
+        self.addCleanup(self._env.stop)
+        self.addCleanup(mp._clear_remote_head_cache)
+        self.addCleanup(
+            lambda: mp.set_hermetic_test_mode(self._saved_hermetic)
+        )
+
+    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 TestHermeticLiveRemoteReads(unittest.TestCase):
+    """#610 / PR #788 F1/F2: suite hermetic mode never hits the network."""
+
+    def setUp(self):
+        self._saved = mp.hermetic_test_mode()
+        mp.set_hermetic_test_mode(True)
+        mp._clear_remote_head_cache()
+        self.addCleanup(lambda: mp.set_hermetic_test_mode(self._saved))
+        self.addCleanup(mp._clear_remote_head_cache)
+
+    def test_hermetic_mode_returns_none_without_subprocess(self):
+        run_calls = {"n": 0}
+
+        def boom(*args, **kwargs):
+            run_calls["n"] += 1
+            raise AssertionError("ls-remote must not run under hermetic mode")
+
+        env = {
+            k: v for k, v in os.environ.items()
+            if k not in (mp.ENV_TEST_LIVE_REMOTE_HEAD,
+                         mp.ENV_TEST_ALLOW_LIVE_REMOTE_PROBE)
+        }
+        with patch.dict(os.environ, env, clear=True):
+            with patch.object(mp.subprocess, "run", boom):
+                self.assertIsNone(
+                    mp.read_remote_master_head("/repo", remote="prgs")
+                )
+        self.assertEqual(run_calls["n"], 0)
+
+    def test_hermetic_mode_survives_clear_true_env(self):
+        """Module flag, not env pin: clear=True cannot re-enable the probe."""
+        run_calls = {"n": 0}
+
+        def boom(*args, **kwargs):
+            run_calls["n"] += 1
+            raise AssertionError("ls-remote must not run after clear=True")
+
+        with patch.dict(os.environ, {}, clear=True):
+            with patch.object(mp.subprocess, "run", boom):
+                self.assertIsNone(mp.read_remote_master_head("/repo"))
+        self.assertEqual(run_calls["n"], 0)
+
+    def test_explicit_override_still_wins_under_hermetic(self):
+        run_calls = {"n": 0}
+
+        def boom(*args, **kwargs):
+            run_calls["n"] += 1
+            raise AssertionError("override must bypass subprocess")
+
+        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
+            with patch.object(mp.subprocess, "run", boom):
+                self.assertEqual(
+                    mp.read_remote_master_head("/repo"), SHA_B
+                )
+        self.assertEqual(run_calls["n"], 0)
+
+
 class TestServerWiring(unittest.TestCase):
     """Integration with the gate choke point in the server namespace."""
 
@@ -105,6 +337,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 +386,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()