From adc61255b2d7eb65d1670e09e561dbd84b652317 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sun, 19 Jul 2026 02:26:58 -0400 Subject: [PATCH] fix(mcp): honor the create-issue bootstrap in anti-stomp preflight (Closes #757) The sanctioned create_issue bootstrap from #749/#750 was unreachable in production. Two guards assessed the same workspace for the same task and reached opposite conclusions: the #274 branches-only guard consulted the bootstrap and permitted a clean canonical control checkout, then the #604 anti-stomp preflight -- which never consulted it -- rejected that same checkout as wrong_worktree. The defect was wiring, not policy: the bootstrap decision was computed in one guard and discarded, while the other re-derived a conflicting answer from a lower-level assessor with no notion of the bootstrap phase. Fix: one computation site, one interpretation site. * create_issue_bootstrap.bootstrap_permits_control_checkout() is the single predicate both guards use to interpret an assessment. It is fail-closed by construction: missing, malformed, refused, incomplete, or contradictory evidence returns False and leaves the ordinary block in force. It also verifies the assessment describes the exact workspace and canonical root being guarded, so a stale or foreign assessment cannot be reused. * _create_issue_bootstrap_assessment() computes the assessment once per preflight from inspected repository state. verify_preflight_purity threads that single result into both guards. * The #604 assessor accepts the assessment and waives ONLY the wrong-worktree verdict. Root checkout, repo, role, stale runtime, lease, head-SHA, workflow-hash, and contamination checks are evaluated independently and still apply. Evidence is server-derived only and travels an internal path: no MCP tool signature gains a bootstrap argument, and no caller-controlled boolean can manufacture eligibility. Behavior is unchanged for callers that supply no evidence, and for every non-create_issue author mutation. No issue or PR number is special-cased in production behavior. Tests: new tests/test_issue_757_bootstrap_guard_agreement.py (38 tests, 26 subtests) covering the shared predicate, the narrow waiver, guard agreement across the full workspace-state matrix, non-forgeable eligibility, and an end-to-end native gitea_create_issue run with the #604 gate LIVE. All 38 fail against unfixed sources; the e2e reproduces the production error text verbatim ("Anti-stomp preflight (#604) blocked mutation [wrong_worktree]"). tests/test_reconciler_close_workspace_guard.py: one case asserted that create_issue stays blocked on the control checkout, which only held because the bootstrap-blind #604 guard was overriding #750 -- it encoded the defect. Re-pointed to lock_issue, which is issue-backed and legitimately still requires a branches/ worktree. Its teardown now restores the module-level preflight task/role so test order cannot leak resolved state. Full suite: 3624 passed, 2 failed, 6 skipped (426 subtests). Baseline at bde5c5fb on a clean detached worktree: 3586 passed, 2 failed, 6 skipped (400 subtests). The same 2 failures reproduce identically on pristine master and are unrelated to this change (test_issue_702_review_findings_f1_f6 F1 worktree recovery; test_reconciler_supersession_close org/repo forwarding). Co-Authored-By: Claude Opus 4.8 (1M context) --- anti_stomp_preflight.py | 22 +- create_issue_bootstrap.py | 70 ++ gitea_mcp_server.py | 104 ++- ...est_issue_757_bootstrap_guard_agreement.py | 613 ++++++++++++++++++ .../test_reconciler_close_workspace_guard.py | 34 +- 5 files changed, 810 insertions(+), 33 deletions(-) create mode 100644 tests/test_issue_757_bootstrap_guard_agreement.py diff --git a/anti_stomp_preflight.py b/anti_stomp_preflight.py index 83f1dae..10cacdd 100644 --- a/anti_stomp_preflight.py +++ b/anti_stomp_preflight.py @@ -33,6 +33,7 @@ from __future__ import annotations from typing import Any import author_mutation_worktree +import create_issue_bootstrap import master_parity_gate import remote_repo_guard import root_checkout_guard @@ -354,6 +355,7 @@ def assess_anti_stomp_preflight( remote_master_sha: str | None = None, check_root_checkout: bool = True, check_worktree: bool = True, + create_issue_bootstrap_assessment: dict[str, Any] | None = None, # stale runtime (master parity) startup_head: str | None = None, current_code_head: str | None = None, @@ -584,12 +586,28 @@ def assess_anti_stomp_preflight( project_root=project_root, current_branch=current_branch, ) + # #757: the #274 guard consults the server-derived create_issue + # bootstrap before blocking the canonical control checkout. Route this + # guard's decision through the *same* predicate on the *same* + # assessment so the two cannot disagree about identical evidence. + # Only the wrong-worktree verdict is waived; every other check in this + # assessment (root checkout, repo, role, stale runtime, lease, ...) is + # evaluated independently and still applies. + bootstrap_waived = wt.get("block") and ( + create_issue_bootstrap.bootstrap_permits_control_checkout( + create_issue_bootstrap_assessment, + task=task_name, + workspace_path=workspace_path, + canonical_repo_root=project_root, + ) + ) checks["worktree"] = { - "block": bool(wt.get("block")), + "block": bool(wt.get("block")) and not bootstrap_waived, "reasons": list(wt.get("reasons") or []), "under_branches": wt.get("under_branches"), + "create_issue_bootstrap_waived": bool(bootstrap_waived), } - if wt.get("block"): + if wt.get("block") and not bootstrap_waived: blockers.append( _blocker( BLOCKER_WRONG_WORKTREE, diff --git a/create_issue_bootstrap.py b/create_issue_bootstrap.py index 74c8e15..0c188f1 100644 --- a/create_issue_bootstrap.py +++ b/create_issue_bootstrap.py @@ -179,6 +179,76 @@ def assess_create_issue_bootstrap( ) +def bootstrap_permits_control_checkout( + assessment: Any, + *, + task: str | None, + workspace_path: str | None, + canonical_repo_root: str | None, +) -> bool: + """Single interpretation of a bootstrap assessment (#757). + + Both author-mutation guards — the #274 branches-only enforcer and the #604 + anti-stomp preflight — route their "may this workspace mutate" decision + through this predicate, so the two can never reach opposite conclusions + about identical evidence. + + Fail-closed by construction. Every proof obligation must be present and + affirmative in *assessment*, and the assessment must describe the very + workspace and canonical root being guarded. A missing, malformed, refused, + incomplete, or contradictory assessment returns ``False``, which leaves the + caller's ordinary block in force. + + ``assessment`` is server-derived only: it is produced by + :func:`assess_create_issue_bootstrap` from inspected repository state. It is + never accepted from an MCP tool argument, so no caller can assert + eligibility it has not proven. + """ + if not isinstance(assessment, dict): + return False + if not is_create_issue_task(task): + return False + + # Positive proof: the assessment must affirmatively allow, with no + # competing refusal or not-applicable disposition recorded alongside it. + if assessment.get("allowed") is not True: + return False + if assessment.get("proven") is not True: + return False + if assessment.get("block") is not False: + return False + if assessment.get("not_applicable") is not False: + return False + if assessment.get("reasons"): + return False + + # Scope proof: only the create_issue bootstrap, only via the clean + # canonical control checkout path. + if assessment.get("task_scope") != "create_issue_only": + return False + if assessment.get("bootstrap_path") != "clean_canonical_control_checkout": + return False + + # State proof: clean, and not a branches/ worktree (those keep #274). + if assessment.get("dirty_files"): + return False + if assessment.get("under_branches") is not False: + return False + + # Binding proof: the assessment must describe *this* workspace and root, + # and that workspace must be exactly the canonical control checkout. + root = os.path.realpath(canonical_repo_root or "") + workspace = os.path.realpath(workspace_path or root or ".") + if not root or workspace != root: + return False + if assessment.get("canonical_repo_root") != root: + return False + if assessment.get("workspace_path") != workspace: + return False + + return True + + def format_create_issue_bootstrap_error(assessment: dict[str, Any]) -> str: """RuntimeError / typed-block message for a failed bootstrap assessment.""" reasons = "; ".join( diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 5e45fff..d495ab6 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -817,10 +817,56 @@ def _enforce_canonical_repository_root( ) +# #757: distinguishes "caller supplied no bootstrap evidence" (compute it) from +# "caller supplied a not-applicable/refused assessment" (honour it, fail closed). +_BOOTSTRAP_UNSET = object() + + +def _create_issue_bootstrap_assessment( + task: str | None, + worktree_path: str | None = None, +) -> dict | None: + """#757: the single server-derived create_issue bootstrap assessment. + + Computed once per preflight from inspected repository state, then consumed + by BOTH the #274 branches-only guard and the #604 anti-stomp preflight so + the two cannot reach opposite conclusions about identical evidence. + + Returns ``None`` when *task* is not a create_issue mutation, which leaves + every other author mutation on the ordinary branches-only path. The result + is derived from inspected repository state only — never from an MCP tool + argument. + """ + import create_issue_bootstrap as _cib + + if not _cib.is_create_issue_task(task): + return None + + ctx = _resolve_namespace_mutation_context(worktree_path) + workspace = ctx["workspace_path"] + git_state = issue_lock_worktree.read_worktree_git_state(workspace) + try: + remote_master_sha = root_checkout_guard.resolve_remote_master_sha( + ctx["canonical_repo_root"] + ) + except Exception: + remote_master_sha = None + return _cib.assess_create_issue_bootstrap( + workspace_path=workspace, + canonical_repo_root=ctx["canonical_repo_root"], + current_branch=git_state.get("current_branch"), + head_sha=git_state.get("head_sha"), + porcelain_status=git_state.get("porcelain_status") or "", + remote_master_sha=remote_master_sha, + task=task, + ) + + def _enforce_branches_only_author_mutation( worktree_path: str | None = None, *, task: str | None = None, + bootstrap_assessment: Any = _BOOTSTRAP_UNSET, ) -> None: """#274: author file/branch mutations must run from a branches/ worktree. @@ -859,27 +905,29 @@ def _enforce_branches_only_author_mutation( return # #749 create_issue bootstrap: narrow phase exemption only. + # #757: consume the caller-computed assessment when one was threaded in, so + # this guard and the later #604 anti-stomp guard judge identical evidence. + # Falling back to computing it here preserves behaviour for callers that + # supply none. import create_issue_bootstrap as _cib - remote_master_sha = None - try: - remote_master_sha = root_checkout_guard.resolve_remote_master_sha( - ctx["canonical_repo_root"] - ) - except Exception: - remote_master_sha = None - bootstrap = _cib.assess_create_issue_bootstrap( + bootstrap = ( + _create_issue_bootstrap_assessment(task, worktree_path) + if bootstrap_assessment is _BOOTSTRAP_UNSET + else bootstrap_assessment + ) + if _cib.bootstrap_permits_control_checkout( + bootstrap, + task=task, workspace_path=workspace, canonical_repo_root=ctx["canonical_repo_root"], - current_branch=git_state.get("current_branch"), - head_sha=git_state.get("head_sha"), - porcelain_status=git_state.get("porcelain_status") or "", - remote_master_sha=remote_master_sha, - task=task, - ) - if bootstrap.get("allowed"): + ): return - if bootstrap.get("block") and not bootstrap.get("not_applicable"): + if ( + isinstance(bootstrap, dict) + and bootstrap.get("block") + and not bootstrap.get("not_applicable") + ): raise RuntimeError(_cib.format_create_issue_bootstrap_error(bootstrap)) raise RuntimeError( author_mutation_worktree.format_author_mutation_worktree_error(assessment) @@ -926,6 +974,7 @@ def _run_anti_stomp_preflight( workflow_hash_valid: bool | None = None, workflow_hash_reasons: list[str] | None = None, raise_on_block: bool = True, + bootstrap_assessment: Any = _BOOTSTRAP_UNSET, ) -> dict | None: """Shared #604 anti-stomp preflight for MCP mutation entrypoints. @@ -1106,6 +1155,14 @@ def _run_anti_stomp_preflight( source_contaminated=source_contaminated, contamination_reasons=contamination_reasons, manual_bypass_attempted=False, + # #757: same server-derived bootstrap decision the #274 guard consumed. + # Computing it here when unsupplied keeps standalone callers consistent + # rather than silently bootstrap-blind. + create_issue_bootstrap_assessment=( + _create_issue_bootstrap_assessment(task, worktree_path) + if bootstrap_assessment is _BOOTSTRAP_UNSET + else bootstrap_assessment + ), ) if not assessment.get("block"): return None @@ -1248,7 +1305,12 @@ def verify_preflight_purity( # Historical path: root + branches after purity-order when dirty paths live. _enforce_canonical_repository_root(worktree_path, remote=remote) _enforce_root_checkout_guard(worktree_path) - _enforce_branches_only_author_mutation(worktree_path, task=task) + # #757: compute the create_issue bootstrap decision ONCE and hand the + # same result to both the #274 and #604 guards, so they cannot disagree. + bootstrap_assessment = _create_issue_bootstrap_assessment(task, worktree_path) + _enforce_branches_only_author_mutation( + worktree_path, task=task, bootstrap_assessment=bootstrap_assessment + ) _enforce_issue_scope_guard( worktree_path, task=task, @@ -1258,6 +1320,7 @@ def verify_preflight_purity( # #604: common anti-stomp preflight after legacy + #683 enforcers. _run_anti_stomp_preflight( task, + bootstrap_assessment=bootstrap_assessment, remote=remote, worktree_path=worktree_path, org=org, @@ -1282,7 +1345,11 @@ def verify_preflight_purity( if production_active: _enforce_canonical_repository_root(worktree_path, remote=remote) _enforce_root_checkout_guard(worktree_path) - _enforce_branches_only_author_mutation(worktree_path, task=task) + # #757: one shared bootstrap decision for both guards (see above). + bootstrap_assessment = _create_issue_bootstrap_assessment(task, worktree_path) + _enforce_branches_only_author_mutation( + worktree_path, task=task, bootstrap_assessment=bootstrap_assessment + ) _enforce_issue_scope_guard( worktree_path, task=task, @@ -1292,6 +1359,7 @@ def verify_preflight_purity( if force_anti_stomp: _run_anti_stomp_preflight( task, + bootstrap_assessment=bootstrap_assessment, remote=remote, worktree_path=worktree_path, org=org, diff --git a/tests/test_issue_757_bootstrap_guard_agreement.py b/tests/test_issue_757_bootstrap_guard_agreement.py new file mode 100644 index 0000000..9b8b05e --- /dev/null +++ b/tests/test_issue_757_bootstrap_guard_agreement.py @@ -0,0 +1,613 @@ +"""Regression tests for #757: #274 and #604 must not disagree. + +The sanctioned ``create_issue`` bootstrap (#749/#750) was unreachable in +production: the #274 branches-only guard consulted the bootstrap and permitted +a clean canonical control checkout, then the bootstrap-blind #604 anti-stomp +preflight rejected the same checkout as ``wrong_worktree``. + +These tests pin the fix: + +* one server-derived assessment, interpreted by one shared predicate; +* the waiver is narrow (only the wrong-worktree verdict, only create_issue, + only from the exact clean canonical control checkout); +* every other guard and rejection reason keeps its fail-closed behaviour; +* eligibility cannot be forged through a public tool signature. +""" + +from __future__ import annotations + +import inspect +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import anti_stomp_preflight as asp # noqa: E402 +import create_issue_bootstrap as cib # noqa: E402 +import gitea_mcp_server as srv # noqa: E402 + +FAKE_AUTH = {"Authorization": "token test-token"} +MASTER_SHA = "a" * 40 +STALE_SHA = "b" * 40 + +current_file_path = Path(__file__).resolve() +if "branches" in current_file_path.parts: + CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[3]) +else: + CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[1]) + +BRANCHES_WORKTREE = str(Path(CONTROL_CHECKOUT_ROOT) / "branches" / "issue-1-x") + + +def proven_bootstrap( + *, + workspace=CONTROL_CHECKOUT_ROOT, + root=CONTROL_CHECKOUT_ROOT, + task="create_issue", + branch="master", + porcelain="", + head=MASTER_SHA, + remote_sha=MASTER_SHA, +): + """Build a real assessment via the production assessor (never hand-rolled).""" + return cib.assess_create_issue_bootstrap( + workspace_path=workspace, + canonical_repo_root=root, + current_branch=branch, + head_sha=head, + porcelain_status=porcelain, + remote_master_sha=remote_sha, + task=task, + ) + + +class TestSharedPredicate(unittest.TestCase): + """The one interpretation both guards consume (AC6, AC7).""" + + def _permits(self, assessment, task="create_issue", workspace=None, root=None): + return cib.bootstrap_permits_control_checkout( + assessment, + task=task, + workspace_path=workspace or CONTROL_CHECKOUT_ROOT, + canonical_repo_root=root or CONTROL_CHECKOUT_ROOT, + ) + + def test_proven_bootstrap_permits(self): + self.assertTrue(self._permits(proven_bootstrap())) + + def test_tool_alias_permits(self): + assessment = proven_bootstrap(task="gitea_create_issue") + self.assertTrue(self._permits(assessment, task="gitea_create_issue")) + + def test_missing_assessment_fails_closed(self): + self.assertFalse(self._permits(None)) + + def test_malformed_assessment_fails_closed(self): + for bogus in ("allowed", 1, True, [], ["allowed"], object()): + with self.subTest(bogus=bogus): + self.assertFalse(self._permits(bogus)) + + def test_refused_assessment_fails_closed(self): + # Dirty control checkout -> assessor blocks -> predicate must refuse. + self.assertFalse( + self._permits(proven_bootstrap(porcelain=" M gitea_mcp_server.py\n")) + ) + + def test_not_applicable_assessment_fails_closed(self): + # branches/ worktree -> not_applicable -> no waiver. + self.assertFalse( + self._permits(proven_bootstrap(workspace=BRANCHES_WORKTREE)) + ) + + def test_incomplete_assessment_fails_closed(self): + incomplete = dict(proven_bootstrap()) + del incomplete["bootstrap_path"] + self.assertFalse(self._permits(incomplete)) + + def test_contradictory_block_and_allow_fails_closed(self): + contradictory = dict(proven_bootstrap(), block=True) + self.assertFalse(self._permits(contradictory)) + + def test_contradictory_dirty_but_allowed_fails_closed(self): + contradictory = dict(proven_bootstrap(), dirty_files=["gitea_mcp_server.py"]) + self.assertFalse(self._permits(contradictory)) + + def test_contradictory_under_branches_but_allowed_fails_closed(self): + contradictory = dict(proven_bootstrap(), under_branches=True) + self.assertFalse(self._permits(contradictory)) + + def test_contradictory_reasons_present_fails_closed(self): + contradictory = dict(proven_bootstrap(), reasons=["something refused"]) + self.assertFalse(self._permits(contradictory)) + + def test_truthy_non_true_values_fail_closed(self): + """Strict identity: no truthy smuggling (1, 'yes') can assert eligibility.""" + for value in (1, "yes", "true", [1]): + with self.subTest(value=value): + self.assertFalse(self._permits(dict(proven_bootstrap(), allowed=value))) + + def test_wrong_task_fails_closed(self): + """An otherwise-proven assessment cannot license a different task.""" + self.assertFalse(self._permits(proven_bootstrap(), task="lock_issue")) + self.assertFalse(self._permits(proven_bootstrap(), task="create_pr")) + self.assertFalse(self._permits(proven_bootstrap(), task=None)) + + def test_foreign_workspace_binding_fails_closed(self): + """Assessment must describe the workspace actually being guarded.""" + other = proven_bootstrap(workspace="/other/clone", root="/other/clone") + self.assertFalse(self._permits(other)) + + def test_scope_and_path_tampering_fails_closed(self): + self.assertFalse( + self._permits(dict(proven_bootstrap(), task_scope="all_tasks")) + ) + self.assertFalse( + self._permits(dict(proven_bootstrap(), bootstrap_path="anything_goes")) + ) + + +class TestAntiStompHonorsBootstrap(unittest.TestCase): + """#604 consumes the same decision, and waives only wrong_worktree.""" + + def _assess(self, *, task="create_issue", bootstrap=None, workspace=None, **kw): + params = dict( + task=task, + profile_role="author", + required_role="author", + workspace_path=workspace or CONTROL_CHECKOUT_ROOT, + project_root=CONTROL_CHECKOUT_ROOT, + current_branch="master", + root_head_sha=MASTER_SHA, + root_porcelain="", + remote_master_sha=MASTER_SHA, + check_repo=False, + create_issue_bootstrap_assessment=bootstrap, + ) + params.update(kw) + return asp.assess_anti_stomp_preflight(**params) + + def _blocker_kinds(self, result): + return {b["kind"] for b in result.get("blockers") or []} + + def test_control_checkout_without_bootstrap_still_blocked(self): + """Baseline: the exact production failure, unwaived.""" + res = self._assess(bootstrap=None) + self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)) + + def test_control_checkout_with_proven_bootstrap_permitted(self): + """AC1/AC2: the #757 fix — same inputs, bootstrap honoured.""" + res = self._assess(bootstrap=proven_bootstrap()) + self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)) + self.assertTrue(res["checks"]["worktree"]["create_issue_bootstrap_waived"]) + self.assertFalse(res["checks"]["worktree"]["block"]) + + def test_tool_alias_permitted(self): + res = self._assess( + task="gitea_create_issue", + bootstrap=proven_bootstrap(task="gitea_create_issue"), + ) + self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)) + + def test_non_create_issue_task_never_waived(self): + """AC5: other author mutations keep the branches-only requirement.""" + for task in ("lock_issue", "create_pr", "commit_files", "mark_issue"): + with self.subTest(task=task): + res = self._assess(task=task, bootstrap=proven_bootstrap()) + self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)) + + def test_dirty_control_checkout_still_blocked(self): + """AC4: dirty root refuses the bootstrap, so no waiver.""" + dirty = " M gitea_mcp_server.py\n" + res = self._assess( + bootstrap=proven_bootstrap(porcelain=dirty), root_porcelain=dirty + ) + self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)) + + def test_detached_control_checkout_still_blocked(self): + res = self._assess( + bootstrap=proven_bootstrap(branch=""), current_branch="" + ) + self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)) + + def test_base_divergence_still_blocked(self): + res = self._assess( + bootstrap=proven_bootstrap(head=STALE_SHA), root_head_sha=STALE_SHA + ) + self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)) + + def test_waiver_does_not_suppress_stale_runtime(self): + """AC: waive only wrong_worktree — stale runtime still fails closed.""" + res = self._assess( + bootstrap=proven_bootstrap(), + startup_head=STALE_SHA, + current_code_head=MASTER_SHA, + ) + self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)) + self.assertIn(asp.BLOCKER_STALE_RUNTIME, self._blocker_kinds(res)) + self.assertTrue(res["block"]) + + def test_waiver_does_not_suppress_wrong_repo(self): + res = self._assess( + bootstrap=proven_bootstrap(), + check_repo=True, + remote="prgs", + resolved_org="Scaled-Tech-Consulting", + resolved_repo="Timesheet", + # Not both-explicit, so the #530 repo guard actually evaluates the + # mismatch against the local remote instead of trusting the caller. + org_explicit=False, + repo_explicit=False, + local_remote_url=( + "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" + ), + ) + self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)) + self.assertIn(asp.BLOCKER_WRONG_REPO, self._blocker_kinds(res)) + + def test_waiver_does_not_suppress_wrong_role(self): + res = self._assess( + bootstrap=proven_bootstrap(), + profile_role="reviewer", + required_role="author", + ) + self.assertIn(asp.BLOCKER_WRONG_ROLE, self._blocker_kinds(res)) + + def test_branches_worktree_unaffected(self): + """AC: ordinary branches/ worktrees keep working, waived or not.""" + for bootstrap in (None, proven_bootstrap(workspace=BRANCHES_WORKTREE)): + with self.subTest(bootstrap=bool(bootstrap)): + res = self._assess( + workspace=BRANCHES_WORKTREE, + bootstrap=bootstrap, + current_branch="fix/issue-1-x", + ) + self.assertNotIn( + asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res) + ) + + def test_forged_assessment_rejected(self): + """AC6: a hand-built 'allowed' dict cannot unlock the waiver.""" + forged = {"allowed": True, "block": False} + res = self._assess(bootstrap=forged) + self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)) + + +class TestGuardAgreement(unittest.TestCase): + """AC7: the regression that would have caught the #757 defect. + + For identical evidence, the #274 guard and the #604 guard must return the + same wrong-worktree verdict across the full workspace-state matrix. + """ + + MATRIX = [ + ( + "clean control + create_issue", + CONTROL_CHECKOUT_ROOT, "create_issue", "master", "", MASTER_SHA, + ), + ( + "clean control + alias", + CONTROL_CHECKOUT_ROOT, "gitea_create_issue", "master", "", MASTER_SHA, + ), + ( + "clean control + lock_issue", + CONTROL_CHECKOUT_ROOT, "lock_issue", "master", "", MASTER_SHA, + ), + ( + "clean control + create_pr", + CONTROL_CHECKOUT_ROOT, "create_pr", "master", "", MASTER_SHA, + ), + ( + "dirty control + create_issue", + CONTROL_CHECKOUT_ROOT, "create_issue", "master", + " M gitea_mcp_server.py\n", MASTER_SHA, + ), + ( + "detached control + create_issue", + CONTROL_CHECKOUT_ROOT, "create_issue", "", "", MASTER_SHA, + ), + ( + "non-base control + create_issue", + CONTROL_CHECKOUT_ROOT, "create_issue", "feat/x", "", MASTER_SHA, + ), + ( + "diverged control + create_issue", + CONTROL_CHECKOUT_ROOT, "create_issue", "master", "", STALE_SHA, + ), + ( + "branches wt + create_issue", + BRANCHES_WORKTREE, "create_issue", "fix/issue-1-x", "", MASTER_SHA, + ), + ( + "branches wt + lock_issue", + BRANCHES_WORKTREE, "lock_issue", "fix/issue-1-x", "", MASTER_SHA, + ), + ] + + def _guard_274_blocks(self, *, workspace, task, branch, porcelain, head, bootstrap): + git_state = { + "current_branch": branch, + "head_sha": head, + "porcelain_status": porcelain, + } + ctx = { + "workspace_path": workspace, + "canonical_repo_root": CONTROL_CHECKOUT_ROOT, + } + with patch.object(srv, "_effective_workspace_role", return_value="author"), \ + patch.object(srv, "_actual_profile_role", return_value="author"), \ + patch.object( + srv, "_resolve_namespace_mutation_context", return_value=ctx + ), \ + patch.object( + srv.issue_lock_worktree, + "read_worktree_git_state", + return_value=git_state, + ): + try: + srv._enforce_branches_only_author_mutation( + workspace, task=task, bootstrap_assessment=bootstrap + ) + return False + except RuntimeError: + return True + + def _guard_604_blocks(self, *, workspace, task, branch, porcelain, head, bootstrap): + res = asp.assess_anti_stomp_preflight( + task=task, + profile_role="author", + required_role="author", + workspace_path=workspace, + project_root=CONTROL_CHECKOUT_ROOT, + current_branch=branch, + root_head_sha=head, + root_porcelain=porcelain, + remote_master_sha=MASTER_SHA, + check_repo=False, + create_issue_bootstrap_assessment=bootstrap, + ) + return asp.BLOCKER_WRONG_WORKTREE in { + b["kind"] for b in res.get("blockers") or [] + } + + def test_guards_agree_across_matrix(self): + for label, workspace, task, branch, porcelain, head in self.MATRIX: + with self.subTest(case=label): + # ONE server-derived assessment, exactly as production computes it. + bootstrap = cib.assess_create_issue_bootstrap( + workspace_path=workspace, + canonical_repo_root=CONTROL_CHECKOUT_ROOT, + current_branch=branch, + head_sha=head, + porcelain_status=porcelain, + remote_master_sha=MASTER_SHA, + task=task, + ) + kwargs = dict( + workspace=workspace, + task=task, + branch=branch, + porcelain=porcelain, + head=head, + bootstrap=bootstrap, + ) + blocked_274 = self._guard_274_blocks(**kwargs) + blocked_604 = self._guard_604_blocks(**kwargs) + self.assertEqual( + blocked_274, + blocked_604, + f"{label}: #274 blocked={blocked_274} " + f"but #604 blocked={blocked_604}", + ) + + def test_clean_control_create_issue_permitted_by_both(self): + """The specific case that was broken: both guards must permit.""" + kwargs = dict( + workspace=CONTROL_CHECKOUT_ROOT, + task="create_issue", + branch="master", + porcelain="", + head=MASTER_SHA, + bootstrap=proven_bootstrap(), + ) + self.assertFalse(self._guard_274_blocks(**kwargs)) + self.assertFalse(self._guard_604_blocks(**kwargs)) + + +class TestNoCallerForgeableEligibility(unittest.TestCase): + """AC6: eligibility is never reachable through a public tool signature.""" + + def test_create_issue_tool_exposes_no_bootstrap_argument(self): + params = set(inspect.signature(srv.gitea_create_issue).parameters) + for forbidden in ( + "bootstrap", + "bootstrap_assessment", + "create_issue_bootstrap", + "create_issue_bootstrap_assessment", + "allow_control_checkout", + "bootstrap_allowed", + ): + self.assertNotIn(forbidden, params) + + def test_no_mcp_tool_exposes_bootstrap_argument(self): + """No public gitea_* tool may take bootstrap evidence from the caller.""" + for name in dir(srv): + if not name.startswith("gitea_"): + continue + fn = getattr(srv, name) + if not callable(fn): + continue + try: + params = set(inspect.signature(fn).parameters) + except (TypeError, ValueError): + continue + leaked = {p for p in params if "bootstrap" in p.lower()} + self.assertFalse(leaked, f"{name} exposes bootstrap args: {leaked}") + + def test_internal_helper_yields_nothing_for_other_tasks(self): + self.assertTrue(hasattr(srv, "_create_issue_bootstrap_assessment")) + self.assertIsNone( + srv._create_issue_bootstrap_assessment("lock_issue"), + "non-create_issue tasks must yield no bootstrap evidence", + ) + + +class TestNativeCreateIssueEndToEnd(unittest.TestCase): + """AC8: the production handler, with the #604 gate LIVE (not patched out). + + The pre-existing #749 e2e test patched ``_run_anti_stomp_preflight`` to a + no-op, which is exactly why this defect reached production. These tests + leave it running. + """ + + def setUp(self): + srv._preflight_whoami_called = True + srv._preflight_capability_called = True + srv._preflight_resolved_role = "author" + srv._preflight_resolved_task = "create_issue" + 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 + srv._preflight_resolved_task = None + + def _git_state(self, branch="master", head=MASTER_SHA, porcelain=""): + return { + "current_branch": branch, + "head_sha": head, + "porcelain_status": porcelain, + } + + def _run_create_issue( + self, + *, + git_state, + remote_sha=MASTER_SHA, + parity=None, + title="Bootstrap issue from clean control", + ): + """Invoke the native handler with the anti-stomp gate live.""" + parity = parity or { + "startup_head": MASTER_SHA, + "current_head": MASTER_SHA, + } + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \ + patch.object(srv, "_auth", return_value=FAKE_AUTH), \ + patch.object(srv, "_profile_permission_block", return_value=None), \ + patch.object(srv, "_namespace_mutation_block", return_value=None), \ + patch.object( + srv.role_session_router, + "check_author_mutation_after_reviewer_stop", + return_value=(True, []), + ), \ + patch.object(srv, "api_get_all", return_value=[]), \ + patch.object(srv, "api_request") as mock_api, \ + patch.object( + srv.root_checkout_guard, + "resolve_remote_master_sha", + return_value=remote_sha, + ), \ + patch.object(srv, "_current_master_parity", return_value=parity), \ + patch.object( + srv, + "get_profile", + return_value={ + "profile_name": "prgs-author", + "allowed_operations": [ + "gitea.issue.create", + "gitea.issue.comment", + "gitea.pr.create", + "gitea.read", + ], + }, + ), \ + patch.object(srv, "_actual_profile_role", return_value="author"), \ + patch.object(srv, "_effective_workspace_role", return_value="author"), \ + patch.object( + srv.issue_lock_worktree, + "read_worktree_git_state", + return_value=git_state, + ), \ + patch.object( + srv, + "_get_workspace_porcelain", + return_value=git_state["porcelain_status"], + ), \ + patch.object(srv, "_enforce_root_checkout_guard"): + mock_api.return_value = { + "number": 99, + "html_url": "https://gitea.example.com/issues/99", + } + try: + result = srv.gitea_create_issue( + title=title, + body="Body text for content gate.", + ) + except RuntimeError as exc: + return {"raised": str(exc)}, mock_api + return result, mock_api + + def test_clean_control_checkout_reaches_api(self): + """The exact production failure that blocked filing #757 and #758.""" + res, mock_api = self._run_create_issue(git_state=self._git_state()) + self.assertNotIn("raised", res, f"guard still blocks: {res.get('raised')}") + self.assertEqual(res.get("number"), 99) + mock_api.assert_called_once() + + def test_dirty_control_checkout_blocked(self): + dirty = " M gitea_mcp_server.py\n" + res, mock_api = self._run_create_issue( + git_state=self._git_state(porcelain=dirty) + ) + self.assertFalse(isinstance(res, dict) and res.get("number") == 99) + mock_api.assert_not_called() + + def test_detached_control_checkout_blocked(self): + res, mock_api = self._run_create_issue(git_state=self._git_state(branch="")) + self.assertFalse(isinstance(res, dict) and res.get("number") == 99) + mock_api.assert_not_called() + + def test_base_mismatch_blocked(self): + res, mock_api = self._run_create_issue( + git_state=self._git_state(head=STALE_SHA) + ) + self.assertFalse(isinstance(res, dict) and res.get("number") == 99) + mock_api.assert_not_called() + + def test_stale_runtime_blocked(self): + res, mock_api = self._run_create_issue( + git_state=self._git_state(), + parity={"startup_head": STALE_SHA, "current_head": MASTER_SHA}, + ) + self.assertFalse(isinstance(res, dict) and res.get("number") == 99) + mock_api.assert_not_called() + + def test_non_create_issue_mutation_still_blocked_from_control(self): + """AC5 through the real preflight, with anti-stomp live.""" + srv._preflight_resolved_task = "lock_issue" + with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \ + patch.object(srv, "_actual_profile_role", return_value="author"), \ + patch.object(srv, "_effective_workspace_role", return_value="author"), \ + patch.object( + srv.issue_lock_worktree, + "read_worktree_git_state", + return_value=self._git_state(), + ), \ + patch.object(srv, "_get_workspace_porcelain", return_value=""), \ + patch.object( + srv.root_checkout_guard, + "resolve_remote_master_sha", + return_value=MASTER_SHA, + ), \ + patch.object(srv, "_enforce_root_checkout_guard"): + with self.assertRaises(RuntimeError) as ctx: + srv.verify_preflight_purity(remote="prgs", task="lock_issue") + self.assertIn("control checkout", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_reconciler_close_workspace_guard.py b/tests/test_reconciler_close_workspace_guard.py index 43682a0..5343cae 100644 --- a/tests/test_reconciler_close_workspace_guard.py +++ b/tests/test_reconciler_close_workspace_guard.py @@ -38,11 +38,17 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase): srv._preflight_capability_violation = False self._orig_in_test = srv._preflight_in_test_mode srv._preflight_in_test_mode = lambda: False + self._orig_resolved_task = srv._preflight_resolved_task + self._orig_resolved_role = srv._preflight_resolved_role self._env_patch = patch.dict(os.environ, {"GITEA_MCP_DISABLE_PARITY_GATE": "1"}, clear=False) self._env_patch.start() def tearDown(self): srv._preflight_in_test_mode = self._orig_in_test + # Preflight task/role are module-level; restore so test order cannot + # leak a resolved task into sibling cases. + srv._preflight_resolved_task = self._orig_resolved_task + srv._preflight_resolved_role = self._orig_resolved_role self._env_patch.stop() @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @@ -81,26 +87,28 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase): "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( + def test_author_non_create_issue_still_blocked_on_control_checkout( self, _git, _get_all, _role, _ns, _prof, _auth ): + """Author mutations other than create_issue keep the branches-only rule. + + #749/#750 sanctioned ``create_issue`` from a clean control checkout, and + #757 made the #604 anti-stomp guard honour that same decision — so + ``create_issue`` is no longer a valid probe for this boundary. This case + previously asserted create_issue stayed blocked, which only held because + the bootstrap-blind #604 guard was overriding #750; that is precisely + the defect #757 fixed. ``lock_issue`` is issue-backed and post-ownership, + so it still requires a ``branches/`` worktree. + """ srv._preflight_resolved_role = "author" + srv._preflight_resolved_task = "lock_issue" with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): try: - res = srv.gitea_create_issue(title="Test", body="body") + srv.verify_preflight_purity(remote="prgs", task="lock_issue") except RuntimeError as exc: - self.assertIn("stable control checkout", str(exc)) + self.assertIn("control checkout", str(exc).lower()) else: - # #683 typed blocker at mutation entrypoint - self.assertFalse(res.get("success")) - blob = " ".join(res.get("reasons") or []) + str( - res.get("blocker_kind") or "" - ) - self.assertTrue( - "stable control checkout" in blob - or "missing_issue_worktree" in blob - or "control checkout" in blob.lower() - ) + self.fail("lock_issue must stay blocked on the control checkout") if __name__ == "__main__":