From 514eae84f9c773aba09c529952b97b18f4beb127 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 17:47:30 -0400 Subject: [PATCH 1/3] 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) --- gitea_mcp_server.py | 122 +++++++++++++++++++------- tests/test_preflight_read_survival.py | 98 +++++++++++++++++++++ 2 files changed, 190 insertions(+), 30 deletions(-) create mode 100644 tests/test_preflight_read_survival.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 9ea9050..2048ce6 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -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) diff --git a/tests/test_preflight_read_survival.py b/tests/test_preflight_read_survival.py new file mode 100644 index 0000000..26156a2 --- /dev/null +++ b/tests/test_preflight_read_survival.py @@ -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() \ No newline at end of file -- 2.43.7 From 2d4ab4e54ecbc40bcd65fd05170ff4a38b60c710 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 17:49:06 -0400 Subject: [PATCH 2/3] test: reset preflight globals in read-survival setUp (#469) Prevents cross-test leakage that masked the missing-capability fail-closed case. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_preflight_read_survival.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/tests/test_preflight_read_survival.py b/tests/test_preflight_read_survival.py index 26156a2..6da8af6 100644 --- a/tests/test_preflight_read_survival.py +++ b/tests/test_preflight_read_survival.py @@ -21,7 +21,15 @@ class TestPreflightReadSurvival(unittest.TestCase): if key in os.environ: del os.environ[key] os.environ["GITEA_TEST_PORCELAIN"] = "" + 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 + mcp_server._preflight_resolved_task = None mcp_server._process_start_porcelain = "" + mcp_server._preflight_whoami_baseline_porcelain = None + mcp_server._preflight_capability_baseline_porcelain = None def tearDown(self): mcp_server._preflight_whoami_called = self.orig_whoami -- 2.43.7 From a8bcbdbcf2bb90ab047dc7df6083e5187ae3cbad Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 02:28:07 -0400 Subject: [PATCH 3/3] fix: clarify capability preflight lifetime across safe reads (Closes #470) Add preflight_contract helpers with explicit re-resolve error messages, document the read-only tool set and consumption rules, and extend tests for close_pr sequencing and task replacement. Validation: ./venv/bin/python -m pytest tests/test_preflight_read_survival.py tests/test_mcp_server.py::TestPreflightVerification -q Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/llm-workflow-runbooks.md | 19 ++++++++ gitea_mcp_server.py | 13 ++++-- preflight_contract.py | 66 +++++++++++++++++++++++++++ tests/test_preflight_read_survival.py | 55 +++++++++++++++++++++- 4 files changed, 147 insertions(+), 6 deletions(-) create mode 100644 preflight_contract.py diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index 634e22c..9247258 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -459,6 +459,25 @@ Root-level matches are listed in `.gitignore` so they never get committed. `gitea_get_runtime_context` and `gitea_lock_issue` surface **warnings** (not hard blocks) when these artifacts are still present. +## Capability preflight lifetime (#470) + +After `gitea_resolve_task_capability(task=…)` proves the mutation is allowed, +interleaved **read-only** calls preserve that proof until a gated mutation +consumes it: + +- Safe reads: `gitea_whoami`, `gitea_view_pr`, `gitea_view_issue`, `gitea_list_*`, + `gitea_get_runtime_context`, `gitea_check_pr_eligibility`, and related + read-only inventory/eligibility tools (see `preflight_contract.py`). +- Each mutation consumes the proof once; call `gitea_resolve_task_capability` + again immediately before the next mutation on the same task. +- Resolving capability for a **different** task replaces the prior task binding. +- Workspace edits before resolve, profile switches, or a dirty whoami baseline + invalidate proof (fail closed). + +If a mutation fails with “capability has not been resolved” or “task mismatch”, +re-run `gitea_resolve_task_capability(task="")` immediately before +retrying — do not guess or skip the resolve step. + Implementation work and review work must use separate branch folders. For example, an implementation branch might live under `branches/fix-issue-123-example`, while a review branch for the resulting PR diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 2048ce6..8348439 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -502,7 +502,7 @@ def verify_preflight_purity( ) if not _preflight_capability_called: raise RuntimeError( - "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)" + preflight_contract.format_missing_capability_error(task) ) if ( task is not None @@ -510,9 +510,9 @@ def verify_preflight_purity( and task != _preflight_resolved_task ): raise RuntimeError( - "Pre-flight task mismatch: " - f"resolved '{_preflight_resolved_task}' but mutation requires " - f"'{task}' (fail closed)" + preflight_contract.format_task_mismatch_error( + _preflight_resolved_task, task + ) ) ctx = _resolve_author_mutation_context(worktree_path) @@ -600,6 +600,7 @@ import issue_lock_adoption # noqa: E402 import merge_approval_gate # noqa: E402 import already_landed_reconcile # noqa: E402 import author_mutation_worktree # noqa: E402 +import preflight_contract # noqa: E402 import issue_claim_heartbeat # noqa: E402 import issue_work_duplicate_gate # noqa: E402 import reviewer_pr_lease # noqa: E402 @@ -6886,7 +6887,9 @@ def gitea_acquire_conflict_fix_lease( task_capability_map.required_permission("comment_issue")) if blocked: return blocked - verify_preflight_purity(remote, worktree_path=worktree_path) + verify_preflight_purity( + remote, worktree_path=worktree_path, task="comment_issue" + ) comments = _list_pr_lease_comments( pr_number, remote=remote, diff --git a/preflight_contract.py b/preflight_contract.py new file mode 100644 index 0000000..60859c7 --- /dev/null +++ b/preflight_contract.py @@ -0,0 +1,66 @@ +"""Capability preflight lifetime contract (#470). + +Defines which read-only MCP tools preserve an existing capability proof and the +canonical sequencing operators must follow before mutations. +""" + +from __future__ import annotations + +# Read-only tools that must not invalidate capability proof (#470). +# gitea_whoami is read-only but re-pins identity; capability is preserved when +# identity was already verified clean (#469). +READ_ONLY_PREFLIGHT_TOOLS = frozenset({ + "gitea_whoami", + "gitea_get_authenticated_user", + "gitea_get_current_user", + "gitea_view_pr", + "gitea_view_issue", + "gitea_list_prs", + "gitea_list_issues", + "gitea_list_issue_comments", + "gitea_get_runtime_context", + "gitea_resolve_task_capability", + "gitea_check_pr_eligibility", + "gitea_get_pr_review_feedback", + "gitea_assess_work_issue_duplicate", + "gitea_route_task_session", + "gitea_audit_config", + "gitea_list_labels", + "gitea_get_profile", + "gitea_list_profiles", +}) + +PREFLIGHT_CONTRACT_SUMMARY = ( + "Capability preflight is session-scoped per MCP process, profile, and task. " + "After gitea_resolve_task_capability(task=...), interleaved read-only calls " + "(whoami, view_*, list_*, get_runtime_context, eligibility checks) preserve " + "the proof until a gated mutation consumes it. Each mutation consumes the proof " + "once; re-resolve immediately before the next mutation. A new resolve for a " + "different task replaces the prior task binding. Profile/session changes or " + "workspace edits before resolve invalidate proof." +) + + +def format_missing_capability_error(task: str | None = None) -> str: + base = ( + "Pre-flight order violation: Task capability " + "(gitea_resolve_task_capability) has not been resolved (fail closed)" + ) + if task: + return ( + f"{base}. Re-run gitea_resolve_task_capability(task=\"{task}\") " + "immediately before this mutation." + ) + return ( + f"{base}. Re-run gitea_resolve_task_capability for the mutation task " + "immediately before acting." + ) + + +def format_task_mismatch_error(resolved: str, required: str) -> str: + return ( + "Pre-flight task mismatch: " + f"resolved '{resolved}' but mutation requires '{required}' (fail closed). " + f"Re-run gitea_resolve_task_capability(task=\"{required}\") " + "immediately before this mutation." + ) \ No newline at end of file diff --git a/tests/test_preflight_read_survival.py b/tests/test_preflight_read_survival.py index 6da8af6..c422fc6 100644 --- a/tests/test_preflight_read_survival.py +++ b/tests/test_preflight_read_survival.py @@ -1,4 +1,4 @@ -"""#469: capability preflight survives interleaved read-only whoami calls.""" +"""#469/#470: capability preflight lifetime across safe read-only calls.""" import os import unittest @@ -101,6 +101,59 @@ class TestPreflightReadSurvival(unittest.TestCase): ) mcp_server.verify_preflight_purity(task="review_pr") + def test_close_pr_sequence_with_interleaved_reads(self): + """resolve(close_pr) → whoami/view reads → close_pr gate (#470).""" + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check( + "capability", resolved_role="reconciler", resolved_task="close_pr" + ) + # Simulate live-state revalidation between resolve and mutation. + 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_fresh_capability_resolve_replaces_prior_task(self): + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check( + "capability", resolved_role="author", resolved_task="create_issue" + ) + mcp_server.record_preflight_check( + "capability", resolved_role="reconciler", resolved_task="close_pr" + ) + self.assertEqual(mcp_server._preflight_resolved_task, "close_pr") + mcp_server.verify_preflight_purity(task="close_pr") + + def test_missing_capability_error_names_re_resolve(self): + mcp_server.record_preflight_check("whoami") + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity(task="close_pr") + msg = str(ctx.exception) + self.assertIn("gitea_resolve_task_capability", msg) + self.assertIn('task="close_pr"', msg) + + def test_task_mismatch_error_names_re_resolve(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") + msg = str(ctx.exception) + self.assertIn("task mismatch", msg) + self.assertIn('task="close_pr"', msg) + + def test_consumed_capability_error_names_re_resolve(self): + mcp_server.record_preflight_check("whoami") + mcp_server.record_preflight_check( + "capability", resolved_role="reconciler", resolved_task="close_pr" + ) + mcp_server.verify_preflight_purity(task="close_pr") + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_preflight_purity(task="close_pr") + self.assertIn('task="close_pr"', str(ctx.exception)) + if __name__ == "__main__": unittest.main() \ No newline at end of file -- 2.43.7