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.

.env.example

@@ -55,12 +55,3 @@ GITEA_MCP_PROFILE=prgs
 # GITEA_MERGER_WORKTREE=/path/to/repo/branches/merge-pr456
 # GITEA_RECONCILER_WORKTREE=/path/to/repo/branches/reconcile-pr456
 # GITEA_ACTIVE_WORKTREE=/path/to/repo/branches/session-override
-
-# Durable HMAC key for irrecoverable decision-lock provenance artifacts (#709 F4).
-# REQUIRED in production native MCP processes that mint or verify recovery
-# authorization (reconciler + merger must share the same key). Hex (preferred),
-# base64, or utf-8 literal (>=16 bytes). Never generate an ephemeral per-process
-# key — cross-process / post-restart verification would fail closed.
-# GITEA_IRRECOVERABLE_AUTH_HMAC_KEY=0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef
-# Optional key version string bound into the signature (for rotation).
-# GITEA_IRRECOVERABLE_AUTH_HMAC_KEY_VERSION=v1

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)
 

final_report_validator.py

@@ -134,22 +134,16 @@ _TARGET_BRANCH_SHA_RE = re.compile(
     r"target branch sha\s*:\s*[0-9a-f]{40}",
     re.IGNORECASE,
 )
-# #698: structured proof is rendered in several equivalent shapes —
-# `workflow_hash: abc...`, `workflow_hash=abc...`, or JSON
-# `"workflow_hash": "abc..."`. Recognize all of them; demanding one exact
-# punctuation style rejects legitimate structured workflow-load proof.
 _WORKFLOW_LOAD_HELPER_RE = re.compile(
-    r"workflow[-_ ]load[-_ ]helper[-_ ]result\s*[:=]",
+    r"workflow[- ]load helper result\s*:",
     re.IGNORECASE,
 )
 _WORKFLOW_LOAD_HASH_RE = re.compile(
-    r"workflow[-_ ]load[-_ ]helper[-_ ]result[\s\S]{0,400}?"
+    r"workflow[- ]load helper result[\s\S]{0,400}?workflow[_ ]hash\s*:\s*[0-9a-f]{12}",
-    r"workflow[_ ]hash\"?\s*[:=]\s*\"?[0-9a-f]{12}",
     re.IGNORECASE,
 )
 _WORKFLOW_LOAD_BOUNDARY_RE = re.compile(
-    r"workflow[-_ ]load[-_ ]helper[-_ ]result[\s\S]{0,400}?"
+    r"workflow[- ]load helper result[\s\S]{0,400}?boundary[_ ]status\s*:\s*(?:clean|violation)",
-    r"boundary[_ ]status\"?\s*[:=]\s*\"?(?:clean|violation)",
     re.IGNORECASE,
 )
 _WORKFLOW_FILE_VIEW_NARRATIVE_RE = re.compile(
@@ -250,59 +244,6 @@ def _normalize_task_kind(task_kind: str | None) -> str:
     return _TASK_KIND_ALIASES.get(raw, raw)
 
 
-def _iter_action_entries(action_log: list | None) -> list[dict]:
-    """Yield only structured (dict) action-log entries (#698).
-
-    Callers must never crash on malformed entries (strings, numbers, null)
-    that reach the validator from LLM-composed or partially parsed logs;
-    :func:`sanitize_action_log` reports them separately.
-    """
-    return [e for e in (action_log or []) if isinstance(e, dict)]
-
-
-def sanitize_action_log(
-    action_log: list | None,
-) -> tuple[list[dict], list[dict[str, str]]]:
-    """Split an action log into structured entries and sanitized findings (#698).
-
-    Malformed entries become clear, sanitized ``warning`` findings — the
-    offending value's content is never echoed back (only its position and
-    type), so secrets or garbage in a broken log cannot leak into validation
-    errors, and validation itself proceeds without secondary exceptions.
-    """
-    if action_log is None:
-        return [], []
-    if not isinstance(action_log, (list, tuple)):
-        return [], [
-            validator_finding(
-                "shared.action_log_malformed",
-                "downgrade",
-                "Action log",
-                "action_log is not a list of structured entries "
-                f"(got {type(action_log).__name__}); it was ignored",
-                "pass action_log as a list of dict entries",
-            )
-        ]
-    entries: list[dict] = []
-    findings: list[dict[str, str]] = []
-    for index, entry in enumerate(action_log):
-        if isinstance(entry, dict):
-            entries.append(entry)
-            continue
-        findings.append(
-            validator_finding(
-                "shared.action_log_malformed",
-                "downgrade",
-                "Action log",
-                f"action_log entry {index} is not a structured mapping "
-                f"(got {type(entry).__name__}); the entry was ignored",
-                "repair the malformed action_log entry or drop it before "
-                "revalidating",
-            )
-        )
-    return entries, findings
-
-
 def validator_finding(
     rule_id: str,
     severity: str,
@@ -480,7 +421,7 @@ def _rule_shared_canonical_comment_post_claim(
     rejected_in_report = bool(_CANONICAL_VALIDATION_REJECTED_RE.search(text))
     rejected_in_log = False
     if action_log:
-        for entry in _iter_action_entries(action_log):
+        for entry in action_log:
             validation = entry.get("canonical_comment_validation") or {}
             if validation.get("allowed") is False:
                 rejected_in_log = True
@@ -534,13 +475,9 @@ def _rule_reviewer_vague_mutations_none(
     action_log: list[dict] | None = None,
     mutations_observed: bool = False,
 ) -> list[dict[str, str]]:
-    # #698: infer review mutations only from authoritative evidence — an
-    # entry proves a mutation only when it affirmatively records
-    # performed=true and was not gated. Read-only diagnostics and pre-API
-    # rejections (entries without a performed flag) are not mutations.
     performed = any(
-        e.get("performed") is True and not e.get("gated_rejected")
+        e.get("performed") is not False and not e.get("gated_rejected")
-        for e in _iter_action_entries(action_log)
+        for e in (action_log or [])
     )
     if not (mutations_observed or performed):
         return []
@@ -599,7 +536,7 @@ def _rule_reviewer_git_fetch_readonly(
     text = report_text or ""
     fetch_observed = any(
         _GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
-        for e in _iter_action_entries(action_log)
+        for e in (action_log or [])
     ) or _GIT_FETCH_RE.search(text)
     if not fetch_observed:
         return []
@@ -1040,7 +977,7 @@ def _rule_reviewer_target_branch_freshness(
     fields = _handoff_fields(text)
     fetch_reported = bool(_GIT_FETCH_RE.search(text)) or any(
         _GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
-        for e in _iter_action_entries(action_log)
+        for e in (action_log or [])
     )
     target_sha_reported = bool(_TARGET_BRANCH_SHA_RE.search(text)) or any(
         "target branch" in key and "sha" in key and _FULL_SHA_RE.search(value)
@@ -1798,12 +1735,6 @@ def assess_final_report_validator(
     checks: dict[str, Any] = {}
     findings: list[dict[str, str]] = []
 
-    # #698: malformed action_log data must never crash validation with a
-    # secondary exception; malformed entries surface as sanitized findings.
-    sanitized_action_log, action_log_findings = sanitize_action_log(action_log)
-    action_log = sanitized_action_log
-    findings.extend(action_log_findings)
-
     if normalized_kind == "issue_filing" and issue_filing_lock is not None:
         checks["issue_filing"] = assess_issue_filing_final_report(
             report_text,
@@ -1832,23 +1763,7 @@ def assess_final_report_validator(
     }
 
     for rule in _RULES_BY_TASK.get(normalized_kind, ()):
-        try:
+        findings.extend(_call_rule(rule, report_text, normalized_kind, rule_kwargs))
-            findings.extend(
-                _call_rule(rule, report_text, normalized_kind, rule_kwargs)
-            )
-        except Exception as exc:  # #698: fail closed with a sanitized error
-            findings.append(
-                validator_finding(
-                    "shared.validator_rule_error",
-                    "block",
-                    "Validator",
-                    f"validator rule '{getattr(rule, '__name__', 'unknown')}' "
-                    f"failed with {type(exc).__name__} (details withheld; "
-                    "sanitized)",
-                    "file a validator defect with the rule name; do not "
-                    "bypass final-report validation",
-                )
-            )
 
     grade, blocked, downgraded = _aggregate_grade(findings)
     reasons = [f"{f['rule_id']}: {f['reason']}" for f in findings]

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

@@ -36,24 +36,7 @@ KIND_REVIEW_DRAFT = "review_draft"
 # other session proofs; a contaminated session fails closed on gated mutations
 # until a reconciler audits and clears it.
 KIND_STABLE_BRANCH_CONTAMINATION = "stable_branch_contamination"
-# #709: archive prior terminal decision ledgers instead of silent overwrite.
-KIND_DECISION_LOCK_ARCHIVE = "review_decision_lock_archive"
-# #709: post-merge cleanup/audit reconciliation-required durable record.
-KIND_POST_MERGE_DECISION_RECOVERY = "post_merge_decision_recovery"
-# #709: truthful record when historical terminal evidence is irrecoverably gone.
-KIND_IRRECOVERABLE_DECISION_PROVENANCE = "irrecoverable_decision_provenance"
-# #709 F1: server-side non-forgeable authorization artifact for recovery.
-KIND_IRRECOVERABLE_PROVENANCE_AUTH = "irrecoverable_provenance_authorization"
 
-# Kinds that must survive the default session-state TTL (forensic / recovery).
-RECOVERY_CRITICAL_KINDS = frozenset(
-    {
-        KIND_DECISION_LOCK_ARCHIVE,
-        KIND_POST_MERGE_DECISION_RECOVERY,
-        KIND_IRRECOVERABLE_DECISION_PROVENANCE,
-        KIND_IRRECOVERABLE_PROVENANCE_AUTH,
-    }
-)
 
 
 _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
@@ -61,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
 
@@ -287,11 +242,7 @@ def identity_match_reasons(
         reasons.append("session state missing recorded_at timestamp (fail closed)")
     else:
         age = _now_utc() - recorded_at
-        kind = (record.get("kind") or "").strip()
+        if age > timedelta(hours=ttl_hours()):
-        ttl_exempt = kind in RECOVERY_CRITICAL_KINDS or bool(
-            record.get("recovery_critical")
-        )
-        if age > timedelta(hours=ttl_hours()) and not ttl_exempt:
             reasons.append(
                 f"session state expired after {ttl_hours():g}h (fail closed)"
             )
@@ -411,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,
@@ -442,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)
@@ -468,177 +397,3 @@ def clear_state(
         profile_identity=profile_identity,
         state_dir=state_dir,
     )
-
-def list_decision_lock_profile_identities(
-    state_dir: str | None = None,
-) -> list[str]:
-    """Return profile identities that have a durable review_decision_lock file (#709).
-
-    Filename form: ``review_decision_lock-<profile>.json`` (see ``state_key``).
-    Does not validate TTL or identity — callers must load via ``load_state``.
-    """
-    root = (state_dir or default_state_dir()).strip()
-    if not root or not os.path.isdir(root):
-        return []
-    prefix = f"{_sanitize_segment(KIND_DECISION_LOCK)}-"
-    suffix = ".json"
-    found: list[str] = []
-    try:
-        names = os.listdir(root)
-    except OSError:
-        return []
-    for name in names:
-        if not name.startswith(prefix) or not name.endswith(suffix):
-            continue
-        if name.endswith(".lock"):
-            continue
-        mid = name[len(prefix) : -len(suffix)]
-        if mid:
-            found.append(mid)
-    return sorted(set(found))
-
-
-def load_state_for_profile(
-    *,
-    kind: str,
-    profile_identity: str,
-    remote: str | None = None,
-    org: str | None = None,
-    repo: str | None = None,
-    state_dir: str | None = None,
-    skip_identity_match: bool = False,
-    enforce_repo_scope: bool = True,
-) -> dict[str, Any] | None:
-    """Load durable state for an explicit profile identity (#709 cross-profile).
-
-    When *skip_identity_match* is True, still requires the file's recorded
-    profile_identity to equal the requested profile (anti-stomp), but does not
-    require the *active* session identity to match — needed so a merger can
-    inspect a reviewer lock after merge.
-
-    #709 F3: remote/org/repo filter reasons are **enforced** (not merely
-    computed). When *enforce_repo_scope* is True (default) and the caller
-    supplies remote/org/repo, mismatches fail closed with ``None``.
-    """
-    # #709 F3: refuse traversal / malformed profile identities.
-    try:
-        from irrecoverable_provenance import assess_profile_path_identity
-
-        path_gate = assess_profile_path_identity(profile_identity)
-        if not path_gate.get("valid"):
-            return None
-    except Exception:
-        # Module may be mid-import in edge bootstraps; fall through to
-        # conservative checks below.
-        raw = str(profile_identity or "")
-        if ".." in raw or "/" in raw or "\\" in raw or "\x00" in raw:
-            return None
-
-    profile = current_profile_identity(profile_identity=profile_identity)
-    root = _ensure_state_dir(state_dir)
-    # Refuse symlink escape of the state root (#709 F3).
-    try:
-        real_root = os.path.realpath(root)
-        path = state_file_path(
-            kind=kind,
-            remote=remote,
-            org=org,
-            repo=repo,
-            profile_identity=profile,
-            state_dir=root,
-        )
-        real_path = os.path.realpath(path) if os.path.exists(path) else path
-        if os.path.exists(path) and not str(real_path).startswith(
-            str(real_root) + os.sep
-        ) and str(real_path) != str(real_root):
-            return None
-    except OSError:
-        return None
-
-    path = state_file_path(
-        kind=kind,
-        remote=remote,
-        org=org,
-        repo=repo,
-        profile_identity=profile,
-        state_dir=root,
-    )
-    lock_path = f"{path}.lock"
-    with _exclusive_file_lock(lock_path):
-        envelope = _read_json(path)
-    if not envelope:
-        return None
-    payload = envelope.get("payload")
-    if not isinstance(payload, dict):
-        return None
-    merged = dict(payload)
-    for key in (
-        "kind",
-        "remote",
-        "org",
-        "repo",
-        "profile_identity",
-        "session_profile_lock",
-        "recorded_at",
-        "updated_at",
-        "writer_pid",
-    ):
-        if key in envelope and key not in merged:
-            merged[key] = envelope[key]
-    stored = (merged.get("profile_identity") or "").strip()
-    if stored and stored != profile:
-        return None
-    if skip_identity_match:
-        # Enforce remote/org/repo when caller provides them (#709 F3).
-        # Drop only *active session* profile-identity mismatches; keep
-        # expiry, spoof, and repository-scope reasons.
-        scope_remote = remote if enforce_repo_scope else None
-        scope_org = org if enforce_repo_scope else None
-        scope_repo = repo if enforce_repo_scope else None
-        reasons = identity_match_reasons(
-            merged,
-            remote=scope_remote,
-            org=scope_org,
-            repo=scope_repo,
-            profile_identity=stored or profile,
-        )
-        filtered = [
-            r
-            for r in reasons
-            if "profile identity mismatch" not in r
-            or (stored and stored != profile)
-        ]
-        # When caller requested a specific remote/org/repo, also fail if the
-        # stored record lacks those identity fields entirely (legacy incomplete).
-        if enforce_repo_scope:
-            for field, want in (
-                ("remote", remote),
-                ("org", org),
-                ("repo", repo),
-            ):
-                want_s = (want or "").strip()
-                if not want_s:
-                    continue
-                have = (str(merged.get(field) or "")).strip()
-                if not have:
-                    filtered.append(
-                        f"session state {field} missing on durable record "
-                        f"(expected={want_s!r}; fail closed, #709 F3)"
-                    )
-                elif have != want_s:
-                    # identity_match_reasons already adds mismatch; ensure kept
-                    pass
-        if filtered:
-            return None
-        return merged
-    reasons = identity_match_reasons(
-        merged,
-        remote=remote,
-        org=org,
-        repo=repo,
-        profile_identity=profile,
-    )
-    if reasons:
-        return None
-    return merged
-

post_merge_cleanup_proof.py

@@ -86,22 +86,6 @@ _WRONG_BRANCH_RE = re.compile(
     r"deleted branch (?:does not match|!=|differs from) (?:merged )?pr head",
     re.IGNORECASE,
 )
-# #698: canonical reviewer lease lifecycle operations are NOT post-merge
-# cleanup. Releasing a reviewer PR lease (or posting its terminal
-# phase=released marker) happens after every review — merged or not — and
-# must never trigger the post-merge branch/worktree cleanup checklist.
-_LEASE_LIFECYCLE_RE = re.compile(
-    r"(?:gitea_release_reviewer_pr_lease|gitea_abandon_workflow_lease|"
-    r"release[d]? (?:the )?(?:reviewer|workflow) (?:pr )?lease|"
-    r"reviewer (?:pr )?lease release[d]?|"
-    r"lease (?:marker|comment).{0,40}phase\s*[:=]\s*released|"
-    r"phase\s*[:=]\s*released)",
-    re.IGNORECASE,
-)
-_CLEANUP_MUTATIONS_VALUE_RE = re.compile(
-    r"cleanup mutations\s*:\s*([^\n]+)",
-    re.IGNORECASE,
-)
 
 
 def _claims_remote_delete(text: str) -> bool:
@@ -220,19 +204,16 @@ def assess_post_merge_cleanup_proof(
             for field in _worktree_cleanup_fields_present(text)
         )
 
+    if (remote_delete or worktree_remove) and not (remote_delete or worktree_remove):
+        pass
+
     if not remote_delete and not worktree_remove:
-        value_match = _CLEANUP_MUTATIONS_VALUE_RE.search(text)
+        cleanup_mutations = re.search(
-        value = (value_match.group(1).strip() if value_match else "")
+            r"cleanup mutations\s*:\s*(?!none\b)\S",
-        value_lower = value.lower()
+            text,
-        substantive = bool(value) and value_lower not in {
+            re.IGNORECASE,
-            "none", "n/a", "not applicable",
-        }
-        # #698: reviewer lease release / terminal lease markers are lease
-        # lifecycle, not post-merge cleanup — no checklist owed.
-        lease_lifecycle_only = substantive and bool(
-            _LEASE_LIFECYCLE_RE.search(value)
         )
-        if substantive and not lease_lifecycle_only:
+        if cleanup_mutations:
             reasons.append(
                 "cleanup mutations reported without post-merge cleanup proof checklist"
             )

pr_work_lease.py

@@ -403,37 +403,10 @@ _REVIEWER_ACTIVE_RE = re.compile(
     r"whether any reviewer was active\s*:\s*(yes|no|true|false)",
     re.IGNORECASE,
 )
-# #698 phase detection for phase-specific head proofs.
-_NO_REVIEWED_HEAD_RE = re.compile(
-    r"(?:reviewed head sha|candidate head sha)\s*:\s*none\b",
-    re.IGNORECASE,
-)
-_VERDICT_RECORDED_RE = re.compile(
-    r"review decision\s*:\s*(?:approve[d]?|request[_ ]changes)\b"
-    r"|review_status\s*:\s*(?:approved|request_changes)\b"
-    r"|terminal review mutation\s*:\s*(?!none\b)\S",
-    re.IGNORECASE,
-)
-_MERGE_ATTEMPTED_RE = re.compile(
-    r"merge result\s*:\s*(?:merged|success|performed|failed|attempted)\b"
-    r"|merge mutations\s*:\s*(?!none\b|not applicable\b)\S",
-    re.IGNORECASE,
-)
-_VALIDATION_STARTED_RE = re.compile(
-    r"validation\s*:\s*(?!none\b|not run\b|not applicable\b|not started\b)"
-    r"[^\n]*(?:pass|fail|ran|executed|\d+\s+passed)",
-    re.IGNORECASE,
-)
 
 
 def assess_reviewer_stale_head_final_report(report_text: str) -> dict[str, Any]:
-    """Final-report proof for reviewed vs live head SHAs (#399 AC 6).
+    """Final-report proof for reviewed vs live head SHAs (#399 AC 6)."""
-
-    #698: head proofs are phase-specific. A legitimately blocked run that
-    never began validation (no reviewed head, no formal verdict, no merge)
-    owes none of them; approval-time and merge-time live-head proofs are
-    owed only once the corresponding phase actually begins.
-    """
     text = report_text or ""
     reasons: list[str] = []
     reviewed = _normalize_sha(_REVIEWED_HEAD_RE.search(text).group(1) if _REVIEWED_HEAD_RE.search(text) else None)
@@ -449,49 +422,15 @@ def assess_reviewer_stale_head_final_report(report_text: str) -> dict[str, Any]:
     )
     push_during = _PUSH_DURING_VALIDATION_RE.search(text)
 
-    # Phase detection from the report's own claims.
+    if not reviewed:
-    no_head_stated = bool(_NO_REVIEWED_HEAD_RE.search(text))
-    verdict_recorded = bool(_VERDICT_RECORDED_RE.search(text))
-    merge_attempted = bool(_MERGE_ATTEMPTED_RE.search(text))
-    validation_started = bool(reviewed) or bool(_VALIDATION_STARTED_RE.search(text))
-    blocked_before_validation = (
-        no_head_stated
-        and not reviewed
-        and not verdict_recorded
-        and not merge_attempted
-        and not validation_started
-    )
-
-    if blocked_before_validation:
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": [],
-            "reviewed_head_sha": None,
-            "live_head_sha_before_approval": None,
-            "live_head_sha_before_merge": None,
-            "push_during_validation": (
-                push_during.group(1).lower() if push_during else None
-            ),
-            "phase": "blocked_before_validation",
-        }
-
-    if not reviewed and not no_head_stated:
-        # The head must always be STATED — either a SHA or an explicit
-        # 'none'. Silence is not a phase claim and fails closed.
-        reasons.append(
-            "reviewed head SHA not stated in final report "
-            "(state the SHA or an explicit 'none')"
-        )
-    elif not reviewed and (validation_started or verdict_recorded or merge_attempted):
         reasons.append("reviewed head SHA not stated in final report")
-    if verdict_recorded and not live_approval:
+    if not live_approval:
         reasons.append("final live head SHA before approval not stated")
-    if merge_attempted and not live_merge:
+    if not live_merge:
         reasons.append("final live head SHA before merge not stated")
-    if validation_started and not push_during:
+    if not push_during:
         reasons.append("whether push occurred during validation not stated")
-    if reviewed and live_approval and reviewed != live_approval:
+    elif reviewed and live_approval and reviewed != live_approval:
         reasons.append("live head before approval differs from reviewed head SHA")
     elif reviewed and live_merge and reviewed != live_merge:
         reasons.append("live head before merge differs from reviewed head SHA")

review_final_report_schema.py

@@ -25,13 +25,8 @@ _REVIEWED_HEAD_RE = re.compile(
     r"(?:pinned reviewed head|reviewed head sha)\s*:\s*([0-9a-f]{7,40})",
     re.IGNORECASE,
 )
-# #698: validation pass proof appears in several legitimate shapes —
-# "Validation: pass", "Validation: focused 50 passed; full 2665 passed",
-# or structured "validation_status: pass". Accept pass evidence anywhere in
-# the Validation field's value, not only as its first token.
 _VALIDATION_PASS_RE = re.compile(
-    r"validation(?:_status)?\s*:[^\n]{0,300}?"
+    r"validation\s*:\s*(?:pass|passed|strong|ok|green)",
-    r"(?:\bpass(?:ed)?\b|\bstrong\b|\bok\b|\bgreen\b|\d+\s+passed)",
     re.IGNORECASE,
 )
 _MERGED_CLAIM_RE = re.compile(

review_proofs.py

@@ -858,16 +858,9 @@ _WALKTHROUGH_ARTIFACT_RE = re.compile(r"walkthrough\.md", re.I)
 
 
 def _performed_file_mutations(action_log: list[dict] | None) -> list[dict]:
-    """Return performed local file mutations, excluding gated rejections.
+    """Return performed local file mutations, excluding gated rejections."""
-
-    Non-dict entries (malformed JSON, LLM mistakes) are ignored instead of
-    raising ``AttributeError`` (#698): a malformed ledger entry can never be
-    authoritative mutation evidence.
-    """
     performed: list[dict] = []
     for entry in action_log or []:
-        if not isinstance(entry, dict):
-            continue
         if entry.get("gated_rejected") or entry.get("performed") is False:
             continue
         action = (entry.get("action") or "").strip().lower()
@@ -2235,31 +2228,24 @@ HANDOFF_REVIEW_MUTATION_FIELDS = (
 )
 
 HANDOFF_ROLE_FIELDS = {
-    # #698: the review/merger required-field sets must stay aligned with the
-    # canonical schema (skills/llm-project-workflow/schemas/
-    # review-merge-final-report.md). The schema explicitly FORBIDS the legacy
-    # fields 'Pinned reviewed head', 'Scratch worktree used', and 'Workspace
-    # mutations' — a validator must never demand a field the schema bans.
     "review": (
         ("Selected PR", ("selected pr",)),
         ("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
-        ("Reviewed head SHA", ("reviewed head sha", "candidate head sha")),
+        ("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
-        ("Review worktree path", ("review worktree path", "worktree path",
+        ("Worktree path", ("worktree path", "starting worktree path")),
-                                  "starting worktree path")),
+        ("Worktree dirty", ("worktree dirty", "whether worktree was dirty")),
-        ("Review worktree dirty", ("review worktree dirty", "worktree dirty",
+        ("Scratch worktree used", ("scratch worktree used", "scratch clone used",
-                                   "whether worktree was dirty")),
+                                   "scratch worktree")),
         ("Unrelated local mutations", ("unrelated local mutations",
-                                       "unrelated files modified",
+                                       "unrelated files modified")),
-                                       "file edits by reviewer")),
         ("Review decision", ("review decision", "decision")),
         ("Merge result", ("merge result",)),
         ("Linked issue status", ("linked issue status", "linked issue")),
         ("Cleanup status", ("cleanup status", "cleanup")),
-        ("Safe next action", ("safe next action", "next")),
     ) + HANDOFF_REVIEW_MUTATION_FIELDS,
     "merger": (
         ("Selected PR", ("selected pr",)),
-        ("Reviewed head SHA", ("reviewed head sha", "candidate head sha")),
+        ("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
         ("Active profile", ("active profile",)),
         ("Role kind", ("role kind",)),
         ("Merge capability source", ("merge capability source",)),
@@ -2447,22 +2433,9 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
         # Issue #320: reviewer and merger handoffs use the precise mutation categories
         # in HANDOFF_REVIEW_MUTATION_FIELDS instead of the legacy ambiguous
         # "Workspace mutations" field, which is rejected below.
-        # Issue #698: the canonical review-merge schema has no 'Mutations',
-        # 'Next', 'Issue/PR', 'Branch/SHA', or 'Files changed' fields — their
-        # content lives in the precise mutation categories, 'Safe next
-        # action', 'Selected PR'/'Linked issue', head-SHA fields, and 'Files
-        # reviewed'. Requiring the legacy names rejects canonical reports.
-        _non_canonical_for_review = {
-            "Workspace mutations",
-            "Mutations",
-            "Next",
-            "Issue/PR",
-            "Branch/SHA",
-            "Files changed",
-        }
         required = [
             field for field in required
-            if field[0] not in _non_canonical_for_review
+            if field[0] != "Workspace mutations"
         ]
         if any(label.startswith("workspace mutations") for label in labels):
             return {

stale_review_decision_lock.py

@@ -438,252 +438,3 @@ def format_cleanup_audit_comment(audit: dict[str, Any]) -> str:
         "This path only clears a lock when the referenced PR is merged/closed.",
     ]
     return "\n".join(lines)
-
-
-# ---------------------------------------------------------------------------
-# #709 cross-profile cleanup, overwrite protection, recovery provenance
-# ---------------------------------------------------------------------------
-
-
-def has_unresolved_terminal_evidence(lock: dict | None) -> bool:
-    """True when *lock* still carries a terminal live mutation ledger."""
-    return last_terminal_mutation(lock) is not None
-
-
-def assess_init_overwrite(
-    existing_lock: dict | None,
-    *,
-    force: bool = False,
-) -> dict[str, Any]:
-    """Whether init_review_decision_lock may replace an existing durable lock (#709 AC2).
-
-    Unresolved terminal evidence must never be silently replaced by an empty
-    initialized lock. *force* does not authorize destruction of terminal
-    ledgers — only explicit cleanup / archive transitions may remove them.
-    """
-    result: dict[str, Any] = {
-        "overwrite_allowed": True,
-        "has_existing": existing_lock is not None,
-        "has_unresolved_terminal": False,
-        "reasons": [],
-        "last_terminal_pr": None,
-        "last_terminal_action": None,
-        "existing_profile_identity": None,
-    }
-    if existing_lock is None:
-        result["reasons"].append("no existing lock; empty init allowed")
-        return result
-    result["existing_profile_identity"] = (
-        existing_lock.get("profile_identity")
-        or existing_lock.get("session_profile_lock")
-        or existing_lock.get("session_profile")
-    )
-    last = last_terminal_mutation(existing_lock)
-    if last is None:
-        result["reasons"].append(
-            "existing lock has no terminal mutations; re-init allowed"
-        )
-        return result
-    result["has_unresolved_terminal"] = True
-    result["overwrite_allowed"] = False
-    result["last_terminal_pr"] = last.get("pr_number")
-    result["last_terminal_action"] = last.get("action")
-    result["reasons"].append(
-        "refuse empty re-init: unresolved terminal decision-lock evidence "
-        f"present ({last.get('action')} on PR #{last.get('pr_number')}); "
-        "use gitea_cleanup_stale_review_decision_lock when moot, or archive "
-        f"via sanctioned recovery — force={force!r} does not authorize overwrite "
-        "(#709 AC2)"
-    )
-    return result
-
-
-def lock_targets_merged_pr_approval(
-    lock: dict | None,
-    *,
-    pr_number: int,
-    expected_head_sha: str | None = None,
-) -> bool:
-    """True when *lock*'s last terminal mutation is approve of *pr_number*.
-
-    When *expected_head_sha* is provided, a **recorded** terminal head must
-    match. Legacy same-repo approve locks with no recorded head do **not**
-    match (fail closed, #709 F3 residual / review 435) — callers must use the
-    strict secondary path that refuses no-head clears.
-    """
-    last = last_terminal_mutation(lock)
-    if last is None:
-        return False
-    if last.get("action") != "approve":
-        return False
-    if last.get("pr_number") != pr_number:
-        return False
-    if expected_head_sha:
-        want = normalize_head_sha(expected_head_sha)
-        if not want:
-            return False
-        locked = mutation_head_sha(last, lock)
-        # Require recorded-head match; unrecorded head is not a match (#709 F3).
-        if not locked or not heads_equal(locked, want):
-            return False
-    return True
-
-
-def build_post_merge_recovery_record(
-    *,
-    pr_number: int,
-    head_sha: str | None,
-    merge_commit_sha: str | None,
-    target_profile_identity: str | None,
-    failed_step: str,
-    error: str | None,
-    remote: str | None,
-    org: str | None,
-    repo: str | None,
-    actor_username: str | None,
-    profile_name: str | None,
-) -> dict[str, Any]:
-    """Durable recovery-required payload after irreversible merge (#709 AC3)."""
-    return {
-        "event": "post_merge_decision_lock_recovery_required",
-        "status": "recovery_required",
-        "issue_ref": "#709",
-        "recovery_critical": True,
-        "applied": False,  # never claim historical cleanup succeeded
-        "timestamp": datetime.now(timezone.utc).isoformat(),
-        "pr_number": pr_number,
-        "head_sha": normalize_head_sha(head_sha),
-        "merge_commit_sha": merge_commit_sha,
-        "target_profile_identity": target_profile_identity,
-        "failed_step": failed_step,
-        "error": error,
-        "remote": remote,
-        "org": org,
-        "repo": repo,
-        "actor_username": actor_username,
-        "profile_name": profile_name,
-        "required_recovery_action": (
-            "retry cross-profile decision-lock cleanup and audit publication "
-            "for the merged PR; if terminal evidence is gone, use "
-            "gitea_record_irrecoverable_decision_lock_provenance"
-        ),
-    }
-
-
-def build_irrecoverable_provenance_record(
-    *,
-    pr_number: int,
-    head_sha: str | None,
-    remote: str | None,
-    org: str | None,
-    repo: str | None,
-    actor_username: str | None,
-    profile_name: str | None,
-    reason: str,
-    incident_ref: str | None = None,
-    # Deprecated kwargs retained only so stale call sites fail closed:
-    operator_authorized: bool | None = None,
-    # Required for merger-acceptable records (#709 F1):
-    authorization: dict[str, Any] | None = None,
-    incident_issue: int | None = None,
-    incident_comment_id: int | None = None,
-    destroyed_subject: str | None = None,
-    historical_provenance_subject: str | None = None,
-) -> dict[str, Any]:
-    """Truthful absence-of-proof record (#709 AC5). Never sets applied=True.
-
-    Caller-supplied ``operator_authorized`` is **ignored** as authorization
-    evidence (review 434 F1). Prefer
-    :func:`irrecoverable_provenance.build_irrecoverable_provenance_record`
-    with a verified server-side authorization artifact.
-    """
-    # Explicitly ignore deprecated self-assertable Boolean.
-    _ = operator_authorized
-    if authorization is not None and incident_issue is not None and incident_comment_id is not None:
-        from irrecoverable_provenance import (
-            build_irrecoverable_provenance_record as _build,
-        )
-
-        return _build(
-            pr_number=int(pr_number),
-            head_sha=str(head_sha or ""),
-            remote=str(remote or ""),
-            org=str(org or ""),
-            repo=str(repo or ""),
-            actor_username=actor_username,
-            profile_name=profile_name,
-            reason=reason,
-            incident_issue=int(incident_issue),
-            incident_comment_id=int(incident_comment_id),
-            authorization=authorization,
-            destroyed_subject=destroyed_subject,
-            historical_provenance_subject=historical_provenance_subject
-            or destroyed_subject,
-        )
-    # Fail-closed skeleton when no server authorization is supplied: never
-    # sets merger_may_accept True (even if operator_authorized was True).
-    return {
-        "event": "irrecoverable_decision_lock_provenance",
-        "status": "provenance_irrecoverable",
-        "record_type": "irrecoverable_decision_provenance",
-        "operator_recovery_required": True,
-        "issue_ref": "#709",
-        "recovery_critical": True,
-        "applied": False,
-        "historical_cleanup_proven": False,
-        "timestamp": datetime.now(timezone.utc).isoformat(),
-        "pr_number": pr_number,
-        "head_sha": normalize_head_sha(head_sha),
-        "remote": remote,
-        "org": org,
-        "repo": repo,
-        "actor_username": actor_username,
-        "profile_name": profile_name,
-        "reason": reason,
-        "incident_ref": incident_ref,
-        "incident_issue": incident_issue,
-        "incident_comment_id": incident_comment_id,
-        "authorization_verified": False,
-        "merger_may_accept": False,
-        "acceptance_rule": (
-            "Merger may accept this record only when a server-side "
-            "authorization artifact verifies for remote/org/repo/PR/exact "
-            "head/incident, the record is durable and read back, and normal "
-            "merge gates still pass. Caller Booleans never authorize. This "
-            "does not prove historical cleanup."
-        ),
-    }
-
-
-def format_irrecoverable_audit_comment(record: dict[str, Any]) -> str:
-    """Markdown body for irrecoverable provenance audit (no applied=true claim)."""
-    from irrecoverable_provenance import (
-        format_irrecoverable_audit_comment as _fmt,
-    )
-
-    return _fmt(record)
-
-
-def format_post_merge_recovery_comment(record: dict[str, Any]) -> str:
-    """Markdown body for post-merge recovery-required audit."""
-    lines = [
-        "## Post-merge decision-lock recovery required (#709)",
-        "",
-        "Status: **RECOVERY_REQUIRED** (merge is irreversible; cleanup/audit incomplete)",
-        "",
-        f"- actor: `{record.get('actor_username')}`",
-        f"- profile: `{record.get('profile_name')}`",
-        f"- timestamp: `{record.get('timestamp')}`",
-        f"- PR: `#{record.get('pr_number')}`",
-        f"- head_sha: `{record.get('head_sha')}`",
-        f"- merge_commit_sha: `{record.get('merge_commit_sha')}`",
-        f"- target_profile_identity: `{record.get('target_profile_identity')}`",
-        f"- failed_step: `{record.get('failed_step')}`",
-        f"- error: `{record.get('error')}`",
-        "",
-        f"Required action: {record.get('required_recovery_action')}",
-        "",
-        "applied=false — do not treat this as successful cleanup evidence.",
-    ]
-    return "\n".join(lines)
-

task_capability_map.py

@@ -127,32 +127,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.review",
         "role": "reviewer",
     },
-    # #709: truthful absence-of-proof recovery (server-side auth + record + consume).
-    # Dedicated mutation capability — gitea.read is insufficient (review 434 F1).
-    "issue_irrecoverable_provenance_authorization": {
-        "permission": "gitea.decision_lock.irrecoverable_recovery",
-        "role": "reconciler",
-    },
-    "gitea_issue_irrecoverable_provenance_authorization": {
-        "permission": "gitea.decision_lock.irrecoverable_recovery",
-        "role": "reconciler",
-    },
-    "record_irrecoverable_decision_lock_provenance": {
-        "permission": "gitea.decision_lock.irrecoverable_recovery",
-        "role": "reconciler",
-    },
-    "gitea_record_irrecoverable_decision_lock_provenance": {
-        "permission": "gitea.decision_lock.irrecoverable_recovery",
-        "role": "reconciler",
-    },
-    "consume_irrecoverable_decision_lock_provenance": {
-        "permission": "gitea.pr.merge",
-        "role": "merger",
-    },
-    "gitea_consume_irrecoverable_decision_lock_provenance": {
-        "permission": "gitea.pr.merge",
-        "role": "merger",
-    },
     "delete_branch": {
         "permission": "gitea.branch.delete",
         "role": "author",

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_issue_698_report_validator_schema_alignment.py

@@ -1,461 +0,0 @@
-"""Regression tests for #698: final-report validator vs canonical schema.
-
-Covers the original #698 lead plus the independent reproduction recorded
-during the PR #703 formal review (issue #698 comment 11246):
-
-1. non-dict ``action_log`` entries must fail structured, never crash;
-2. the validator must not demand legacy fields the canonical schema forbids
-   (``Pinned reviewed head``, ``Scratch worktree used``, ``Worktree path``,
-   ``Worktree dirty``, ``Mutations``, ``Next``);
-3. a legitimately blocked report (``Candidate head SHA: none``, no formal
-   verdict) must not owe approval/merge live-head proofs;
-4. canonical reviewer lease release must not be misclassified as post-merge
-   cleanup;
-5. structured workflow-load and validation proof must be recognized;
-6. review mutations are inferred only from authoritative evidence.
-"""
-
-from __future__ import annotations
-
-import sys
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-import final_report_validator as frv  # noqa: E402
-import post_merge_cleanup_proof as pmcp  # noqa: E402
-import pr_work_lease as pwl  # noqa: E402
-import review_proofs as rp  # noqa: E402
-from review_final_report_schema import (  # noqa: E402
-    assess_review_final_report_schema,
-)
-
-REVIEWED_HEAD = "a" * 40
-LIVE_HEAD = "a" * 40
-
-
-def _pr703_style_report(
-    *,
-    decision: str = "request_changes",
-    cleanup_mutations: str = (
-        "released reviewer PR lease via gitea_release_reviewer_pr_lease "
-        "(terminal lease marker phase=released posted)"
-    ),
-) -> str:
-    """Canonical-schema report modeled on the PR #703 formal review handoff."""
-    return f"""
-Formal review completed with a REQUEST_CHANGES verdict submitted and read
-back via the native review API.
-
-## Controller Handoff
-
-- Task: review-merge-pr
-- Repo: Scaled-Tech-Consulting/Gitea-Tools
-- Role: reviewer
-- Identity: sysadmin / prgs-reviewer
-- Active profile: prgs-reviewer
-- Runtime context: neutral workspace binding
-- Selected PR: 703
-- Linked issue: #702 open
-- Eligibility class: reviewable
-- Queue ordering policy: oldest eligible first
-- Inventory pagination proof: has_more=false, total_count=8
-- Earlier PRs skipped: none
-- Candidate head SHA: {REVIEWED_HEAD}
-- Reviewed head SHA: {REVIEWED_HEAD}
-- Target branch: master
-- Target branch SHA: {"2" * 40}
-- Already-landed gate: not landed
-- Author-safety result: pass (author differs from reviewer)
-- Prior request-changes state: none
-- Review worktree used: true
-- Review worktree path: branches/review-pr-703-independent
-- Review worktree inside branches: true
-- Review worktree HEAD state: detached at pinned head
-- Review worktree dirty before validation: clean
-- Review worktree dirty after validation: clean
-- Baseline worktree used: false
-- Baseline worktree path: none
-- Files reviewed: 4
-- Validation: focused 50 passed; related 94 passed; full 2665 passed, 6 skipped
-- Official validation integrity status: intact
-- Terminal review mutation: one REQUEST_CHANGES review submitted and read back
-- Review decision: {decision}
-- Merge preflight: not run
-- Merge result: none
-- Linked issue status: open (live fetch proof: gitea_view_issue)
-- Main checkout branch: master
-- Main checkout dirty state: clean
-- Main checkout updated: false
-- File edits by reviewer: none
-- Worktree/index mutations: none
-- Git ref mutations: git fetch prgs (recorded)
-- MCP/Gitea mutations: review submission and lease comments only
-- Review mutations: one formal REQUEST_CHANGES verdict
-- Merge mutations: none
-- Cleanup mutations: {cleanup_mutations}
-- External-state mutations: none
-- Read-only diagnostics: gitea_view_pr, gitea_get_pr_review_feedback
-- Blockers: findings F1-F6 recorded on the PR thread
-- Current status: review complete; author remediation required
-- Safe next action: author addresses findings and pushes a new head
-- Safety statement: no merge attempted; no self-review; no root-checkout edits
-- Workflow-load helper result: workflow_hash=da045d1e1f1f boundary_status=clean
-- Live head SHA before approval: {LIVE_HEAD}
-- Pushes occurred during validation: no
-"""
-
-
-def _blocked_preflight_report() -> str:
-    """Blocked-run report modeled on the #702 comment 11164 reproduction."""
-    return """
-Fresh review preflight stopped before any worktree or validation work.
-
-## Controller Handoff
-
-- Task: review-merge-pr
-- Repo: Scaled-Tech-Consulting/Gitea-Tools
-- Role: reviewer
-- Identity: sysadmin / prgs-reviewer
-- Active profile: prgs-reviewer
-- Runtime context: stale workspace binding detected
-- Selected PR: 701
-- Linked issue: #699 open
-- Eligibility class: blocked-before-validation
-- Queue ordering policy: oldest eligible first
-- Inventory pagination proof: has_more=false, total_count=8
-- Earlier PRs skipped: none
-- Candidate head SHA: none
-- Reviewed head SHA: none
-- Target branch: master
-- Target branch SHA: none
-- Already-landed gate: not run
-- Author-safety result: not run
-- Prior request-changes state: none
-- Review worktree used: false
-- Review worktree path: none
-- Review worktree inside branches: not applicable
-- Review worktree HEAD state: not applicable
-- Review worktree dirty before validation: not applicable
-- Review worktree dirty after validation: not applicable
-- Baseline worktree used: false
-- Baseline worktree path: none
-- Files reviewed: 0
-- Validation: not run
-- Official validation integrity status: not applicable
-- Terminal review mutation: none
-- Review decision: none
-- Merge preflight: not run
-- Merge result: none
-- Linked issue status: open (live fetch proof: gitea_view_issue)
-- Main checkout branch: master
-- Main checkout dirty state: clean
-- Main checkout updated: false
-- File edits by reviewer: none
-- Worktree/index mutations: none
-- Git ref mutations: none
-- MCP/Gitea mutations: none
-- Review mutations: none
-- Merge mutations: none
-- Cleanup mutations: none
-- External-state mutations: none
-- Read-only diagnostics: gitea_view_pr, gitea_get_runtime_context
-- Blockers: runtime bound to a foreign task worktree; mutation prohibited
-- Current status: stopped before validation began
-- Next actor: operator
-- Next action: repair the runtime workspace binding, then rerun the full
-  review workflow in a fresh reviewer session
-- Next prompt: Act as REVIEWER for PR 701 after the operator repairs the
-  runtime binding; acquire the lease before any validation.
-- Safe next action: operator repairs runtime binding, then a fresh reviewer
-  reruns the full workflow
-- Safety statement: no lease acquired; no verdict recorded; no source edits
-- Workflow-load helper result: workflow_hash=da045d1e1f1f boundary_status=clean
-"""
-
-
-class TestActionLogRobustness(unittest.TestCase):
-    """#698 original lead: non-dict action_log must not crash validation."""
-
-    MALFORMED = [
-        "git fetch prgs",
-        42,
-        None,
-        {"action": "edit", "path": "x.py", "performed": True, "tracked": True},
-    ]
-
-    def test_assess_final_report_validator_survives_malformed_entries(self):
-        result = frv.assess_final_report_validator(
-            _pr703_style_report(),
-            "review_pr",
-            action_log=self.MALFORMED,
-        )
-        self.assertIsInstance(result, dict)
-        rule_ids = {f["rule_id"] for f in result["findings"]}
-        self.assertIn("shared.action_log_malformed", rule_ids)
-
-    def test_malformed_entry_errors_are_sanitized(self):
-        _entries, findings = frv.sanitize_action_log(["secret-token-abc123"])
-        self.assertEqual(len(findings), 1)
-        reason = findings[0]["reason"]
-        self.assertNotIn("secret-token-abc123", reason)
-        self.assertIn("str", reason)
-        self.assertIn("entry 0", reason)
-
-    def test_non_list_action_log_is_reported_not_raised(self):
-        entries, findings = frv.sanitize_action_log("not-a-list")
-        self.assertEqual(entries, [])
-        self.assertEqual(len(findings), 1)
-        self.assertIn("not a list", findings[0]["reason"])
-
-    def test_performed_file_mutations_skips_non_dict_entries(self):
-        performed = rp._performed_file_mutations(
-            ["oops", {"action": "edited", "path": "a.py"}]
-        )
-        self.assertEqual(len(performed), 1)
-        self.assertEqual(performed[0]["path"], "a.py")
-
-    def test_schema_entrypoint_survives_string_only_log(self):
-        result = assess_review_final_report_schema(
-            _pr703_style_report(),
-            action_log=["just a string", "another string"],
-        )
-        self.assertIsInstance(result, dict)
-
-
-class TestLegacyFieldRequirementsRemoved(unittest.TestCase):
-    """#698: prohibited legacy fields must not be REQUIRED of reports."""
-
-    PROHIBITED = (
-        "Pinned reviewed head",
-        "Scratch worktree used",
-        "Worktree path",
-        "Worktree dirty",
-        "Workspace mutations",
-        "Mutations",
-        "Next",
-        "Issue/PR",
-        "Branch/SHA",
-        "Files changed",
-    )
-
-    def test_review_role_field_table_has_no_prohibited_requirements(self):
-        names = [name for name, _ in rp.HANDOFF_ROLE_FIELDS["review"]]
-        for prohibited in ("Pinned reviewed head", "Scratch worktree used",
-                           "Worktree path", "Worktree dirty"):
-            self.assertNotIn(prohibited, names)
-
-    def test_merger_role_field_table_has_no_pinned_reviewed_head(self):
-        names = [name for name, _ in rp.HANDOFF_ROLE_FIELDS["merger"]]
-        self.assertNotIn("Pinned reviewed head", names)
-
-    def test_canonical_report_missing_fields_never_include_prohibited(self):
-        result = rp.assess_controller_handoff(
-            _pr703_style_report(), role="review"
-        )
-        for prohibited in self.PROHIBITED:
-            self.assertNotIn(prohibited, result.get("missing_fields") or [])
-
-    def test_canonical_pr703_report_satisfies_required_fields(self):
-        result = rp.assess_controller_handoff(
-            _pr703_style_report(), role="review"
-        )
-        self.assertEqual(result.get("missing_fields") or [], [])
-        self.assertEqual(result.get("verdict"), "complete")
-
-
-class TestBlockedReportAccepted(unittest.TestCase):
-    """#698: blocked run with no reviewed head / verdict is legitimate."""
-
-    def test_stale_head_proof_waived_before_validation(self):
-        result = pwl.assess_reviewer_stale_head_final_report(
-            _blocked_preflight_report()
-        )
-        self.assertTrue(result["proven"])
-        self.assertEqual(result.get("phase"), "blocked_before_validation")
-
-    def test_blocked_report_passes_schema_validation(self):
-        result = assess_review_final_report_schema(_blocked_preflight_report())
-        blocking = [
-            f for f in result["findings"] if f["severity"] == "block"
-        ]
-        self.assertEqual(blocking, [], blocking)
-
-    def test_verdict_phase_still_demands_approval_head_proof(self):
-        report = _blocked_preflight_report().replace(
-            "- Review decision: none",
-            "- Review decision: approve",
-        ).replace(
-            "- Candidate head SHA: none",
-            f"- Candidate head SHA: {REVIEWED_HEAD}",
-        )
-        result = pwl.assess_reviewer_stale_head_final_report(report)
-        self.assertFalse(result["proven"])
-        joined = " ".join(result["reasons"])
-        self.assertIn("before approval", joined)
-
-    def test_merge_phase_still_demands_merge_head_proof(self):
-        report = _pr703_style_report().replace(
-            "- Merge result: none",
-            "- Merge result: merged",
-        )
-        result = pwl.assess_reviewer_stale_head_final_report(report)
-        self.assertFalse(result["proven"])
-        self.assertIn(
-            "final live head SHA before merge not stated",
-            result["reasons"],
-        )
-
-    def test_validation_phase_demands_push_disclosure(self):
-        report = _pr703_style_report().replace(
-            "- Pushes occurred during validation: no\n", ""
-        )
-        result = pwl.assess_reviewer_stale_head_final_report(report)
-        self.assertFalse(result["proven"])
-        self.assertIn(
-            "whether push occurred during validation not stated",
-            result["reasons"],
-        )
-
-
-class TestLeaseReleaseVsPostMergeCleanup(unittest.TestCase):
-    """#698 (PR #703 review reproduction): lease release is not cleanup."""
-
-    def test_lease_release_cleanup_mutations_do_not_demand_checklist(self):
-        result = pmcp.assess_post_merge_cleanup_proof(_pr703_style_report())
-        self.assertFalse(result["block"], result["reasons"])
-
-    def test_release_tool_name_alone_is_recognized(self):
-        report = _pr703_style_report(
-            cleanup_mutations="gitea_release_reviewer_pr_lease comment 11244"
-        )
-        result = pmcp.assess_post_merge_cleanup_proof(report)
-        self.assertFalse(result["block"], result["reasons"])
-
-    def test_substantive_non_lease_cleanup_still_demands_checklist(self):
-        report = _pr703_style_report(
-            cleanup_mutations="deleted stale scratch directory manually"
-        )
-        result = pmcp.assess_post_merge_cleanup_proof(report)
-        self.assertTrue(result["block"])
-
-    def test_remote_branch_delete_claims_still_demand_full_proof(self):
-        report = _pr703_style_report(
-            cleanup_mutations="gitea_delete_branch removed the remote branch"
-        )
-        result = pmcp.assess_post_merge_cleanup_proof(report)
-        self.assertTrue(result["block"])
-        self.assertTrue(
-            any("remote branch deletion missing" in r for r in result["reasons"])
-        )
-
-    def test_full_schema_run_accepts_lease_release_report(self):
-        result = assess_review_final_report_schema(_pr703_style_report())
-        lease_cleanup_blocks = [
-            f for f in result["findings"]
-            if f["rule_id"] == "reviewer.post_merge_cleanup_proof"
-        ]
-        self.assertEqual(lease_cleanup_blocks, [], lease_cleanup_blocks)
-
-
-class TestStructuredProofRecognition(unittest.TestCase):
-    """#698: structured workflow-load and validation proof must be accepted."""
-
-    def test_key_value_workflow_proof_recognized(self):
-        findings = frv._rule_reviewer_workflow_load_boundary(
-            _pr703_style_report()
-        )
-        self.assertEqual(findings, [], findings)
-
-    def test_colon_form_workflow_proof_still_recognized(self):
-        report = _pr703_style_report().replace(
-            "- Workflow-load helper result: workflow_hash=da045d1e1f1f "
-            "boundary_status=clean",
-            "- Workflow-load helper result: workflow_hash: da045d1e1f1f, "
-            "boundary_status: clean",
-        )
-        findings = frv._rule_reviewer_workflow_load_boundary(report)
-        self.assertEqual(findings, [], findings)
-
-    def test_incomplete_structured_proof_still_blocks(self):
-        report = _pr703_style_report().replace(
-            "workflow_hash=da045d1e1f1f boundary_status=clean",
-            "workflow_hash=da045d1e1f1f",
-        )
-        findings = frv._rule_reviewer_workflow_load_boundary(report)
-        self.assertTrue(findings)
-        self.assertIn("boundary_status", findings[0]["reason"])
-
-    def test_validation_counts_accepted_as_pass_proof(self):
-        # "Validation: focused 50 passed; ..." must satisfy the reviewed-head
-        # validation-proof rule (PR #703 reproduction).
-        result = assess_review_final_report_schema(_pr703_style_report())
-        head_blocks = [
-            f for f in result["findings"]
-            if f["rule_id"] == "reviewer.reviewed_head_without_validation"
-        ]
-        self.assertEqual(head_blocks, [], head_blocks)
-
-
-class TestAuthoritativeMutationInference(unittest.TestCase):
-    """#698: review mutations inferred only from authoritative evidence."""
-
-    def test_read_only_entries_do_not_imply_mutations(self):
-        report = "## Controller Handoff\n- Mutations: none\n"
-        findings = frv._rule_reviewer_vague_mutations_none(
-            report,
-            action_log=[
-                {"action": "gitea_view_pr"},
-                {"action": "gitea_get_pr_review_feedback", "performed": False},
-            ],
-        )
-        self.assertEqual(findings, [], findings)
-
-    def test_performed_mutation_still_blocks_vague_none(self):
-        report = "## Controller Handoff\n- Mutations: none\n"
-        findings = frv._rule_reviewer_vague_mutations_none(
-            report,
-            action_log=[{"action": "edit", "path": "a.py", "performed": True}],
-        )
-        self.assertTrue(findings)
-
-    def test_gated_rejection_is_not_a_mutation(self):
-        report = "## Controller Handoff\n- Mutations: none\n"
-        findings = frv._rule_reviewer_vague_mutations_none(
-            report,
-            action_log=[
-                {"action": "edit", "path": "a.py", "performed": True,
-                 "gated_rejected": True},
-            ],
-        )
-        self.assertEqual(findings, [], findings)
-
-
-class TestValidatorRuleErrorContainment(unittest.TestCase):
-    """#698: a defective rule fails closed with a sanitized error."""
-
-    def test_rule_exception_becomes_sanitized_block_finding(self):
-        def _boom(report_text):
-            raise ValueError("raw secret detail that must not leak")
-
-        original = frv._RULES_BY_TASK["review_pr"]
-        frv._RULES_BY_TASK["review_pr"] = [_boom]
-        try:
-            result = frv.assess_final_report_validator(
-                "report body", "review_pr"
-            )
-        finally:
-            frv._RULES_BY_TASK["review_pr"] = original
-        self.assertTrue(result["blocked"])
-        finding = next(
-            f for f in result["findings"]
-            if f["rule_id"] == "shared.validator_rule_error"
-        )
-        self.assertNotIn("raw secret detail", finding["reason"])
-        self.assertIn("ValueError", finding["reason"])
-        self.assertIn("_boom", finding["reason"])
-
-
-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()

tests/test_review_proofs.py

@@ -957,20 +957,17 @@ class TestControllerHandoff(unittest.TestCase):
             if not line.startswith("- Workspace mutations:"))
         result = assess_controller_handoff(review_base, role="review")
         self.assertEqual(result["verdict"], "incomplete")
-        # #698: the canonical schema forbids the legacy fields, so the
+        self.assertIn("Pinned reviewed head", result["missing_fields"])
-        # validator must demand the canonical names instead.
+        self.assertIn("Worktree path", result["missing_fields"])
-        self.assertIn("Reviewed head SHA", result["missing_fields"])
-        self.assertIn("Review worktree path", result["missing_fields"])
         self.assertIn("Merge result", result["missing_fields"])
-        for legacy in ("Pinned reviewed head", "Scratch worktree used"):
-            self.assertNotIn(legacy, result["missing_fields"])
 
         complete = review_base + "\n" + "\n".join([
             "- Selected PR: #999",
             "- Reviewer eligibility: passed",
-            "- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+            "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
-            "- Review worktree path: /repo/branches/review-pr-999",
+            "- Worktree path: /repo/branches/review-pr-999",
-            "- Review worktree dirty before validation: no",
+            "- Worktree dirty: no",
+            "- Scratch worktree used: yes (/repo/branches/review-pr-999)",
             "- Unrelated local mutations: none",
             "- Review decision: approve",
             "- Merge result: merged",
@@ -1128,9 +1125,10 @@ class TestReviewHandoffPreciseMutationCategories(unittest.TestCase):
         "- Safety: no self-review; no self-merge; no secrets",
         "- Selected PR: #999",
         "- Reviewer eligibility: passed",
-        "- Reviewed head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
         "- Worktree path: /repo/branches/review-pr-999",
         "- Worktree dirty: no",
+        "- Scratch worktree used: yes (/repo/branches/review-pr-999)",
         "- Unrelated local mutations: none",
         "- Review decision: approve",
         "- Merge result: none",