fix(guard): pin session-state authority and reject direct-import mutation (#695)

Close remaining AC1/AC2/AC6-8 gaps after REQUEST_CHANGES and the PR #701
recurrence: GITEA_ALLOW_DIRECT_MCP_IMPORT never authorizes mutations;
production transport bind pins GITEA_MCP_SESSION_STATE_DIR so redirected
dirs cannot forge independent decision locks; mark/submit fail closed
offline; quarantine continues to void contaminated merge eligibility.

Regression tests reproduce the PR #701 direct-import + state-dir override
sequence. Full suite: 2665 passed, 6 skipped.

Closes #695 (remediation on PR #696)

mcp_daemon_guard.py

@@ -48,6 +48,11 @@ _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."""
@@ -211,10 +216,22 @@ def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]:
             "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()
 
@@ -264,6 +281,50 @@ def clear_native_runtime_for_tests() -> None:
     _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)."""
     if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
@@ -329,9 +390,12 @@ def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
     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")
@@ -367,7 +431,8 @@ def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
         "shell, offline runner, or ad-hoc script after native MCP failure. "
         "Stop and reconnect the official MCP daemon (mcp_server.py) over "
         "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}"
     )
 
@@ -412,9 +477,13 @@ def native_runtime_status() -> dict[str, Any]:
         "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,
     }
 
@@ -443,4 +512,6 @@ def mutation_provenance_fields() -> dict[str, Any]:
         "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")),
     }