feat(preflight): block workspace edits before identity/capability verification (Closes #210)

mcp_server.py

@@ -31,6 +31,82 @@ 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
+
+    is_dirty = False
+    in_test = "pytest" in sys.modules or "unittest" in sys.modules
+    if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
+        is_dirty = False
+    else:
+        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)")
+
+    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
@@ -324,6 +400,7 @@ def gitea_create_issue(
     Returns:
         dict with 'number' of the created issue ('url' only with the reveal opt-in).
     """
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/issues"
@@ -366,6 +443,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)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls"
@@ -976,6 +1054,7 @@ def gitea_submit_pr_review(
         authenticated user, profile name, PR author, PR number, head SHA
         checked, and the reasons/gates passed or blocked. Never secrets.
     """
+    verify_preflight_purity(remote)
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -1199,6 +1278,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"
@@ -1293,6 +1373,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,
@@ -2557,6 +2638,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"]
@@ -3377,6 +3459,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()
 

review_proofs.py

@@ -530,6 +530,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",)),
@@ -580,7 +581,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:
@@ -605,10 +606,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 "", ()))
@@ -627,6 +632,19 @@ def assess_controller_handoff(report_text, role=None):
             "reasons": [f"handoff missing required field: {m}"
                         for m in missing],
         }
+
+    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,

tests/test_mcp_server.py

@@ -2296,3 +2296,88 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
             "gitea.issue.comment", reviewer["allowed_operations"],
             reviewer.get("forbidden_operations", []))
         self.assertTrue(ok)
+
+
+# ---------------------------------------------------------------------------
+# 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

@@ -710,6 +710,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",
@@ -796,6 +797,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 TestPRInventoryTrustGate(unittest.TestCase):
     """Issue #194: unit tests for the PR inventory trust gate."""