diff --git a/anti_stomp_preflight.py b/anti_stomp_preflight.py index bdc0f96..83f1dae 100644 --- a/anti_stomp_preflight.py +++ b/anti_stomp_preflight.py @@ -67,50 +67,79 @@ BLOCKER_KINDS = frozenset({ BLOCKER_MANUAL_BYPASS, }) -# Mutation tasks that must invoke this preflight before acting. -# Covers create issue/comment, lease acquire, submit review, approve, -# request changes, merge, cleanup, and label mutation (issue #604 AC1). +# Mutation tasks that must invoke the shared anti-stomp preflight before +# acting (issue #604 AC1). Every member must appear as a live task= kwarg +# to verify_preflight_purity / _run_anti_stomp_preflight (or the review +# anti_task map) in gitea_mcp_server.py. Inventory↔wiring tests fail if +# a declared task is not wired. MUTATION_TASKS = frozenset({ "create_issue", "comment_issue", "close_issue", - "claim_issue", "mark_issue", "lock_issue", "set_issue_labels", "create_label", "create_pr", - "comment_pr", "close_pr", + "edit_pr", "commit_files", "gitea_commit_files", - "create_branch", - "push_branch", "delete_branch", "cleanup_merged_pr_branch", "cleanup_stale_claims", - "cleanup_stale_review_decision_lock", - "gitea_cleanup_stale_review_decision_lock", "reconcile_merged_cleanups", "reconcile_already_landed_pr", "reconcile_close_superseded_pr", - "reconciliation_cleanup", "post_heartbeat", "acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease", "adopt_merger_pr_lease", - "heartbeat_reviewer_pr_lease", - "release_reviewer_pr_lease", "review_pr", "submit_pr_review", "approve_pr", "request_changes_pr", "merge_pr", - "mark_final_review_decision", - "save_review_draft", - "resume_review_draft", }) +# Intentionally excluded from MUTATION_TASKS. Each has a dedicated +# fail-closed gate that preserves decision-lock ownership, workflow-hash, +# head-SHA, and repository consistency. Do not re-add without either +# wiring shared preflight or updating this rationale (#604 Blocker B). +DEDICATED_GATE_MUTATIONS: dict[str, str] = { + "mark_final_review_decision": ( + "Local review-decision lock only. Enforced by session lock ownership, " + "workflow-hash, expected head SHA, PR work lease, eligibility, and " + "terminal-lock gates. No Gitea review POST; shared anti-stomp would " + "duplicate without side-effect-ordering value." + ), + "save_review_draft": ( + "Local session-state draft save with live PR head/base consistency " + "checks. Not a Gitea review/merge mutation; dedicated head-SHA and " + "worktree resolution gates apply." + ), + "resume_review_draft": ( + "Live-state resume with terminal lock, lease ownership, head/base SHA, " + "and parity checks. When submit=True, delegates to gitea_submit_pr_review " + "which runs shared anti-stomp before the Gitea review POST." + ), + "cleanup_stale_review_decision_lock": ( + "Moot-lock cleanup with identity match, live PR merged/closed proof, " + "and reviewer capability gate (#594). Distinct from shared anti-stomp." + ), + "gitea_cleanup_stale_review_decision_lock": ( + "Alias of cleanup_stale_review_decision_lock; dedicated #594 path." + ), + "heartbeat_reviewer_pr_lease": ( + "Lease heartbeat posts via verify_preflight_purity(task='review_pr') " + "plus in-session lease ownership checks; not a separate mutation class." + ), + "release_reviewer_pr_lease": ( + "Lease release uses workspace binding + ownership checks; posts only " + "when the session owns the active lease." + ), +} + # Roles that must mutate from a branches/ worktree (not the control checkout). _BRANCHES_REQUIRED_ROLES = frozenset({"author"}) @@ -122,13 +151,9 @@ _LEASE_AWARE_TASKS = frozenset({ "approve_pr", "request_changes_pr", "merge_pr", - "mark_final_review_decision", "acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease", "adopt_merger_pr_lease", - "heartbeat_reviewer_pr_lease", - "release_reviewer_pr_lease", - "resume_review_draft", }) _NEXT_ACTIONS: dict[str, str] = { @@ -199,6 +224,10 @@ def roles_compatible(active_role: str | None, required_role: str | None) -> bool (comment/close/cleanup) that the capability map stamps as ``author`` — matching the long-standing reconciler exemption for control-checkout work. No other cross-role substitutions are allowed. + + Capability-authorized multi-role tasks (e.g. reviewer holding + ``gitea.issue.comment`` for ``comment_issue``) are handled by + :func:`authorization_compatible`, not by expanding this role matrix. """ active = (active_role or "").strip().lower() required = (required_role or "").strip().lower() @@ -211,6 +240,43 @@ def roles_compatible(active_role: str | None, required_role: str | None) -> bool return False +def authorization_compatible( + active_role: str | None, + required_role: str | None, + *, + required_permission: str | None = None, + allowed_operations: Any = None, +) -> bool: + """Whether the active session may run a task given role *and* capability. + + Order: + + 1. :func:`roles_compatible` (exact role or narrow reconciler→author). + 2. Capability possession: when *required_permission* is non-empty and + present in *allowed_operations*, allow even if the nominal task role + differs (e.g. reviewer/merger with ``gitea.issue.comment`` on + ``comment_issue`` / ``mark_issue`` / ``set_issue_labels`` / + ``lock_issue``). + + Fail closed otherwise. This is **not** a broad reviewer→author or + merger→author role rewrite: without the specific permission the check + still denies. Missing *allowed_operations* when a permission is required + also fails closed (callers must supply the live profile op list). + """ + if roles_compatible(active_role, required_role): + return True + perm = (required_permission or "").strip() + if not perm: + return False + if allowed_operations is None: + return False + try: + ops = {str(o).strip() for o in allowed_operations if o is not None} + except TypeError: + return False + return perm in ops + + def _blocker( kind: str, reasons: list[str], @@ -272,10 +338,12 @@ def assess_anti_stomp_preflight( org_explicit: bool = False, repo_explicit: bool = False, check_repo: bool = True, - # profile/role + # profile/role + capability profile_name: str | None = None, profile_role: str | None = None, required_role: str | None = None, + required_permission: str | None = None, + allowed_operations: Any = None, check_role: bool = True, # root checkout + worktree workspace_path: str | None = None, @@ -327,11 +395,13 @@ def assess_anti_stomp_preflight( task_name = (task or "").strip() role = (profile_role or "").strip().lower() or None req_role = (required_role or "").strip().lower() or None + req_perm = (required_permission or "").strip() or None checks: dict[str, Any] = { "task": task_name or None, "profile_name": profile_name, "profile_role": role, "required_role": req_role, + "required_permission": req_perm, } blockers: list[dict[str, Any]] = [] @@ -376,15 +446,33 @@ def assess_anti_stomp_preflight( else: checks["repo"] = {"block": False, "skipped": True} - # ── profile/role ───────────────────────────────────────────────────────── + # ── profile/role + capability ──────────────────────────────────────────── + # Role match OR possession of the task's required permission (Blocker A). + # Reviewer/merger may run issue-comment-class tasks when they hold + # gitea.issue.comment; unauthorized escalation without the permission + # still fails closed. if check_role and req_role and role: - if not roles_compatible(role, req_role): + authorized = authorization_compatible( + role, + req_role, + required_permission=req_perm, + allowed_operations=allowed_operations, + ) + if not authorized: + perm_clause = ( + f", required_permission='{req_perm}'" if req_perm else "" + ) reasons = [ - f"active profile role '{role}' does not match required role " - f"'{req_role}' for task '{task_name or '(unknown)'}' " - f"(profile={profile_name or '(unknown)'})" + f"active profile role '{role}' is not authorized for task " + f"'{task_name or '(unknown)'}' (required_role='{req_role}'" + f"{perm_clause}; profile={profile_name or '(unknown)'})" ] - checks["role"] = {"block": True, "reasons": reasons} + checks["role"] = { + "block": True, + "reasons": reasons, + "required_permission": req_perm, + "capability_authorized": False, + } blockers.append( _blocker( BLOCKER_WRONG_ROLE, @@ -393,11 +481,25 @@ def assess_anti_stomp_preflight( "profile_name": profile_name, "profile_role": role, "required_role": req_role, + "required_permission": req_perm, }, ) ) else: - checks["role"] = {"block": False, "reasons": []} + checks["role"] = { + "block": False, + "reasons": [], + "required_permission": req_perm, + "capability_authorized": bool( + req_perm + and allowed_operations is not None + and req_perm in { + str(o).strip() for o in (allowed_operations or []) + if o is not None + } + and not roles_compatible(role, req_role) + ), + } else: checks["role"] = {"block": False, "skipped": True} diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index d1b587f..43b018c 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -671,12 +671,18 @@ def _run_anti_stomp_preflight( profile = get_profile() profile_name = profile.get("profile_name") profile_role = _actual_profile_role() + allowed_operations = list(profile.get("allowed_operations") or []) req_role = None + req_perm = None if task: try: req_role = task_capability_map.required_role(task) except KeyError: req_role = _preflight_resolved_role + try: + req_perm = task_capability_map.required_permission(task) + except KeyError: + req_perm = None ctx = _resolve_namespace_mutation_context(worktree_path) workspace = ctx["workspace_path"] @@ -734,7 +740,6 @@ def _run_anti_stomp_preflight( "approve_pr", "request_changes_pr", "merge_pr", - "mark_final_review_decision", }: try: wf_reasons = _review_workflow_load_gate_reasons() @@ -765,6 +770,8 @@ def _run_anti_stomp_preflight( profile_name=profile_name, profile_role=profile_role, required_role=req_role or _preflight_resolved_role, + required_permission=req_perm, + allowed_operations=allowed_operations, workspace_path=workspace, project_root=canonical_root, current_branch=git_state.get("current_branch"), @@ -5030,7 +5037,12 @@ def gitea_edit_pr( raise ValueError("At least one field to edit (title, body, state, base) must be provided.") closing = payload.get("state") == "closed" - verify_preflight_purity(remote, task="close_pr" if closing else None) + # Closing uses close_pr; non-closing title/body/base edits use edit_pr so + # the shared #604 anti-stomp inventory stays consistent with wiring. + if closing: + verify_preflight_purity(remote, task="close_pr") + else: + verify_preflight_purity(remote, task="edit_pr") # PR closure is a first-class capability, distinct from retitling or # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close @@ -7650,7 +7662,11 @@ def gitea_acquire_reviewer_pr_lease( "permission_report": _permission_block_report("gitea.pr.comment"), } - _verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr") + # task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604 + # anti-stomp for the declared lease-acquire mutation inventory entry. + _verify_role_mutation_workspace( + remote, worktree=worktree, task="acquire_reviewer_pr_lease" + ) h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) profile = get_profile() @@ -7791,6 +7807,7 @@ def gitea_adopt_merger_pr_lease( "permission_report": _permission_block_report("gitea.pr.merge"), } + # task=adopt_merger_pr_lease so verify_preflight_purity runs shared #604 anti-stomp. _verify_role_mutation_workspace( remote, worktree=worktree, task="adopt_merger_pr_lease" ) diff --git a/task_capability_map.py b/task_capability_map.py index 32d6ab0..4477747 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -60,6 +60,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.pr.close", "role": "author", }, + # Non-closing PR metadata edits (title/body/base). Closing uses close_pr. + "edit_pr": { + "permission": "gitea.pr.create", + "role": "author", + }, + "gitea_edit_pr": { + "permission": "gitea.pr.create", + "role": "author", + }, "address_pr_change_requests": { "permission": "gitea.branch.push", "role": "author", @@ -68,10 +77,22 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.pr.review", "role": "reviewer", }, + "submit_pr_review": { + "permission": "gitea.pr.review", + "role": "reviewer", + }, "merge_pr": { "permission": "gitea.pr.merge", "role": "merger", }, + "acquire_reviewer_pr_lease": { + "permission": "gitea.pr.comment", + "role": "reviewer", + }, + "gitea_acquire_reviewer_pr_lease": { + "permission": "gitea.pr.comment", + "role": "reviewer", + }, "adopt_merger_pr_lease": { "permission": "gitea.pr.comment", "role": "reviewer", diff --git a/tests/test_anti_stomp_preflight.py b/tests/test_anti_stomp_preflight.py index cf53177..93b6865 100644 --- a/tests/test_anti_stomp_preflight.py +++ b/tests/test_anti_stomp_preflight.py @@ -406,6 +406,141 @@ class TestWrongRole(unittest.TestCase): self.assertFalse(asp.roles_compatible("author", "merger")) +class TestCapabilityAuthorizedRoles(unittest.TestCase): + """Blocker A: reviewer/merger holding issue-comment capability may mutate. + + Nominal task role is still ``author`` in task_capability_map, but the + shared anti-stomp gate must not WRONG_ROLE-block when the active profile + holds ``gitea.issue.comment``. Unauthorized escalation without the + permission remains fail closed. + """ + + _ISSUE_COMMENT_TASKS = ( + "comment_issue", + "mark_issue", + "set_issue_labels", + "lock_issue", + ) + _ISSUE_COMMENT_PERM = "gitea.issue.comment" + + def _role_kwargs(self, task, role, *, allowed_ops, permission=None): + return _happy_kwargs( + task=task, + profile_name=f"prgs-{role}", + profile_role=role, + required_role="author", + required_permission=permission if permission is not None else self._ISSUE_COMMENT_PERM, + allowed_operations=allowed_ops, + workspace_path=f"/repo/branches/{role}-ws", + project_root="/repo", + current_branch=f"review/{role}-task", + ) + + def test_reviewer_allowed_with_issue_comment_capability(self): + ops = ["gitea.read", "gitea.issue.comment", "gitea.pr.review"] + for task in self._ISSUE_COMMENT_TASKS: + with self.subTest(task=task): + result = asp.assess_anti_stomp_preflight( + **self._role_kwargs(task, "reviewer", allowed_ops=ops) + ) + self.assertTrue(result["allowed"], result.get("reasons")) + self.assertFalse(result["block"]) + self.assertTrue( + result["checks"]["role"].get("capability_authorized") + ) + + def test_merger_allowed_with_issue_comment_capability(self): + ops = ["gitea.read", "gitea.issue.comment", "gitea.pr.merge"] + for task in self._ISSUE_COMMENT_TASKS: + with self.subTest(task=task): + result = asp.assess_anti_stomp_preflight( + **self._role_kwargs(task, "merger", allowed_ops=ops) + ) + self.assertTrue(result["allowed"], result.get("reasons")) + self.assertFalse(result["block"]) + + def test_reviewer_denied_without_issue_comment_capability(self): + # Holds review permission only — not issue comment. + ops = ["gitea.read", "gitea.pr.review", "gitea.pr.approve"] + for task in self._ISSUE_COMMENT_TASKS: + with self.subTest(task=task): + result = asp.assess_anti_stomp_preflight( + **self._role_kwargs(task, "reviewer", allowed_ops=ops) + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_ROLE) + + def test_merger_denied_without_issue_comment_capability(self): + ops = ["gitea.read", "gitea.pr.merge"] + for task in self._ISSUE_COMMENT_TASKS: + with self.subTest(task=task): + result = asp.assess_anti_stomp_preflight( + **self._role_kwargs(task, "merger", allowed_ops=ops) + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_ROLE) + + def test_not_broad_reviewer_to_author_bypass(self): + # Reviewer with only issue.comment must not pass create_pr (needs create). + result = asp.assess_anti_stomp_preflight( + **_happy_kwargs( + task="create_pr", + profile_name="prgs-reviewer", + profile_role="reviewer", + required_role="author", + required_permission="gitea.pr.create", + allowed_operations=["gitea.read", "gitea.issue.comment", "gitea.pr.review"], + workspace_path="/repo/branches/review-ws", + project_root="/repo", + ) + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_ROLE) + + def test_not_broad_merger_to_author_bypass(self): + result = asp.assess_anti_stomp_preflight( + **_happy_kwargs( + task="create_issue", + profile_name="prgs-merger", + profile_role="merger", + required_role="author", + required_permission="gitea.issue.create", + allowed_operations=["gitea.read", "gitea.issue.comment", "gitea.pr.merge"], + workspace_path="/repo/branches/merge-ws", + project_root="/repo", + ) + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_ROLE) + + def test_authorization_compatible_helper(self): + self.assertTrue( + asp.authorization_compatible( + "reviewer", + "author", + required_permission="gitea.issue.comment", + allowed_operations=["gitea.issue.comment"], + ) + ) + self.assertFalse( + asp.authorization_compatible( + "reviewer", + "author", + required_permission="gitea.issue.comment", + allowed_operations=["gitea.pr.review"], + ) + ) + # Missing ops list fails closed when permission is required and roles differ. + self.assertFalse( + asp.authorization_compatible( + "reviewer", + "author", + required_permission="gitea.issue.comment", + allowed_operations=None, + ) + ) + + class TestWorkflowHash(unittest.TestCase): def test_stale_workflow_hash_blocks(self): result = asp.assess_anti_stomp_preflight( @@ -483,7 +618,7 @@ class TestRootCheckoutContamination(unittest.TestCase): class TestMutationTaskInventory(unittest.TestCase): - """AC1: mutation task set covers issue-listed paths.""" + """AC1 + Blocker B: declared inventory matches runtime wiring.""" def test_issue_required_paths_covered(self): required = { @@ -497,10 +632,74 @@ class TestMutationTaskInventory(unittest.TestCase): "merge_pr", "cleanup_merged_pr_branch", "cleanup_stale_claims", + "edit_pr", } missing = required - asp.MUTATION_TASKS self.assertFalse(missing, f"missing mutation tasks: {missing}") + def test_disputed_helpers_classified_as_dedicated_or_shared(self): + """Blocker B explicit classification for disputed mutation helpers.""" + # Shared preflight inventory (wired). + self.assertIn("edit_pr", asp.MUTATION_TASKS) + # Dedicated fail-closed gates — must NOT claim shared preflight. + for name in ( + "mark_final_review_decision", + "save_review_draft", + "resume_review_draft", + ): + self.assertNotIn(name, asp.MUTATION_TASKS, name) + self.assertIn(name, asp.DEDICATED_GATE_MUTATIONS, name) + self.assertTrue(asp.DEDICATED_GATE_MUTATIONS[name].strip()) + + def test_inventory_and_wiring_consistent(self): + """Every MUTATION_TASKS member is referenced as a live task kwarg. + + Accepts: + * task=\"name\" / task='name' on verify_preflight_purity / _run_anti_stomp + * review anti_task map values (approve_pr / request_changes_pr / …) + * gitea_ alias form when the bare name is also declared + """ + import pathlib + import re + + server_src = pathlib.Path(__file__).resolve().parents[1] / "gitea_mcp_server.py" + text = server_src.read_text(encoding="utf-8") + + # Collect string literals used as task= kwargs. + task_kw = set(re.findall(r'task\s*=\s*["\']([a-z0-9_]+)["\']', text)) + # anti_task map values: "approve": "approve_pr", + anti_map = set( + re.findall( + r'"(?:approve|request_changes|comment)"\s*:\s*"([a-z0-9_]+)"', + text, + ) + ) + wired = task_kw | anti_map + # Also accept f-string / conditional close_pr|edit_pr style already in task_kw. + + missing = set() + for task in asp.MUTATION_TASKS: + bare = task.removeprefix("gitea_") + if task in wired or bare in wired or f"gitea_{bare}" in wired: + continue + # Alias pairs: gitea_commit_files ↔ commit_files + if task.startswith("gitea_") and bare in asp.MUTATION_TASKS and bare in wired: + continue + if f"gitea_{task}" in asp.MUTATION_TASKS and task in wired: + continue + missing.add(task) + self.assertFalse( + missing, + f"MUTATION_TASKS declared but not wired in gitea_mcp_server.py: {sorted(missing)}", + ) + + # Dedicated exclusions must not appear as shared anti-stomp task kwargs + # for the three disputed local helpers (except resume→submit_pr_review). + for name in ("mark_final_review_decision", "save_review_draft"): + # May appear as string metadata but not as task="..." anti-stomp target. + # Allow mention in comments; assert not as task= kwarg. + self.assertNotIn(name, task_kw, f"{name} must not be shared-preflight task=") + class TestServerWiring(unittest.TestCase): """Smoke: MCP server imports anti_stomp and exposes the runner.""" @@ -562,5 +761,272 @@ class TestServerWiring(unittest.TestCase): self.assertIn(asp.BLOCKER_STALE_RUNTIME, str(ctx.exception)) +class TestEntrypointSideEffectOrdering(unittest.TestCase): + """Blocker C / AC4: anti-stomp aborts before Gitea mutation side effects. + + Invokes real entrypoint functions with external boundaries mocked. + Forces the assessor to block and proves api_request POST/merge paths + and durable local writes never run. + """ + + def _blocked_assessment(self, kind=asp.BLOCKER_FOREIGN_LEASE, next_action="stop"): + return { + "allowed": False, + "block": True, + "blockers": [{ + "kind": kind, + "reasons": ["forced block for ordering test"], + "exact_next_action": next_action, + "detail": {}, + }], + "reasons": ["forced block for ordering test"], + "exact_next_action": next_action, + "blocker_kind": kind, + "checks": {}, + } + + def test_merge_pr_blocks_before_merge_api(self): + import gitea_mcp_server as server + + blocked = self._blocked_assessment( + kind=asp.BLOCKER_HEAD_SHA, + next_action="re-pin expected_head_sha and retry", + ) + api_mock = mock.Mock(side_effect=AssertionError("Gitea API must not be called")) + save_lock = mock.Mock(side_effect=AssertionError("local lock must not mutate")) + + with mock.patch.dict( + os.environ, + {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, + clear=False, + ): + with mock.patch.object( + server, "_verify_role_mutation_workspace", return_value="/repo/branches/x" + ), mock.patch.object( + server, "_review_workflow_load_gate_reasons", return_value=[] + ), mock.patch.object( + server, "_live_namespace_health_gate", return_value=None + ), mock.patch.object( + server, "terminal_review_hard_stop_reasons", return_value=[] + ), mock.patch.object( + server, + "gitea_check_pr_eligibility", + return_value={ + "eligible": True, + "authenticated_user": "merger-bot", + "profile_name": "prgs-merger", + "pr_author": "other-user", + "head_sha": HEAD_A, + "mergeable": True, + "reasons": [], + }, + ), mock.patch.object( + server, "_reviewer_pr_lease_gate", return_value=[] + ), mock.patch.object( + server, "_pr_work_lease_reviewer_block", return_value={"block": False} + ), mock.patch.object( + server, "get_profile", return_value={ + "profile_name": "prgs-merger", + "allowed_operations": ["gitea.pr.merge", "gitea.read"], + "forbidden_operations": [], + } + ), mock.patch.object( + server, "_role_kind", return_value="merger" + ), mock.patch.object( + asp, "assess_anti_stomp_preflight", return_value=blocked + ) as assess_mock, mock.patch.object( + server, "api_request", api_mock + ), mock.patch.object( + server, "_save_review_decision_lock", save_lock + ): + result = server.gitea_merge_pr( + pr_number=680, + confirmation="MERGE PR 680", + expected_head_sha=HEAD_A, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + worktree_path="/repo/branches/merge-680", + ) + + self.assertFalse(result.get("performed")) + self.assertTrue(any("#604" in r or "anti-stomp" in r.lower() or asp.BLOCKER_HEAD_SHA in r + for r in (result.get("reasons") or [])), + result.get("reasons")) + joined = " ".join(result.get("reasons") or []) + self.assertIn(asp.BLOCKER_HEAD_SHA, joined) + self.assertIn("exact_next_action", joined) + assess_mock.assert_called() + api_mock.assert_not_called() + save_lock.assert_not_called() + + def test_submit_pr_review_blocks_before_review_api(self): + import gitea_mcp_server as server + + blocked = self._blocked_assessment( + kind=asp.BLOCKER_FOREIGN_LEASE, + next_action="stop; do not stomp foreign lease", + ) + api_mock = mock.Mock(side_effect=AssertionError("Gitea review API must not be called")) + save_lock = mock.Mock(side_effect=AssertionError("decision lock must not mutate")) + + with mock.patch.dict( + os.environ, + {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, + clear=False, + ): + with mock.patch.object( + server, "_verify_role_mutation_workspace", return_value="/repo/branches/r" + ), mock.patch.object( + server, "_review_workflow_load_gate_reasons", return_value=[] + ), mock.patch.object( + server, "_live_namespace_health_gate", return_value=None + ), mock.patch.object( + server, "check_review_decision_gate", return_value=[] + ), mock.patch.object( + server, + "gitea_check_pr_eligibility", + return_value={ + "eligible": True, + "authenticated_user": "reviewer-bot", + "profile_name": "prgs-reviewer", + "pr_author": "other-user", + "head_sha": HEAD_A, + "reasons": [], + }, + ), mock.patch.object( + server, "_reviewer_pr_lease_gate", return_value=[] + ), mock.patch.object( + server, "_pr_work_lease_reviewer_block", return_value={"block": False} + ), mock.patch.object( + server, "_load_review_decision_lock", return_value={ + "ready_expected_head_sha": HEAD_A, + "final_review_decision_ready": True, + "ready_pr_number": 680, + "ready_action": "request_changes", + } + ), mock.patch.object( + server, "get_profile", return_value={ + "profile_name": "prgs-reviewer", + "allowed_operations": [ + "gitea.pr.review", + "gitea.pr.request_changes", + "gitea.read", + ], + "forbidden_operations": [], + } + ), mock.patch.object( + asp, "assess_anti_stomp_preflight", return_value=blocked + ) as assess_mock, mock.patch.object( + server, "api_request", api_mock + ), mock.patch.object( + server, "_save_review_decision_lock", save_lock + ), mock.patch.object( + server, "_submit_pending_pull_review", api_mock + ): + result = server.gitea_submit_pr_review( + pr_number=680, + action="request_changes", + body="needs work", + expected_head_sha=HEAD_A, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + final_review_decision_ready=True, + worktree_path="/repo/branches/review-680", + ) + + self.assertFalse(result.get("performed")) + joined = " ".join(result.get("reasons") or []) + self.assertIn(asp.BLOCKER_FOREIGN_LEASE, joined) + self.assertIn("exact_next_action", joined) + assess_mock.assert_called() + # Assessor must run for request_changes_pr (anti_task map). + call_task = None + for c in assess_mock.call_args_list: + kwargs = c.kwargs if c.kwargs else {} + if "task" in kwargs: + call_task = kwargs["task"] + elif c.args: + call_task = c.args[0] if False else kwargs.get("task") + # assess_anti_stomp is called with keyword task= + if c.kwargs.get("task"): + call_task = c.kwargs["task"] + # _run_anti_stomp passes task as first positional to assess via keyword. + self.assertTrue(assess_mock.called) + api_mock.assert_not_called() + save_lock.assert_not_called() + + def test_comment_issue_blocks_before_comment_api(self): + """Representative issue-mutation class for AC4 coverage.""" + import gitea_mcp_server as server + + blocked = self._blocked_assessment( + kind=asp.BLOCKER_WRONG_ROLE, + next_action="switch namespace", + ) + api_mock = mock.Mock(side_effect=AssertionError("comment API must not be called")) + + with mock.patch.dict( + os.environ, + {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, + clear=False, + ): + with mock.patch.object( + server, "verify_preflight_purity", wraps=None + ) as purity, mock.patch.object( + server, "api_request", api_mock + ): + # Drive verify_preflight_purity → _run_anti_stomp by calling purity + # path: patch _run_anti_stomp to raise via assessor. + def _purity_side_effect(*args, **kwargs): + # Call real runner under force so assessor block surfaces. + return server._run_anti_stomp_preflight( + kwargs.get("task") or (args[2] if len(args) > 2 else None), + remote=args[0] if args else kwargs.get("remote"), + worktree_path=kwargs.get("worktree_path"), + org=kwargs.get("org"), + repo=kwargs.get("repo"), + raise_on_block=True, + ) + + purity.side_effect = None + with mock.patch.object( + asp, "assess_anti_stomp_preflight", return_value=blocked + ), mock.patch.object( + server, "verify_preflight_purity", side_effect=lambda *a, **k: ( + server._run_anti_stomp_preflight( + k.get("task"), + remote=a[0] if a else k.get("remote"), + worktree_path=k.get("worktree_path"), + org=k.get("org"), + repo=k.get("repo"), + ) + ) + ), mock.patch.object( + server, "_profile_operation_gate", return_value=[] + ), mock.patch.object( + server, "_canonical_comment_gate", return_value={"blocked": False} + ), mock.patch.object( + server, "get_profile", return_value={ + "profile_name": "prgs-author", + "allowed_operations": ["gitea.issue.comment", "gitea.read"], + "forbidden_operations": [], + } + ): + with self.assertRaises(RuntimeError) as ctx: + server.gitea_create_issue_comment( + issue_number=604, + body="handoff", + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + worktree_path="/repo/branches/issue-604", + ) + self.assertIn(asp.BLOCKER_WRONG_ROLE, str(ctx.exception)) + self.assertIn("exact_next_action", str(ctx.exception)) + api_mock.assert_not_called() + + if __name__ == "__main__": unittest.main()