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) <[email protected]>

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,

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(

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,

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()

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__":