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

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

docs/mcp-daemon-import-guard.md

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

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]

gitea_mcp_server.py

@@ -1118,32 +1118,6 @@ import workflow_scope_guard  # noqa: E402  # #683 production scope / force-on gu
 import stable_branch_push_guard  # noqa: E402
 import remote_repo_guard  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
-import session_context_binding as session_ctx  # noqa: E402  # #714 immutable session context
-
-
-def _seed_session_context(
-    *,
-    profile: dict,
-    remote: str | None,
-    host: str | None,
-    identity: str | None,
-    repository: str | None = None,
-    org: str | None = None,
-    source: str = "seed",
-) -> dict:
-    """Seed immutable session context once for the current process (#714)."""
-    expected = (profile.get("username") or "").strip() or None
-    return session_ctx.seed_session_context_if_unbound(
-        profile_name=profile.get("profile_name") or "",
-        remote=remote,
-        host=host,
-        identity=identity,
-        repository=repository,
-        org=org,
-        role_kind=_profile_role_kind(profile),
-        expected_username=expected,
-        source=source,
-    )
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
@@ -1872,44 +1846,61 @@ def _authenticated_username(host: str):
     return user
 
 
-def _ensure_matching_profile(
+def _ensure_matching_profile(required_permission: str, required_role: str, remote: str | None, host: str | None = None) -> str | None:
-    required_permission: str,
+    """Check if the active profile is allowed to perform *required_permission*.
-    required_role: str,
+    If not, automatically switch to the first matching usable configured profile.
-    remote: str | None,
-    host: str | None = None,
-) -> str | None:
-    """Return the active profile name only when it is allowed for *permission*.
-
-    #714: never silently switch profiles (including cross-host substitution).
-    Capability resolution and mutation gates evaluate only the active profile
-    for the requested remote. Explicit ``gitea_activate_profile`` is the sole
-    in-process switch path.
     """
-    del required_role, host  # kept for call-site compatibility
     try:
         profile = get_profile()
     except Exception:
         return None
     active_profile = profile.get("profile_name")
-    # Cross-host denial: mdcps profile cannot serve a prgs remote (and vice versa).
-    config = None
-    try:
-        config = gitea_config.load_config()
-    except Exception:
-        config = None
-    contexts = (config or {}).get("contexts") if config else None
-    remote_ok = session_ctx.profile_allowed_for_remote(
-        profile, remote, REMOTES, contexts=contexts
-    )
-    if remote_ok.get("block"):
-        return None
     active_allowed = profile.get("allowed_operations") or []
     active_forbidden = profile.get("forbidden_operations") or []
-    allowed, _ = gitea_config.check_operation(
+    allowed, _ = gitea_config.check_operation(required_permission, active_allowed, active_forbidden)
-        required_permission, active_allowed, active_forbidden
-    )
     if allowed:
         return active_profile
+
+    # Try to find a matching usable profile in config
+    if gitea_config.is_runtime_switching_enabled():
+        config = gitea_config.load_config()
+        if config and "profiles" in config:
+            for p_name, p_data in config["profiles"].items():
+                p_allowed = p_data.get("allowed_operations") or []
+                p_forbidden = p_data.get("forbidden_operations") or []
+                p_allowed_n = []
+                for op in p_allowed:
+                    try:
+                        p_allowed_n.append(gitea_config.normalize_operation(op))
+                    except Exception:
+                        pass
+                p_forbidden_n = []
+                for op in p_forbidden:
+                    try:
+                        p_forbidden_n.append(gitea_config.normalize_operation(op))
+                    except Exception:
+                        pass
+                ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
+                if ok:
+                    # Verify credentials/token are available
+                    try:
+                        tok = gitea_config.resolve_token(p_data)
+                        if tok:
+                            # Perform automatic switch
+                            gitea_config._active_profile_override = p_name
+                            h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
+                            if h:
+                                _IDENTITY_CACHE.pop(h, None)
+                            username = _authenticated_username(h) if h else None
+                            # Update mutation authority
+                            global _MUTATION_AUTHORITY
+                            if _MUTATION_AUTHORITY is not None:
+                                _MUTATION_AUTHORITY["current_profile"] = p_name
+                                _MUTATION_AUTHORITY["current_identity"] = username
+                                _MUTATION_AUTHORITY["role_pivot_authorized"] = True
+                            return p_name
+                    except Exception:
+                        pass
     return None
 
 
@@ -1931,12 +1922,6 @@ def _audit(action: str, *, host, remote, result, org=None, repo=None,
         if mutation_task:
             ns_ctx = role_namespace_gate.mutation_audit_context(
                 mutation_task, profile, remote=remote, repository=repo)
-        # #714: always record the bound session context at mutation time.
-        session_audit = session_ctx.mutation_context_audit_fields()
-        if isinstance(request_metadata, dict):
-            request_metadata = {**request_metadata, **session_audit}
-        elif request_metadata is None:
-            request_metadata = dict(session_audit)
         event = gitea_audit.build_event(
             action=action,
             result=result,
@@ -2095,10 +2080,6 @@ def gitea_create_issue(
     blocked = _profile_permission_block(
         task_capability_map.required_permission("create_issue"),
         number=None,
-        remote=remote,
-        host=h,
-        org=o,
-        repo=r,
     )
     if blocked:
         return blocked
@@ -2539,10 +2520,6 @@ def gitea_create_pr(
     blocked = _profile_permission_block(
         task_capability_map.required_permission("create_pr"),
         number=None,
-        remote=remote,
-        host=host,
-        org=org,
-        repo=repo,
     )
     if blocked:
         return blocked
@@ -3328,18 +3305,6 @@ def _save_review_decision_lock(data):
     payload["profile_identity"] = binding["profile_identity"]
     if binding.get("remote") and not payload.get("remote"):
         payload["remote"] = binding["remote"]
-    # #695 AC6: stamp native transport provenance on durable decision locks.
-    try:
-        payload.update(
-            {
-                k: v
-                for k, v in mcp_daemon_guard.mutation_provenance_fields().items()
-                if v is not None
-            }
-        )
-    except Exception:
-        payload.setdefault("transport", "untrusted")
-        payload.setdefault("native_mcp_transport", False)
     persisted = mcp_session_state.save_state(
         kind=mcp_session_state.KIND_DECISION_LOCK,
         payload=payload,
@@ -4076,15 +4041,6 @@ def _evaluate_pr_review_submission(
         reasons.extend(review_workflow_load.recovery_handoff_without_replay())
         return result
     if live:
-        # #695 AC1/AC2: offline direct-import submit (PR #701 run_submit.py) fails closed.
-        try:
-            mcp_daemon_guard.assert_sanctioned_mutation_runtime(
-                "gitea_submit_pr_review"
-            )
-            mcp_daemon_guard.assert_no_direct_import_bypass("gitea_submit_pr_review")
-        except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
-            reasons.append(str(exc))
-            return result
         ns_gate = _live_namespace_health_gate("review_pr")
         if ns_gate:
             reasons.extend(ns_gate)
@@ -4272,16 +4228,6 @@ def gitea_mark_final_review_decision(
     repo: str | None = None,
 ) -> dict:
     """Mark validation complete; the final review decision is ready to submit."""
-    # #695 AC1/AC2: direct import / redirected session state cannot mark final.
-    try:
-        mcp_daemon_guard.assert_sanctioned_mutation_runtime(
-            "gitea_mark_final_review_decision"
-        )
-        mcp_daemon_guard.assert_no_direct_import_bypass(
-            "gitea_mark_final_review_decision"
-        )
-    except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
-        return {"marked_ready": False, "reasons": [str(exc)]}
     action = (action or "").strip().lower()
     lock = _load_review_decision_lock()
     if lock is None:
@@ -5379,7 +5325,7 @@ def gitea_commit_files(
         return blocked
     blocked = _profile_permission_block(
         task_capability_map.required_permission("commit_files"),
-        commit="", branch="", remote=remote, host=host, org=org, repo=repo,
+        commit="", branch="",
     )
     if blocked:
         return blocked
@@ -7522,13 +7468,48 @@ def _role_for_operation(op: str) -> str | None:
 
 
 def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
-    """#714: automatic profile substitution is removed (fail closed).
+    """Try to find a profile in config that allows op and has valid credentials.
 
-    Always returns False. Callers must use explicit ``gitea_activate_profile``
+    If found, switch to it, clear identity cache, and return True.
-    to change roles; capability and mutation gates evaluate only the active
+    Otherwise return False.
-    profile. Parameters retained for call-site compatibility.
     """
-    del op, host
+    role = _role_for_operation(op)
+    if not role:
+        return False
+    if not gitea_config.is_runtime_switching_enabled():
+        return False
+    config = gitea_config.load_config()
+    if not config or "profiles" not in config:
+        return False
+    for p_name, p_data in config["profiles"].items():
+        # Role classification matching
+        p_role = p_data.get("role") or _role_kind(p_data.get("allowed_operations", []), p_data.get("forbidden_operations", []))
+        if p_role != role:
+            continue
+        p_allowed = p_data.get("allowed_operations") or []
+        p_forbidden = p_data.get("forbidden_operations") or []
+        p_allowed_n = []
+        for op_val in p_allowed:
+            try:
+                p_allowed_n.append(gitea_config.normalize_operation(op_val))
+            except Exception:
+                pass
+        p_forbidden_n = []
+        for op_val in p_forbidden:
+            try:
+                p_forbidden_n.append(gitea_config.normalize_operation(op_val))
+            except Exception:
+                pass
+        ok, _ = gitea_config.check_operation(op, p_allowed_n, p_forbidden_n)
+        if ok:
+            try:
+                tok = gitea_config.resolve_token(p_data)
+                if tok:
+                    gitea_config._active_profile_override = p_name
+                    _IDENTITY_CACHE.clear()
+                    return True
+            except Exception:
+                pass
     return False
 
 
@@ -7595,109 +7576,6 @@ def _profile_operation_gate(op: str) -> list[str]:
     return [f"profile is not allowed to {op}"]
 
 
-def _session_context_mutation_block(
-    *,
-    remote: str | None,
-    host: str | None = None,
-    org: str | None = None,
-    repo: str | None = None,
-    username: str | None = None,
-    **extra_fields,
-) -> dict | None:
-    """#714: fail closed on session context / identity / cross-host drift."""
-    try:
-        profile = get_profile()
-    except Exception as exc:
-        return {
-            "success": False,
-            "performed": False,
-            "reasons": [
-                f"profile could not be resolved (fail closed): {_redact(str(exc))}"
-            ],
-            "blocker_kind": "session_context",
-            **extra_fields,
-        }
-    profile_name = profile.get("profile_name")
-    expected = (profile.get("username") or "").strip() or None
-    remote_config = REMOTES.get(remote or "", {}) if remote else {}
-    h = host or remote_config.get("host")
-    resolved_org = org or remote_config.get("org")
-    resolved_repo = repo or remote_config.get("repo")
-    identity = username
-    # Legacy env-only profiles do not declare an expected identity. Their
-    # first mutation still pins remote/host/repository, while existing audit
-    # paths resolve identity at the actual write. Configured identities are
-    # always verified here before a mutation can proceed.
-    if identity is None and expected and h:
-        try:
-            identity = _authenticated_username(h)
-        except Exception:
-            identity = None
-
-    config = None
-    try:
-        config = gitea_config.load_config()
-    except Exception:
-        config = None
-    contexts = (config or {}).get("contexts") if config else None
-    remote_gate = session_ctx.profile_allowed_for_remote(
-        profile, remote, REMOTES, contexts=contexts
-    )
-    reasons: list[str] = list(remote_gate.get("reasons") or [])
-
-    id_gate = session_ctx.assess_identity_match(
-        authenticated=identity, expected_username=expected
-    )
-    reasons.extend(id_gate.get("reasons") or [])
-
-    # Seed if unbound so subsequent tools share one context; then assess.
-    _seed_session_context(
-        profile=profile,
-        remote=remote,
-        host=h,
-        identity=identity,
-        repository=resolved_repo,
-        org=resolved_org,
-        source="mutation-gate-seed",
-    )
-    ctx_gate = session_ctx.assess_session_context(
-        profile_name=profile_name,
-        remote=remote,
-        host=h,
-        identity=identity,
-        repository=resolved_repo,
-        org=resolved_org,
-        expected_username=expected,
-        require_bound=True,
-    )
-    reasons.extend(ctx_gate.get("reasons") or [])
-    # De-dupe while preserving order
-    seen: set[str] = set()
-    uniq: list[str] = []
-    for r in reasons:
-        if r not in seen:
-            seen.add(r)
-            uniq.append(r)
-    if not uniq:
-        return None
-    blocked = {
-        "success": False,
-        "performed": False,
-        "reasons": uniq,
-        "blocker_kind": "session_context",
-        "session_context_audit": session_ctx.mutation_context_audit_fields(),
-        "exact_next_action": (
-            "BLOCKED + DIAGNOSE: session profile/remote/host/identity context "
-            "is inconsistent or unauthorized for this mutation. Re-pin the "
-            "correct profile with gitea_activate_profile, re-verify with "
-            "gitea_whoami for the intended remote, and do not use cross-host "
-            "fallback profiles."
-        ),
-    }
-    blocked.update(extra_fields)
-    return blocked
-
-
 def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
     """Structured permission denial for gated tools (#69, #142).
 
@@ -7707,37 +7585,25 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
     req_role = "reviewer" if any(required_operation.startswith(p) for p in (
         "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes", "gitea.pr.review"
     )) else "author"
-    # #714: evaluate active profile only — never auto-switch.
     _ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
 
     reasons = _profile_operation_gate(required_operation)
-    if reasons:
+    if not reasons:
+        return None
     blocked = {
         "success": False,
         "performed": False,
         "reasons": reasons,
         "permission_report": _permission_block_report(required_operation),
-            "session_context_audit": session_ctx.mutation_context_audit_fields(),
     }
     blocked.update(extra_fields)
     return blocked
 
-    return _session_context_mutation_block(
-        remote=extra_fields.get("remote"),
-        host=extra_fields.get("host"),
-        org=extra_fields.get("org"),
-        repo=extra_fields.get("repo"),
-        number=extra_fields.get("number"),
-        issue_number=extra_fields.get("issue_number"),
-        pr_number=extra_fields.get("pr_number"),
-    )
-
 
 def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None:
     """Reviewer/author namespace alignment gate (#209)."""
     required_permission = task_capability_map.required_permission(mutation_task)
     required_role = task_capability_map.required_role(mutation_task)
-    # #714: evaluate active profile only — never auto-switch.
     _ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
 
     try:
@@ -9724,22 +9590,9 @@ def gitea_whoami(
     # name is the addressing surface; 'server' appears only under the
     # GITEA_MCP_REVEAL_ENDPOINTS admin opt-in.
     profile = get_profile()
-    identity = data.get("login")
-    expected_username = (profile.get("username") or "").strip() or None
-    # #714: seed immutable session context so later tools share one pin.
-    _seed_session_context(
-        profile=profile,
-        remote=remote,
-        host=h,
-        identity=identity,
-        source="gitea_whoami",
-    )
-    id_match = session_ctx.assess_identity_match(
-        authenticated=identity, expected_username=expected_username
-    )
     result = {
         "authenticated": True,
-        "username": identity,
+        "username": data.get("login"),
         "display_name": data.get("full_name") or None,
         "user_id": data.get("id"),
         "email": data.get("email") or None,
@@ -9756,11 +9609,7 @@ def gitea_whoami(
             "execution_profile": profile.get("execution_profile"),
             "audit_label": profile.get("audit_label"),
             "auth_source_type": profile.get("auth_source_type"),
-            "expected_username": expected_username,
         },
-        "session_context_audit": session_ctx.mutation_context_audit_fields(),
-        "identity_match": not id_match.get("block"),
-        "identity_match_reasons": id_match.get("reasons") or [],
     }
     if _reveal_endpoints():
         result["server"] = gitea_url(h, "").rstrip("/")
@@ -9886,32 +9735,14 @@ _RUNTIME_CAPABILITY_TASKS = (
 def _matching_configured_profiles(
     config: dict | None,
     required_permission: str,
-    remote: str | None = None,
 ) -> list[str]:
-    """Profile names that allow *required_permission* (redacted metadata only).
+    """Profile names that allow *required_permission* (redacted metadata only)."""
-
-    #714: when *remote* is provided, only profiles bound to that remote's host
-    are returned — a dadeschools request never lists prgs profiles.
-    """
     if not config or "profiles" not in config:
         return []
-    contexts = config.get("contexts") or {}
-    remote_ok_names: set[str] | None = None
-    if remote:
-        remote_ok_names = set(
-            session_ctx.filter_profiles_for_remote(config, remote, REMOTES)
-        )
     matches: list[str] = []
     for p_name, p_data in config["profiles"].items():
         if not p_data.get("enabled", True):
             continue
-        if remote_ok_names is not None and p_name not in remote_ok_names:
-            continue
-        # Also require host/context match when remote given (defensive).
-        if remote and not session_ctx.profile_matches_remote(
-            p_data, remote, REMOTES, contexts=contexts
-        ):
-            continue
         p_allowed = p_data.get("allowed_operations") or []
         p_forbidden = p_data.get("forbidden_operations") or []
         p_allowed_n = []
@@ -9938,9 +9769,8 @@ def _build_runtime_task_capabilities(
     allowed: list[str],
     forbidden: list[str],
     config: dict | None,
-    remote: str | None = None,
 ) -> dict:
-    """Per-task capability summary for role-aware runtime context (#139, #714)."""
+    """Per-task capability summary for role-aware runtime context (#139)."""
     task_entries = []
     flags: dict[str, bool] = {}
     flag_keys = {
@@ -9964,7 +9794,7 @@ def _build_runtime_task_capabilities(
             "required_role_kind": task_capability_map.required_role(task),
             "allowed_in_current_session": allowed_here,
             "matching_configured_profiles": _matching_configured_profiles(
-                config, permission, remote=remote
+                config, permission
             ),
         }
         task_entries.append(entry)
@@ -10416,9 +10246,8 @@ def gitea_get_runtime_context(
             "or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
         )
 
-    # #714: filter matching profiles by requested remote when building capability summary
     session_capabilities = _build_runtime_task_capabilities(
-        allowed, forbidden, config, remote=remote if remote in REMOTES else None
+        allowed, forbidden, config
     )
 
     preflight = assess_preflight_status(worktree_path)
@@ -10429,16 +10258,6 @@ def gitea_get_runtime_context(
             f"Blocked: {'; '.join(preflight['preflight_block_reasons'])}"
         )
 
-    expected_username = (profile.get("username") or "").strip() or None
-    h_rt = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
-    _seed_session_context(
-        profile=profile,
-        remote=remote if remote in REMOTES else None,
-        host=h_rt,
-        identity=username,
-        source="gitea_get_runtime_context",
-    )
-
     result = {
         "active_profile": profile["profile_name"],
         "authenticated_username": username,
@@ -10449,8 +10268,6 @@ def gitea_get_runtime_context(
         "forbidden_operations": forbidden,
         "runtime_switching_supported": switching,
         "profile_mode": profile_mode,
-        "session_context_audit": session_ctx.mutation_context_audit_fields(),
-        "auto_profile_substitution": False,
         "review_merge_allowed": review_merge_allowed,
         "review_merge_blocked_reasons": blocked_reasons,
         "suggested_fix": suggested_fix,
@@ -10812,10 +10629,8 @@ def gitea_activate_profile(
         _IDENTITY_CACHE.pop(h, None)
 
     # 4. Resolve fresh identity
-    after_profile_data = get_profile()
+    after_profile = get_profile()["profile_name"]
-    after_profile = after_profile_data["profile_name"]
     after_identity = _authenticated_username(h) if h else None
-    expected_username = (after_profile_data.get("username") or "").strip() or None
 
     # 4.5 Record the authorized pivot in the in-process mutation authority
     # and keep the session profile lock in sync — this is the ONLY path that
@@ -10830,32 +10645,15 @@ def gitea_activate_profile(
             "from_identity": before_identity,
             "to_identity": after_identity,
         }
-        _MUTATION_AUTHORITY["remote"] = remote
     if os.environ.get(SESSION_PROFILE_LOCK_ENV) and after_profile:
         os.environ[SESSION_PROFILE_LOCK_ENV] = after_profile
 
-    # 4.6 #714: re-bind immutable session context after explicit activation.
-    session_ctx.bind_session_context(
-        profile_name=after_profile or profile_name,
-        remote=remote,
-        host=h,
-        identity=after_identity,
-        role_kind=_profile_role_kind(after_profile_data),
-        expected_username=expected_username,
-        source="gitea_activate_profile",
-    )
-
     # 5. Audit the switch if auditing is on
     _audit(
         "activate_profile",
         host=h,
         remote=remote,
-        result={
+        result={"success": True, "before": before_profile, "after": after_profile},
-            "success": True,
-            "before": before_profile,
-            "after": after_profile,
-            "session_context": session_ctx.mutation_context_audit_fields(),
-        },
         username=after_identity,
     )
 
@@ -10866,8 +10664,6 @@ def gitea_activate_profile(
         "before_identity": before_identity,
         "after_profile": after_profile,
         "after_identity": after_identity,
-        "session_context_audit": session_ctx.mutation_context_audit_fields(),
-        "auto_profile_substitution": False,
     }
 
 
@@ -11968,44 +11764,21 @@ def gitea_resolve_task_capability(
 
     record_preflight_check("capability", required_role, resolved_task=task)
 
-    # #714: never auto-switch profiles during capability resolution.
+    # Try automatic dispatch switching
+    _ensure_matching_profile(required_permission, required_role, remote, host)
+
     profile = get_profile()
     config = gitea_config.load_config()
-    contexts = (config or {}).get("contexts") if config else None
 
     h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
     username = _authenticated_username(h) if h else None
-    expected_username = (profile.get("username") or "").strip() or None
-
-    # Seed / re-check immutable session context for consistent multi-tool reads.
-    _seed_session_context(
-        profile=profile,
-        remote=remote,
-        host=h,
-        identity=username,
-        source="resolve_task_capability",
-    )
-    ctx_assess = session_ctx.assess_session_context(
-        profile_name=profile.get("profile_name"),
-        remote=remote,
-        host=h,
-        identity=username,
-        expected_username=expected_username,
-        require_bound=False,
-    )
-    id_assess = session_ctx.assess_identity_match(
-        authenticated=username, expected_username=expected_username
-    )
-    remote_assess = session_ctx.profile_allowed_for_remote(
-        profile, remote, REMOTES, contexts=contexts
-    )
 
     # Load active permissions
     active_allowed = profile.get("allowed_operations") or []
     active_forbidden = profile.get("forbidden_operations") or []
     active_role_kind = _profile_role_kind(profile)
 
-    # Check if allowed in current session (active profile only — #714)
+    # Check if allowed in current session
     permission_allowed_in_current_session, _ = gitea_config.check_operation(
         required_permission, active_allowed, active_forbidden
     )
@@ -12020,15 +11793,8 @@ def gitea_resolve_task_capability(
             f"{required_role} task '{task}' even if nearby permissions are "
             "present (fail closed)."
         )
-    cross_host_block = bool(remote_assess.get("block"))
-    identity_block = bool(id_assess.get("block"))
-    drift_block = bool(ctx_assess.get("block"))
     allowed_in_current_session = (
-        permission_allowed_in_current_session
+        permission_allowed_in_current_session and role_matches_current_session
-        and role_matches_current_session
-        and not cross_host_block
-        and not identity_block
-        and not drift_block
     )
 
     switching = gitea_config.is_runtime_switching_enabled()
@@ -12037,19 +11803,33 @@ def gitea_resolve_task_capability(
     restart_required = False
     reason_msg = None
 
-    # Matching profiles: same remote/host only (#714) — advisory, never activated.
+    # Find matching configured profiles
-    matching_profiles = _matching_configured_profiles(
+    matching_profiles = []
-        config, required_permission, remote=remote
+    if config and "profiles" in config:
-    )
+        for p_name, p_data in config["profiles"].items():
-    # Role filter for exclusive tasks
+            p_allowed = p_data.get("allowed_operations") or []
-    if config and "profiles" in config and task in role_exclusive_tasks:
+            p_forbidden = p_data.get("forbidden_operations") or []
-        filtered = []
+            p_allowed_n = []
-        for p_name in matching_profiles:
+            for op in p_allowed:
-            p_data = (config.get("profiles") or {}).get(p_name) or {}
+                try:
+                    p_allowed_n.append(gitea_config.normalize_operation(op))
+                except Exception:
+                    pass
+            p_forbidden_n = []
+            for op in p_forbidden:
+                try:
+                    p_forbidden_n.append(gitea_config.normalize_operation(op))
+                except Exception:
+                    pass
+            ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
             p_role = (p_data.get("role") or "").strip()
-            if not p_role or p_role == required_role:
+            p_role_ok = (
-                filtered.append(p_name)
+                task not in role_exclusive_tasks
-        matching_profiles = filtered
+                or not p_role
+                or p_role == required_role
+            )
+            if ok and p_role_ok:
+                matching_profiles.append(p_name)
 
     configured = len(matching_profiles) > 0
     available_in_session = allowed_in_current_session
@@ -12065,35 +11845,16 @@ def gitea_resolve_task_capability(
             )
 
     if not allowed_in_current_session:
-        deny_parts: list[str] = []
+        if configured and switching:
-        if cross_host_block:
+            restart_required = True
-            deny_parts.extend(remote_assess.get("reasons") or [])
-        if identity_block:
-            deny_parts.extend(id_assess.get("reasons") or [])
-        if drift_block:
-            deny_parts.extend(ctx_assess.get("reasons") or [])
-        if not permission_allowed_in_current_session and not deny_parts:
-            deny_parts.append(
-                f"Active profile '{profile.get('profile_name')}' is not allowed "
-                f"to '{required_permission}' (no substitute profile activated; fail closed)."
-            )
-        if role_mismatch_reason:
-            deny_parts.append(role_mismatch_reason)
-        if deny_parts:
-            reason_msg = "; ".join(deny_parts)
-        elif configured and switching:
-            # Same-remote profile exists but is not the active one — explicit switch only.
-            restart_required = False
             available_in_session = False
             reason_msg = (
-                f"Active profile cannot perform '{required_permission}'. "
+                f"{required_role.capitalize()} profile exists but MCP server "
-                f"Same-remote matching profiles (advisory only, not activated): "
+                "was added after session startup and is not attached."
-                f"{matching_profiles}."
             )
         elif not configured:
             reason_msg = (
-                f"No same-remote profile configured with permission "
+                f"No profile configured with permission '{required_permission}'."
-                f"'{required_permission}' for remote '{remote}'."
             )
         elif role_mismatch_reason:
             reason_msg = role_mismatch_reason
@@ -12101,22 +11862,9 @@ def gitea_resolve_task_capability(
     next_safe_action = "None; ready for operations."
 
     if not allowed_in_current_session:
-        if cross_host_block or identity_block or drift_block:
+        if switching:
             different_namespace_required = False
-            next_safe_action = (
+            next_safe_action = f"Switch to a profile that has the required permission by calling gitea_activate_profile (matching configured profiles: {matching_profiles})."
-                "BLOCKED + DIAGNOSE: session context/host/identity is inconsistent "
-                "or the active profile cannot serve this remote. Re-pin the correct "
-                "profile with gitea_activate_profile for the intended remote, verify "
-                "with gitea_whoami, and do not use cross-host profile substitution."
-            )
-        elif switching:
-            different_namespace_required = False
-            next_safe_action = (
-                f"Switch explicitly with gitea_activate_profile to a same-remote "
-                f"profile that allows '{required_permission}' "
-                f"(matching configured profiles: {matching_profiles}). "
-                "Capability resolution never auto-substitutes profiles (#714)."
-            )
         else:
             different_namespace_required = True
             if required_role == "reviewer":
@@ -12201,13 +11949,6 @@ def gitea_resolve_task_capability(
         "runtime_switching_supported": switching,
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
-        # #714: immutable session context proof (must match whoami / runtime).
-        "requested_remote": remote,
-        "resolved_host": h,
-        "session_context_audit": session_ctx.mutation_context_audit_fields(),
-        "profile_remote_compatible": not cross_host_block,
-        "identity_match": not identity_block,
-        "auto_profile_substitution": False,
     }
     if reason_msg:
         result["reason"] = reason_msg
@@ -13102,10 +12843,9 @@ def gitea_quarantine_contaminated_review(
     427 until an independent adversarial reviewer and controller deployment
     of this tooling have completed.
     """
-    # Fail closed outside production native MCP before any mutation assessment.
+    # Fail closed outside native MCP before any mutation assessment.
-    # Test-mode bootstrap must never reach this production mutation endpoint (#695).
     try:
-        mcp_daemon_guard.assert_production_mutation_runtime(
+        mcp_daemon_guard.assert_sanctioned_mutation_runtime(
             "gitea_quarantine_contaminated_review"
         )
     except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
@@ -13311,12 +13051,10 @@ def gitea_quarantine_contaminated_review(
 # ── Entry point ───────────────────────────────────────────────────────────────
 
 if __name__ == "__main__":
-    # #558 / #695: claim the resolved canonical entrypoint, then bind the live
+    # #558 / #695: mark this process as the official native MCP daemon before
-    # native MCP transport lifecycle before any tool dispatch. Env vars,
+    # any tool dispatch. Env vars alone cannot reconstruct native transport;
-    # basename-only stack frames, and import-only launch cannot reconstruct
+    # offline imports / standalone scripts fail closed on mutations.
-    # native transport; offline imports / standalone scripts fail closed.
     mcp_daemon_guard.mark_sanctioned_daemon()
-    mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
     # Lock this session's launch profile into the environment so child CLI
     # processes (e.g. review_pr.py) can detect and refuse profile
     # side-channel overrides (#199).

mcp_daemon_guard.py

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

mcp_server.py

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

mcp_session_state.py

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

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 {

session_context_binding.py

@@ -1,419 +0,0 @@
-"""Session-immutable MCP mutation context (#714).
-
-Pins profile, remote, host, repository, identity, and role for the life of an
-MCP process (or until an explicit ``gitea_activate_profile`` re-bind).
-
-Capability resolution and mutation gates must evaluate only the active
-profile for the requested remote. Silent cross-host / cross-profile
-substitution is forbidden.
-"""
-
-from __future__ import annotations
-
-import os
-import threading
-import urllib.parse
-from dataclasses import dataclass
-from typing import Any, Mapping
-
-
-@dataclass(frozen=True, slots=True)
-class _SessionContext:
-    """One atomic, immutable process-session binding."""
-
-    profile_name: str | None
-    remote: str | None
-    host: str | None
-    identity: str | None
-    repository: str | None
-    org: str | None
-    role_kind: str | None
-    expected_username: str | None
-    source: str
-    pid: int
-
-    def as_dict(self) -> dict[str, Any]:
-        return {
-            "profile_name": self.profile_name,
-            "remote": self.remote,
-            "host": self.host,
-            "identity": self.identity,
-            "repository": self.repository,
-            "org": self.org,
-            "role_kind": self.role_kind,
-            "expected_username": self.expected_username,
-            "source": self.source,
-            "pid": self.pid,
-        }
-
-
-# Process-local only — never a shared file (same rationale as mutation authority).
-# The frozen value prevents partial mutation, while the lock makes first-bind and
-# sanctioned rebind atomic across concurrent MCP calls.
-_SESSION_CONTEXT: _SessionContext | None = None
-_SESSION_CONTEXT_LOCK = threading.RLock()
-
-
-def _reset_session_context_for_testing() -> None:
-    """Reset at a pytest test boundary; unavailable to production callers.
-
-    Production sessions transition only through process startup/PID change or
-    the explicit profile-activation rebind.  Keeping this helper private and
-    requiring pytest's per-test marker prevents it from becoming an MCP/runtime
-    bypass.
-    """
-    if "PYTEST_CURRENT_TEST" not in os.environ:
-        raise RuntimeError("session context reset is restricted to pytest boundaries")
-    global _SESSION_CONTEXT
-    with _SESSION_CONTEXT_LOCK:
-        _SESSION_CONTEXT = None
-
-
-def get_session_context() -> dict[str, Any] | None:
-    """Return a detached snapshot of the bound context, or None if unbound."""
-    with _SESSION_CONTEXT_LOCK:
-        if _SESSION_CONTEXT is None:
-            return None
-        return _SESSION_CONTEXT.as_dict()
-
-
-def profile_host(profile: dict | None) -> str | None:
-    """Hostname from profile base_url, lowercased, or None."""
-    if not profile:
-        return None
-    base = (profile.get("base_url") or "").strip()
-    if not base:
-        return None
-    try:
-        parsed = urllib.parse.urlparse(base)
-        host = (parsed.netloc or parsed.path or "").strip().lower()
-        return host or None
-    except Exception:
-        return None
-
-
-def remote_host(remote: str | None, remotes: dict | None) -> str | None:
-    """Hostname for a known remote key."""
-    if not remote or not remotes:
-        return None
-    entry = remotes.get(remote) or {}
-    return (entry.get("host") or "").strip().lower() or None
-
-
-def profile_matches_remote(
-    profile: dict | None,
-    remote: str | None,
-    remotes: dict | None,
-    *,
-    contexts: dict | None = None,
-) -> bool:
-    """True when *profile* is bound to the same host/context as *remote*."""
-    if not profile or not remote:
-        return False
-    r_host = remote_host(remote, remotes)
-    p_host = profile_host(profile)
-    if r_host and p_host:
-        return r_host == p_host
-    # Fall back to context name heuristics when base_url missing.
-    ctx = (profile.get("context") or "").strip().lower()
-    if not ctx or not contexts:
-        return False
-    ctx_data = contexts.get(ctx) or {}
-    gitea = ctx_data.get("gitea") or {}
-    base = (gitea.get("base_url") or "").strip()
-    if not base or not r_host:
-        return False
-    try:
-        parsed = urllib.parse.urlparse(base)
-        c_host = (parsed.netloc or parsed.path or "").strip().lower()
-    except Exception:
-        return False
-    return bool(c_host) and c_host == r_host
-
-
-def bind_session_context(
-    *,
-    profile_name: str,
-    remote: str | None,
-    host: str | None,
-    identity: str | None,
-    repository: str | None = None,
-    org: str | None = None,
-    role_kind: str | None = None,
-    expected_username: str | None = None,
-    source: str = "bind",
-) -> dict[str, Any]:
-    """Atomically bind/re-bind context (the explicit activation path)."""
-    with _SESSION_CONTEXT_LOCK:
-        return _bind_session_context_unlocked(
-            profile_name=profile_name,
-            remote=remote,
-            host=host,
-            identity=identity,
-            repository=repository,
-            org=org,
-            role_kind=role_kind,
-            expected_username=expected_username,
-            source=source,
-        )
-
-
-def _bind_session_context_unlocked(
-    *,
-    profile_name: str,
-    remote: str | None,
-    host: str | None,
-    identity: str | None,
-    repository: str | None,
-    org: str | None,
-    role_kind: str | None,
-    expected_username: str | None,
-    source: str,
-) -> dict[str, Any]:
-    """Store a complete immutable context while the caller holds the lock."""
-    global _SESSION_CONTEXT
-    _SESSION_CONTEXT = _SessionContext(
-        profile_name=(profile_name or "").strip() or None,
-        remote=(remote or "").strip() or None,
-        host=(host or "").strip().lower() or None,
-        identity=(identity or "").strip() or None,
-        repository=(repository or "").strip() or None,
-        org=(org or "").strip() or None,
-        role_kind=(role_kind or "").strip() or None,
-        expected_username=(expected_username or "").strip() or None,
-        source=source,
-        pid=os.getpid(),
-    )
-    return _SESSION_CONTEXT.as_dict()
-
-
-def seed_session_context_if_unbound(
-    *,
-    profile_name: str,
-    remote: str | None,
-    host: str | None,
-    identity: str | None,
-    repository: str | None = None,
-    org: str | None = None,
-    role_kind: str | None = None,
-    expected_username: str | None = None,
-    source: str = "seed",
-) -> dict[str, Any]:
-    """Atomically bind only when this process has no current context.
-
-    A changed environment or an interleaved call is not a session boundary and
-    therefore cannot replace an established binding.  A newly started/forked
-    process is recognized by PID; explicit ``gitea_activate_profile`` uses
-    :func:`bind_session_context` as its sanctioned logical-session transition.
-    """
-    with _SESSION_CONTEXT_LOCK:
-        if _SESSION_CONTEXT is None or _SESSION_CONTEXT.pid != os.getpid():
-            return _bind_session_context_unlocked(
-                profile_name=profile_name,
-                remote=remote,
-                host=host,
-                identity=identity,
-                repository=repository,
-                org=org,
-                role_kind=role_kind,
-                expected_username=expected_username,
-                source=source,
-            )
-        return _SESSION_CONTEXT.as_dict()
-
-
-def assess_session_context(
-    *,
-    profile_name: str | None,
-    remote: str | None,
-    host: str | None = None,
-    identity: str | None = None,
-    repository: str | None = None,
-    org: str | None = None,
-    expected_username: str | None = None,
-    require_bound: bool = False,
-) -> dict[str, Any]:
-    """Compare live values against the bound session context.
-
-    Returns ``proven`` / ``block`` / ``reasons``. When unbound and
-    ``require_bound`` is false, does not block (caller may seed). When
-    unbound and ``require_bound`` is true, fails closed.
-    """
-    reasons: list[str] = []
-    with _SESSION_CONTEXT_LOCK:
-        bound = _SESSION_CONTEXT
-        ctx = bound.as_dict() if bound is not None else None
-    if ctx is None or ctx.get("pid") != os.getpid():
-        if require_bound:
-            reasons.append(
-                "session mutation context is unbound; call gitea_whoami or "
-                "gitea_activate_profile before mutating (fail closed)"
-            )
-            return _assessment(False, reasons, ctx)
-        return _assessment(True, reasons, ctx)
-
-    live_profile = (profile_name or "").strip() or None
-    live_remote = (remote or "").strip() or None
-    live_host = (host or "").strip().lower() or None
-    live_identity = (identity or "").strip() or None
-    live_repo = (repository or "").strip() or None
-    live_org = (org or "").strip() or None
-
-    if ctx.get("profile_name") and live_profile and live_profile != ctx.get("profile_name"):
-        reasons.append(
-            f"profile drift: live '{live_profile}' != bound "
-            f"'{ctx.get('profile_name')}' (fail closed)"
-        )
-    if ctx.get("remote") and live_remote and live_remote != ctx.get("remote"):
-        reasons.append(
-            f"remote drift: live '{live_remote}' != bound "
-            f"'{ctx.get('remote')}' (fail closed)"
-        )
-    if ctx.get("host") and live_host and live_host != ctx.get("host"):
-        reasons.append(
-            f"host drift: live '{live_host}' != bound "
-            f"'{ctx.get('host')}' (fail closed)"
-        )
-    if (
-        ctx.get("identity")
-        and live_identity
-        and live_identity != ctx.get("identity")
-    ):
-        reasons.append(
-            f"identity drift: live '{live_identity}' != bound "
-            f"'{ctx.get('identity')}' (fail closed)"
-        )
-    if ctx.get("repository") and live_repo and live_repo != ctx.get("repository"):
-        reasons.append(
-            f"repository drift: live '{live_repo}' != bound "
-            f"'{ctx.get('repository')}' (fail closed)"
-        )
-    if ctx.get("org") and live_org and live_org != ctx.get("org"):
-        reasons.append(
-            f"org drift: live '{live_org}' != bound "
-            f"'{ctx.get('org')}' (fail closed)"
-        )
-
-    expected = expected_username or ctx.get("expected_username")
-    if expected and live_identity and live_identity != expected:
-        reasons.append(
-            f"identity mismatch: authenticated '{live_identity}' != "
-            f"profile expected '{expected}' (fail closed)"
-        )
-
-    return _assessment(not reasons, reasons, ctx)
-
-
-def assess_identity_match(
-    *,
-    authenticated: str | None,
-    expected_username: str | None,
-) -> dict[str, Any]:
-    """Fail closed when profile declares a username that does not match live auth."""
-    reasons: list[str] = []
-    auth = (authenticated or "").strip() or None
-    expected = (expected_username or "").strip() or None
-    if expected and auth and auth != expected:
-        reasons.append(
-            f"identity mismatch: authenticated '{auth}' != "
-            f"profile expected '{expected}' (fail closed)"
-        )
-    return {
-        "proven": not reasons,
-        "block": bool(reasons),
-        "reasons": reasons,
-        "authenticated": auth,
-        "expected_username": expected,
-    }
-
-
-def filter_profiles_for_remote(
-    config: dict | None,
-    remote: str | None,
-    remotes: dict | None,
-) -> list[str]:
-    """Profile names whose host/context matches *remote* (enabled only)."""
-    if not config or not remote:
-        return []
-    profiles = config.get("profiles") or {}
-    contexts = config.get("contexts") or {}
-    names: list[str] = []
-    for name, data in profiles.items():
-        if not isinstance(data, dict):
-            continue
-        if not data.get("enabled", True):
-            continue
-        if profile_matches_remote(data, remote, remotes, contexts=contexts):
-            names.append(name)
-    return sorted(names)
-
-
-def profile_allowed_for_remote(
-    profile: dict | None,
-    remote: str | None,
-    remotes: dict | None,
-    *,
-    contexts: dict | None = None,
-) -> dict[str, Any]:
-    """Assess whether the active profile may serve *remote*."""
-    reasons: list[str] = []
-    if not profile:
-        reasons.append("active profile unresolved (fail closed)")
-        return {"proven": False, "block": True, "reasons": reasons}
-    if not remote:
-        return {"proven": True, "block": False, "reasons": reasons}
-    # Legacy env-only profiles have no configured base URL or v2 context. Their
-    # first call may establish the process remote/host pin; after that,
-    # assess_session_context rejects any drift. Configured v2 profiles still
-    # require positive host/context alignment here.
-    if not profile_host(profile) and not (profile.get("context") or "").strip():
-        return {"proven": True, "block": False, "reasons": reasons}
-    if not profile_matches_remote(profile, remote, remotes, contexts=contexts):
-        p_name = profile.get("profile_name") or profile.get("name") or "(unknown)"
-        p_host = profile_host(profile) or "(none)"
-        r_host = remote_host(remote, remotes) or "(none)"
-        reasons.append(
-            f"cross-host profile denial: profile '{p_name}' (host '{p_host}') "
-            f"cannot serve remote '{remote}' (host '{r_host}') (fail closed)"
-        )
-    return {"proven": not reasons, "block": bool(reasons), "reasons": reasons}
-
-
-def mutation_context_audit_fields(
-    ctx: Mapping[str, Any] | None = None,
-) -> dict[str, Any]:
-    """Fields to include in pre-mutation audit records."""
-    data = ctx if ctx is not None else get_session_context()
-    if not data:
-        return {
-            "session_context_bound": False,
-            "session_profile": None,
-            "session_remote": None,
-            "session_host": None,
-            "session_identity": None,
-            "session_repository": None,
-            "session_org": None,
-        }
-    return {
-        "session_context_bound": True,
-        "session_profile": data.get("profile_name"),
-        "session_remote": data.get("remote"),
-        "session_host": data.get("host"),
-        "session_identity": data.get("identity"),
-        "session_repository": data.get("repository"),
-        "session_org": data.get("org"),
-        "session_role_kind": data.get("role_kind"),
-        "session_context_source": data.get("source"),
-    }
-
-
-def _assessment(
-    proven: bool, reasons: list[str], ctx: Mapping[str, Any] | None
-) -> dict[str, Any]:
-    return {
-        "proven": proven,
-        "block": not proven,
-        "reasons": list(reasons),
-        "bound_context": dict(ctx) if ctx else None,
-        "audit": mutation_context_audit_fields(ctx),
-    }

tests/conftest.py

@@ -26,14 +26,6 @@ def _reset_mutation_authority(monkeypatch):
     Pin ``default_state_dir`` / ``DEFAULT_STATE_DIR`` to a per-test temp dir
     so durable load/save never touches host state even after env clears.
     """
-    import session_context_binding as session_ctx
-
-    # Each pytest item is an independent logical MCP session. Reset both before
-    # and after the item; the post-yield reset is in a finally block so an
-    # assertion, exception, or unittest teardown failure cannot pollute the
-    # next item. Production code has no automatic per-call reset path.
-    session_ctx._reset_session_context_for_testing()
-
     for env_key in [
         "GITEA_SESSION_PROFILE_LOCK",
         "GITEA_ACTIVE_WORKTREE",
@@ -66,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
 
@@ -88,15 +69,9 @@ def _reset_mutation_authority(monkeypatch):
     try:
         import mcp_server
     except Exception:
-        try:
-            yield
-        finally:
-            session_ctx._reset_session_context_for_testing()
         _state_tmp.cleanup()
+        yield
         return
-    import gitea_config
-
-    monkeypatch.setattr(gitea_config, "_active_profile_override", None)
     monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
     monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
     monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
@@ -127,9 +102,7 @@ def _reset_mutation_authority(monkeypatch):
         capability_stop_terminal.clear()
     except Exception:
         pass
-    try:
     yield
-    finally:
     try:
         import capability_stop_terminal
         capability_stop_terminal.clear()
@@ -144,6 +117,4 @@ def _reset_mutation_authority(monkeypatch):
         mcp_server._REVIEW_DECISION_LOCK = None
     except Exception:
         pass
-        gitea_config._active_profile_override = None
-        session_ctx._reset_session_context_for_testing()
     _state_tmp.cleanup()

tests/test_commit_payloads.py

@@ -227,7 +227,6 @@ class TestCommitPayloads(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_commit_files_traversal_blocked(self, _auth, mock_api):
-        mock_api.return_value = {"login": "author-user"}
         # Remove active lock file to ensure it fails on traversal/invalid locks
         os.remove(self.lock_file_path)
 
@@ -249,7 +248,6 @@ class TestCommitPayloads(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_commit_files_outside_scope_blocked(self, _auth, mock_api):
-        mock_api.return_value = {"login": "author-user"}
         with patch.dict(os.environ, self._env("full-author"), clear=True):
             with self.assertRaises(ValueError) as ctx:
                 mcp_server.gitea_commit_files(
@@ -268,7 +266,6 @@ class TestCommitPayloads(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_commit_files_multiple_sources_blocked(self, _auth, mock_api):
-        mock_api.return_value = {"login": "author-user"}
         with patch.dict(os.environ, self._env("full-author"), clear=True):
             with self.assertRaises(ValueError) as ctx:
                 mcp_server.gitea_commit_files(

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,11 +133,6 @@ 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.
-            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):
@@ -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_issue_714_session_context_immutability.py

@@ -1,490 +0,0 @@
-"""#714: fail closed on cross-host MCP profile drift and capability substitution."""
-
-from __future__ import annotations
-
-import json
-import os
-import sys
-import tempfile
-import threading
-import unittest
-from pathlib import Path
-from unittest.mock import patch
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-import gitea_config  # noqa: E402
-import mcp_server  # noqa: E402
-import session_context_binding as session_ctx  # noqa: E402
-
-
-CONFIG_714 = {
-    "version": 2,
-    "allow_runtime_switching": True,
-    "contexts": {
-        "prgs": {
-            "enabled": True,
-            "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
-        },
-        "mdcps": {
-            "enabled": True,
-            "gitea": {"enabled": True, "base_url": "https://gitea.dadeschools.net"},
-        },
-    },
-    "profiles": {
-        "prgs-author": {
-            "enabled": True,
-            "context": "prgs",
-            "role": "author",
-            "username": "jcwalker3",
-            "base_url": "https://gitea.prgs.cc",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_PRGS_AUTHOR"},
-            "allowed_operations": [
-                "gitea.read",
-                "gitea.issue.create",
-                "gitea.issue.comment",
-                "gitea.issue.close",
-                "gitea.pr.create",
-                "gitea.pr.comment",
-                "gitea.branch.create",
-                "gitea.branch.push",
-                "gitea.repo.commit",
-            ],
-            "forbidden_operations": [
-                "gitea.pr.approve",
-                "gitea.pr.merge",
-                "gitea.pr.request_changes",
-            ],
-            "execution_profile": "prgs-author",
-        },
-        "mdcps-reviewer": {
-            "enabled": True,
-            "context": "mdcps",
-            "role": "reviewer",
-            "username": "913443",
-            "base_url": "https://gitea.dadeschools.net",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_MDCPS_REVIEWER"},
-            "allowed_operations": [
-                "gitea.read",
-                "gitea.pr.review",
-                "gitea.pr.comment",
-                "gitea.pr.approve",
-                "gitea.pr.request_changes",
-            ],
-            "forbidden_operations": [
-                "gitea.branch.create",
-                "gitea.branch.push",
-                "gitea.pr.create",
-                "gitea.pr.merge",
-                "gitea.repo.commit",
-            ],
-            "execution_profile": "mdcps-reviewer",
-        },
-        "mdcps-author": {
-            "enabled": True,
-            "context": "mdcps",
-            "role": "author",
-            "username": "913443",
-            "base_url": "https://gitea.dadeschools.net",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_MDCPS_AUTHOR"},
-            "allowed_operations": [
-                "gitea.read",
-                "gitea.issue.create",
-                "gitea.issue.comment",
-                "gitea.pr.create",
-                "gitea.pr.comment",
-                "gitea.branch.create",
-                "gitea.branch.push",
-                "gitea.repo.commit",
-            ],
-            "forbidden_operations": [
-                "gitea.pr.approve",
-                "gitea.pr.merge",
-                "gitea.pr.request_changes",
-            ],
-            "execution_profile": "mdcps-author",
-        },
-    },
-}
-
-
-class TestIssue714SessionContextImmutability(unittest.TestCase):
-    def setUp(self):
-        self._dir = tempfile.TemporaryDirectory()
-        self.config_path = os.path.join(self._dir.name, "profiles.json")
-        with open(self.config_path, "w", encoding="utf-8") as fh:
-            fh.write(json.dumps(CONFIG_714))
-        self._remotes = patch.dict(
-            mcp_server.REMOTES,
-            {
-                "dadeschools": {
-                    "host": "gitea.dadeschools.net",
-                    "org": "913443",
-                    "repo": "eAgenda",
-                },
-                "prgs": {
-                    "host": "gitea.prgs.cc",
-                    "org": "Scaled-Tech-Consulting",
-                    "repo": "Gitea-Tools",
-                },
-            },
-            clear=False,
-        )
-        self._remotes.start()
-        mcp_server._IDENTITY_CACHE.clear()
-        gitea_config._active_profile_override = None
-        mcp_server._MUTATION_AUTHORITY = None
-        self._env = {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": "mdcps-reviewer",
-            "GITEA_TOKEN_MDCPS_REVIEWER": "mdcps-reviewer-token",
-            "GITEA_TOKEN_MDCPS_AUTHOR": "mdcps-author-token",
-            "GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token",
-        }
-
-    def tearDown(self):
-        self._remotes.stop()
-        mcp_server._IDENTITY_CACHE.clear()
-        gitea_config._active_profile_override = None
-        mcp_server._MUTATION_AUTHORITY = None
-        self._dir.cleanup()
-
-    def _api_side_effect(self, method, url, header):
-        # Token identity mapping for whoami endpoint
-        auth = str(header)
-        if "mdcps-reviewer-token" in auth or "mdcps-author-token" in auth:
-            login = "913443"
-        else:
-            login = "jcwalker3"
-        return {"login": login, "full_name": "Test", "id": 1, "email": "[email protected]"}
-
-    def test_pin_mdcps_reviewer_never_drifts_to_prgs_author(self):
-        """Reproduce the incident: pin mdcps-reviewer then resolve comment_issue."""
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                act = mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                self.assertTrue(act["success"])
-                self.assertEqual(act["after_profile"], "mdcps-reviewer")
-
-                who1 = mcp_server.gitea_whoami(remote="dadeschools")
-                self.assertEqual(who1["profile"]["profile_name"], "mdcps-reviewer")
-                self.assertEqual(who1["username"], "913443")
-
-                # Capability that mdcps-reviewer lacks — must NOT switch to prgs-author
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="comment_issue", remote="dadeschools"
-                )
-                self.assertEqual(res["active_profile"], "mdcps-reviewer")
-                self.assertFalse(res["allowed_in_current_session"])
-                self.assertFalse(res.get("auto_profile_substitution", True))
-                self.assertNotIn("prgs-author", res["matching_configured_profile"])
-                self.assertIn("mdcps-author", res["matching_configured_profile"])
-
-                who2 = mcp_server.gitea_whoami(remote="dadeschools")
-                rt = mcp_server.gitea_get_runtime_context(remote="dadeschools")
-                self.assertEqual(who2["profile"]["profile_name"], "mdcps-reviewer")
-                self.assertEqual(rt["active_profile"], "mdcps-reviewer")
-                # Still pinned — never prgs-author
-                self.assertEqual(
-                    gitea_config.selected_profile_name(), "mdcps-reviewer"
-                )
-
-    def test_repeated_calls_remain_on_mdcps_reviewer(self):
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                for _ in range(5):
-                    res = mcp_server.gitea_resolve_task_capability(
-                        task="review_pr", remote="dadeschools"
-                    )
-                    self.assertEqual(res["active_profile"], "mdcps-reviewer")
-                    who = mcp_server.gitea_whoami(remote="dadeschools")
-                    self.assertEqual(who["profile"]["profile_name"], "mdcps-reviewer")
-                    rt = mcp_server.gitea_get_runtime_context(remote="dadeschools")
-                    self.assertEqual(rt["active_profile"], "mdcps-reviewer")
-
-    def test_dadeschools_request_cannot_resolve_through_prgs_author(self):
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="comment_issue", remote="dadeschools"
-                )
-                self.assertNotEqual(res["active_profile"], "prgs-author")
-                self.assertNotIn("prgs-author", res["matching_configured_profile"])
-                # Gate must not silently switch either
-                blocked = mcp_server._profile_permission_block(
-                    "gitea.issue.comment", remote="dadeschools"
-                )
-                self.assertIsNotNone(blocked)
-                self.assertFalse(blocked.get("success", True))
-                self.assertEqual(
-                    gitea_config.selected_profile_name(), "mdcps-reviewer"
-                )
-
-    def test_unsupported_capability_fails_closed_structured(self):
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="comment_issue", remote="dadeschools"
-                )
-                self.assertFalse(res["allowed_in_current_session"])
-                self.assertTrue(res["stop_required"])
-                self.assertIn("reason", res)
-                self.assertIn("exact_safe_next_action", res)
-                self.assertIn("activate_profile", res["exact_safe_next_action"])
-                # Unknown task still fail closed
-                with self.assertRaises(ValueError):
-                    mcp_server.gitea_resolve_task_capability(
-                        task="reopen_issue", remote="dadeschools"
-                    )
-
-    def test_identity_mismatch_blocks_mutation(self):
-        """Profile expects 913443 but authenticated as jcwalker3."""
-
-        def wrong_identity(method, url, header):
-            return {
-                "login": "jcwalker3",
-                "full_name": "Wrong",
-                "id": 9,
-                "email": "[email protected]",
-            }
-
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=wrong_identity):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                blocked = mcp_server._session_context_mutation_block(
-                    remote="dadeschools"
-                )
-                self.assertIsNotNone(blocked)
-                self.assertTrue(
-                    any("identity mismatch" in r for r in blocked["reasons"])
-                )
-
-    def test_repository_host_mismatch_blocks_mutation(self):
-        """mdcps-reviewer cannot serve prgs remote."""
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                blocked = mcp_server._session_context_mutation_block(remote="prgs")
-                self.assertIsNotNone(blocked)
-                self.assertTrue(
-                    any("cross-host" in r for r in blocked["reasons"])
-                )
-
-    def test_drift_cannot_reach_write(self):
-        """Simulated mid-session profile override cannot pass permission gate."""
-        with patch.dict(os.environ, self._env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                mcp_server.gitea_activate_profile(
-                    profile_name="mdcps-reviewer", remote="dadeschools"
-                )
-                # Bind session as mdcps-reviewer
-                self.assertEqual(
-                    session_ctx.get_session_context()["profile_name"],
-                    "mdcps-reviewer",
-                )
-                # Hostile override without activate_profile
-                gitea_config._active_profile_override = "prgs-author"
-                blocked = mcp_server._session_context_mutation_block(
-                    remote="dadeschools"
-                )
-                self.assertIsNotNone(blocked)
-                self.assertTrue(any("drift" in r or "cross-host" in r for r in blocked["reasons"]))
-
-    def test_static_prgs_author_still_works(self):
-        env = {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": "prgs-author",
-            "GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token",
-        }
-        with patch.dict(os.environ, env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                gitea_config._active_profile_override = None
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="create_issue", remote="prgs"
-                )
-                self.assertEqual(res["active_profile"], "prgs-author")
-                self.assertTrue(res["allowed_in_current_session"])
-                self.assertEqual(res["active_identity"], "jcwalker3")
-
-    def test_static_mdcps_author_still_works(self):
-        env = {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": "mdcps-author",
-            "GITEA_TOKEN_MDCPS_AUTHOR": "mdcps-author-token",
-        }
-        with patch.dict(os.environ, env, clear=False):
-            with patch("mcp_server.api_request", side_effect=self._api_side_effect):
-                gitea_config._active_profile_override = None
-                res = mcp_server.gitea_resolve_task_capability(
-                    task="comment_issue", remote="dadeschools"
-                )
-                self.assertEqual(res["active_profile"], "mdcps-author")
-                self.assertTrue(res["allowed_in_current_session"])
-                self.assertNotIn("prgs-author", res["matching_configured_profile"])
-
-
-class TestSessionContextBindingUnit(unittest.TestCase):
-    def test_profile_matches_remote_by_base_url(self):
-        remotes = {
-            "dadeschools": {"host": "gitea.dadeschools.net"},
-            "prgs": {"host": "gitea.prgs.cc"},
-        }
-        mdcps = {"base_url": "https://gitea.dadeschools.net", "context": "mdcps"}
-        prgs = {"base_url": "https://gitea.prgs.cc", "context": "prgs"}
-        self.assertTrue(
-            session_ctx.profile_matches_remote(mdcps, "dadeschools", remotes)
-        )
-        self.assertFalse(session_ctx.profile_matches_remote(mdcps, "prgs", remotes))
-        self.assertTrue(session_ctx.profile_matches_remote(prgs, "prgs", remotes))
-
-    def test_bind_and_detect_drift(self):
-        session_ctx.bind_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-            source="test",
-        )
-        ok = session_ctx.assess_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-        )
-        self.assertTrue(ok["proven"])
-        bad = session_ctx.assess_session_context(
-            profile_name="prgs-author",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="jcwalker3",
-        )
-        self.assertTrue(bad["block"])
-        self.assertTrue(any("drift" in r for r in bad["reasons"]))
-
-    def test_bound_context_survives_multiple_calls_and_exceptions(self):
-        original = session_ctx.bind_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-            source="test",
-        )
-
-        try:
-            for _ in range(3):
-                observed = session_ctx.seed_session_context_if_unbound(
-                    profile_name="prgs-author",
-                    remote="prgs",
-                    host="gitea.prgs.cc",
-                    identity="jcwalker3",
-                    source="interleaved-test-call",
-                )
-                self.assertEqual(observed, original)
-            raise RuntimeError("simulated caller failure")
-        except RuntimeError:
-            pass
-
-        self.assertEqual(session_ctx.get_session_context(), original)
-        drift = session_ctx.assess_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            require_bound=True,
-        )
-        self.assertTrue(drift["block"])
-
-    def test_parallel_calls_cannot_overwrite_established_binding(self):
-        original = session_ctx.bind_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-            source="test",
-        )
-        barrier = threading.Barrier(9)
-        results = []
-        results_lock = threading.Lock()
-
-        def competing_seed(index: int) -> None:
-            barrier.wait()
-            result = session_ctx.seed_session_context_if_unbound(
-                profile_name=f"prgs-author-{index}",
-                remote="prgs",
-                host="gitea.prgs.cc",
-                identity=f"other-{index}",
-                source="parallel-test-call",
-            )
-            with results_lock:
-                results.append(result)
-
-        threads = [threading.Thread(target=competing_seed, args=(i,)) for i in range(8)]
-        for thread in threads:
-            thread.start()
-        barrier.wait()
-        for thread in threads:
-            thread.join(timeout=5)
-
-        self.assertFalse(any(thread.is_alive() for thread in threads))
-        self.assertEqual(results, [original] * 8)
-        self.assertEqual(session_ctx.get_session_context(), original)
-
-    def test_returned_snapshot_cannot_mutate_binding(self):
-        session_ctx.bind_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-            source="test",
-        )
-        snapshot = session_ctx.get_session_context()
-        snapshot["profile_name"] = "prgs-author"
-        self.assertEqual(
-            session_ctx.get_session_context()["profile_name"], "mdcps-reviewer"
-        )
-
-
-class TestSessionContextTestBoundaryIsolation(unittest.TestCase):
-    def test_mdcps_binding_starts_clean_and_does_not_escape_test(self):
-        self.assertIsNone(session_ctx.get_session_context())
-        session_ctx.bind_session_context(
-            profile_name="mdcps-reviewer",
-            remote="dadeschools",
-            host="gitea.dadeschools.net",
-            identity="913443",
-            source="test-boundary",
-        )
-
-    def test_prgs_binding_starts_clean_and_does_not_escape_test(self):
-        self.assertIsNone(session_ctx.get_session_context())
-        session_ctx.bind_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            source="test-boundary",
-        )
-
-    def test_reset_helper_is_rejected_outside_pytest_boundary(self):
-        with patch.dict(os.environ, {}, clear=True):
-            with self.assertRaises(RuntimeError):
-                session_ctx._reset_session_context_for_testing()
-
-
-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_operation_scoped_roles.py

@@ -1,9 +1,4 @@
-"""Tests for operation-scoped role selection (#228, updated by #714).
+"""Tests for operation-scoped role selection and automatic dispatch switching (#228)."""
-
-#714 removes silent automatic profile substitution. Capability resolution
-and mutation gates evaluate only the active profile; explicit
-``gitea_activate_profile`` is required to switch roles.
-"""
 import os
 import sys
 import json
@@ -20,7 +15,6 @@ from reviewer_worktree import assess_author_worktree_continuity
 
 CONFIG_TEST = {
     "version": 2,
-    "allow_runtime_switching": True,
     "contexts": {
         "ctx": {
             "enabled": True,
@@ -36,7 +30,6 @@ CONFIG_TEST = {
             "context": "ctx",
             "role": "author",
             "username": "author-user",
-            "base_url": "https://gitea.example.com",
             "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
             "allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"],
             "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
@@ -47,7 +40,6 @@ CONFIG_TEST = {
             "context": "ctx",
             "role": "reviewer",
             "username": "reviewer-user",
-            "base_url": "https://gitea.example.com",
             "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
             "allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.merge", "gitea.issue.comment"],
             "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
@@ -93,71 +85,62 @@ class TestOperationScopedRoles(unittest.TestCase):
         return env
 
     @patch("mcp_server.api_request")
-    def test_no_auto_switch_to_reviewer(self, mock_api):
+    def test_auto_switch_to_reviewer(self, mock_api):
-        # #714: capability resolution must not silently switch author → reviewer
+        # mock identity resolution to return username matching profile
         mock_api.side_effect = lambda method, url, header: (
             {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
         )
         with patch.dict(os.environ, self._env("author-profile")):
+            # initially we are author-profile
             self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
             self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile")
 
+            # resolve a reviewer task (review_pr)
             res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
             
-            self.assertFalse(res["allowed_in_current_session"])
+            # verify it automatically switched to reviewer-profile
-            self.assertFalse(res["available_in_session"])
+            self.assertTrue(res["allowed_in_current_session"])
-            self.assertEqual(res["active_profile"], "author-profile")
+            self.assertTrue(res["available_in_session"])
-            self.assertEqual(res["active_identity"], "author-user")
+            self.assertEqual(res["active_profile"], "reviewer-profile")
-            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+            self.assertEqual(res["active_identity"], "reviewer-user")
-            self.assertIn("reviewer-profile", res["matching_configured_profile"])
+            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
-            self.assertFalse(res.get("auto_profile_substitution", True))
 
     @patch("mcp_server.api_request")
-    def test_no_auto_switch_to_author(self, mock_api):
+    def test_auto_switch_to_author(self, mock_api):
         mock_api.side_effect = lambda method, url, header: (
             {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
         )
         with patch.dict(os.environ, self._env("reviewer-profile")):
+            # initially we are reviewer-profile
             self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
 
+            # resolve an author task (create_issue)
             res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
             
-            self.assertFalse(res["allowed_in_current_session"])
+            # verify it automatically switched to author-profile
-            self.assertFalse(res["available_in_session"])
-            self.assertEqual(res["active_profile"], "reviewer-profile")
-            self.assertEqual(res["active_identity"], "reviewer-user")
-            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
-            self.assertIn("author-profile", res["matching_configured_profile"])
-
-    @patch("mcp_server.api_request")
-    def test_explicit_activate_profile_still_works(self, mock_api):
-        mock_api.side_effect = lambda method, url, header: (
-            {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
-        )
-        with patch.dict(os.environ, self._env("author-profile")):
-            act = mcp_server.gitea_activate_profile(
-                profile_name="reviewer-profile", remote="prgs"
-            )
-            self.assertTrue(act["success"])
-            self.assertEqual(act["after_profile"], "reviewer-profile")
-            res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
             self.assertTrue(res["allowed_in_current_session"])
-            self.assertEqual(res["active_profile"], "reviewer-profile")
+            self.assertTrue(res["available_in_session"])
+            self.assertEqual(res["active_profile"], "author-profile")
+            self.assertEqual(res["active_identity"], "author-user")
+            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
 
     @patch("mcp_server.api_request")
-    def test_denied_when_matching_profile_token_missing(self, mock_api):
+    def test_restart_required_when_unattached(self, mock_api):
         mock_api.side_effect = lambda method, url, header: {"login": "author-user"}
         # launch without reviewer token in env
         with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)):
             self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
 
+            # resolve a reviewer task (review_pr)
             res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
             
+            # verify it did NOT switch and reports restart_required
             self.assertFalse(res["allowed_in_current_session"])
             self.assertFalse(res["available_in_session"])
             self.assertTrue(res["configured"])
+            self.assertTrue(res["restart_required"])
             self.assertTrue(res["stop_required"])
-            self.assertEqual(res["active_profile"], "author-profile")
+            self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"])
 
     def test_author_continuity_dirty_worktree(self):
         # author is allowed to keep dirty worktree
@@ -178,21 +161,20 @@ class TestOperationScopedRoles(unittest.TestCase):
         self.assertFalse(res2["proven"])
 
     @patch("mcp_server.api_request")
-    def test_mutating_actions_do_not_auto_switch(self, mock_api):
+    def test_mutating_actions_auto_switch(self, mock_api):
         mock_api.side_effect = lambda method, url, header: (
             {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
         )
         with patch.dict(os.environ, self._env("author-profile")):
+            # verify we are author-profile
             self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
 
-            blocked = mcp_server._profile_permission_block(
+            # call a reviewer mutation check helper (like _profile_permission_block with reviewer permission)
-                "gitea.pr.merge", remote="prgs"
+            blocked = mcp_server._profile_permission_block("gitea.pr.merge", remote="prgs")
-            )
+            self.assertIsNone(blocked) # should switch to reviewer-profile and allow it (no permission block)
-            self.assertIsNotNone(blocked)
-            self.assertFalse(blocked.get("success", True))
 
-            # profile must remain author — no silent substitution
+            # verify we dynamically switched to reviewer-profile
-            self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
+            self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
 
 
 if __name__ == "__main__":

tests/test_resolve_task_capability.py

@@ -188,8 +188,8 @@ class TestResolveTaskCapability(unittest.TestCase):
                 self.assertEqual(res["required_role_kind"], "author")
                 self.assertTrue(res["allowed_in_current_session"])
 
-    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
-    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_work_issue_reviewer_profile_blocked(self, _auth, _api):
         with patch.dict(os.environ, self._env("reviewer-profile")):
             res = mcp_server.gitea_resolve_task_capability(

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",