fix(mcp): invalidate review session state on cross-profile activation (Closes #690)

A mid-run profile switch (reviewer -> author -> reviewer) left workflow-load
proof, reviewer lease binding, the review decision lock, live namespace
health, and preflight identity/capability stamps intact, so a formal verdict
could be recorded under contaminated session state.

- gitea_activate_profile now invalidates all review-critical session state
  on a cross-profile switch, in memory and in durable state keyed by either
  profile identity, and reports the invalidation + re-preflight requirement.
- Full reviewer preflight (whoami, load_review_workflow,
  resolve_task_capability(review_pr), head re-pin, lease re-acquire) is
  required before any formal verdict after a switch; switching back cannot
  resurrect the stale run.
- Namespace provenance: optional launcher-declared GITEA_MCP_NAMESPACE is
  reported by whoami/runtime context/capability resolution, and a declared
  namespace that disagrees with a task's required namespace fails closed.
- Docs: supported pattern is separate session/namespace per role, not
  in-process profile hopping mid-review.

docs/gitea-execution-profiles.md

@@ -589,6 +589,47 @@ When dynamic profile switching is enabled and a profile is activated via `gitea_
 2. Call `gitea_whoami` with the target remote to prove and verify the fresh Gitea authenticated identity.
 This guarantees the active profile operations align with the actual Gitea authenticated user credential.
 
+### 4. Review-State Invalidation on Profile Switch (#690)
+
+A cross-profile activation (e.g. reviewer → author → reviewer) is a session
+boundary for formal review state. On any switch where the activated profile
+differs from the previous one, `gitea_activate_profile` invalidates, in
+memory **and** in durable session state (for both the old and new profile
+identities):
+
+- preflight identity/capability stamps (`gitea_whoami` / `gitea_resolve_task_capability` proof),
+- review workflow-load proof (`gitea_load_review_workflow`),
+- the review decision lock (including `final_review_decision_ready` markers),
+- the reviewer PR session lease binding,
+- live namespace-health assessments.
+
+Before any formal verdict (`gitea_mark_final_review_decision` /
+`gitea_submit_pr_review`) the full reviewer preflight must be re-established
+under the new profile: `gitea_whoami`, `gitea_load_review_workflow`,
+`gitea_resolve_task_capability(review_pr)`, live head re-pin, and lease
+re-acquire/adopt. Switching back to the earlier profile does **not**
+resurrect the prior run — durable state keyed by either profile identity is
+cleared at switch time.
+
+The supported pattern remains **separate session/namespace per role**
+(dual-namespace, §2): file author-side follow-ups from an author session,
+not by hopping profiles inside a formal review run. Runtime profile
+switching is the operator-approved exception and always carries the
+re-preflight cost above.
+
+### 5. Namespace Provenance (#690)
+
+The server cannot derive its own client-managed MCP namespace name, so a
+launcher may declare it via the `GITEA_MCP_NAMESPACE` environment variable
+(e.g. `gitea-reviewer`). `gitea_whoami`, `gitea_get_runtime_context`, and
+`gitea_resolve_task_capability` report `namespace_provenance` — the
+configured client namespace, the active execution profile, and, for tasks
+with a required namespace (`review_pr` → `gitea-reviewer`, `merge_pr` →
+`gitea-merger`), a mismatch verdict. A declared namespace that disagrees
+with the requested task's required namespace **fails closed**. An
+undeclared namespace is reported as `unknown` and is never treated as
+proof either way.
+
 ## Gitea MCP Runtime Isolation and Worktree Safety
 
 To ensure high availability and prevent broken feature worktrees from disabling essential security/identity controls, the Gitea MCP server implements runtime isolation:

docs/mcp-namespace-health.md

@@ -86,3 +86,24 @@ When a namespace returns EOF, follow
 
 When blocked, repair the IDE namespace and re-record a healthy
 `client_namespace` assessment before retrying the mutation.
+
+## Namespace provenance (#690)
+
+A server process cannot derive the name of the client-managed namespace it is
+registered under, so the launcher may declare it with the
+`GITEA_MCP_NAMESPACE` environment variable (e.g. `GITEA_MCP_NAMESPACE=gitea-reviewer`).
+
+- `gitea_whoami`, `gitea_get_runtime_context`, and
+  `gitea_resolve_task_capability` report `namespace_provenance`: the declared
+  client namespace, the active execution profile, and — for tasks with a
+  required namespace (`review_pr`/`submit_review` → `gitea-reviewer`,
+  `merge_pr` → `gitea-merger`) — a `mismatch` verdict.
+- A declared namespace that disagrees with the requested task's required
+  namespace **fails closed** (`allowed_in_current_session=false` with a STOP
+  guidance entry).
+- An undeclared namespace is reported as `namespace_source="unknown"` and is
+  never treated as proof either way.
+- A profile switch via `gitea_activate_profile` clears all recorded live
+  namespace-health assessments; re-probe through the client before further
+  review/merge mutations.
+

gitea_mcp_server.py

@@ -24,8 +24,6 @@ import subprocess
 import uuid
 from datetime import datetime, timedelta, timezone
 from typing import Any
-import hashlib
-
 
 
 
@@ -841,6 +839,75 @@ def _invalidate_preflight_identity_state() -> None:
     _clear_preflight_capability_state()
 
 
+# #690: session-boundary invalidation record for the most recent cross-profile
+# activation. Surfaced in runtime diagnostics so a formal review run can prove
+# its state was reset by a profile switch and must be fully re-established.
+_PROFILE_SWITCH_INVALIDATION: dict | None = None
+
+
+def _invalidate_review_state_on_profile_switch(
+    before_profile: str,
+    after_profile: str,
+) -> dict:
+    """Invalidate review-critical session state on a profile switch (#690).
+
+    Workflow-load proof, reviewer lease binding, the review decision lock,
+    live namespace health, and preflight identity/capability stamps recorded
+    under the prior profile are contaminated for the new role. Durable state
+    keyed by *either* profile identity is cleared so a reviewer → author →
+    reviewer hop cannot resurrect a stale review run: the full reviewer
+    preflight (gitea_whoami, gitea_load_review_workflow,
+    gitea_resolve_task_capability(review_pr), live head re-pin, lease
+    re-acquire/adopt) must be re-established before any formal verdict.
+    """
+    global _PROFILE_SWITCH_INVALIDATION
+    invalidated: list[str] = []
+
+    _invalidate_preflight_identity_state()
+    invalidated.append("preflight_identity_capability")
+
+    review_workflow_load.clear_review_workflow_load()
+    invalidated.append("review_workflow_load")
+
+    _save_review_decision_lock(None)
+    invalidated.append("review_decision_lock")
+
+    reviewer_pr_lease.clear_session_lease()
+    invalidated.append("reviewer_session_lease")
+
+    if _LIVE_NAMESPACE_HEALTH:
+        _LIVE_NAMESPACE_HEALTH.clear()
+        invalidated.append("live_namespace_health")
+
+    # Durable records keyed by either profile identity must not survive the
+    # switch, or activating author → reviewer → author could revive a stale
+    # review run without re-preflight.
+    for identity in {before_profile, after_profile}:
+        if not identity:
+            continue
+        try:
+            mcp_session_state.clear_state(
+                kind=mcp_session_state.KIND_DECISION_LOCK,
+                profile_identity=identity,
+            )
+            mcp_session_state.clear_state(
+                kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+                profile_identity=identity,
+            )
+        except Exception:
+            pass  # best-effort durable cleanup; in-memory state already reset
+    invalidated.append("durable_profile_state")
+
+    _PROFILE_SWITCH_INVALIDATION = {
+        "from_profile": before_profile,
+        "to_profile": after_profile,
+        "invalidated": invalidated,
+        "invalidated_at": datetime.now(timezone.utc).isoformat(),
+        "re_preflight_required": True,
+    }
+    return dict(_PROFILE_SWITCH_INVALIDATION)
+
+
 def record_preflight_check(
     type_name: str,
     resolved_role: str | None = None,
@@ -2260,12 +2327,7 @@ def _seed_session_context(
         expected_username=expected,
         source=source,
         canonical_repository_root=canonical_root_pin,
-        cohort_id=_COHORT_ID,
-        startup_sha=_STARTUP_PARITY.get("startup_head"),
-        endpoint=host or (profile.get("base_url") or "").strip() or None,
-        config_fingerprint=_CONFIG_FINGERPRINT,
     )
-
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
 import terminal_pr_label_cleanup  # noqa: E402  # #780 status:pr-open terminal rule
@@ -2294,11 +2356,6 @@ import stable_control_runtime  # noqa: E402
 # master has advanced past the running code and fail closed until restart.
 # Read-only operations are never blocked by staleness.
 _STARTUP_PARITY = master_parity_gate.capture_startup_parity(PROJECT_ROOT)
-_COHORT_ID: str = f"cohort-p{os.getpid()}-{_STARTUP_PARITY.get('startup_head') or 'unknown'}"
-_CONFIG_FINGERPRINT: str = hashlib.sha256(
-    (PROJECT_ROOT + str(_STARTUP_PARITY.get("startup_head"))).encode("utf-8")
-).hexdigest()[:16]
-
 
 # Stable-control runtime facts (#615): which runtime this process serves from.
 # These are the *immutable* facts -- process root, branch, head, checkout-ness --
@@ -14209,10 +14266,8 @@ def _current_master_parity() -> dict:
     current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
     live_head = master_parity_gate.read_remote_master_head(
         PROJECT_ROOT, remote=_git_default_remote_name(PROJECT_ROOT))
-    bound_context = session_ctx.get_session_context()
     return master_parity_gate.assess_master_parity(
-        _STARTUP_PARITY, current_head, live_remote_head=live_head, bound_cohort=bound_context)
+        _STARTUP_PARITY, current_head, live_remote_head=live_head)
-
 
 
 def _current_runtime_mode_report(refresh: bool = False) -> dict:
@@ -17224,6 +17279,11 @@ def gitea_whoami(
         "session_context_audit": session_ctx.mutation_context_audit_fields(),
         "identity_match": not id_match.get("block"),
         "identity_match_reasons": id_match.get("reasons") or [],
+        # #690 AC4: report launcher-declared client namespace alongside the
+        # active execution profile so drift is visible in diagnostics.
+        "namespace_provenance": mcp_namespace_health.namespace_provenance(
+            active_profile=profile["profile_name"]
+        ),
     }
     if id_match.get("block"):
         _invalidate_preflight_identity_state()
@@ -18122,6 +18182,11 @@ def gitea_get_runtime_context(
         "shell_health": native_mcp_preference.shell_health_status(),
         "workflow_load_proof": review_workflow_load.workflow_load_status(
             PROJECT_ROOT),
+        # #690: namespace provenance + profile-switch invalidation evidence.
+        "namespace_provenance": mcp_namespace_health.namespace_provenance(
+            active_profile=profile["profile_name"]
+        ),
+        "profile_switch_invalidation": _PROFILE_SWITCH_INVALIDATION,
     }
 
     # #702: read-only visibility into the inherited GITEA_ACTIVE_WORKTREE
@@ -18625,6 +18690,17 @@ def gitea_activate_profile(
         source="gitea_activate_profile",
     )
 
+    # 4.7 #690: a profile switch is a session-boundary event for review state.
+    # Any workflow-load proof, reviewer lease, decision lock, namespace
+    # health, or preflight stamp recorded under the prior profile is
+    # contaminated for the new role and must be re-established under the new
+    # profile before any formal review verdict.
+    switch_invalidation = None
+    if before_profile != after_profile:
+        switch_invalidation = _invalidate_review_state_on_profile_switch(
+            before_profile, after_profile
+        )
+
     # 5. Audit the switch if auditing is on
     _audit(
         "activate_profile",
@@ -18635,11 +18711,12 @@ def gitea_activate_profile(
             "before": before_profile,
             "after": after_profile,
             "session_context": session_ctx.mutation_context_audit_fields(),
+            "review_state_invalidated": bool(switch_invalidation),
         },
         username=after_identity,
     )
 
-    return {
+    result = {
         "success": True,
         "message": f"Successfully activated profile '{profile_name}' (fresh identity verification complete).",
         "before_profile": before_profile,
@@ -18649,6 +18726,18 @@ def gitea_activate_profile(
         "session_context_audit": session_ctx.mutation_context_audit_fields(),
         "auto_profile_substitution": False,
     }
+    if switch_invalidation is not None:
+        result["review_state_invalidation"] = switch_invalidation
+        result["re_preflight_required"] = True
+        result["exact_next_action"] = (
+            "Profile switch invalidated workflow-load proof, reviewer lease, "
+            "decision lock, and preflight stamps (#690). Before any formal "
+            "review verdict, re-run the full reviewer preflight: "
+            "gitea_whoami, gitea_load_review_workflow, "
+            "gitea_resolve_task_capability(review_pr), live head re-pin, and "
+            "lease re-acquire/adopt."
+        )
+    return result
 
 
 @mcp.tool()
@@ -20893,12 +20982,22 @@ def gitea_resolve_task_capability(
             f"{required_role} task '{task}' even if nearby permissions are "
             "present (fail closed)."
         )
+    # #690 AC4: when the launcher declares a client namespace, a task with a
+    # required namespace must fail closed on mismatch (e.g. review_pr served
+    # from an author namespace).
+    ns_provenance = mcp_namespace_health.namespace_provenance(
+        task=task_key, active_profile=profile.get("profile_name")
+    )
+    ns_mismatch_reason = None
+    if ns_provenance.get("mismatch"):
+        ns_mismatch_reason = "; ".join(ns_provenance.get("reasons") or [])
     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
         and role_matches_current_session
+        and not ns_provenance.get("mismatch")
         and not cross_host_block
         and not identity_block
         and not drift_block
@@ -20949,6 +21048,8 @@ def gitea_resolve_task_capability(
             )
         if role_mismatch_reason:
             deny_parts.append(role_mismatch_reason)
+        if ns_mismatch_reason:
+            deny_parts.append(ns_mismatch_reason)
         if deny_parts:
             reason_msg = "; ".join(deny_parts)
         elif configured and switching:
@@ -21026,6 +21127,8 @@ def gitea_resolve_task_capability(
     task_role_guidance = []
     if role_mismatch_reason:
         task_role_guidance.append(f"STOP: {role_mismatch_reason}")
+    if ns_mismatch_reason:
+        task_role_guidance.append(f"STOP: {ns_mismatch_reason}")
     if required_role == "reviewer":
         if allowed_in_current_session:
             task_role_guidance.append(
@@ -21092,6 +21195,7 @@ def gitea_resolve_task_capability(
         "session_context_audit": session_ctx.mutation_context_audit_fields(),
         "profile_remote_compatible": not cross_host_block,
         "identity_match": not identity_block,
+        "namespace_provenance": ns_provenance,
         "auto_profile_substitution": False,
     }
     # #685: report typed reconnect blocker without mutating config or exiting.

master_parity_gate.py

@@ -183,7 +183,6 @@ def assess_master_parity(
     startup: dict | None,
     current_head: str | None,
     live_remote_head: str | None = None,
-    bound_cohort: dict | None = None,
 ) -> dict:
     """Compare the startup baseline against the current on-disk ``HEAD``.
 
@@ -193,8 +192,7 @@ def assess_master_parity(
       could not be determined, which is not treated as stale).
     - ``stale``        -- the on-disk master has definitively advanced past the
       running process.
-    - ``restart_required`` -- ``stale``, ``live_stale``, or ``cohort_stale``; the
+    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
-      recovery action.
     - ``determinable`` -- whether both local HEADs were known well enough to
       compare.
     - ``startup_head`` / ``current_head`` / ``reasons``.
@@ -211,69 +209,24 @@ def assess_master_parity(
     - ``live_known`` -- whether the live remote target was resolved.
     - ``live_stale`` -- the live remote master has advanced past the running
       process (daemon is behind live master) even if local parity is green.
-    - ``bound_cohort`` -- metadata describing the bound MCP cohort.
+    - ``mutation_safe`` -- the daemon code, local checkout, and live remote
-    - ``cohort_parity_match`` -- whether bound cohort startup SHA matches parity.
+      target all agree; the only state in which a mutation may rely on parity.
-    - ``cohort_stale`` -- bound cohort startup SHA is stale relative to parity.
-    - ``mutation_safe`` -- the daemon code, local checkout, live remote target,
-      and bound cohort all agree; the only state in which a mutation may rely
-      on parity.
     """
     startup_head = (startup or {}).get("startup_head")
     reasons: list[str] = []
 
-    cohort_info: dict | None = None
-    cohort_parity_match = True
-    cohort_stale = False
-    if bound_cohort:
-        c_id = str(bound_cohort.get("cohort_id") or "").strip() or None
-        c_pid = bound_cohort.get("pid")
-        c_sha = str(
-            bound_cohort.get("startup_sha")
-            or bound_cohort.get("git_head")
-            or ""
-        ).strip() or None
-        c_endpoint = str(bound_cohort.get("endpoint") or "").strip() or None
-        c_fingerprint = str(
-            bound_cohort.get("config_fingerprint") or ""
-        ).strip() or None
-
-        cohort_info = {
-            "cohort_id": c_id,
-            "pid": c_pid,
-            "startup_sha": c_sha,
-            "endpoint": c_endpoint,
-            "config_fingerprint": c_fingerprint,
-        }
-
-        parity_ref = live_remote_head or current_head or startup_head
-        if c_sha and parity_ref:
-            if c_sha.lower() != parity_ref.lower():
-                cohort_parity_match = False
-                cohort_stale = True
-                reasons.append(
-                    f"bound cohort startup SHA '{_short(c_sha)}' does not "
-                    f"match authoritative parity SHA '{_short(parity_ref)}' "
-                    "(stale cohort refused)"
-                )
-
     if startup_head is None:
         reasons.append(
             "startup commit was not captured; code parity cannot be enforced")
         return _result(True, False, False, startup_head, current_head,
-                       live_remote_head, False, reasons,
+                       live_remote_head, False, reasons)
-                       bound_cohort=cohort_info,
-                       cohort_parity_match=cohort_parity_match,
-                       cohort_stale=cohort_stale)
 
     if current_head is None:
         reasons.append(
             "current workspace HEAD could not be read; code parity cannot be "
             "enforced")
         return _result(True, False, False, startup_head, current_head,
-                       live_remote_head, False, reasons,
+                       live_remote_head, False, reasons)
-                       bound_cohort=cohort_info,
-                       cohort_parity_match=cohort_parity_match,
-                       cohort_stale=cohort_stale)
 
     local_in_parity = startup_head == current_head
     local_stale = not local_in_parity
@@ -293,28 +246,18 @@ def assess_master_parity(
 
     return _result(
         local_in_parity, local_stale, True, startup_head, current_head,
-        live_remote_head, live_stale, reasons,
+        live_remote_head, live_stale, reasons)
-        bound_cohort=cohort_info,
-        cohort_parity_match=cohort_parity_match,
-        cohort_stale=cohort_stale)
 
 
 def _result(in_parity, stale, determinable, startup_head, current_head,
-            live_remote_head, live_stale, reasons, bound_cohort=None,
+            live_remote_head, live_stale, reasons):
-            cohort_parity_match=True, cohort_stale=False):
     live_known = live_remote_head is not None
     mutation_safe = (
-        determinable
+        determinable and in_parity and live_known and not live_stale)
-        and in_parity
-        and live_known
-        and not live_stale
-        and cohort_parity_match
-        and not cohort_stale
-    )
     return {
         "in_parity": in_parity,
         "stale": stale,
-        "restart_required": stale or live_stale or cohort_stale,
+        "restart_required": stale or live_stale,
         "determinable": determinable,
         "startup_head": startup_head,
         "current_head": current_head,
@@ -324,15 +267,11 @@ def _result(in_parity, stale, determinable, startup_head, current_head,
         "live_remote_head": live_remote_head,
         "live_known": live_known,
         "live_stale": live_stale,
-        "cohort_parity_match": cohort_parity_match,
-        "cohort_stale": cohort_stale,
-        "bound_cohort": bound_cohort,
         "mutation_safe": mutation_safe,
         "reasons": list(reasons),
     }
 
 
-
 def gate_disabled() -> bool:
     """Whether the parity gate is disabled by env escape hatch."""
     return bool((os.environ.get(ENV_DISABLE) or "").strip())

mcp_namespace_health.py

@@ -16,6 +16,7 @@ Probe sources
 
 from __future__ import annotations
 
+import os
 from typing import Any
 
 
@@ -57,8 +58,56 @@ SAFE_ENV_KEYS = (
     "GITEA_SERVICE",
     "GITEA_EXECUTION_ROLE",
     "GITEA_MCP_CONFIG",
+    "GITEA_MCP_NAMESPACE",
 )
 
+# Optional launcher-provided env declaring the client-managed MCP namespace
+# this process is registered under (e.g. ``gitea-reviewer``). The server
+# cannot derive its own IDE namespace name, so the launcher declares it; when
+# declared, reviewers/mergers can fail closed on a namespace/task mismatch
+# (#690 AC4). Absence means "unknown" — reported, never guessed.
+NAMESPACE_ENV = "GITEA_MCP_NAMESPACE"
+
+
+def configured_client_namespace(env: dict[str, str] | None = None) -> str | None:
+    """Return the launcher-declared client namespace, or None when unknown."""
+    source = os.environ if env is None else env
+    value = (source.get(NAMESPACE_ENV) or "").strip()
+    return value or None
+
+
+def namespace_provenance(
+    task: str | None = None,
+    *,
+    active_profile: str | None = None,
+    env: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """Report configured client namespace vs active execution profile (#690).
+
+    When *task* carries a required namespace (``TASK_REQUIRED_NAMESPACES``)
+    and the launcher declared a different one, ``mismatch`` is True and the
+    caller must fail closed for that task. An undeclared namespace is
+    reported as unknown — never treated as proof either way.
+    """
+    configured = configured_client_namespace(env)
+    required = TASK_REQUIRED_NAMESPACES.get(task or "")
+    mismatch = bool(configured and required and configured != required)
+    reasons: list[str] = []
+    if mismatch:
+        reasons.append(
+            f"configured client namespace '{configured}' does not match "
+            f"required namespace '{required}' for task '{task}' (fail closed)"
+        )
+    return {
+        "configured_namespace": configured,
+        "namespace_source": NAMESPACE_ENV if configured else "unknown",
+        "active_profile": active_profile,
+        "requested_task": task,
+        "required_namespace": required,
+        "mismatch": mismatch,
+        "reasons": reasons,
+    }
+
 
 def _as_list(value: Any) -> list[str] | None:
     if value is None:
@@ -104,8 +153,6 @@ def classify_namespace_probe(
     profile: str | None = None,
     configured: bool = True,
     probe_source: str | None = None,
-    expected_parity_sha: str | None = None,
-    bound_cohort: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Classify whether a required tool is callable through a live namespace.
 
@@ -137,50 +184,20 @@ def classify_namespace_probe(
         else:
             error_type = "namespace_call_failed"
 
-    # Extract cohort metadata
-    cohort_meta = bound_cohort or probe.get("cohort") or probe.get("bound_cohort") or {}
-    cohort_id = str(
-        cohort_meta.get("cohort_id") or probe.get("cohort_id") or ""
-    ).strip() or None
-    startup_sha = str(
-        cohort_meta.get("startup_sha")
-        or cohort_meta.get("git_head")
-        or probe.get("startup_sha")
-        or probe.get("git_head")
-        or ""
-    ).strip() or None
-    endpoint = str(
-        cohort_meta.get("endpoint") or probe.get("endpoint") or ""
-    ).strip() or None
-    config_fingerprint = str(
-        cohort_meta.get("config_fingerprint") or probe.get("config_fingerprint") or ""
-    ).strip() or None
-
-    expected_sha = (expected_parity_sha or "").strip().lower() or None
-    stale_cohort = False
-    if expected_sha and startup_sha:
-        if startup_sha.lower() != expected_sha:
-            stale_cohort = True
-            error_type = "stale_cohort_refused"
-
     if not configured:
         error_type = "namespace_not_configured"
     elif registered is False:
         error_type = "tool_missing"
     elif not probe_result:
         error_type = "live_probe_missing"
-    elif stale_cohort:
-        error_type = "stale_cohort_refused"
     elif not probe_success and not error_type:
         error_type = "namespace_call_failed"
 
-    callable_live = bool(configured and probe_result and probe_success and not stale_cohort)
+    callable_live = bool(configured and probe_result and probe_success)
     # Probe-path health (spawn or client). IDE-proven only for client path.
     healthy = bool(configured and registered is not False and callable_live)
     ide_namespace_proven = bool(healthy and source == PROBE_SOURCE_CLIENT)
-    process_pid = process.get("pid") if isinstance(process, dict) else (
+    process_pid = process.get("pid") if isinstance(process, dict) else None
-        cohort_meta.get("pid") if isinstance(cohort_meta, dict) else None
-    )
     profile_name = profile or (
         process.get("profile") if isinstance(process, dict) else None
     )
@@ -193,13 +210,7 @@ def classify_namespace_probe(
         reasons.append(
             f"Required tool '{tool}' is not registered in namespace '{ns}'."
         )
-    if error_type == "stale_cohort_refused":
+    if error_type == "live_probe_missing":
-        reasons.append(
-            f"Bound cohort startup SHA '{startup_sha[:12] if startup_sha else 'unknown'}' "
-            f"does not match expected parity SHA '{expected_sha[:12] if expected_sha else 'unknown'}' "
-            "(stale cohort refused)."
-        )
-    elif error_type == "live_probe_missing":
         reasons.append(
             f"No live client invocation proof was supplied for '{ns}.{tool}'."
         )
@@ -286,10 +297,6 @@ def classify_namespace_probe(
             "env": env_summary,
             "config_path": config_path,
             "probe_source": source,
-            "cohort_id": cohort_id,
-            "startup_sha": startup_sha,
-            "endpoint": endpoint,
-            "config_fingerprint": config_fingerprint,
         },
         "blocks_merge_workflow": blocks,
     }

post_restart_reconcile.py

@@ -151,16 +151,12 @@ class RestartCompletionProof:
     unresolved_count: int
     skipped_count: int
     note: str
-    binding_unchanged: bool = False
-    prior_reconcile_id: str | None = None
 
     def as_dict(self) -> dict[str, Any]:
         return {
             "schema_version": self.schema_version,
             "reconcile_version": self.reconcile_version,
             "reconcile_id": self.reconcile_id,
-            "binding_unchanged": self.binding_unchanged,
-            "prior_reconcile_id": self.prior_reconcile_id,
             "started_at": self.started_at,
             "finished_at": self.finished_at,
             "boot_head_sha": self.boot_head_sha,
@@ -177,7 +173,6 @@ class RestartCompletionProof:
             "skipped_count": self.skipped_count,
             "note": self.note,
             "links": {
-
                 "umbrella": 655,
                 "vision": 652,
                 "roadmap": 653,
@@ -364,9 +359,7 @@ def reconcile_after_restart(
     now: datetime | None = None,
     mode: str = MODE_LOG_ONLY,
     reconcile_id: str | None = None,
-    prior_reconcile_id: str | None = None,
 ) -> RestartCompletionProof:
-
     """Classify a post-restart inventory into a completion proof (#662).
 
     Parameters
@@ -757,26 +750,12 @@ def reconcile_after_restart(
         f"Mode={mode_norm}."
     )
 
-    prior_id = str(
-        prior_reconcile_id
-        or inventory.get("prior_reconcile_id")
-        or ""
-    ).strip() or None
-    target_rec_id = str(reconcile_id or "").strip() or None
-    binding_unchanged = bool(
-        prior_id and target_rec_id and target_rec_id == prior_id
-    )
-    final_reconcile_id = target_rec_id or f"reconcile-{uuid4().hex[:12]}"
-
     return RestartCompletionProof(
         schema_version=SCHEMA_VERSION,
         reconcile_version=RECONCILE_VERSION,
-        reconcile_id=final_reconcile_id,
+        reconcile_id=(reconcile_id or f"reconcile-{uuid4().hex[:12]}"),
-        binding_unchanged=binding_unchanged,
-        prior_reconcile_id=prior_id,
         started_at=_ts(started),
         finished_at=_ts(finished),
-
         boot_head_sha=(
             str(inventory.get("boot_head_sha")).strip()
             if inventory.get("boot_head_sha")

session_context_binding.py

@@ -33,10 +33,6 @@ class _SessionContext:
     source: str
     pid: int
     canonical_repository_root: str | None = None
-    cohort_id: str | None = None
-    startup_sha: str | None = None
-    endpoint: str | None = None
-    config_fingerprint: str | None = None
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -51,14 +47,9 @@ class _SessionContext:
             "source": self.source,
             "pid": self.pid,
             "canonical_repository_root": self.canonical_repository_root,
-            "cohort_id": self.cohort_id,
-            "startup_sha": self.startup_sha,
-            "endpoint": self.endpoint,
-            "config_fingerprint": self.config_fingerprint,
         }
 
 
-
 # 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.
@@ -81,13 +72,6 @@ def _reset_session_context_for_testing() -> None:
         _SESSION_CONTEXT = None
 
 
-def clear_session_context() -> None:
-    """Purge process-session context and cohort bindings on disconnect."""
-    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:
@@ -162,10 +146,6 @@ def bind_session_context(
     expected_username: str | None = None,
     source: str = "bind",
     canonical_repository_root: str | None = None,
-    cohort_id: str | None = None,
-    startup_sha: str | None = None,
-    endpoint: str | None = None,
-    config_fingerprint: str | None = None,
 ) -> dict[str, Any]:
     """Atomically bind/re-bind context (the explicit activation path)."""
     with _SESSION_CONTEXT_LOCK:
@@ -180,10 +160,6 @@ def bind_session_context(
             expected_username=expected_username,
             source=source,
             canonical_repository_root=canonical_repository_root,
-            cohort_id=cohort_id,
-            startup_sha=startup_sha,
-            endpoint=endpoint,
-            config_fingerprint=config_fingerprint,
         )
 
 
@@ -199,10 +175,6 @@ def _bind_session_context_unlocked(
     expected_username: str | None,
     source: str,
     canonical_repository_root: str | None = None,
-    cohort_id: str | None = None,
-    startup_sha: str | None = None,
-    endpoint: str | None = None,
-    config_fingerprint: str | None = None,
 ) -> dict[str, Any]:
     """Store a complete immutable context while the caller holds the lock."""
     global _SESSION_CONTEXT
@@ -218,10 +190,6 @@ def _bind_session_context_unlocked(
         source=source,
         pid=os.getpid(),
         canonical_repository_root=(canonical_repository_root or "").strip() or None,
-        cohort_id=(cohort_id or "").strip() or None,
-        startup_sha=(startup_sha or "").strip() or None,
-        endpoint=(endpoint or "").strip() or None,
-        config_fingerprint=(config_fingerprint or "").strip() or None,
     )
     return _SESSION_CONTEXT.as_dict()
 
@@ -238,10 +206,6 @@ def seed_session_context_if_unbound(
     expected_username: str | None = None,
     source: str = "seed",
     canonical_repository_root: str | None = None,
-    cohort_id: str | None = None,
-    startup_sha: str | None = None,
-    endpoint: str | None = None,
-    config_fingerprint: str | None = None,
 ) -> dict[str, Any]:
     """Atomically bind only when this process has no current context.
 
@@ -263,15 +227,10 @@ def seed_session_context_if_unbound(
                 expected_username=expected_username,
                 source=source,
                 canonical_repository_root=canonical_repository_root,
-                cohort_id=cohort_id,
-                startup_sha=startup_sha,
-                endpoint=endpoint,
-                config_fingerprint=config_fingerprint,
             )
         return _SESSION_CONTEXT.as_dict()
 
 
-
 def assess_session_context(
     *,
     profile_name: str | None,
@@ -627,10 +586,6 @@ def mutation_context_audit_fields(
             "session_identity": None,
             "session_repository": None,
             "session_org": None,
-            "session_cohort_id": None,
-            "session_startup_sha": None,
-            "session_endpoint": None,
-            "session_config_fingerprint": None,
         }
     return {
         "session_context_bound": True,
@@ -643,14 +598,9 @@ def mutation_context_audit_fields(
         "session_role_kind": data.get("role_kind"),
         "session_context_source": data.get("source"),
         "session_canonical_repository_root": data.get("canonical_repository_root"),
-        "session_cohort_id": data.get("cohort_id"),
-        "session_startup_sha": data.get("startup_sha"),
-        "session_endpoint": data.get("endpoint"),
-        "session_config_fingerprint": data.get("config_fingerprint"),
     }
 
 
-
 def _assessment(
     proven: bool, reasons: list[str], ctx: Mapping[str, Any] | None
 ) -> dict[str, Any]:

tests/conftest.py

@@ -41,6 +41,7 @@ def _reset_mutation_authority(monkeypatch):
         "GITEA_REVIEWER_WORKTREE",
         "GITEA_MERGER_WORKTREE",
         "GITEA_RECONCILER_WORKTREE",
+        "GITEA_MCP_NAMESPACE",
     ]:
         monkeypatch.delenv(env_key, raising=False)
 
@@ -115,6 +116,7 @@ def _reset_mutation_authority(monkeypatch):
         monkeypatch.setattr(mcp_server, "_ACTOR_IDENTITY_CACHE", {})
     monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
     monkeypatch.setattr(mcp_server, "_LIVE_NAMESPACE_HEALTH", {})
+    monkeypatch.setattr(mcp_server, "_PROFILE_SWITCH_INVALIDATION", None)
     monkeypatch.setattr(mcp_server, "_preflight_whoami_called", False)
     monkeypatch.setattr(mcp_server, "_preflight_capability_called", False)
     monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)

tests/test_issue_689_deterministic_cohort_attachment.py

@@ -1,253 +0,0 @@
-"""Tests for Issue #689: Deterministic MCP namespace attachment.
-
-Verifies cohort identity exposure, stale cohort refusal, parity matching,
-reconcile_id freshness, session context cleanup, and regression scenarios.
-"""
-
-from __future__ import annotations
-
-import os
-import unittest
-from unittest.mock import patch
-
-import master_parity_gate
-import mcp_namespace_health
-import post_restart_reconcile
-import session_context_binding as session_ctx
-
-
-class TestIssue689DeterministicCohortAttachment(unittest.TestCase):
-    """Suite covering Issue #689 acceptance criteria."""
-
-    def setUp(self) -> None:
-        session_ctx._reset_session_context_for_testing()
-
-    def tearDown(self) -> None:
-        session_ctx._reset_session_context_for_testing()
-
-    def test_ac1_session_context_exposes_cohort_identity(self) -> None:
-        """AC1: Session context exposes cohort ID, startup SHA, endpoint, and config fingerprint."""
-        ctx = session_ctx.bind_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            repository="Gitea-Tools",
-            org="Scaled-Tech-Consulting",
-            role_kind="author",
-            cohort_id="cohort-p1234-abc123456789",
-            startup_sha="abc123456789def",
-            endpoint="gitea.prgs.cc",
-            config_fingerprint="fingerprint12345",
-        )
-        self.assertEqual(ctx["cohort_id"], "cohort-p1234-abc123456789")
-        self.assertEqual(ctx["startup_sha"], "abc123456789def")
-        self.assertEqual(ctx["endpoint"], "gitea.prgs.cc")
-        self.assertEqual(ctx["config_fingerprint"], "fingerprint12345")
-
-        fetched = session_ctx.get_session_context()
-        self.assertIsNotNone(fetched)
-        self.assertEqual(fetched["cohort_id"], "cohort-p1234-abc123456789")
-        self.assertEqual(fetched["startup_sha"], "abc123456789def")
-
-    def test_ac2_stale_cohort_refused_by_probe_classifier(self) -> None:
-        """AC2: Probe classifier refuses binding to a stale cohort as stale_cohort_refused."""
-        probe_res = {
-            "success": True,
-            "cohort": {
-                "cohort_id": "cohort-obsolete-1",
-                "startup_sha": "22698c1000000000000000000000000000000000",
-                "endpoint": "gitea.prgs.cc",
-                "config_fingerprint": "fp-old",
-            },
-        }
-        res = mcp_namespace_health.classify_namespace_probe(
-            "gitea-author",
-            probe_result=probe_res,
-            probe_source="client_namespace",
-            expected_parity_sha="a4c73766f4b0cc32f7c3808688eceeb6fee74335",
-        )
-        self.assertFalse(res["healthy"])
-        self.assertFalse(res["success"])
-        self.assertEqual(res["error_type"], "stale_cohort_refused")
-        self.assertIn("stale cohort refused", " ".join(res["reasons"]))
-        self.assertEqual(
-            res["diagnostics"]["startup_sha"],
-            "22698c1000000000000000000000000000000000",
-        )
-
-    def test_ac3_reconnection_parity_matching_and_fail_closed(self) -> None:
-        """AC3: Parity gate fails closed when bound cohort startup SHA mismatches parity."""
-        startup = {"startup_head": "a4c73766f4b0cc32f7c3808688eceeb6fee74335"}
-        current = "a4c73766f4b0cc32f7c3808688eceeb6fee74335"
-        live_remote = "a4c73766f4b0cc32f7c3808688eceeb6fee74335"
-
-        # Matching cohort
-        matching_cohort = {
-            "cohort_id": "cohort-fresh",
-            "startup_sha": "a4c73766f4b0cc32f7c3808688eceeb6fee74335",
-        }
-        res_matching = master_parity_gate.assess_master_parity(
-            startup, current, live_remote_head=live_remote, bound_cohort=matching_cohort
-        )
-        self.assertTrue(res_matching["cohort_parity_match"])
-        self.assertFalse(res_matching["cohort_stale"])
-        self.assertTrue(res_matching["mutation_safe"])
-
-        # Mismatched obsolete cohort
-        obsolete_cohort = {
-            "cohort_id": "cohort-obsolete-22698c1",
-            "startup_sha": "22698c1000000000000000000000000000000000",
-        }
-        res_stale = master_parity_gate.assess_master_parity(
-            startup, current, live_remote_head=live_remote, bound_cohort=obsolete_cohort
-        )
-        self.assertFalse(res_stale["cohort_parity_match"])
-        self.assertTrue(res_stale["cohort_stale"])
-        self.assertTrue(res_stale["restart_required"])
-        self.assertFalse(res_stale["mutation_safe"])
-
-    def test_ac4_reconcile_id_freshness(self) -> None:
-        """AC4: Re-attachment distinguishes new reconcile_id from preserved binding."""
-        inventory = {"inventory_complete": True}
-
-        # New attachment generates fresh reconcile_id
-        proof1 = post_restart_reconcile.reconcile_after_restart(inventory)
-        self.assertFalse(proof1.binding_unchanged)
-        self.assertTrue(proof1.reconcile_id.startswith("reconcile-"))
-
-        # Preserved binding reports binding_unchanged=True
-        proof2 = post_restart_reconcile.reconcile_after_restart(
-            inventory,
-            reconcile_id=proof1.reconcile_id,
-            prior_reconcile_id=proof1.reconcile_id,
-        )
-        self.assertTrue(proof2.binding_unchanged)
-        self.assertEqual(proof2.reconcile_id, proof1.reconcile_id)
-
-        # Disconnected re-attachment gets new reconcile_id
-        proof3 = post_restart_reconcile.reconcile_after_restart(
-            inventory,
-            prior_reconcile_id=proof1.reconcile_id,
-        )
-        self.assertFalse(proof3.binding_unchanged)
-        self.assertNotEqual(proof3.reconcile_id, proof1.reconcile_id)
-
-    def test_ac5_session_disconnect_clears_bindings(self) -> None:
-        """AC5: clear_session_context purges session context on disconnect."""
-        session_ctx.bind_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            cohort_id="cohort-1",
-        )
-        self.assertIsNotNone(session_ctx.get_session_context())
-
-        session_ctx.clear_session_context()
-        self.assertIsNone(session_ctx.get_session_context())
-
-    def test_ac6_bound_cohort_in_diagnostics(self) -> None:
-        """AC6: Bound cohort identity appears in audit diagnostics."""
-        session_ctx.bind_session_context(
-            profile_name="prgs-author",
-            remote="prgs",
-            host="gitea.prgs.cc",
-            identity="jcwalker3",
-            cohort_id="cohort-test-99",
-            startup_sha="sha99999",
-            endpoint="gitea.prgs.cc",
-            config_fingerprint="fp999",
-        )
-        audit = session_ctx.mutation_context_audit_fields()
-        self.assertTrue(audit["session_context_bound"])
-        self.assertEqual(audit["session_cohort_id"], "cohort-test-99")
-        self.assertEqual(audit["session_startup_sha"], "sha99999")
-        self.assertEqual(audit["session_endpoint"], "gitea.prgs.cc")
-        self.assertEqual(audit["session_config_fingerprint"], "fp999")
-
-    def test_ac7_regression_n_reconnects_never_bind_to_obsolete_daemon(self) -> None:
-        """AC7: N reconnects against a daemon set containing obsolete daemons never bind obsolete ones."""
-        live_master = "master-head-latest-12345"
-        daemons = [
-            {"id": "d1", "startup_sha": "obsolete-head-11111"},
-            {"id": "d2", "startup_sha": "obsolete-head-22698c1"},
-            {"id": "d3", "startup_sha": live_master},
-            {"id": "d4", "startup_sha": "obsolete-head-33333"},
-        ]
-
-        for _ in range(5):
-            for daemon in daemons:
-                res = master_parity_gate.assess_master_parity(
-                    {"startup_head": live_master},
-                    live_master,
-                    live_remote_head=live_master,
-                    bound_cohort=daemon,
-                )
-                if daemon["startup_sha"] != live_master:
-                    self.assertFalse(res["mutation_safe"])
-                    self.assertTrue(res["cohort_stale"])
-                else:
-                    self.assertTrue(res["mutation_safe"])
-                    self.assertFalse(res["cohort_stale"])
-
-    def test_ac8_regression_incident_shape_reproduction(self) -> None:
-        """AC8: Reproduce incident shape — obsolete cohort 22698c1 resident vs newer daemon."""
-        live_master = "a4c73766f4b0cc32f7c3808688eceeb6fee74335"
-        obsolete_cohort = {
-            "cohort_id": "cohort-resident-22698c1",
-            "startup_sha": "22698c1000000000000000000000000000000000",
-        }
-        new_cohort = {
-            "cohort_id": "cohort-spawned-new",
-            "startup_sha": live_master,
-        }
-
-        # Obsolete cohort fails parity check
-        obs_res = mcp_namespace_health.classify_namespace_probe(
-            "gitea-author",
-            probe_result={"success": True, "cohort": obsolete_cohort},
-            probe_source="client_namespace",
-            expected_parity_sha=live_master,
-        )
-        self.assertFalse(obs_res["healthy"])
-        self.assertEqual(obs_res["error_type"], "stale_cohort_refused")
-
-        # Fresh cohort succeeds
-        new_res = mcp_namespace_health.classify_namespace_probe(
-            "gitea-author",
-            probe_result={"success": True, "cohort": new_cohort},
-            probe_source="client_namespace",
-            expected_parity_sha=live_master,
-        )
-        self.assertTrue(new_res["healthy"])
-
-    def test_ac9_regression_bound_cohort_going_stale_detected(self) -> None:
-        """AC9: A bound cohort that later goes stale is detected on next attachment check."""
-        initial_master = "sha-v1-initial"
-        cohort = {"cohort_id": "c1", "startup_sha": initial_master}
-
-        # Initial state: in parity
-        res1 = master_parity_gate.assess_master_parity(
-            {"startup_head": initial_master},
-            initial_master,
-            live_remote_head=initial_master,
-            bound_cohort=cohort,
-        )
-        self.assertTrue(res1["mutation_safe"])
-
-        # Master advances to sha-v2-advanced while cohort remains at sha-v1-initial
-        advanced_master = "sha-v2-advanced"
-        res2 = master_parity_gate.assess_master_parity(
-            {"startup_head": initial_master},
-            advanced_master,
-            live_remote_head=advanced_master,
-            bound_cohort=cohort,
-        )
-        self.assertFalse(res2["mutation_safe"])
-        self.assertTrue(res2["restart_required"])
-        self.assertTrue(res2["cohort_stale"])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_issue_690_profile_switch_review_guard.py

@@ -0,0 +1,274 @@
+"""Regression coverage for #690: cross-role profile activation invalidation.
+
+A mid-run profile switch (e.g. reviewer → author → reviewer) must invalidate
+workflow-load proof, reviewer lease binding, review decision lock, live
+namespace health, and preflight identity/capability stamps, and must require
+a full reviewer preflight before any formal verdict. Namespace provenance
+must be reported and fail closed on task/namespace mismatch.
+"""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_config
+import mcp_namespace_health
+import mcp_server
+import mcp_session_state
+import review_workflow_load
+import reviewer_pr_lease
+
+from tests.test_runtime_clarity import CONFIG_SWITCHING_ENABLED
+
+
+class TestProfileSwitchReviewGuard(unittest.TestCase):
+    def setUp(self):
+        self._remotes_patch = patch.dict(mcp_server.REMOTES, {
+            "dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
+        })
+        self._remotes_patch.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        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_SWITCHING_ENABLED))
+
+    def tearDown(self):
+        self._remotes_patch.stop()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir.cleanup()
+
+    def _env(self, profile="reviewer-profile"):
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+            "GITEA_TOKEN_MERGER": "merger-pass",
+        }
+
+    def _seed_contaminated_review_state(self):
+        """Simulate an in-flight reviewer run under reviewer-profile."""
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "reviewer"
+        mcp_server._preflight_resolved_task = "review_pr"
+        review_workflow_load._REVIEW_WORKFLOW_LOAD = {"loaded": True}
+        mcp_server._REVIEW_DECISION_LOCK = {
+            "session_profile": "reviewer-profile",
+            "final_review_decision_ready": True,
+            "ready_pr_number": 688,
+        }
+        reviewer_pr_lease.record_session_lease(
+            {"session_id": "lease-session-1", "pr_number": 688}
+        )
+        mcp_server._LIVE_NAMESPACE_HEALTH["gitea-reviewer"] = {
+            "namespace": "gitea-reviewer",
+            "healthy": True,
+            "ide_namespace_proven": True,
+        }
+        # Durable records keyed by the reviewer identity must also be cleared.
+        mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+            payload={"loaded": True},
+            profile_identity="reviewer-profile",
+        )
+        mcp_session_state.save_state(
+            kind=mcp_session_state.KIND_DECISION_LOCK,
+            payload={"final_review_decision_ready": True, "ready_pr_number": 688},
+            profile_identity="reviewer-profile",
+        )
+
+    def _activate(self, target, logins):
+        with patch.object(
+            mcp_server, "get_auth_header", side_effect=[f"token p" for _ in logins]
+        ), patch.object(
+            mcp_server, "api_request", side_effect=[{"login": l} for l in logins]
+        ), patch.object(
+            mcp_server,
+            "_workspace_repository_slug",
+            return_value="Example-Org/Example-Repo",
+        ), patch.object(
+            mcp_server, "_canonical_repository_slug", return_value=(None, [])
+        ):
+            return mcp_server.gitea_activate_profile(profile_name=target)
+
+    # -----------------------------------------------------------------
+    # AC1/AC2/AC3: switch invalidates review state; re-preflight required
+    # -----------------------------------------------------------------
+    def test_switch_invalidates_review_state_and_blocks_verdict(self):
+        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
+            self._seed_contaminated_review_state()
+            res = self._activate("author-profile", ["reviewer-user", "author-user"])
+
+            self.assertTrue(res["success"])
+            self.assertTrue(res["re_preflight_required"])
+            inv = res["review_state_invalidation"]
+            self.assertEqual(inv["from_profile"], "reviewer-profile")
+            self.assertEqual(inv["to_profile"], "author-profile")
+            for item in (
+                "preflight_identity_capability",
+                "review_workflow_load",
+                "review_decision_lock",
+                "reviewer_session_lease",
+                "live_namespace_health",
+            ):
+                self.assertIn(item, inv["invalidated"])
+
+            # In-memory state cleared.
+            self.assertFalse(mcp_server._preflight_whoami_called)
+            self.assertFalse(mcp_server._preflight_capability_called)
+            self.assertIsNone(mcp_server._preflight_resolved_task)
+            self.assertIsNone(review_workflow_load._REVIEW_WORKFLOW_LOAD)
+            self.assertIsNone(mcp_server._REVIEW_DECISION_LOCK)
+            self.assertIsNone(reviewer_pr_lease.get_session_lease())
+            self.assertEqual(mcp_server._LIVE_NAMESPACE_HEALTH, {})
+            self.assertIsNotNone(mcp_server._PROFILE_SWITCH_INVALIDATION)
+
+            # Durable records keyed by the reviewer identity are gone.
+            self.assertIsNone(
+                mcp_session_state.load_state(
+                    kind=mcp_session_state.KIND_WORKFLOW_LOAD,
+                    profile_identity="reviewer-profile",
+                )
+            )
+            self.assertIsNone(
+                mcp_session_state.load_state(
+                    kind=mcp_session_state.KIND_DECISION_LOCK,
+                    profile_identity="reviewer-profile",
+                )
+            )
+
+            # A formal verdict without re-preflight fails closed.
+            reasons = mcp_server.check_review_decision_gate(
+                688, "APPROVE", final_review_decision_ready=True
+            )
+            self.assertTrue(reasons)
+
+    def test_switch_back_cannot_resurrect_stale_review_run(self):
+        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
+            self._seed_contaminated_review_state()
+            self._activate("author-profile", ["reviewer-user", "author-user"])
+            res = self._activate("reviewer-profile", ["author-user", "reviewer-user"])
+
+            self.assertTrue(res["success"])
+            # The pre-switch review run must not reappear.
+            self.assertIsNone(mcp_server._REVIEW_DECISION_LOCK)
+            self.assertIsNone(review_workflow_load._REVIEW_WORKFLOW_LOAD)
+            self.assertIsNone(reviewer_pr_lease.get_session_lease())
+            status = review_workflow_load.workflow_load_status()
+            self.assertFalse(status["workflow_load_valid"])
+            reasons = mcp_server.check_review_decision_gate(
+                688, "APPROVE", final_review_decision_ready=True
+            )
+            self.assertTrue(reasons)
+
+    def test_same_profile_reactivation_keeps_state(self):
+        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
+            self._seed_contaminated_review_state()
+            res = self._activate("reviewer-profile", ["reviewer-user", "reviewer-user"])
+            self.assertTrue(res["success"], res)
+            self.assertNotIn("review_state_invalidation", res)
+            self.assertIsNotNone(mcp_server._REVIEW_DECISION_LOCK)
+            self.assertTrue(mcp_server._preflight_whoami_called)
+
+    def test_clean_repreflight_after_switch_allows_gate(self):
+        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
+            self._seed_contaminated_review_state()
+            self._activate("author-profile", ["reviewer-user", "author-user"])
+            self._activate("reviewer-profile", ["author-user", "reviewer-user"])
+
+            # Re-establish the full reviewer preflight under the new profile.
+            mcp_server.record_preflight_check("whoami")
+            mcp_server.record_preflight_check(
+                "capability", resolved_role="reviewer", resolved_task="review_pr"
+            )
+            mcp_server.init_review_decision_lock("dadeschools", "review_pr")
+            lock = mcp_server._load_review_decision_lock()
+            self.assertIsNotNone(lock)
+            lock.update(
+                {
+                    "final_review_decision_ready": True,
+                    "ready_pr_number": 688,
+                    "ready_action": "APPROVE",
+                    "ready_remote": "dadeschools",
+                    "ready_org": "Example-Org",
+                    "ready_repo": "Example-Repo",
+                }
+            )
+            mcp_server._save_review_decision_lock(lock)
+
+            with patch.object(
+                mcp_server, "_review_workflow_load_gate_reasons", return_value=[]
+            ):
+                reasons = mcp_server.check_review_decision_gate(
+                    688,
+                    "APPROVE",
+                    final_review_decision_ready=True,
+                    remote="dadeschools",
+                )
+            self.assertEqual(reasons, [])
+
+    # -----------------------------------------------------------------
+    # AC4: namespace provenance reporting + fail-closed mismatch
+    # -----------------------------------------------------------------
+    def test_namespace_provenance_mismatch_detection(self):
+        prov = mcp_namespace_health.namespace_provenance(
+            task="review_pr",
+            active_profile="reviewer-profile",
+            env={"GITEA_MCP_NAMESPACE": "gitea-author"},
+        )
+        self.assertTrue(prov["mismatch"])
+        self.assertEqual(prov["required_namespace"], "gitea-reviewer")
+
+        prov_ok = mcp_namespace_health.namespace_provenance(
+            task="review_pr",
+            active_profile="reviewer-profile",
+            env={"GITEA_MCP_NAMESPACE": "gitea-reviewer"},
+        )
+        self.assertFalse(prov_ok["mismatch"])
+
+        prov_unknown = mcp_namespace_health.namespace_provenance(
+            task="review_pr", active_profile="reviewer-profile", env={}
+        )
+        self.assertIsNone(prov_unknown["configured_namespace"])
+        self.assertFalse(prov_unknown["mismatch"])
+        self.assertEqual(prov_unknown["namespace_source"], "unknown")
+
+    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_whoami_reports_namespace_provenance(self, _auth, _api):
+        env = self._env("reviewer-profile")
+        env["GITEA_MCP_NAMESPACE"] = "gitea-reviewer"
+        with patch.dict(os.environ, env, clear=True):
+            res = mcp_server.gitea_whoami(remote="dadeschools")
+        prov = res["namespace_provenance"]
+        self.assertEqual(prov["configured_namespace"], "gitea-reviewer")
+        self.assertEqual(prov["active_profile"], "reviewer-profile")
+        self.assertFalse(prov["mismatch"])
+
+    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_resolve_fails_closed_on_namespace_mismatch(self, _auth, _api):
+        env = self._env("reviewer-profile")
+        env["GITEA_MCP_NAMESPACE"] = "gitea-author"
+        with patch.dict(os.environ, env, clear=True):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="review_pr", kwargs="{}", remote="dadeschools"
+            )
+        self.assertFalse(res["allowed_in_current_session"])
+        self.assertTrue(res["namespace_provenance"]["mismatch"])
+        self.assertTrue(
+            any("namespace" in g for g in res["task_role_guidance"])
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()