fix: preserve capability preflight across interleaved read-only whoami (#469)

Read-only gitea_whoami calls no longer clear a valid capability proof when whoami was already verified clean. Capability is consumed on mutation (one resolve per mutation), and verify_preflight_purity rejects task mismatches. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-08 02:26:02 -04:00
parent 7af73e1539
commit 514eae84f9
2 changed files with 190 additions and 30 deletions
@@ -165,6 +165,7 @@ _preflight_capability_called = False
 _preflight_whoami_violation = False
 _preflight_capability_violation = False
 _preflight_resolved_role = None
+_preflight_resolved_task: str | None = None
 _process_start_porcelain: str | None = None
 _preflight_whoami_baseline_porcelain: str | None = None
 _preflight_capability_baseline_porcelain: str | None = None
@@ -379,11 +380,30 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
    }


-def record_preflight_check(type_name: str, resolved_role: str | None = None):
+def _clear_preflight_capability_state() -> None:
+    """Drop resolved capability proof (consumed by a mutation or fresh resolve)."""
+    global _preflight_capability_called, _preflight_capability_violation
+    global _preflight_resolved_task
+    global _preflight_capability_baseline_porcelain, _preflight_capability_violation_files
+    global _preflight_reviewer_violation_files
+
+    _preflight_capability_called = False
+    _preflight_capability_violation = False
+    _preflight_capability_violation_files = []
+    _preflight_capability_baseline_porcelain = None
+    _preflight_resolved_task = None
+    _preflight_reviewer_violation_files = []
+
+
+def record_preflight_check(
+    type_name: str,
+    resolved_role: str | None = None,
+    resolved_task: str | None = None,
+):
    """Record a pre-flight check (whoami or capability) with session-scoped deltas."""
    global _preflight_whoami_called, _preflight_capability_called
    global _preflight_whoami_violation, _preflight_capability_violation
-    global _preflight_resolved_role
+    global _preflight_resolved_role, _preflight_resolved_task
    global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
    global _preflight_whoami_violation_files, _preflight_capability_violation_files
    global _preflight_reviewer_violation_files
@@ -391,13 +411,24 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
    current = _get_workspace_porcelain()

    if type_name == "whoami":
-        # Fresh whoami restarts the capability step and re-evaluates violations
-        # instead of replaying a sticky record (#252).
-        _preflight_capability_called = False
-        _preflight_capability_violation = False
-        _preflight_capability_violation_files = []
-        _preflight_capability_baseline_porcelain = None
-        _preflight_reviewer_violation_files = []
+        # Re-evaluate whoami violations instead of replaying sticky state (#252).
+        # Interleaved read-only whoami must not clear a valid capability proof (#469).
+        saved_capability = None
+        preserve_capability = (
+            _preflight_capability_called
+            and _preflight_whoami_called
+            and not _preflight_whoami_violation
+        )
+        if preserve_capability:
+            saved_capability = (
+                _preflight_capability_violation,
+                list(_preflight_capability_violation_files),
+                _preflight_capability_baseline_porcelain,
+                _preflight_resolved_role,
+                _preflight_resolved_task,
+            )
+        else:
+            _clear_preflight_capability_state()

        process_start = _ensure_process_start_porcelain()
        whoami_delta = _new_tracked_changes_since(process_start, current)
@@ -405,6 +436,20 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
        _preflight_whoami_violation_files = whoami_delta
        _preflight_whoami_baseline_porcelain = current
        _preflight_whoami_called = True
+
+        if (
+            preserve_capability
+            and saved_capability is not None
+            and not whoami_delta
+        ):
+            (
+                _preflight_capability_violation,
+                _preflight_capability_violation_files,
+                _preflight_capability_baseline_porcelain,
+                _preflight_resolved_role,
+                _preflight_resolved_task,
+            ) = saved_capability
+            _preflight_capability_called = True
    elif type_name == "capability":
        baseline = _preflight_whoami_baseline_porcelain or ""
        capability_delta = _new_tracked_changes_since(baseline, current)
@@ -414,6 +459,8 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
        _preflight_capability_called = True
        if resolved_role:
            _preflight_resolved_role = resolved_role
+        if resolved_task:
+            _preflight_resolved_task = resolved_task


 def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
@@ -434,7 +481,11 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
        )


-def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
+def verify_preflight_purity(
+    remote: str | None = None,
+    worktree_path: str | None = None,
+    task: str | None = None,
+):
    """Verify that identity and capability were verified prior to session edits."""
    global _preflight_reviewer_violation_files

@@ -453,6 +504,16 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
        raise RuntimeError(
            "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
        )
+    if (
+        task is not None
+        and _preflight_resolved_task is not None
+        and task != _preflight_resolved_task
+    ):
+        raise RuntimeError(
+            "Pre-flight task mismatch: "
+            f"resolved '{_preflight_resolved_task}' but mutation requires "
+            f"'{task}' (fail closed)"
+        )

    ctx = _resolve_author_mutation_context(worktree_path)
    workspace = ctx["workspace_path"]
@@ -508,6 +569,7 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
                )

    _enforce_branches_only_author_mutation(worktree_path)
+    _clear_preflight_capability_state()

 from mcp.server.fastmcp import FastMCP  # noqa: E402

@@ -1222,7 +1284,7 @@ def gitea_create_issue(
    )
    if blocked:
        return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
    base = repo_api_url(h, o, r)
    open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
    closed_issues = api_get_all(
@@ -1310,7 +1372,7 @@ def gitea_lock_issue(
        raise RuntimeError(active_lease_block)

    git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
-    verify_preflight_purity(remote, worktree_path=resolved_worktree)
+    verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
    lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
        worktree_path=resolved_worktree,
        current_branch=git_state.get("current_branch"),
@@ -1517,7 +1579,7 @@ def gitea_create_pr(
    )
    if blocked:
        return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
    h, o, r = _resolve(remote, host, org, repo)

    # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
@@ -2560,7 +2622,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)
+    verify_preflight_purity(remote, task="review_pr")
    action = (action or "").strip().lower()
    result = {
        "requested_action": action,
@@ -3067,12 +3129,12 @@ 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)
+    closing = payload.get("state") == "closed"
+    verify_preflight_purity(remote, task="close_pr" if closing else None)

    # 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.
-    closing = payload.get("state") == "closed"
    if closing:
        gate_reasons = _profile_operation_gate("gitea.pr.close")
        if gate_reasons:
@@ -3317,7 +3379,7 @@ def gitea_commit_files(
            branch="",
        )

-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="commit_files")
    processed_files, source_proofs = _prepare_commit_payload_files(files)

    h, o, r = _resolve(remote, host, org, repo)
@@ -3418,7 +3480,7 @@ def gitea_merge_pr(
        reasons/gates passed or blocked, and merge result / merge commit if
        available. Never secrets.
    """
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="merge_pr")
    do = (do or "").strip().lower()
    result = {
        "performed": False,
@@ -3947,7 +4009,7 @@ def gitea_delete_branch(
            "permission_report": _permission_block_report("gitea.branch.delete"),
        }

-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="delete_branch")
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    import urllib.parse
@@ -4056,7 +4118,7 @@ def gitea_reconcile_merged_cleanups(
        report["executed"] = False
        return {"success": True, "performed": False, **report}

-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="reconcile_merged_cleanups")
    actions: list[dict] = []
    for entry in report.get("entries") or []:
        head_branch = entry.get("head_branch") or ""
@@ -4370,7 +4432,7 @@ def gitea_reconcile_already_landed_pr(
        )
        return result

-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="reconcile_already_landed_pr")

    if post_comment and comment_body.strip():
        comment_block = _profile_operation_gate("gitea.pr.comment")
@@ -4473,7 +4535,7 @@ def gitea_close_issue(
        task_capability_map.required_permission("close_issue"))
    if blocked:
        return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="close_issue")
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -4901,7 +4963,7 @@ def gitea_acquire_reviewer_pr_lease(
            "permission_report": _permission_block_report("gitea.pr.comment"),
        }

-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="review_pr")
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    profile = get_profile()
@@ -5002,7 +5064,7 @@ def gitea_heartbeat_reviewer_pr_lease(
            ],
        }

-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="review_pr")
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    body = reviewer_pr_lease.format_lease_body(
@@ -5173,7 +5235,7 @@ def gitea_create_issue_comment(
        (permission blocks also carry a structured 'permission_report',
        #142).
    """
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="comment_issue")
    gate_reasons = _profile_operation_gate("gitea.issue.comment")
    reasons = list(gate_reasons)
    if not (body or "").strip():
@@ -6707,7 +6769,7 @@ def gitea_mark_issue(
        task_capability_map.required_permission("mark_issue"))
    if blocked:
        return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    base = repo_api_url(h, o, r)
@@ -6785,7 +6847,7 @@ def gitea_post_heartbeat(
        task_capability_map.required_permission("post_heartbeat"))
    if blocked:
        return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="post_heartbeat")
    active_profile = profile or get_profile().get("profile_name")
    body = issue_claim_heartbeat.format_heartbeat_body(
        kind="progress",
@@ -7028,7 +7090,7 @@ def gitea_cleanup_stale_claims(
        task_capability_map.required_permission("cleanup_stale_claims"))
    if blocked:
        return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="cleanup_stale_claims")

    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
@@ -7182,7 +7244,7 @@ def gitea_set_issue_labels(
        task_capability_map.required_permission("set_issue_labels"))
    if blocked:
        return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="set_issue_labels")
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    base = repo_api_url(h, o, r)
@@ -7419,7 +7481,7 @@ def gitea_resolve_task_capability(
                "exact_safe_next_action": next_safe_action,
            }

-    record_preflight_check("capability", required_role)
+    record_preflight_check("capability", required_role, resolved_task=task)

    # Try automatic dispatch switching
    _ensure_matching_profile(required_permission, required_role, remote, host)
@@ -0,0 +1,98 @@
+"""#469: capability preflight survives interleaved read-only whoami calls."""
+
+import os
+import unittest
+
+import gitea_mcp_server as mcp_server
+
+
+class TestPreflightReadSurvival(unittest.TestCase):
+    def setUp(self):
+        self.orig_whoami = mcp_server._preflight_whoami_called
+        self.orig_capability = 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
+        self.orig_resolved_task = mcp_server._preflight_resolved_task
+        self.orig_process_start = mcp_server._process_start_porcelain
+        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
+        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
+        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+            if key in os.environ:
+                del os.environ[key]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server._process_start_porcelain = ""
+
+    def tearDown(self):
+        mcp_server._preflight_whoami_called = self.orig_whoami
+        mcp_server._preflight_capability_called = self.orig_capability
+        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
+        mcp_server._preflight_resolved_task = self.orig_resolved_task
+        mcp_server._process_start_porcelain = self.orig_process_start
+        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
+        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
+        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+            if key in os.environ:
+                del os.environ[key]
+
+    def test_interleaved_whoami_preserves_capability(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reconciler", resolved_task="close_pr"
+        )
+        self.assertTrue(mcp_server._preflight_capability_called)
+        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
+
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_capability_called)
+        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
+
+        mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertFalse(mcp_server._preflight_capability_called)
+
+    def test_missing_capability_still_fails_closed(self):
+        mcp_server.record_preflight_check("whoami")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertIn("has not been resolved", str(ctx.exception))
+
+    def test_task_mismatch_fails_closed(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="author", resolved_task="create_issue"
+        )
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertIn("task mismatch", str(ctx.exception))
+
+    def test_capability_consumed_after_mutation_gate(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="author", resolved_task="create_issue"
+        )
+        mcp_server.verify_preflight_purity(task="create_issue")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="create_issue")
+        self.assertIn("has not been resolved", str(ctx.exception))
+
+    def test_whoami_recovery_after_violation_clears_capability(self):
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_whoami_violation)
+
+        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server.record_preflight_check("whoami")
+        self.assertFalse(mcp_server._preflight_whoami_violation)
+        self.assertFalse(mcp_server._preflight_capability_called)
+
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reviewer", resolved_task="review_pr"
+        )
+        mcp_server.verify_preflight_purity(task="review_pr")
+
+
+if __name__ == "__main__":
+    unittest.main()