feat: enforce capability stop terminal mode after reviewer denial (Issue #197)

Enter terminal mode when review_pr/merge_pr capability is denied. Block
reviewer queue tools (list_prs, eligibility checks) and add report-purity
validators for forbidden PR selection, fallback, and empty-queue claims.

Closes #197

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

capability_stop_terminal.py

@@ -0,0 +1,202 @@
+"""Hard-stop terminal mode after reviewer capability denial (#197)."""
+
+from __future__ import annotations
+
+import re
+
+TERMINAL_REPORT_HEADING = (
+    "Cannot perform reviewer task under current profile. "
+    "No reviewer mutations performed."
+)
+
+REVIEWER_CAPABILITY_TASKS = frozenset({
+    "review_pr",
+    "merge_pr",
+    "blind_pr_queue_review",
+    "request_changes_pr",
+    "approve_pr",
+})
+
+BLOCKED_QUEUE_TOOLS = frozenset({
+    "list_prs",
+    "check_pr_eligibility",
+    "view_pr",
+    "submit_pr_review",
+    "dry_run_pr_review",
+    "merge_pr",
+    "review_pr",
+})
+
+_session_terminal: dict | None = None
+
+
+def enter_from_capability_result(capability: dict) -> dict | None:
+    """Enter terminal mode when a reviewer/merge task is denied."""
+    global _session_terminal
+    task = (capability or {}).get("requested_task", "")
+    required_role = (capability or {}).get("required_role_kind")
+    if not capability.get("stop_required"):
+        return None
+    if required_role != "reviewer" and task not in REVIEWER_CAPABILITY_TASKS:
+        return None
+    record = {
+        "active": True,
+        "requested_task": task,
+        "required_role_kind": required_role,
+        "active_profile": capability.get("active_profile"),
+        "active_identity": capability.get("active_identity"),
+        "stop_required": True,
+        "exact_safe_next_action": capability.get("exact_safe_next_action"),
+        "terminal_message": TERMINAL_REPORT_HEADING,
+    }
+    _session_terminal = record
+    return dict(record)
+
+
+def enter_from_route_result(route: dict) -> dict | None:
+    """Enter terminal mode from a role router wrong_role_stop (#206 compat)."""
+    if (route or {}).get("route_result") != "wrong_role_stop":
+        return None
+    if route.get("required_role") != "reviewer":
+        return None
+    return enter_from_capability_result({
+        "requested_task": route.get("task_type"),
+        "required_role_kind": "reviewer",
+        "stop_required": True,
+        "active_profile": route.get("active_profile"),
+        "active_identity": None,
+        "exact_safe_next_action": route.get("message"),
+    })
+
+
+def is_active() -> bool:
+    return bool(_session_terminal and _session_terminal.get("active"))
+
+
+def active_record() -> dict | None:
+    if not is_active():
+        return None
+    return dict(_session_terminal)
+
+
+def clear():
+    global _session_terminal
+    _session_terminal = None
+
+
+def check_reviewer_queue_tool(tool_name: str) -> tuple[bool, list[str]]:
+    """Return (allowed, reasons). False when terminal mode blocks queue work."""
+    if not is_active():
+        return True, []
+    name = (tool_name or "").strip().lower().removeprefix("gitea_")
+    if name in BLOCKED_QUEUE_TOOLS:
+        return False, [
+            TERMINAL_REPORT_HEADING,
+            f"Reviewer queue tool '{tool_name}' is blocked after "
+            "capability denial (fail closed).",
+            "Relaunch a reviewer MCP namespace to perform reviewer work.",
+        ]
+    return True, []
+
+
+def validate_eligibility_wording(text: str) -> tuple[bool, list[str]]:
+    """Reject session-based eligibility reasoning (#197)."""
+    lower = (text or "").lower()
+    violations = []
+    if "not authored by this session" in lower:
+        violations.append(
+            "eligibility must use authenticated account identity, not "
+            "'this session' wording"
+        )
+    if re.search(r"not (?:self-)?authored by (?:the )?session", lower):
+        violations.append("session-based eligibility reasoning is invalid")
+    return (len(violations) == 0), violations
+
+
+def assess_capability_stop_report(
+    report_text: str,
+    *,
+    trust_gate_status: str | None = None,
+    capability_denied: bool = True,
+) -> dict:
+    """Validate final report purity after reviewer capability denial."""
+    text = report_text or ""
+    lower = text.lower()
+    violations = []
+
+    if capability_denied and TERMINAL_REPORT_HEADING.lower() not in lower:
+        violations.append("missing required terminal report heading")
+
+    forbidden_patterns = [
+        ("pr selection", re.compile(
+            r"selected pr|pr #\d+ (?:to review|selected)|eligible pr|"
+            r"next pr to review", re.I)),
+        ("sibling repo inventory", re.compile(
+            r"sibling repo|other repo|mcp-control-plane|gitea-tools and", re.I)),
+        ("author fallback", re.compile(
+            r"rebase conflicted|author-side fallback|have me rebase|"
+            r"implement the fix|push a branch|open a pr for", re.I)),
+        ("invalid session eligibility", re.compile(
+            r"not authored by this session", re.I)),
+    ]
+    for label, pattern in forbidden_patterns:
+        if pattern.search(text):
+            violations.append(f"forbidden after hard stop: {label}")
+
+    empty_queue_patterns = re.compile(
+        r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
+        r"inventory empty",
+        re.I,
+    )
+    if empty_queue_patterns.search(text):
+        if trust_gate_status != "trusted_empty":
+            violations.append(
+                "empty-queue claim after capability stop without "
+                "pr_inventory_trust_gate.status == trusted_empty"
+            )
+
+    ok, elig_violations = validate_eligibility_wording(text)
+    violations.extend(elig_violations)
+
+    if violations:
+        return {
+            "pure": False,
+            "downgraded": True,
+            "violations": violations,
+            "reasons": violations,
+        }
+    return {
+        "pure": True,
+        "downgraded": False,
+        "violations": [],
+        "reasons": [],
+    }
+
+
+def build_terminal_report(capability: dict) -> dict:
+    """Minimal allowed report fields after hard stop."""
+    return {
+        "terminal_mode": True,
+        "heading": TERMINAL_REPORT_HEADING,
+        "authenticated_profile": capability.get("active_profile"),
+        "authenticated_identity": capability.get("active_identity"),
+        "denied_task": capability.get("requested_task"),
+        "required_role_kind": capability.get("required_role_kind"),
+        "stop_required": capability.get("stop_required"),
+        "required_action": capability.get("exact_safe_next_action"),
+        "mutations_performed": False,
+        "allowed_sections": [
+            "authenticated identity/profile",
+            "denied capability result",
+            "reason task cannot proceed",
+            "required reviewer profile/identity",
+            "mutation confirmation (none)",
+        ],
+        "forbidden_sections": [
+            "PR selection",
+            "sibling-repo queue recommendations",
+            "author-side fallback suggestions",
+            "empty-queue claims without trusted_empty",
+            "session-based eligibility wording",
+        ],
+    }

mcp_server.py

@@ -157,6 +157,89 @@ PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
 if PROJECT_ROOT not in sys.path:
     sys.path.insert(0, PROJECT_ROOT)
 
+import json
+
+_preflight_whoami_called = False
+_preflight_capability_called = False
+_preflight_whoami_violation = False
+_preflight_capability_violation = False
+_preflight_resolved_role = None
+
+def record_preflight_check(type_name: str, resolved_role: str | None = None):
+    """Record a pre-flight check (whoami or capability) and check for workspace edits."""
+    global _preflight_whoami_called, _preflight_capability_called
+    global _preflight_whoami_violation, _preflight_capability_violation
+    global _preflight_resolved_role
+
+    in_test = "pytest" in sys.modules or "unittest" in sys.modules
+    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
+        is_dirty = True
+    elif in_test:
+        is_dirty = False
+    else:
+        is_dirty = False
+        try:
+            res = subprocess.run(
+                ["git", "status", "--porcelain"],
+                capture_output=True, text=True, cwd=PROJECT_ROOT
+            )
+            for line in res.stdout.splitlines():
+                if line and not line.startswith("??"):
+                    is_dirty = True
+                    break
+        except Exception:
+            pass
+
+    if is_dirty:
+        if type_name == "whoami" and not _preflight_whoami_called:
+            _preflight_whoami_violation = True
+        if type_name == "capability" and not _preflight_capability_called:
+            _preflight_capability_violation = True
+
+    if type_name == "whoami":
+        _preflight_whoami_called = True
+    elif type_name == "capability":
+        _preflight_capability_called = True
+        if resolved_role:
+            _preflight_resolved_role = resolved_role
+
+def verify_preflight_purity(remote: str | None = None):
+    """Verify that identity and capability were verified prior to edits, and that reviewers made no edits."""
+    in_test = "pytest" in sys.modules or "unittest" in sys.modules
+    if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
+        return
+
+    if not _preflight_whoami_called:
+        raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)")
+    if not _preflight_capability_called:
+        raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)")
+
+    if _preflight_whoami_violation:
+        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)")
+    if _preflight_capability_violation:
+        raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)")
+
+    if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
+        is_dirty = True
+    elif in_test:
+        is_dirty = False
+    else:
+        is_dirty = False
+        try:
+            res = subprocess.run(
+                ["git", "status", "--porcelain"],
+                capture_output=True, text=True, cwd=PROJECT_ROOT
+            )
+            for line in res.stdout.splitlines():
+                if line and not line.startswith("??"):
+                    is_dirty = True
+                    break
+        except Exception:
+            pass
+
+    if _preflight_resolved_role == "reviewer" and is_dirty:
+        raise RuntimeError("Reviewer role violation: Reviewer profile is forbidden from modifying tracked workspace files (fail closed)")
+
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
 from gitea_auth import (  # noqa: E402
@@ -171,6 +254,7 @@ from gitea_auth import (  # noqa: E402
 )
 import gitea_audit  # noqa: E402
 import gitea_config  # noqa: E402
+import capability_stop_terminal  # noqa: E402
 import issue_duplicate_gate  # noqa: E402
 import role_session_router  # noqa: E402
 
@@ -471,6 +555,7 @@ def gitea_create_issue(
             "number": None,
             "reasons": block_reasons,
         }
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -615,6 +700,7 @@ def gitea_create_pr(
     Returns:
         dict with 'number' of the created PR ('url' only with the reveal opt-in).
     """
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196) ──
@@ -690,6 +776,11 @@ def gitea_list_prs(
         'mergeable', 'updated_at' ('url' only with the reveal opt-in).
         The additional 'updated_at' aids stale/conflicting queue detection.
     """
+    allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
+        "list_prs"
+    )
+    if not allowed:
+        raise RuntimeError("; ".join(block_reasons))
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls?state={state}"
@@ -801,6 +892,17 @@ def gitea_check_pr_eligibility(
         'permission_report' (#142).
     """
     action = (action or "").strip().lower()
+    if action in ("review", "approve", "request_changes", "merge"):
+        allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
+            "check_pr_eligibility"
+        )
+        if not allowed:
+            return {
+                "eligible": False,
+                "requested_action": action,
+                "reasons": block_reasons,
+                "terminal_mode": True,
+            }
     profile = get_profile()
     result = {
         "eligible": False,
@@ -1388,6 +1490,7 @@ def _evaluate_pr_review_submission(
     final_review_decision_ready: bool = False,
 ) -> dict:
     """Shared gate chain for live submit and dry-run review tools."""
+    verify_preflight_purity(remote)
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -1763,6 +1866,8 @@ def gitea_edit_pr(
     if not payload:
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
+    verify_preflight_purity(remote)
+
     # PR closure is a first-class capability, distinct from retitling or
     # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
     # never touches the network.
@@ -1880,6 +1985,7 @@ def gitea_commit_files(
     Returns:
         dict with success status and commit/branch information.
     """
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/contents"
@@ -1974,6 +2080,7 @@ def gitea_merge_pr(
         reasons/gates passed or blocked, and merge result / merge commit if
         available. Never secrets.
     """
+    verify_preflight_purity(remote)
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -2355,6 +2462,7 @@ def gitea_delete_branch(
     Returns:
         dict with 'success' and 'message'.
     """
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     import urllib.parse
@@ -2386,6 +2494,7 @@ def gitea_close_issue(
     Returns:
         dict with 'success' boolean and 'message'.
     """
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -2684,6 +2793,7 @@ def gitea_create_issue_comment(
         (permission blocks also carry a structured 'permission_report',
         #142).
     """
+    verify_preflight_purity(remote)
     gate_reasons = _profile_operation_gate("gitea.issue.comment")
     reasons = list(gate_reasons)
     if not (body or "").strip():
@@ -3252,6 +3362,7 @@ def gitea_whoami(
         'email', 'server', 'remote', and 'profile' (safe runtime profile
         metadata: profile_name + allowed_operations; never the token).
     """
+    record_preflight_check("whoami")
     if remote not in REMOTES:
         raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
     h = host or REMOTES[remote]["host"]
@@ -3771,6 +3882,7 @@ def gitea_mark_issue(
     if action not in ("start", "done"):
         raise ValueError(f"action must be 'start' or 'done', got '{action}'")
 
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -3853,6 +3965,7 @@ def gitea_create_label(
     Returns:
         dict containing the created label details.
     """
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -3892,6 +4005,7 @@ def gitea_set_issue_labels(
     Returns:
         list of all labels currently applied to the issue.
     """
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -4162,6 +4276,8 @@ def gitea_resolve_task_capability(
     required_permission = TASK_MAP[task]["permission"]
     required_role = TASK_MAP[task]["role"]
 
+    record_preflight_check("capability", required_role)
+
     profile = get_profile()
     config = gitea_config.load_config()
 
@@ -4269,7 +4385,7 @@ def gitea_resolve_task_capability(
 
     record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
 
-    return {
+    result = {
         "requested_task": task,
         "required_operation_permission": required_permission,
         "required_role_kind": required_role,
@@ -4284,6 +4400,33 @@ def gitea_resolve_task_capability(
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
     }
+    if stop_required:
+        terminal = capability_stop_terminal.enter_from_capability_result(result)
+        if terminal:
+            result["terminal_mode"] = True
+            result["terminal_report"] = (
+                capability_stop_terminal.build_terminal_report(result)
+            )
+    return result
+
+
+@mcp.tool()
+def gitea_capability_stop_terminal_report() -> dict:
+    """Read-only: terminal report template after reviewer capability denial (#197)."""
+    record = capability_stop_terminal.active_record()
+    if not record:
+        return {
+            "terminal_mode": False,
+            "reasons": ["capability stop terminal mode is not active"],
+        }
+    return capability_stop_terminal.build_terminal_report({
+        "requested_task": record.get("requested_task"),
+        "required_role_kind": record.get("required_role_kind"),
+        "active_profile": record.get("active_profile"),
+        "active_identity": record.get("active_identity"),
+        "stop_required": record.get("stop_required"),
+        "exact_safe_next_action": record.get("exact_safe_next_action"),
+    })
 
 
 # ── Entry point ───────────────────────────────────────────────────────────────

review_proofs.py

@@ -822,6 +822,7 @@ HANDOFF_BASE_FIELDS = (
     ("Files changed", ("files changed", "changed", "files")),
     ("Validation", ("validation",)),
     ("Mutations", ("mutations",)),
+    ("Workspace mutations", ("workspace mutations",)),
     ("Current status", ("current status", "status")),
     ("Blockers", ("blockers",)),
     ("Next", ("next",)),
@@ -872,7 +873,7 @@ def _handoff_section_lines(report_text):
     return lines[start:]
 
 
-def assess_controller_handoff(report_text, role=None):
+def assess_controller_handoff(report_text, role=None, local_edits=False):
     """Issue #182: final reports without a Controller Handoff downgrade.
 
     Verdicts:
@@ -897,10 +898,14 @@ def assess_controller_handoff(report_text, role=None):
         }
 
     labels = []
+    fields_dict = {}
     for line in section:
         stripped = line.strip().lstrip("-*").strip()
         if ":" in stripped:
-            labels.append(stripped.split(":", 1)[0].strip().lower())
+            k, v = stripped.split(":", 1)
+            label = k.strip().lower()
+            labels.append(label)
+            fields_dict[label] = v.strip()
 
     required = list(HANDOFF_BASE_FIELDS)
     required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
@@ -919,15 +924,7 @@ def assess_controller_handoff(report_text, role=None):
             "reasons": [f"handoff missing required field: {m}"
                         for m in missing],
         }
-
     # Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
-    fields_dict = {}
-    for line in section:
-        stripped = line.strip().lstrip("-*").strip()
-        if ":" in stripped:
-            k, v = stripped.split(":", 1)
-            fields_dict[k.strip().lower()] = v.strip()
-
     for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
         val = fields_dict.get(alias)
         if val:
@@ -948,6 +945,18 @@ def assess_controller_handoff(report_text, role=None):
                     ],
                 }
 
+    if local_edits:
+        workspace_mutations_val = fields_dict.get("workspace mutations", "").strip().lower()
+        if not workspace_mutations_val or workspace_mutations_val == "none":
+            return {
+                "verdict": "incomplete",
+                "downgraded": True,
+                "missing_fields": ["Workspace mutations"],
+                "reasons": [
+                    "Workspace mutations cannot be 'none' or empty when local edits exist"
+                ],
+            }
+
     return {
         "verdict": "complete",
         "downgraded": False,
@@ -993,6 +1002,22 @@ def assess_role_route_handoff(report_text, route_result=None):
     }
 
 
+def assess_capability_stop_terminal_report(
+    report_text,
+    *,
+    trust_gate_status=None,
+    capability_denied=True,
+):
+    """Issue #197: reports after reviewer capability denial must stay pure."""
+    from capability_stop_terminal import assess_capability_stop_report
+
+    return assess_capability_stop_report(
+        report_text,
+        trust_gate_status=trust_gate_status,
+        capability_denied=capability_denied,
+    )
+
+
 # ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
 #
 # A reviewer agent may not convert an empty PR list response into a definitive

tests/conftest.py

@@ -26,4 +26,14 @@ def _reset_mutation_authority(monkeypatch):
     monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
     monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
     monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
+    try:
+        import capability_stop_terminal
+        capability_stop_terminal.clear()
+    except Exception:
+        pass
     yield
+    try:
+        import capability_stop_terminal
+        capability_stop_terminal.clear()
+    except Exception:
+        pass

tests/test_capability_stop_terminal.py

@@ -0,0 +1,164 @@
+"""Tests for capability stop terminal mode (#197)."""
+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 capability_stop_terminal
+import gitea_config
+import mcp_server
+from review_proofs import assess_capability_stop_terminal_report
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "prgs-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "jcwalker3",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.pr.create",
+                "gitea.branch.push",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
+            ],
+            "execution_profile": "prgs-author",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+
+class TestCapabilityStopTerminal(unittest.TestCase):
+    def setUp(self):
+        capability_stop_terminal.clear()
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.example.com",
+                "org": "Scaled-Tech-Consulting",
+                "repo": "Gitea-Tools",
+            },
+        })
+        self._remotes.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))
+
+    def tearDown(self):
+        self._remotes.stop()
+        capability_stop_terminal.clear()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir.cleanup()
+
+    def _env(self):
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": "prgs-author",
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_review_pr_stop_enters_terminal_mode(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="review_pr", remote="prgs"
+            )
+        self.assertTrue(res["stop_required"])
+        self.assertTrue(res.get("terminal_mode"))
+        self.assertTrue(capability_stop_terminal.is_active())
+        self.assertIn(
+            capability_stop_terminal.TERMINAL_REPORT_HEADING,
+            res["terminal_report"]["heading"],
+        )
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_list_prs_blocked_after_capability_stop(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.gitea_list_prs(remote="prgs")
+        self.assertIn("Cannot perform reviewer task", str(ctx.exception))
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_eligibility_check_blocked_after_stop(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            res = mcp_server.gitea_check_pr_eligibility(
+                pr_number=193, action="review", remote="prgs"
+            )
+        self.assertFalse(res["eligible"])
+        self.assertTrue(res.get("terminal_mode"))
+
+    def test_report_with_pr_selection_impure(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Selected PR #193 for review anyway."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertFalse(result["pure"])
+
+    def test_session_eligibility_wording_blocked(self):
+        ok, violations = capability_stop_terminal.validate_eligibility_wording(
+            "PR 193 is not authored by this session so it is eligible."
+        )
+        self.assertFalse(ok)
+        self.assertTrue(violations)
+
+    def test_rebase_fallback_blocked_in_report(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Or have me rebase conflicted PR 193."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertFalse(result["pure"])
+        self.assertTrue(
+            any("author fallback" in v for v in result["violations"])
+        )
+
+    def test_empty_queue_without_trusted_empty_blocked(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "The repo has 0 open PRs."
+        )
+        result = assess_capability_stop_terminal_report(
+            report, trust_gate_status="untrusted_empty"
+        )
+        self.assertFalse(result["pure"])
+
+    def test_pure_terminal_report_passes(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Identity: jcwalker3 / prgs-author.\n"
+            "Required: prgs-reviewer.\n"
+            "No mutations performed."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertTrue(result["pure"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_server.py

@@ -2813,3 +2813,87 @@ class TestIssueLocking(unittest.TestCase):
         with self.assertRaises(ValueError) as ctx:
             gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
         self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
+
+
+# ---------------------------------------------------------------------------
+# Pre-flight ordering and workspace edit block (#210)
+# ---------------------------------------------------------------------------
+class TestPreflightVerification(unittest.TestCase):
+    """Assert that pre-flight verification gates order correctly."""
+
+    def setUp(self):
+        # Save real global variables
+        import mcp_server
+        self.orig_whoami_called = mcp_server._preflight_whoami_called
+        self.orig_capability_called = mcp_server._preflight_capability_called
+        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
+        self.orig_capability_violation = mcp_server._preflight_capability_violation
+        self.orig_resolved_role = mcp_server._preflight_resolved_role
+
+        # Reset state for each test
+        mcp_server._preflight_whoami_called = False
+        mcp_server._preflight_capability_called = False
+        mcp_server._preflight_whoami_violation = False
+        mcp_server._preflight_capability_violation = False
+        mcp_server._preflight_resolved_role = None
+
+    def tearDown(self):
+        # Restore real global variables
+        import mcp_server
+        mcp_server._preflight_whoami_called = self.orig_whoami_called
+        mcp_server._preflight_capability_called = self.orig_capability_called
+        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
+        mcp_server._preflight_capability_violation = self.orig_capability_violation
+        mcp_server._preflight_resolved_role = self.orig_resolved_role
+        if "GITEA_TEST_FORCE_DIRTY" in os.environ:
+            del os.environ["GITEA_TEST_FORCE_DIRTY"]
+
+    def test_preflight_whoami_violation(self):
+        import mcp_server
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server._preflight_capability_called = True
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_whoami_violation)
+
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity()
+        self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
+
+    def test_preflight_capability_violation(self):
+        import mcp_server
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server._preflight_whoami_called = True
+        mcp_server.record_preflight_check("capability", resolved_role="author")
+        self.assertTrue(mcp_server._preflight_capability_violation)
+
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity()
+        self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception))
+
+    def test_preflight_not_called_fails_closed(self):
+        import mcp_server
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity()
+        self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
+
+        mcp_server._preflight_whoami_called = True
+        with self.assertRaises(RuntimeError) as ctx2:
+            mcp_server.verify_preflight_purity()
+        self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx2.exception))
+
+    def test_preflight_reviewer_mutation_violation(self):
+        import mcp_server
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "reviewer"
+
+        # When dirty, reviewer edits are blocked
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity()
+        self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
+
+        # When clean, reviewer is allowed
+        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        mcp_server.verify_preflight_purity()

tests/test_review_proofs.py

@@ -789,6 +789,7 @@ class TestControllerHandoff(unittest.TestCase):
         "- Files changed: review_proofs.py",
         "- Validation: 700 passed, 6 skipped",
         "- Mutations: one PR opened",
+        "- Workspace mutations: none",
         "- Current status: PR open",
         "- Blockers: none",
         "- Next: review PR #999",
@@ -909,6 +910,29 @@ class TestControllerHandoff(unittest.TestCase):
         self.assertIn("assess_controller_handoff", skill)
         self.assertIn("issue #182", skill)
 
+    def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
+        # 1. Workspace mutations: none is rejected when local_edits is True
+        incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
+            "- Selected issue: #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #203",
+            "- No review/merge: confirmed",
+        ])
+        res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
+        self.assertEqual(res["verdict"], "incomplete")
+        self.assertIn("Workspace mutations", res["missing_fields"])
+
+        # 2. Workspace mutations: edited files is allowed when local_edits is True
+        complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
+            "- Selected issue: #196",
+            "- Claim/comment status: comment-claimed",
+            "- PR number opened: #203",
+            "- No review/merge: confirmed",
+        ])
+        res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
+        self.assertEqual(res2["verdict"], "complete")
+
+
 
 class TestReviewMutationFinalReport(unittest.TestCase):
     """Final reports must list exactly one live review mutation."""