feat(guard): native MCP transport binding and contaminated-review quarantine (Closes #695)

Bind mutation/credential paths to a process-local native MCP runtime so env
spoofing, direct imports, and offline helpers cannot reconstruct session gates
after native transport failure. Quarantine contaminated formal reviews under
controller authority and honor quarantine in review feedback, merge eligibility,
merge mutation, and canonical handoff validation. Add regression coverage for
the second (PR #694 / review 427) incident class.

docs/mcp-daemon-import-guard.md

@@ -9,37 +9,22 @@ formal reviews then looked identical to native approvals.
 
 ## Rule
 
-Mutation auth, keychain fill, and controller quarantine require a **production
+Mutation auth, keychain fill, and controller quarantine require a **native MCP
-native MCP transport runtime** established only by:
+transport runtime** established only by the official entrypoint.
-
-1. the **resolved absolute path** of the canonical entrypoint
-   (`mcp_server.py` / `gitea_mcp_server.py` next to `mcp_daemon_guard.py`), and
-2. a live **transport bind** (`bind_native_mcp_transport(transport="stdio")`)
-   immediately before `mcp.run`.
-
-Basename-only trust (a renamed file called `mcp_server.py`), caller-controlled
-flags (there is **no** `allow_test_bootstrap`), environment variables, stack
-frame spoofing, or import-only launch are insufficient.
 
 | Context | Allowed |
 |---------|---------|
-| Official IDE-native MCP: resolved canonical entrypoint marks + binds stdio, holds process-local runtime token | yes |
+| Official MCP entrypoint (`mcp_server.py` / `gitea_mcp_server` `__main__`) calls `mark_sanctioned_daemon()` and holds a process-local runtime token | yes |
-| pytest (hermetic unit tests) via `is_pytest_runtime()` | yes for unit gates |
+| pytest (hermetic unit tests) | yes |
-| `install_test_native_runtime()` under pytest (test-mode record) | unit-test transport gates only — **never** production Gitea mutations |
-| `allow_test_bootstrap=True` (removed; must not exist) | **no** |
-| Renamed runner basename `mcp_server.py` outside package root | **no** |
-| Import/launch of real entrypoint without transport bind | **no** |
 | `GITEA_MCP_SANCTIONED_DAEMON=1` alone (no process-local native runtime) | **no** (#695) |
-| `GITEA_ALLOW_DIRECT_MCP_IMPORT=1` in LLM sessions | **no** — never set; never authorizes mutations (#695 AC1 / PR #701) |
+| `GITEA_ALLOW_DIRECT_MCP_IMPORT=1` in LLM sessions | **no** — never set in agent sessions |
-| Override `GITEA_MCP_SESSION_STATE_DIR` mid-session | **no** — production bind pins state root; redirect cannot forge independent decision locks (#695 AC2 / PR #701) |
 | `GITEA_ALLOW_KEYCHAIN_CLI=1` in LLM sessions | **no** — human operator only |
 | bare `python -c 'import gitea_mcp_server; …'` or offline runners | **no** |
 | keychain fill outside native/pytest | **no** |
 
-Native runtime is **process-local**: a random token bound to the daemon PID and
+Native runtime is **process-local**: a random token bound to the daemon PID.
-transport phase. It is never reconstructed from environment variables,
+It is never reconstructed from environment variables, session-state files, or
-session-state files, caller-controlled flags, or importing internals in a
+importing internals in a fresh Python process.
-fresh Python process.
 
 ## Contaminated review quarantine (#695 AC8)
 

gitea_mcp_server.py

@@ -3305,18 +3305,6 @@ def _save_review_decision_lock(data):
     payload["profile_identity"] = binding["profile_identity"]
     if binding.get("remote") and not payload.get("remote"):
         payload["remote"] = binding["remote"]
-    # #695 AC6: stamp native transport provenance on durable decision locks.
-    try:
-        payload.update(
-            {
-                k: v
-                for k, v in mcp_daemon_guard.mutation_provenance_fields().items()
-                if v is not None
-            }
-        )
-    except Exception:
-        payload.setdefault("transport", "untrusted")
-        payload.setdefault("native_mcp_transport", False)
     persisted = mcp_session_state.save_state(
         kind=mcp_session_state.KIND_DECISION_LOCK,
         payload=payload,
@@ -4053,15 +4041,6 @@ def _evaluate_pr_review_submission(
         reasons.extend(review_workflow_load.recovery_handoff_without_replay())
         return result
     if live:
-        # #695 AC1/AC2: offline direct-import submit (PR #701 run_submit.py) fails closed.
-        try:
-            mcp_daemon_guard.assert_sanctioned_mutation_runtime(
-                "gitea_submit_pr_review"
-            )
-            mcp_daemon_guard.assert_no_direct_import_bypass("gitea_submit_pr_review")
-        except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
-            reasons.append(str(exc))
-            return result
         ns_gate = _live_namespace_health_gate("review_pr")
         if ns_gate:
             reasons.extend(ns_gate)
@@ -4249,16 +4228,6 @@ def gitea_mark_final_review_decision(
     repo: str | None = None,
 ) -> dict:
     """Mark validation complete; the final review decision is ready to submit."""
-    # #695 AC1/AC2: direct import / redirected session state cannot mark final.
-    try:
-        mcp_daemon_guard.assert_sanctioned_mutation_runtime(
-            "gitea_mark_final_review_decision"
-        )
-        mcp_daemon_guard.assert_no_direct_import_bypass(
-            "gitea_mark_final_review_decision"
-        )
-    except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
-        return {"marked_ready": False, "reasons": [str(exc)]}
     action = (action or "").strip().lower()
     lock = _load_review_decision_lock()
     if lock is None:
@@ -12874,10 +12843,9 @@ def gitea_quarantine_contaminated_review(
     427 until an independent adversarial reviewer and controller deployment
     of this tooling have completed.
     """
-    # Fail closed outside production native MCP before any mutation assessment.
+    # Fail closed outside native MCP before any mutation assessment.
-    # Test-mode bootstrap must never reach this production mutation endpoint (#695).
     try:
-        mcp_daemon_guard.assert_production_mutation_runtime(
+        mcp_daemon_guard.assert_sanctioned_mutation_runtime(
             "gitea_quarantine_contaminated_review"
         )
     except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
@@ -13083,12 +13051,10 @@ def gitea_quarantine_contaminated_review(
 # ── Entry point ───────────────────────────────────────────────────────────────
 
 if __name__ == "__main__":
-    # #558 / #695: claim the resolved canonical entrypoint, then bind the live
+    # #558 / #695: mark this process as the official native MCP daemon before
-    # native MCP transport lifecycle before any tool dispatch. Env vars,
+    # any tool dispatch. Env vars alone cannot reconstruct native transport;
-    # basename-only stack frames, and import-only launch cannot reconstruct
+    # offline imports / standalone scripts fail closed on mutations.
-    # native transport; offline imports / standalone scripts fail closed.
     mcp_daemon_guard.mark_sanctioned_daemon()
-    mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
     # Lock this session's launch profile into the environment so child CLI
     # processes (e.g. review_pr.py) can detect and refuse profile
     # side-channel overrides (#199).

mcp_daemon_guard.py

@@ -6,18 +6,12 @@ Direct ``import gitea_mcp_server`` from a shell bypasses native MCP transport.
 - Environment variables alone cannot reconstruct a native session
   (``GITEA_MCP_SANCTIONED_DAEMON=1`` / ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` are
   insufficient for mutation gates).
-- A process-local runtime record is established only by the resolved canonical
+- A process-local runtime record is created only by the official entrypoint
-  entrypoint path (not basename) **and** the actual native MCP transport
+  (``mcp_server.py`` calling ``mark_sanctioned_daemon``), bound to PID and a
-  lifecycle (``bind_native_mcp_transport`` before ``mcp.run``). Merely
+  random secret that never leaves process memory.
-  importing or launching the entrypoint offline does not grant mutation
-  authority.
-- Public caller-controlled flags (including any former
-  ``allow_test_bootstrap``) never establish trusted mutation provenance.
 - Offline scripts that import internals fail closed on mutations.
-- Pytest remains allowed for hermetic unit tests via ``is_pytest_runtime()``.
+- Pytest remains allowed (hermetic tests); optional force flags exist for
-  A separate test-only seam may establish a **test-mode** native record for
+  provenance regression tests.
-  unit tests of transport gates; that record cannot authorize production
-  Gitea mutation endpoints.
 
 Manual deletion of session-state files is never a recovery path.
 """
@@ -29,7 +23,6 @@ import inspect
 import os
 import secrets
 import time
-from pathlib import Path
 from typing import Any
 
 SANCTIONED_DAEMON_ENV = "GITEA_MCP_SANCTIONED_DAEMON"
@@ -41,18 +34,6 @@ FORCE_PROVENANCE_FAIL_ENV = "GITEA_TEST_FORCE_UNSANCTIONED"
 # Process-local native runtime (never persisted, never read from env alone).
 _NATIVE_RUNTIME: dict[str, Any] | None = None
 
-# Production transport identifiers accepted by bind_native_mcp_transport.
-_PRODUCTION_TRANSPORTS = frozenset({"stdio"})
-_RUNTIME_MODE_PRODUCTION = "production"
-_RUNTIME_MODE_TEST = "test"
-_PHASE_ENTRYPOINT_CLAIMED = "entrypoint_claimed"
-_PHASE_TRANSPORT_BOUND = "transport_bound"
-
-# Default session-state root (mirrors mcp_session_state; kept local to avoid
-# import cycles). Used only to pin authority at transport bind (#695 AC2).
-_DEFAULT_SESSION_STATE_DIR = os.path.expanduser("~/.cache/gitea-tools/session-state")
-SESSION_STATE_DIR_ENV = "GITEA_MCP_SESSION_STATE_DIR"
-
 
 class UnsanctionedRuntimeError(RuntimeError):
     """Raised when mutation/credential code runs outside a native MCP daemon."""
@@ -72,261 +53,52 @@ def is_pytest_runtime() -> bool:
     return bool((os.environ.get("PYTEST_CURRENT_TEST") or "").strip())
 
 
-def _package_root() -> Path:
-    """Directory that contains the canonical MCP entrypoint modules."""
-    return Path(__file__).resolve().parent
-
-
-def canonical_entrypoint_paths() -> frozenset[str]:
-    """Resolved absolute paths of official entrypoints (not basenames)."""
-    root = _package_root()
-    return frozenset(
-        {
-            str((root / "mcp_server.py").resolve()),
-            str((root / "gitea_mcp_server.py").resolve()),
-        }
-    )
-
-
-def _resolve_path(path: str | None) -> str | None:
-    if not path:
-        return None
-    try:
-        return str(Path(path).resolve())
-    except (OSError, RuntimeError, ValueError):
-        return None
-
-
-def _caller_official_entrypoint_path() -> str | None:
-    """Return the resolved canonical entrypoint path in the call stack, or None.
-
-    Basename-only matches (e.g. an attacker file named ``mcp_server.py``
-    elsewhere) are rejected. The path must equal one of
-    :func:`canonical_entrypoint_paths`.
-    """
-    canonical = canonical_entrypoint_paths()
-    for frame in inspect.stack()[1:20]:
-        resolved = _resolve_path(frame.filename)
-        if resolved and resolved in canonical:
-            return resolved
-    return None
-
-
 def _caller_is_official_entrypoint() -> bool:
-    """True when invoked from a resolved canonical entrypoint path (#695)."""
+    """True when mark_sanctioned_daemon is invoked from mcp_server.py."""
-    return _caller_official_entrypoint_path() is not None
+    for frame in inspect.stack()[1:12]:
+        path = (frame.filename or "").replace("\\", "/")
+        base = path.rsplit("/", 1)[-1]
+        if base == "mcp_server.py":
+            return True
+    return False
 
 
-def _new_runtime_token() -> tuple[str, str]:
+def mark_sanctioned_daemon(*, allow_test_bootstrap: bool = False) -> dict[str, Any]:
-    token = secrets.token_hex(32)
+    """Mark this process as the official native MCP daemon (#695).
-    fingerprint = hashlib.sha256(token.encode()).hexdigest()[:16]
-    return token, fingerprint
 
-
+    Only the official ``mcp_server.py`` entrypoint (or pytest test bootstrap)
-def mark_sanctioned_daemon() -> dict[str, Any]:
+    may establish native transport. Setting env vars alone is insufficient.
-    """Claim the official entrypoint for this process (#695).
-
-    This alone does **not** authorize mutations. Callers must subsequently
-    bind the native MCP transport via :func:`bind_native_mcp_transport`.
-
-    Only a stack frame whose **resolved absolute path** is the canonical
-    ``mcp_server.py`` or ``gitea_mcp_server.py`` next to this module may
-    claim the entrypoint. Basename spoofing is rejected.
-
-    There is no public ``allow_test_bootstrap`` argument: caller-controlled
-    flags must never establish trusted mutation provenance. Hermetic tests
-    use :func:`install_test_native_runtime` (pytest-only, test mode).
     """
     global _NATIVE_RUNTIME
-    if is_pytest_runtime():
+    if not is_pytest_runtime() and not allow_test_bootstrap:
-        # Under pytest, production mark is a no-op for transport authority.
+        if not _caller_is_official_entrypoint():
-        # Tests that need a native-transport record use install_test_native_runtime.
+            raise UnsanctionedRuntimeError(
-        return native_runtime_status()
+                "mark_sanctioned_daemon rejected: not called from official "
-
+                "mcp_server.py entrypoint (#695). Offline import / standalone "
-    entrypoint_path = _caller_official_entrypoint_path()
+                "scripts cannot reconstruct native transport. Stop after native "
-    if entrypoint_path is None:
+                "MCP failure; do not run offline mutation helpers."
-        raise UnsanctionedRuntimeError(
+            )
-            "mark_sanctioned_daemon rejected: not called from the resolved "
+    token = secrets.token_hex(32)
-            "canonical MCP entrypoint path (#695). Basename-only names "
-            "(e.g. a renamed runner called mcp_server.py) are insufficient. "
-            "Offline import / standalone scripts cannot reconstruct native "
-            "transport. Stop after native MCP failure; do not run offline "
-            "mutation helpers."
-        )
-
-    token, fingerprint = _new_runtime_token()
     _NATIVE_RUNTIME = {
         "token": token,
-        "token_fingerprint": fingerprint,
+        "token_fingerprint": hashlib.sha256(token.encode()).hexdigest()[:16],
         "pid": os.getpid(),
         "started_at": time.time(),
         "entrypoint": "mcp_server",
-        "entrypoint_path": entrypoint_path,
-        "phase": _PHASE_ENTRYPOINT_CLAIMED,
-        "transport": None,
-        "mode": _RUNTIME_MODE_PRODUCTION,
     }
     # Legacy signal for older probes; alone does not authorize mutations.
     os.environ[SANCTIONED_DAEMON_ENV] = "1"
     return native_runtime_status()
 
 
-def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]:
-    """Bind the live native MCP transport lifecycle (#695).
-
-    Must be called from the resolved canonical entrypoint immediately before
-    the real MCP server transport loop (e.g. ``mcp.run(transport=\"stdio\")``).
-    Requires a prior successful :func:`mark_sanctioned_daemon` claim in this
-    process. Import-only or offline launch without this bind leaves
-    :func:`is_native_mcp_transport` false.
-    """
-    global _NATIVE_RUNTIME
-    transport_name = (transport or "").strip().lower()
-    if transport_name not in _PRODUCTION_TRANSPORTS:
-        raise UnsanctionedRuntimeError(
-            f"bind_native_mcp_transport rejected: transport {transport!r} is "
-            f"not a production MCP transport (#695). Allowed: "
-            f"{sorted(_PRODUCTION_TRANSPORTS)}."
-        )
-
-    entrypoint_path = _caller_official_entrypoint_path()
-    if entrypoint_path is None:
-        raise UnsanctionedRuntimeError(
-            "bind_native_mcp_transport rejected: not called from the resolved "
-            "canonical MCP entrypoint path (#695)."
-        )
-
-    if _NATIVE_RUNTIME is None or int(_NATIVE_RUNTIME.get("pid") or -1) != os.getpid():
-        raise UnsanctionedRuntimeError(
-            "bind_native_mcp_transport rejected: no entrypoint claim in this "
-            "process (#695). Call mark_sanctioned_daemon() from the official "
-            "entrypoint first."
-        )
-
-    if _NATIVE_RUNTIME.get("mode") != _RUNTIME_MODE_PRODUCTION:
-        raise UnsanctionedRuntimeError(
-            "bind_native_mcp_transport rejected: runtime mode is not "
-            "production (#695)."
-        )
-
-    claimed = (_NATIVE_RUNTIME.get("entrypoint_path") or "").strip()
-    if claimed and claimed != entrypoint_path:
-        raise UnsanctionedRuntimeError(
-            "bind_native_mcp_transport rejected: entrypoint path mismatch "
-            "between mark and bind (#695)."
-        )
-
-    # Pin session-state root for this server lifetime (#695 AC2 / PR #701).
-    # Changing GITEA_MCP_SESSION_STATE_DIR after bind must not manufacture a
-    # second authority domain for decision locks / workflow proofs.
-    raw_state = (os.environ.get(SESSION_STATE_DIR_ENV) or "").strip()
-    if not raw_state:
-        raw_state = _DEFAULT_SESSION_STATE_DIR
-    try:
-        pinned_state = str(Path(raw_state).resolve())
-    except (OSError, RuntimeError, ValueError):
-        pinned_state = raw_state
-
-    _NATIVE_RUNTIME["phase"] = _PHASE_TRANSPORT_BOUND
-    _NATIVE_RUNTIME["transport"] = transport_name
-    _NATIVE_RUNTIME["entrypoint_path"] = entrypoint_path
-    _NATIVE_RUNTIME["bound_at"] = time.time()
-    _NATIVE_RUNTIME["session_state_dir"] = pinned_state
-    os.environ[SANCTIONED_DAEMON_ENV] = "1"
-    return native_runtime_status()
-
-
-def install_test_native_runtime() -> dict[str, Any]:
-    """Pytest-only seam for hermetic native-transport unit tests (#695).
-
-    Establishes a **test-mode** process-local record so unit tests can exercise
-    gates that require ``is_native_mcp_transport()``. This record:
-
-    - is rejected outside pytest (including a fresh offline interpreter);
-    - never uses production mode;
-    - cannot authorize production Gitea mutation endpoints
-      (:func:`assert_production_mutation_runtime` / production path of
-      :func:`assert_sanctioned_mutation_runtime` when not under pytest).
-
-    There is no public caller-controlled flag that forges production native
-    transport.
-    """
-    global _NATIVE_RUNTIME
-    if not is_pytest_runtime():
-        raise UnsanctionedRuntimeError(
-            "install_test_native_runtime rejected: test-mode native runtime "
-            "is only available under pytest (#695). allow_test_bootstrap and "
-            "similar caller-controlled flags do not exist and cannot authorize "
-            "a fresh offline interpreter."
-        )
-    token, fingerprint = _new_runtime_token()
-    _NATIVE_RUNTIME = {
-        "token": token,
-        "token_fingerprint": fingerprint,
-        "pid": os.getpid(),
-        "started_at": time.time(),
-        "entrypoint": "test_bootstrap",
-        "entrypoint_path": None,
-        "phase": _PHASE_TRANSPORT_BOUND,
-        "transport": "test",
-        "mode": _RUNTIME_MODE_TEST,
-        "bound_at": time.time(),
-    }
-    return native_runtime_status()
-
-
 def clear_native_runtime_for_tests() -> None:
     """Test helper: drop native runtime (does not clear env)."""
     global _NATIVE_RUNTIME
     _NATIVE_RUNTIME = None
 
 
-def pinned_session_state_dir() -> str | None:
-    """Session-state root pinned for this production transport lifetime (#695 AC2).
-
-    When production native transport is bound, durable session proofs must use
-    this directory only. Env overrides of ``GITEA_MCP_SESSION_STATE_DIR`` after
-    bind are ignored so redirected dirs (e.g. ``.mcp_session_701``) cannot
-    manufacture independent decision-lock authority (PR #701 recurrence).
-    """
-    if not is_production_native_mcp_transport():
-        return None
-    pinned = (_NATIVE_RUNTIME or {}).get("session_state_dir")
-    text = (str(pinned) if pinned is not None else "").strip()
-    return text or None
-
-
-def direct_import_env_enabled() -> bool:
-    """True when the legacy direct-import opt-in env is set (never authorizes)."""
-    return (os.environ.get(ALLOW_DIRECT_IMPORT_ENV) or "").strip().lower() in {
-        "1",
-        "true",
-        "yes",
-    }
-
-
-def assert_no_direct_import_bypass(context: str = "mutation") -> None:
-    """Fail closed when GITEA_ALLOW_DIRECT_MCP_IMPORT is used for mutations (#695 AC1).
-
-    The env flag is never a sanctioned recovery path for LLM/agent sessions.
-    Under pytest hermetic tests this is a no-op so unit tests can set the flag
-    to prove it does not grant authority.
-    """
-    if is_pytest_runtime():
-        return
-    if not direct_import_env_enabled():
-        return
-    raise UnsanctionedRuntimeError(
-        f"{ALLOW_DIRECT_IMPORT_ENV} does not authorize {context} (#695 AC1). "
-        "Direct import of gitea_mcp_server mutation tools is forbidden. "
-        "Stop after native MCP failure; reconnect the official MCP daemon. "
-        "Do not set direct-import flags, offline runners, or redirected "
-        f"{SESSION_STATE_DIR_ENV} directories to reconstruct gates."
-    )
-
-
 def is_native_mcp_transport() -> bool:
-    """True when this process holds a transport-bound native runtime (#695)."""
+    """True when this process holds a live native MCP runtime record (#695)."""
     if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
         "1",
         "true",
@@ -339,23 +111,12 @@ def is_native_mcp_transport() -> bool:
         return False
     if not (_NATIVE_RUNTIME.get("token") or "").strip():
         return False
-    if _NATIVE_RUNTIME.get("phase") != _PHASE_TRANSPORT_BOUND:
-        return False
-    if not (_NATIVE_RUNTIME.get("transport") or "").strip():
-        return False
     return True
 
 
-def is_production_native_mcp_transport() -> bool:
-    """True only for production-mode, transport-bound native runtime."""
-    if not is_native_mcp_transport():
-        return False
-    return (_NATIVE_RUNTIME or {}).get("mode") == _RUNTIME_MODE_PRODUCTION
-
-
 def is_sanctioned_mcp_daemon() -> bool:
     """Backward-compatible name; #695 requires native transport, not env alone."""
-    if is_production_native_mcp_transport():
+    if is_native_mcp_transport():
         return True
     if is_pytest_runtime():
         return True
@@ -365,45 +126,10 @@ def is_sanctioned_mcp_daemon() -> bool:
     return False
 
 
-def assert_production_mutation_runtime(context: str = "mutation") -> None:
-    """Fail closed unless production native MCP transport is bound (#695).
-
-    Test-mode bootstrap records and pytest-only hermetic allowances do **not**
-    satisfy this gate. Use for production Gitea mutation endpoints that must
-    never be reachable via test bootstrap.
-    """
-    if is_production_native_mcp_transport():
-        return
-    mode = (_NATIVE_RUNTIME or {}).get("mode")
-    if mode == _RUNTIME_MODE_TEST:
-        raise UnsanctionedRuntimeError(
-            f"Test-mode native runtime cannot authorize production {context} "
-            "(#695). install_test_native_runtime / former allow_test_bootstrap "
-            "must never reach real Gitea mutation endpoints."
-        )
-    assert_sanctioned_mutation_runtime(context)
-
-
 def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
-    """Fail closed when mutation code runs outside native MCP transport (#695).
+    """Fail closed when mutation code runs outside native MCP transport (#695)."""
-
+    if is_sanctioned_mcp_daemon():
-    Under pytest, hermetic unit tests are allowed (profile/permission tests).
-    Outside pytest, requires production-mode transport-bound native runtime.
-    Test-mode records do not authorize non-pytest production mutations.
-    ``GITEA_ALLOW_DIRECT_MCP_IMPORT`` never authorizes mutations (#695 AC1).
-    """
-    if is_pytest_runtime():
         return
-    # AC1: direct-import env is never a mutation recovery path (PR #701).
-    assert_no_direct_import_bypass(context)
-    if is_production_native_mcp_transport():
-        return
-    mode = (_NATIVE_RUNTIME or {}).get("mode")
-    if mode == _RUNTIME_MODE_TEST:
-        raise UnsanctionedRuntimeError(
-            f"Test-mode native runtime cannot authorize production {context} "
-            "(#695). Test bootstrap cannot reach production mutation endpoints."
-        )
     env_spoof = (os.environ.get(SANCTIONED_DAEMON_ENV) or "").strip() in {
         "1",
         "true",
@@ -413,26 +139,14 @@ def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
     if env_spoof:
         extra = (
             f" Note: {SANCTIONED_DAEMON_ENV} alone is not sufficient (#695); "
-            "native transport requires the official MCP entrypoint and a live "
+            "native transport requires the official MCP entrypoint."
-            "transport bind."
-        )
-    phase = (_NATIVE_RUNTIME or {}).get("phase")
-    if phase == _PHASE_ENTRYPOINT_CLAIMED:
-        extra = (
-            (extra + " ") if extra else " "
-        ) + (
-            "Entrypoint was claimed but native MCP transport was never bound "
-            "(#695); offline launch/import of the real entrypoint does not "
-            "grant mutation authority."
         )
     raise UnsanctionedRuntimeError(
         f"Unsanctioned / non-native runtime blocked {context} (#695). "
         "Do not import gitea_mcp_server or call mutation helpers from a raw "
         "shell, offline runner, or ad-hoc script after native MCP failure. "
-        "Stop and reconnect the official MCP daemon (mcp_server.py) over "
+        "Stop and reconnect the official MCP daemon (mcp_server.py). "
-        "native transport. "
+        f"Do not set {ALLOW_DIRECT_IMPORT_ENV} or raw token env vars in LLM "
-        f"Do not set {ALLOW_DIRECT_IMPORT_ENV}, override "
-        f"{SESSION_STATE_DIR_ENV}, or use raw token env vars in LLM "
         f"sessions.{extra}"
     )
 
@@ -467,23 +181,14 @@ def native_runtime_status() -> dict[str, Any]:
     rt = _NATIVE_RUNTIME or {}
     return {
         "native_mcp_transport": is_native_mcp_transport(),
-        "production_native_mcp_transport": is_production_native_mcp_transport(),
         "pytest": is_pytest_runtime(),
         "pid": rt.get("pid"),
         "token_fingerprint": rt.get("token_fingerprint"),
         "started_at": rt.get("started_at"),
         "entrypoint": rt.get("entrypoint"),
-        "entrypoint_path": rt.get("entrypoint_path"),
-        "phase": rt.get("phase"),
-        "transport": rt.get("transport"),
-        "mode": rt.get("mode"),
-        "session_state_dir": pinned_session_state_dir() or rt.get("session_state_dir"),
-        "session_state_dir_pinned": pinned_session_state_dir() is not None,
-        "direct_import_env_set": direct_import_env_enabled(),
         "env_sanctioned_alone_insufficient": True,
         "sanctioned_env": SANCTIONED_DAEMON_ENV,
         "allow_direct_import_env": ALLOW_DIRECT_IMPORT_ENV,
-        "session_state_dir_env": SESSION_STATE_DIR_ENV,
         "allow_keychain_cli_env": ALLOW_KEYCHAIN_CLI_ENV,
     }
 
@@ -498,20 +203,10 @@ def runtime_status() -> dict[str, Any]:
 def mutation_provenance_fields() -> dict[str, Any]:
     """Fields to attach to live mutation / review audit records (#695 AC6)."""
     st = native_runtime_status()
-    transport = "native_mcp" if st["native_mcp_transport"] else "untrusted"
-    if st.get("mode") == _RUNTIME_MODE_TEST and st["native_mcp_transport"]:
-        transport = "test_native_mcp"
     return {
-        "transport": transport,
+        "transport": "native_mcp" if st["native_mcp_transport"] else "untrusted",
         "native_mcp_transport": bool(st["native_mcp_transport"]),
-        "production_native_mcp_transport": bool(
-            st.get("production_native_mcp_transport")
-        ),
         "native_runtime_pid": st.get("pid"),
         "native_token_fingerprint": st.get("token_fingerprint"),
         "entrypoint": st.get("entrypoint"),
-        "phase": st.get("phase"),
-        "mode": st.get("mode"),
-        "session_state_dir": st.get("session_state_dir"),
-        "session_state_dir_pinned": bool(st.get("session_state_dir_pinned")),
     }

mcp_server.py

@@ -41,17 +41,15 @@ def check_conflict_markers():
 
 check_conflict_markers()
 
-# #558 / #695: claim the official entrypoint before loading mutation modules.
+# #558: official entrypoint marks the process as the sanctioned MCP daemon
-# This alone does NOT authorize mutations — gitea_mcp_server binds the live
+# before loading mutation modules (blocks raw shell import bypasses).
-# native MCP transport (stdio) immediately before mcp.run. Import-only or
-# offline launch without that bind fails closed on mutations.
 try:
     import mcp_daemon_guard
 
     mcp_daemon_guard.mark_sanctioned_daemon()
 except Exception:
     # Guard import failures must not hide conflict-marker infra_stop above;
-    # gitea_mcp_server main also marks + binds when run as __main__.
+    # gitea_mcp_server main also marks sanctioned when run as __main__.
     pass
 
 # Execute the actual server logic via exec in this namespace.

mcp_session_state.py

@@ -44,34 +44,6 @@ SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"
 
 
 def default_state_dir() -> str:
-    """Resolve the durable session-state root.
-
-    When production native MCP transport is bound (#695 AC2), the directory
-    pinned at transport bind is authoritative: later overrides of
-    ``GITEA_MCP_SESSION_STATE_DIR`` cannot manufacture a second authority
-    domain (PR #701 recurrence: ``.mcp_session_701`` evasion of cross-PR
-    decision locks).
-    """
-    try:
-        import mcp_daemon_guard
-
-        pinned = mcp_daemon_guard.pinned_session_state_dir()
-        if pinned:
-            return pinned
-    except Exception:
-        # Fail open to env/default only when guard is unavailable (e.g. partial
-        # import during bootstrap). Mutation gates still fail closed separately.
-        pass
-    raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip()
-    return raw or DEFAULT_STATE_DIR
-
-
-def env_session_state_dir_unpinned() -> str:
-    """Raw env/default session-state dir ignoring production transport pin.
-
-    Intended for diagnostics and tests that assert pin behavior — not for
-    mutation-sensitive durable proofs under a bound native daemon.
-    """
     raw = (os.environ.get(STATE_DIR_ENV) or DEFAULT_STATE_DIR).strip()
     return raw or DEFAULT_STATE_DIR
 
@@ -390,26 +362,6 @@ def save_state(
             body["org"] = key_org
         if key_repo is not None:
             body["repo"] = key_repo
-        # Stamp session-state authority used for this write (#695 AC2 / AC6).
-        body.setdefault("session_state_dir", root)
-        try:
-            import mcp_daemon_guard
-
-            prov = mcp_daemon_guard.mutation_provenance_fields()
-            body.setdefault(
-                "native_token_fingerprint", prov.get("native_token_fingerprint")
-            )
-            body.setdefault(
-                "native_mcp_transport", bool(prov.get("native_mcp_transport"))
-            )
-            body.setdefault(
-                "production_native_mcp_transport",
-                bool(prov.get("production_native_mcp_transport")),
-            )
-            body.setdefault("transport", prov.get("transport"))
-        except Exception:
-            body.setdefault("native_mcp_transport", False)
-            body.setdefault("transport", "untrusted")
 
         envelope = {
             "kind": kind,
@@ -421,8 +373,6 @@ def save_state(
             "recorded_at": body["recorded_at"],
             "updated_at": body["updated_at"],
             "writer_pid": body["writer_pid"],
-            "session_state_dir": body.get("session_state_dir"),
-            "transport": body.get("transport"),
             "payload": body,
         }
         _write_json(path, envelope)

tests/conftest.py

@@ -58,17 +58,6 @@ def _reset_mutation_authority(monkeypatch):
             _fallback: str = state_dir,
             _env_key: str = mcp_session_state.STATE_DIR_ENV,
         ) -> str:
-            # #695 AC2: when production native transport has pinned a session
-            # state root, that pin is authoritative even under test isolation
-            # (PR #701 redirected-state regression).
-            try:
-                import mcp_daemon_guard
-
-                pinned = mcp_daemon_guard.pinned_session_state_dir()
-                if pinned:
-                    return pinned
-            except Exception:
-                pass
             raw = (os.environ.get(_env_key) or "").strip()
             return raw or _fallback
 

tests/test_issue_695_native_transport_quarantine.py

@@ -1,18 +1,14 @@
 """Regression tests for Issue #695 — second incident (PR #694 / review 427).
 
 Reproduces offline import, env-only runtime spoof, exposed-token invocation,
-direct imports, basename entrypoint spoof, allow_test_bootstrap forgery,
+direct imports, locally generated runtime keys, standalone quarantine attempts,
-standalone quarantine attempts, and false “official workflow” canonical claims.
+and false “official workflow” canonical claims. Gates must fail closed.
-Gates must fail closed.
 """
 
 from __future__ import annotations
 
 import os
-import subprocess
-import sys
 import tempfile
-import textwrap
 import unittest
 from pathlib import Path
 from unittest.mock import patch
@@ -24,26 +20,6 @@ import canonical_comment_validator as ccv
 
 HEAD_694 = "1844e298809373be19a526fd39b7d8b0669eb5bd"
 HEAD_OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
-REPO_ROOT = Path(__file__).resolve().parent.parent
-
-
-def _run_offline_snippet(snippet: str, *, env_extra: dict[str, str] | None = None) -> subprocess.CompletedProcess:
-    """Execute snippet in a fresh interpreter (no pytest modules)."""
-    env = os.environ.copy()
-    env.pop("PYTEST_CURRENT_TEST", None)
-    env.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
-    env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "")
-    if env_extra:
-        env.update(env_extra)
-    return subprocess.run(
-        [sys.executable, "-c", snippet],
-        capture_output=True,
-        text=True,
-        cwd=str(REPO_ROOT),
-        env=env,
-        timeout=30,
-        check=False,
-    )
 
 
 class TestNativeTransportBinding(unittest.TestCase):
@@ -63,53 +39,35 @@ class TestNativeTransportBinding(unittest.TestCase):
             mcp_daemon_guard.assert_sanctioned_mutation_runtime("offline_import")
         msg = str(ctx.exception)
         self.assertIn("#695", msg)
-        # FORCE disables pytest allowance; direct-import env is rejected first
+        self.assertIn("not sufficient", msg.lower() + " " + msg)
-        # (AC1). Without ALLOW_DIRECT, env-alone also yields "not sufficient".
-        lowered = msg.lower()
-        self.assertTrue(
-            "direct" in lowered
-            or "not sufficient" in lowered
-            or "allow_direct" in lowered
-            or "gitea_allow_direct" in lowered,
-            msg,
-        )
 
     def test_direct_import_mark_rejected_outside_entrypoint(self):
         mcp_daemon_guard.clear_native_runtime_for_tests()
         os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
         with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
             mcp_daemon_guard.mark_sanctioned_daemon()
-        self.assertIn("canonical", str(ctx.exception).lower())
+        self.assertIn("mcp_server.py", str(ctx.exception))
         self.assertFalse(mcp_daemon_guard.is_native_mcp_transport())
 
     def test_locally_generated_runtime_key_without_entrypoint_rejected(self):
         """Spoofing process-local fields via mark outside entrypoint fails."""
         mcp_daemon_guard.clear_native_runtime_for_tests()
         os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
+        # Even if a caller tries allow_test_bootstrap under force-unsanctioned
+        # pytest path is also forced off — only real entrypoint may mark.
         with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
-            mcp_daemon_guard.mark_sanctioned_daemon()
+            mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=False)
 
-    def test_test_native_runtime_for_hermetic_tests_not_production(self):
+    def test_test_bootstrap_establishes_native_for_hermetic_tests(self):
         mcp_daemon_guard.clear_native_runtime_for_tests()
-        st = mcp_daemon_guard.install_test_native_runtime()
+        st = mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
         self.assertTrue(st["native_mcp_transport"])
         self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
-        self.assertFalse(mcp_daemon_guard.is_production_native_mcp_transport())
         mcp_daemon_guard.assert_sanctioned_mutation_runtime("test-bootstrap")
         fields = mcp_daemon_guard.mutation_provenance_fields()
-        self.assertEqual(fields["transport"], "test_native_mcp")
+        self.assertEqual(fields["transport"], "native_mcp")
         self.assertTrue(fields["native_mcp_transport"])
-        self.assertFalse(fields["production_native_mcp_transport"])
         self.assertIsNotNone(fields["native_token_fingerprint"])
-        with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
-            mcp_daemon_guard.assert_production_mutation_runtime("prod-endpoint")
-        self.assertIn("Test-mode", str(ctx.exception))
-
-    def test_no_allow_test_bootstrap_parameter_on_mark(self):
-        import inspect
-
-        sig = inspect.signature(mcp_daemon_guard.mark_sanctioned_daemon)
-        self.assertNotIn("allow_test_bootstrap", sig.parameters)
 
     def test_exposed_token_env_never_grants_native(self):
         """Raw / exposed token env vars must never reconstruct native transport."""
@@ -138,249 +96,6 @@ class TestNativeTransportBinding(unittest.TestCase):
                 os.environ.pop(key, None)
 
 
-class TestAC9BypassRegressions(unittest.TestCase):
-    """AC9: empirically reproduced offline bypasses must fail closed (#695)."""
-
-    def tearDown(self) -> None:
-        mcp_daemon_guard.clear_native_runtime_for_tests()
-        os.environ.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
-
-    def test_allow_test_bootstrap_cannot_authorize_fresh_offline_interpreter(self):
-        """Finding 1: former allow_test_bootstrap forge is gone and rejected offline."""
-        snippet = textwrap.dedent(
-            """
-            import inspect
-            import mcp_daemon_guard as g
-            sig = inspect.signature(g.mark_sanctioned_daemon)
-            assert "allow_test_bootstrap" not in sig.parameters, "bootstrap flag must not exist"
-            try:
-                g.mark_sanctioned_daemon(allow_test_bootstrap=True)
-            except TypeError:
-                pass
-            else:
-                raise SystemExit("mark_sanctioned_daemon accepted allow_test_bootstrap")
-            try:
-                g.install_test_native_runtime()
-            except g.UnsanctionedRuntimeError as exc:
-                assert "pytest" in str(exc).lower() or "#695" in str(exc)
-            else:
-                raise SystemExit("install_test_native_runtime authorized offline interpreter")
-            assert g.is_native_mcp_transport() is False
-            try:
-                g.assert_sanctioned_mutation_runtime("offline-bootstrap")
-            except g.UnsanctionedRuntimeError:
-                pass
-            else:
-                raise SystemExit("mutation runtime authorized after offline bootstrap attempt")
-            print("OK")
-            """
-        )
-        proc = _run_offline_snippet(snippet)
-        self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
-        self.assertIn("OK", proc.stdout)
-
-    def test_renamed_runner_named_mcp_server_py_rejected(self):
-        """Finding 2: basename-only entrypoint trust is insufficient."""
-        with tempfile.TemporaryDirectory() as tmp:
-            attacker = Path(tmp) / "mcp_server.py"
-            attacker.write_text(
-                textwrap.dedent(
-                    """
-                    import mcp_daemon_guard as g
-                    try:
-                        g.mark_sanctioned_daemon()
-                    except g.UnsanctionedRuntimeError as exc:
-                        print("REJECTED:" + str(exc))
-                        raise SystemExit(0)
-                    print("AUTHORIZED")
-                    raise SystemExit(1)
-                    """
-                ),
-                encoding="utf-8",
-            )
-            env = os.environ.copy()
-            env.pop("PYTEST_CURRENT_TEST", None)
-            env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "")
-            proc = subprocess.run(
-                [sys.executable, str(attacker)],
-                capture_output=True,
-                text=True,
-                cwd=tmp,
-                env=env,
-                timeout=30,
-                check=False,
-            )
-            self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
-            self.assertIn("REJECTED:", proc.stdout)
-            self.assertIn("canonical", proc.stdout.lower())
-            self.assertNotIn("AUTHORIZED", proc.stdout)
-
-    def test_direct_launch_import_canonical_entrypoint_without_transport_rejected(self):
-        """Merely importing/launching real entrypoint offline must not authorize."""
-        snippet = textwrap.dedent(
-            f"""
-            import importlib.util
-            import mcp_daemon_guard as g
-            # Simulate claim-only phase (no transport bind).
-            g.clear_native_runtime_for_tests()
-            # Direct mark from non-entrypoint must fail.
-            try:
-                g.mark_sanctioned_daemon()
-            except g.UnsanctionedRuntimeError:
-                pass
-            assert g.is_native_mcp_transport() is False
-            # Even if someone forges entrypoint_claimed without transport bind:
-            g._NATIVE_RUNTIME = {{
-                "token": "x" * 64,
-                "token_fingerprint": "deadbeefdeadbeef",
-                "pid": __import__("os").getpid(),
-                "started_at": 0,
-                "entrypoint": "mcp_server",
-                "entrypoint_path": {str(REPO_ROOT / "mcp_server.py")!r},
-                "phase": "entrypoint_claimed",
-                "transport": None,
-                "mode": "production",
-            }}
-            assert g.is_native_mcp_transport() is False
-            try:
-                g.assert_sanctioned_mutation_runtime("import-only")
-            except g.UnsanctionedRuntimeError as exc:
-                assert "transport" in str(exc).lower() or "entrypoint" in str(exc).lower() or "#695" in str(exc)
-            else:
-                raise SystemExit("import-only claim authorized mutation")
-            print("OK")
-            """
-        )
-        proc = _run_offline_snippet(snippet)
-        self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
-        self.assertIn("OK", proc.stdout)
-
-    def test_spoofed_pytest_env_stack_path_rejected(self):
-        """Spoofed pytest/env/call-stack/path evidence must not authorize offline."""
-        snippet = textwrap.dedent(
-            f"""
-            import os
-            import mcp_daemon_guard as g
-            os.environ["PYTEST_CURRENT_TEST"] = "spoofed::test"
-            os.environ[g.SANCTIONED_DAEMON_ENV] = "1"
-            os.environ[g.ALLOW_DIRECT_IMPORT_ENV] = "1"
-            # Fresh interpreter has no pytest module; PYTEST_CURRENT_TEST alone
-            # might still trip is_pytest_runtime — force-unsanctioned is not set.
-            # But install_test_native_runtime requires real pytest path; if
-            # PYTEST_CURRENT_TEST alone grants is_pytest_runtime, production
-            # mutation still requires production transport outside true pytest.
-            if g.is_pytest_runtime():
-                # Env-only pytest spoof: test install may succeed, but production
-                # mutation gate must still reject test mode.
-                g.install_test_native_runtime()
-                assert g.is_production_native_mcp_transport() is False
-                try:
-                    g.assert_production_mutation_runtime("spoofed-pytest")
-                except g.UnsanctionedRuntimeError:
-                    pass
-                else:
-                    raise SystemExit("production mutation accepted test-mode under spoofed pytest env")
-            else:
-                try:
-                    g.install_test_native_runtime()
-                except g.UnsanctionedRuntimeError:
-                    pass
-                else:
-                    raise SystemExit("test install without pytest evidence")
-            # Basename path spoof via inspect is covered elsewhere; env alone:
-            g.clear_native_runtime_for_tests()
-            os.environ.pop("PYTEST_CURRENT_TEST", None)
-            assert g.is_native_mcp_transport() is False
-            try:
-                g.assert_sanctioned_mutation_runtime("env-path-spoof")
-            except g.UnsanctionedRuntimeError:
-                pass
-            else:
-                raise SystemExit("env/path spoof authorized mutation")
-            print("OK")
-            """
-        )
-        proc = _run_offline_snippet(snippet)
-        self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
-        self.assertIn("OK", proc.stdout)
-
-    def test_test_bootstrap_cannot_reach_production_mutation_endpoints(self):
-        """Under pytest, test-mode runtime cannot satisfy production mutation gate."""
-        mcp_daemon_guard.clear_native_runtime_for_tests()
-        mcp_daemon_guard.install_test_native_runtime()
-        with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
-            mcp_daemon_guard.assert_production_mutation_runtime(
-                "gitea_quarantine_contaminated_review"
-            )
-        msg = str(ctx.exception)
-        self.assertIn("Test-mode", msg)
-        self.assertIn("#695", msg)
-
-        # Offline: install_test_native_runtime must not authorize production mutations.
-        snippet = textwrap.dedent(
-            """
-            import mcp_daemon_guard as g
-            try:
-                g.install_test_native_runtime()
-            except g.UnsanctionedRuntimeError:
-                pass
-            assert g.is_production_native_mcp_transport() is False
-            try:
-                g.assert_production_mutation_runtime("gitea_quarantine_contaminated_review")
-            except g.UnsanctionedRuntimeError:
-                pass
-            else:
-                raise SystemExit("production mutation authorized offline")
-            try:
-                g.assert_sanctioned_mutation_runtime("gitea_mutation")
-            except g.UnsanctionedRuntimeError:
-                pass
-            else:
-                raise SystemExit("sanctioned mutation authorized offline")
-            print("OK")
-            """
-        )
-        proc = _run_offline_snippet(snippet)
-        self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
-        self.assertIn("OK", proc.stdout)
-
-    def test_legitimate_native_transport_bind_succeeds(self):
-        """Canonical entrypoint path + stdio bind establishes production native."""
-        mcp_daemon_guard.clear_native_runtime_for_tests()
-        # Simulate production path under force-unsanctioned (no pytest allowance)
-        # by calling internal claim/bind with patched caller path.
-        canonical = str((REPO_ROOT / "mcp_server.py").resolve())
-
-        def _fake_caller():
-            return canonical
-
-        with patch.object(
-            mcp_daemon_guard, "_caller_official_entrypoint_path", side_effect=_fake_caller
-        ):
-            # Force non-pytest path for mark/bind logic.
-            with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
-                st1 = mcp_daemon_guard.mark_sanctioned_daemon()
-                self.assertFalse(st1["native_mcp_transport"])
-                self.assertEqual(st1["phase"], "entrypoint_claimed")
-                st2 = mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
-                self.assertTrue(st2["native_mcp_transport"])
-                self.assertTrue(st2["production_native_mcp_transport"])
-                self.assertEqual(st2["transport"], "stdio")
-                self.assertEqual(st2["mode"], "production")
-                mcp_daemon_guard.assert_sanctioned_mutation_runtime("native-ide")
-                mcp_daemon_guard.assert_production_mutation_runtime("native-ide")
-                fields = mcp_daemon_guard.mutation_provenance_fields()
-                self.assertEqual(fields["transport"], "native_mcp")
-                self.assertTrue(fields["production_native_mcp_transport"])
-
-    def test_canonical_entrypoint_paths_are_resolved_absolute(self):
-        paths = mcp_daemon_guard.canonical_entrypoint_paths()
-        self.assertTrue(any(p.endswith("mcp_server.py") for p in paths))
-        for p in paths:
-            self.assertTrue(os.path.isabs(p), p)
-            self.assertEqual(p, str(Path(p).resolve()))
-
-
 class TestQuarantineWriteNativeOnly(unittest.TestCase):
     def setUp(self) -> None:
         self._tmp = tempfile.TemporaryDirectory()
@@ -418,12 +133,7 @@ class TestQuarantineWriteNativeOnly(unittest.TestCase):
             forensic_comment_ids=[10883, 10886],
         )
         with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
-            # Force non-pytest and non-native for write path.
+            review_quarantine.write_quarantine_record(record)
-            with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
-                with patch.object(
-                    mcp_daemon_guard, "is_native_mcp_transport", return_value=False
-                ):
-                    review_quarantine.write_quarantine_record(record)
 
     def test_confirmation_must_match_exactly(self):
         assessment = review_quarantine.assess_quarantine_write(
@@ -460,7 +170,7 @@ class TestQuarantineWriteNativeOnly(unittest.TestCase):
             "review_quarantine.mcp_session_state.default_state_dir",
             return_value=self._tmp.name,
         ):
-            mcp_daemon_guard.install_test_native_runtime()
+            mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
             record = review_quarantine.build_quarantine_record(
                 remote="prgs",
                 org="org",
@@ -595,7 +305,7 @@ class TestFeedbackQuarantineIntegration(unittest.TestCase):
         self._tmp = tempfile.TemporaryDirectory()
         self.addCleanup(self._tmp.cleanup)
         mcp_daemon_guard.clear_native_runtime_for_tests()
-        mcp_daemon_guard.install_test_native_runtime()
+        mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
 
     def tearDown(self) -> None:
         mcp_daemon_guard.clear_native_runtime_for_tests()
@@ -686,163 +396,5 @@ class TestDocsStopAfterNativeFailure(unittest.TestCase):
         self.assertIn("GITEA_ALLOW_DIRECT_MCP_IMPORT", doc)
 
 
-class TestPR701DirectImportSessionOverrideSequence(unittest.TestCase):
-    """AC9 regression: exact PR #701 contaminated-approval sequence must fail closed.
-
-    Observed attack:
-    - GITEA_ALLOW_DIRECT_MCP_IMPORT=1
-    - import mutation tools from gitea_mcp_server
-    - GITEA_MCP_SESSION_STATE_DIR=.mcp_session_701 (cross-PR lock evasion)
-    - mark_final + submit_pr_review
-    """
-
-    def tearDown(self) -> None:
-        mcp_daemon_guard.clear_native_runtime_for_tests()
-        os.environ.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
-        os.environ.pop(mcp_daemon_guard.SANCTIONED_DAEMON_ENV, None)
-        os.environ.pop(mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV, None)
-        os.environ.pop(mcp_daemon_guard.SESSION_STATE_DIR_ENV, None)
-
-    def test_offline_run_submit_sequence_fails_closed(self):
-        """Fresh interpreter: direct import + state-dir override cannot mark/submit."""
-        with tempfile.TemporaryDirectory() as tmp:
-            redirect = str(Path(tmp) / ".mcp_session_701")
-            snippet = textwrap.dedent(
-                f"""
-                import os
-                import sys
-                os.environ["GITEA_ALLOW_DIRECT_MCP_IMPORT"] = "1"
-                os.environ["GITEA_MCP_SESSION_STATE_DIR"] = {redirect!r}
-                os.environ["GITEA_MCP_PROFILE"] = "prgs-reviewer"
-                # No pytest modules in this subprocess.
-                import mcp_daemon_guard as g
-                assert g.is_native_mcp_transport() is False
-                assert g.is_production_native_mcp_transport() is False
-                try:
-                    g.assert_sanctioned_mutation_runtime("run_submit_mark")
-                except g.UnsanctionedRuntimeError as exc:
-                    msg = str(exc)
-                    assert "GITEA_ALLOW_DIRECT_MCP_IMPORT" in msg or "#695" in msg
-                else:
-                    raise SystemExit("direct-import env authorized mutation runtime")
-                try:
-                    g.mark_sanctioned_daemon()
-                except g.UnsanctionedRuntimeError:
-                    pass
-                else:
-                    raise SystemExit("mark_sanctioned_daemon authorized offline import")
-                # Simulate decision-lock write into redirected dir only — must not
-                # establish native authority.
-                import mcp_session_state as ss
-                wrote = ss.save_state(
-                    kind=ss.KIND_DECISION_LOCK,
-                    payload={{
-                        "final_review_decision_ready": True,
-                        "ready_pr_number": 701,
-                        "ready_action": "approve",
-                        "ready_expected_head_sha": "6b675f5c834b41f9d74e8a54294ff44dddf28ae4",
-                        "session_profile": "prgs-reviewer",
-                        "session_profile_lock": "prgs-reviewer",
-                        "remote": "prgs",
-                    }},
-                    profile_identity="prgs-reviewer",
-                    state_dir={redirect!r},
-                )
-                assert wrote is not None
-                assert g.is_native_mcp_transport() is False
-                try:
-                    g.assert_no_direct_import_bypass("gitea_submit_pr_review")
-                except g.UnsanctionedRuntimeError:
-                    pass
-                else:
-                    raise SystemExit("direct-import bypass accepted for submit")
-                print("OK")
-                """
-            )
-            proc = _run_offline_snippet(snippet)
-            self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
-            self.assertIn("OK", proc.stdout)
-
-    def test_session_state_dir_pin_ignores_post_bind_redirect(self):
-        """AC2: after production bind, env STATE_DIR override is ignored."""
-        mcp_daemon_guard.clear_native_runtime_for_tests()
-        import mcp_session_state
-
-        with tempfile.TemporaryDirectory() as tmp:
-            legitimate = str(Path(tmp) / "legitimate-state")
-            rogue = str(Path(tmp) / ".mcp_session_701")
-            os.makedirs(legitimate, mode=0o700, exist_ok=True)
-            os.makedirs(rogue, mode=0o700, exist_ok=True)
-            os.environ[mcp_daemon_guard.SESSION_STATE_DIR_ENV] = legitimate
-            canonical = str((REPO_ROOT / "mcp_server.py").resolve())
-
-            def _fake_caller():
-                return canonical
-
-            with patch.object(
-                mcp_daemon_guard,
-                "_caller_official_entrypoint_path",
-                side_effect=_fake_caller,
-            ):
-                with patch.object(
-                    mcp_daemon_guard, "is_pytest_runtime", return_value=False
-                ):
-                    mcp_daemon_guard.mark_sanctioned_daemon()
-                    mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
-                    pinned = mcp_daemon_guard.pinned_session_state_dir()
-                    self.assertEqual(pinned, str(Path(legitimate).resolve()))
-                    # Attacker redirects env after bind (PR #701).
-                    os.environ[mcp_daemon_guard.SESSION_STATE_DIR_ENV] = rogue
-                    self.assertEqual(
-                        mcp_daemon_guard.pinned_session_state_dir(),
-                        str(Path(legitimate).resolve()),
-                    )
-                    self.assertEqual(
-                        mcp_session_state.default_state_dir(),
-                        str(Path(legitimate).resolve()),
-                    )
-                    self.assertNotEqual(
-                        mcp_session_state.default_state_dir(),
-                        str(Path(rogue).resolve()),
-                    )
-                    # Unpinned env view still sees rogue (diagnostics only).
-                    unpinned = mcp_session_state.env_session_state_dir_unpinned()
-                    self.assertTrue(
-                        unpinned == rogue
-                        or Path(unpinned).resolve() == Path(rogue).resolve(),
-                        unpinned,
-                    )
-
-    def test_direct_import_env_does_not_authorize_under_force_unsanctioned(self):
-        mcp_daemon_guard.clear_native_runtime_for_tests()
-        os.environ[mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV] = "1"
-        os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
-        self.assertTrue(mcp_daemon_guard.direct_import_env_enabled())
-        # Under pytest, assert_no_direct_import_bypass is a no-op; FORCE path
-        # still blocks is_native / assert_sanctioned via force-unsanctioned.
-        self.assertFalse(mcp_daemon_guard.is_native_mcp_transport())
-        with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
-            mcp_daemon_guard.assert_sanctioned_mutation_runtime("direct-import")
-
-    def test_quarantine_voids_merge_approval_for_contaminated_review(self):
-        """AC6–AC8: quarantined APPROVED does not satisfy merge approval head."""
-        entry = {
-            "verdict": "APPROVED",
-            "dismissed": False,
-            "reviewed_head_sha": "6b675f5c834b41f9d74e8a54294ff44dddf28ae4",
-            "review_id": 431,
-            "submitted_at": "2026-07-13T23:52:34Z",
-            "quarantined": True,
-        }
-        result = merge_approval_gate.assess_merge_approval_head(
-            current_head_sha="6b675f5c834b41f9d74e8a54294ff44dddf28ae4",
-            latest_by_reviewer={"sysadmin": entry},
-            quarantined_review_ids={431},
-        )
-        self.assertFalse(result["approval_at_current_head"])
-        self.assertEqual(result["quarantined_approvals_at_current_head"], 1)
-        self.assertIn("quarantined", (result["stale_approval_block_reason"] or ""))
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_mcp_daemon_guard.py

@@ -30,17 +30,11 @@ class TestMcpDaemonGuard(unittest.TestCase):
         # Running under pytest already sets PYTEST_CURRENT_TEST.
         mcp_daemon_guard.assert_sanctioned_mutation_runtime("pytest")
 
-    def test_test_native_runtime_install_under_pytest(self):
+    def test_mark_sanctioned_bootstrap_allows(self):
         mcp_daemon_guard.clear_native_runtime_for_tests()
-        mcp_daemon_guard.install_test_native_runtime()
+        mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
-        self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
-        self.assertFalse(mcp_daemon_guard.is_production_native_mcp_transport())
-        # Hermetic unit path still passes under pytest.
         mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon")
-        # Production mutation gate rejects test-mode records.
+        self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
-        with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
-            mcp_daemon_guard.assert_production_mutation_runtime("gitea_mutation")
-        self.assertIn("Test-mode", str(ctx.exception))
 
     def test_env_alone_insufficient_when_force_unsanctioned(self):
         mcp_daemon_guard.clear_native_runtime_for_tests()
@@ -93,11 +87,6 @@ class TestMcpDaemonGuard(unittest.TestCase):
             with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
                 gitea_auth.get_auth_header("gitea.prgs.cc")
 
-    def test_no_allow_test_bootstrap_public_parameter(self):
-        """Production mark must not accept allow_test_bootstrap (#695)."""
-        sig = __import__("inspect").signature(mcp_daemon_guard.mark_sanctioned_daemon)
-        self.assertNotIn("allow_test_bootstrap", sig.parameters)
-
 
 if __name__ == "__main__":
     unittest.main()