From c7b462a034b7799d8881fe6fd1378a7e866a2697 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 17:36:33 -0400 Subject: [PATCH 1/4] fix: exempt reconciler close_pr from branches-only worktree guard (#468) Reconciler profiles perform Gitea metadata mutations (close_pr) and must not be blocked when the MCP workspace resolves to the stable control checkout without GITEA_AUTHOR_WORKTREE. Author file mutations remain guarded. Add regression tests for reconciler close vs author create_issue. Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 9 +- .../test_reconciler_close_workspace_guard.py | 89 +++++++++++++++++++ 2 files changed, 96 insertions(+), 2 deletions(-) create mode 100644 tests/test_reconciler_close_workspace_guard.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index cd24d8d..8fabf90 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -399,8 +399,13 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None): def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None: - """#274: author mutations must run from a branches/ session worktree.""" - if _preflight_resolved_role == "reviewer": + """#274: author file/branch mutations must run from a branches/ worktree. + + Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a + Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE`` + (#468). + """ + if _preflight_resolved_role in ("reviewer", "reconciler"): return workspace = author_mutation_worktree.resolve_mutation_workspace( worktree_path, diff --git a/tests/test_reconciler_close_workspace_guard.py b/tests/test_reconciler_close_workspace_guard.py new file mode 100644 index 0000000..54a24e4 --- /dev/null +++ b/tests/test_reconciler_close_workspace_guard.py @@ -0,0 +1,89 @@ +"""Reconciler close_pr must not require author branches/ worktree (#468).""" +import os +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import gitea_mcp_server as srv + +FAKE_AUTH = "token test" +CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3]) + +RECONCILER_PROFILE = { + "profile_name": "prgs-reconciler", + "allowed_operations": ["gitea.read", "gitea.pr.close", "gitea.pr.comment"], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.review", + "gitea.pr.create", + "gitea.branch.push", + "gitea.repo.commit", + ], + "audit_label": "prgs-reconciler", +} + + +class TestReconcilerCloseWorkspaceGuard(unittest.TestCase): + def setUp(self): + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_whoami_violation = False + srv._preflight_capability_violation = False + self._orig_in_test = srv._preflight_in_test_mode + srv._preflight_in_test_mode = lambda: False + + def tearDown(self): + srv._preflight_in_test_mode = self._orig_in_test + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE) + @patch("gitea_mcp_server.api_request") + def test_reconciler_close_pr_from_control_checkout_succeeds( + self, mock_api, _profile, _ns, _auth + ): + srv._preflight_resolved_role = "reconciler" + mock_api.return_value = { + "number": 414, + "title": "old", + "body": "", + "state": "closed", + "html_url": "https://gitea.example.com/pulls/414", + } + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("GITEA_AUTHOR_WORKTREE", None) + os.environ.pop("GITEA_ACTIVE_WORKTREE", None) + result = srv.gitea_edit_pr(414, state="closed", remote="prgs") + self.assertTrue(result["success"]) + self.assertEqual(result["state"], "closed") + mock_api.assert_called_once() + + @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) + @patch("gitea_mcp_server._profile_permission_block", return_value=None) + @patch("gitea_mcp_server._namespace_mutation_block", return_value=None) + @patch( + "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, []), + ) + @patch("gitea_mcp_server.api_get_all", return_value=[]) + @patch( + "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", + return_value={"current_branch": "master"}, + ) + def test_author_create_issue_still_blocked_on_control_checkout( + self, _git, _get_all, _role, _ns, _prof, _auth + ): + srv._preflight_resolved_role = "author" + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): + with self.assertRaises(RuntimeError) as ctx: + srv.gitea_create_issue(title="Test", body="body") + self.assertIn("stable control checkout", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 08202f7eaa6d09b2eca8f2960126994a4b22646b Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 02:16:52 -0400 Subject: [PATCH 2/4] fix: guard PR/issue comment listing against non-list API payloads (Closes #485) _list_pr_lease_comments and gitea_list_issue_comments now fail safe to an empty list when the comments API returns a non-list payload (e.g. an error object), preventing KeyError: slice on the lease and listing paths. Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 14 +++- .../test_pr_lease_comments_non_list_guard.py | 76 +++++++++++++++++++ 2 files changed, 88 insertions(+), 2 deletions(-) create mode 100644 tests/test_pr_lease_comments_non_list_guard.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 9ea9050..cbc999d 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2515,7 +2515,12 @@ def _list_pr_lease_comments( h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments" - comments = api_request("GET", api, auth) or [] + comments = api_request("GET", api, auth) + # Fail safe to no lease comments when the API returns a non-list payload + # (e.g. an error object such as an HTTP 401 body): lease state can only be + # proven from real comment entries, never inferred from an error shape (#485). + if not isinstance(comments, list): + return [] return list(comments[:limit]) @@ -5119,7 +5124,12 @@ def gitea_list_issue_comments( h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments" - comments = api_request("GET", api, auth) or [] + comments = api_request("GET", api, auth) + # Fail safe to no comments when the API returns a non-list payload (e.g. an + # error object such as an HTTP 401 body) so listing never crashes on a + # malformed response (#485). + if not isinstance(comments, list): + comments = [] reveal = _reveal_endpoints() out = [] for c in comments[:limit]: diff --git a/tests/test_pr_lease_comments_non_list_guard.py b/tests/test_pr_lease_comments_non_list_guard.py new file mode 100644 index 0000000..08c9239 --- /dev/null +++ b/tests/test_pr_lease_comments_non_list_guard.py @@ -0,0 +1,76 @@ +"""Regression tests for non-list API payloads on PR/issue comment listing (#485).""" +import os +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +from mcp_server import ( # noqa: E402 + _list_pr_lease_comments, + gitea_list_issue_comments, +) + +FAKE_AUTH = "Basic dGVzdDp0ZXN0" +AUTHOR_ENV = { + "GITEA_PROFILE_NAME": "gitea-author", + "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment", +} + + +class TestPrLeaseCommentsNonListGuard(unittest.TestCase): + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_pr_lease_comments_non_list_payload_returns_empty(self, _auth, mock_api): + mock_api.return_value = {"message": "Unauthorized"} + result = _list_pr_lease_comments( + 12, remote="prgs", host=None, org=None, repo=None) + self.assertEqual(result, []) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_pr_lease_comments_none_returns_empty(self, _auth, mock_api): + mock_api.return_value = None + result = _list_pr_lease_comments( + 12, remote="prgs", host=None, org=None, repo=None) + self.assertEqual(result, []) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_pr_lease_comments_list_payload_unchanged(self, _auth, mock_api): + comment = {"id": 7, "body": ""} + mock_api.return_value = [comment] + result = _list_pr_lease_comments( + 12, remote="prgs", host=None, org=None, repo=None, limit=5) + self.assertEqual(result, [comment]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_issue_comments_non_list_payload_returns_empty(self, _auth, mock_api): + mock_api.return_value = {"message": "Unauthorized"} + with patch.dict(os.environ, AUTHOR_ENV, clear=True): + result = gitea_list_issue_comments(issue_number=9, remote="prgs") + self.assertTrue(result["success"]) + self.assertEqual(result["comments"], []) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_list_issue_comments_list_payload_unchanged(self, _auth, mock_api): + mock_api.return_value = [ + { + "id": 101, + "user": {"login": "alice"}, + "body": "hello", + "created_at": "2026-07-03T00:00:00Z", + "updated_at": "2026-07-03T01:00:00Z", + } + ] + with patch.dict(os.environ, AUTHOR_ENV, clear=True): + result = gitea_list_issue_comments(issue_number=9, remote="prgs") + self.assertTrue(result["success"]) + self.assertEqual(len(result["comments"]), 1) + self.assertEqual(result["comments"][0]["author"], "alice") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From afb4ba563cb0eb293d6554d10b060faf84ca7bd0 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 17:47:30 -0400 Subject: [PATCH 3/4] 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 b5681e9..8d994ac 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 @@ -1223,7 +1285,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( @@ -1356,7 +1418,7 @@ def gitea_lock_issue( ) else: 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"), @@ -1572,7 +1634,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) ── @@ -2633,7 +2695,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, @@ -3140,12 +3202,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: @@ -3390,7 +3452,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) @@ -3491,7 +3553,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, @@ -4020,7 +4082,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 @@ -4129,7 +4191,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 "" @@ -4443,7 +4505,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") @@ -4546,7 +4608,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}" @@ -4974,7 +5036,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() @@ -5075,7 +5137,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( @@ -5246,7 +5308,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(): @@ -6780,7 +6842,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) @@ -6858,7 +6920,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", @@ -7101,7 +7163,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) @@ -7255,7 +7317,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) @@ -7492,7 +7554,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 From 3bce0a55fb53456f5b946785c1f8711f254967b0 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 17:49:06 -0400 Subject: [PATCH 4/4] 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