Merge pull request 'fix(scope): wire author bootstrap scope into workflow scope guard' (#942) from fix/issue-941-scope-guard-bootstrap-wiring into master

gitea_mcp_server.py

@@ -1546,6 +1546,7 @@ def verify_preflight_purity(
             task=task,
             target_issue_number=target_issue_number,
             require_author_lock=require_author_lock,
+            bootstrap_assessment=bootstrap_assessment,
         )
         # #604: common anti-stomp preflight after legacy + #683 enforcers.
         _run_anti_stomp_preflight(
@@ -1585,6 +1586,7 @@ def verify_preflight_purity(
             task=task,
             target_issue_number=target_issue_number,
             require_author_lock=require_author_lock,
+            bootstrap_assessment=bootstrap_assessment,
         )
         if force_anti_stomp:
             _run_anti_stomp_preflight(
@@ -1652,8 +1654,15 @@ def _enforce_issue_scope_guard(
     task: str | None = None,
     target_issue_number: int | None = None,
     require_author_lock: bool = False,
+    bootstrap_assessment: object = _BOOTSTRAP_UNSET,
 ) -> None:
-    """#683: fail closed on missing/out-of-scope issue ownership for mutations."""
+    """#683: fail closed on missing/out-of-scope issue ownership for mutations.
+
+    #941: the shared bootstrap assessment is threaded in so this guard judges
+    the author issue-worktree bootstrap on the same server-derived evidence as
+    the #274 branches-only and #604 anti-stomp guards. Callers that supply
+    none fall back to computing it here, which preserves behaviour.
+    """
     ctx = _resolve_namespace_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
     git_state = issue_lock_worktree.read_worktree_git_state(workspace)
@@ -1703,11 +1712,32 @@ def _enforce_issue_scope_guard(
     import create_issue_bootstrap as _cib
 
     is_create_issue = _cib.is_create_issue_task(task)
+    # #941: consume the caller-computed bootstrap assessment when one was
+    # threaded in, so this guard and the #274/#604 guards judge identical
+    # evidence. Falling back preserves behaviour for callers that supply none.
+    bootstrap = (
+        _create_issue_bootstrap_assessment(task, worktree_path)
+        if bootstrap_assessment is _BOOTSTRAP_UNSET
+        else bootstrap_assessment
+    )
+    # #941: the author issue-worktree bootstrap is pre-ownership for the same
+    # reason create_issue is — it exists to break the lock<->worktree cycle, so
+    # no owning lock can exist yet. The exemption is granted by the canonical
+    # shared decision over server-derived evidence, never by a task-name list,
+    # and fails closed on missing, malformed, cross-scope, dirty, drifted, or
+    # wrongly bound evidence.
+    bootstrap_waives_ownership = _cib.bootstrap_permits_control_checkout(
+        bootstrap,
+        task=task,
+        workspace_path=workspace,
+        canonical_repo_root=ctx["canonical_repo_root"],
+    )
     require_lock = bool(require_author_lock) or (
         authorish
         and workflow_scope_guard.production_guards_forced()
         and role == "author"
         and not is_create_issue
+        and not bootstrap_waives_ownership
     )
     assessment = workflow_scope_guard.assess_production_mutation_guards(
         workspace_path=workspace,
@@ -1720,6 +1750,7 @@ def _enforce_issue_scope_guard(
         require_author_lock=require_lock,
         in_test_mode=_preflight_in_test_mode(),
         mutation_task=task,
+        bootstrap_assessment=bootstrap,
     )
     workflow_scope_guard.raise_if_blocked(assessment)
 

tests/test_issue_941_scope_guard_bootstrap_wiring.py

@@ -0,0 +1,346 @@
+"""Regression: author bootstrap scope reaches workflow_scope_guard (#941).
+
+PR #926 (#892) made ``bootstrap_permits_control_checkout`` accept
+``task_scope=author_issue_bootstrap`` and wired that canonical decision into
+the #274 branches-only enforcer and the #604 anti-stomp preflight. A third
+enforcement path was left unwired.
+
+``workflow_scope_guard.assess_root_source_mutation`` kept its own copy of the
+clean-root author decision, gated on ``create_issue_bootstrap.is_create_issue_task``
+— a task-name allowlist that never contained ``bootstrap_author_issue_worktree``.
+So the real call path
+
+    gitea_bootstrap_author_issue_worktree
+      -> verify_preflight_purity
+        -> _enforce_issue_scope_guard
+          -> workflow_scope_guard.assess_production_mutation_guards
+
+raised ProductionGuardError(missing_issue_worktree) before
+``assess_author_issue_bootstrap`` was ever consulted.
+
+These tests drive the real enforcer, not the authorization helper in
+isolation. A helper-only test cannot observe this defect: #892's own predicate
+tests all passed while the live bootstrap stayed blocked.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import tempfile
+import unittest
+from unittest import mock
+
+import author_issue_bootstrap as aib
+import create_issue_bootstrap as cib
+import workflow_scope_guard
+
+BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
+BOOTSTRAP_TOOL = "gitea_bootstrap_author_issue_worktree"
+
+
+def _make_control_repo(tmp: str) -> tuple[str, str]:
+    """Create a clean control checkout on master and return (path, head)."""
+    repo = os.path.join(tmp, "repo")
+    os.makedirs(os.path.join(repo, "branches"))
+    subprocess.check_call(
+        ["git", "init", "-b", "master", repo],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    subprocess.check_call(
+        [
+            "git", "-C", repo,
+            "-c", "user.email=t@t", "-c", "user.name=t",
+            "commit", "--allow-empty", "-m", "init",
+        ],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    head = subprocess.check_output(
+        ["git", "-C", repo, "rev-parse", "HEAD"], text=True
+    ).strip()
+    return repo, head
+
+
+def _assessment(
+    repo: str,
+    head: str,
+    *,
+    task: str = BOOTSTRAP_TASK,
+    porcelain: str = "",
+    remote: str | None = None,
+) -> dict:
+    return aib.assess_author_issue_bootstrap(
+        workspace_path=repo,
+        canonical_repo_root=repo,
+        current_branch="master",
+        head_sha=head,
+        porcelain_status=porcelain,
+        remote_master_sha=head if remote is None else remote,
+        task=task,
+    )
+
+
+class _ControlCheckoutHarness(unittest.TestCase):
+    """Drive the real server guard against a temporary clean control checkout."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.repo, self.head = _make_control_repo(self._tmp.name)
+
+        # #683 force-on: production guards must execute under pytest.
+        patcher = mock.patch.dict(
+            os.environ,
+            {workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _enforce(
+        self,
+        task: str,
+        *,
+        porcelain: str = "",
+        assessment: object = "auto",
+        role_kind: str = "author",
+    ):
+        """Call the real _enforce_issue_scope_guard for *task*."""
+        import gitea_mcp_server as srv
+
+        if assessment == "auto":
+            assessment = _assessment(
+                self.repo, self.head, task=task, porcelain=porcelain
+            )
+
+        ctx = {
+            "workspace_path": self.repo,
+            "canonical_repo_root": self.repo,
+            "workspace_role_kind": role_kind,
+            "workspace_binding_source": "process_project_root",
+        }
+        git_state = {
+            "current_branch": "master",
+            "head_sha": self.head,
+            "porcelain_status": porcelain,
+        }
+
+        with mock.patch.object(
+            srv, "_resolve_namespace_mutation_context", return_value=ctx
+        ), mock.patch.object(
+            srv.issue_lock_worktree,
+            "read_worktree_git_state",
+            return_value=git_state,
+        ), mock.patch.object(
+            srv,
+            "_session_issue_lock_snapshot",
+            return_value={
+                "locked_issue_number": None,
+                "lock_branch_name": None,
+                "worktrees_match": False,
+            },
+        ), mock.patch.object(
+            srv, "_actual_profile_role", return_value=role_kind
+        ), mock.patch.object(
+            srv, "_effective_workspace_role", return_value=role_kind
+        ), mock.patch.object(
+            srv, "_create_issue_bootstrap_assessment", return_value=assessment
+        ):
+            srv._enforce_issue_scope_guard(None, task=task)
+
+
+class TestRealPathBootstrapReachesGuard(_ControlCheckoutHarness):
+    """The defect and its fix, observed through the real enforcer."""
+
+    def test_bootstrap_task_passes_scope_guard_from_clean_control(self):
+        # Pre-fix this raises ProductionGuardError(missing_issue_worktree)
+        # because the guard consulted a task-name allowlist instead of the
+        # canonical authorization decision.
+        self._enforce(BOOTSTRAP_TASK)
+
+    def test_bootstrap_tool_alias_passes_scope_guard(self):
+        self._enforce(BOOTSTRAP_TOOL)
+
+    def test_guard_consults_canonical_predicate(self):
+        """The guard must reach bootstrap_permits_control_checkout, not a name list."""
+        real = cib.bootstrap_permits_control_checkout
+        seen: list[str | None] = []
+
+        def _spy(assessment, *, task, workspace_path, canonical_repo_root):
+            seen.append(task)
+            return real(
+                assessment,
+                task=task,
+                workspace_path=workspace_path,
+                canonical_repo_root=canonical_repo_root,
+            )
+
+        with mock.patch.object(
+            cib, "bootstrap_permits_control_checkout", side_effect=_spy
+        ):
+            self._enforce(BOOTSTRAP_TASK)
+
+        self.assertIn(
+            BOOTSTRAP_TASK,
+            seen,
+            "workflow_scope_guard did not consult the canonical bootstrap "
+            "authorization decision",
+        )
+
+
+class TestFailClosedOnBadEvidence(_ControlCheckoutHarness):
+    """Missing, malformed, or mismatched scope evidence must still block."""
+
+    def _assert_blocked(self, **kwargs):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce(BOOTSTRAP_TASK, **kwargs)
+
+    def test_missing_assessment_fails_closed(self):
+        self._assert_blocked(assessment=None)
+
+    def test_malformed_assessment_fails_closed(self):
+        self._assert_blocked(assessment={"allowed": True})
+
+    def test_non_dict_assessment_fails_closed(self):
+        self._assert_blocked(assessment="allowed")
+
+    def test_wrong_task_scope_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["task_scope"] = "create_issue_only"
+        self._assert_blocked(assessment=bad)
+
+    def test_nonempty_reasons_fail_closed(self):
+        bad = dict(_assessment(self.repo, self.head), reasons=["note"])
+        self._assert_blocked(assessment=bad)
+
+    def test_mismatched_base_tips_fail_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["remote_master_sha"] = "b" * 40
+        self._assert_blocked(assessment=bad)
+
+    def test_unverified_base_tips_fail_closed(self):
+        bad = dict(_assessment(self.repo, self.head), base_tips_verified=False)
+        self._assert_blocked(assessment=bad)
+
+    def test_mismatched_workspace_binding_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["workspace_path"] = os.path.join(self.repo, "elsewhere")
+        self._assert_blocked(assessment=bad)
+
+    def test_mismatched_repo_root_binding_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head))
+        bad["canonical_repo_root"] = os.path.join(self.repo, "other-root")
+        self._assert_blocked(assessment=bad)
+
+    def test_blocked_assessment_fails_closed(self):
+        bad = dict(_assessment(self.repo, self.head), block=True, allowed=False)
+        self._assert_blocked(assessment=bad)
+
+
+class TestOrdinaryControlCheckoutMutationStillForbidden(_ControlCheckoutHarness):
+    """The waiver must not leak to ordinary author work."""
+
+    def test_ordinary_author_task_still_blocked(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce("commit_files", assessment=None)
+
+    def test_lock_issue_still_blocked_from_control(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce("lock_issue", assessment=None)
+
+    def test_bootstrap_assessment_cannot_license_other_task(self):
+        # Cross-task smuggling: valid bootstrap evidence must not waive a
+        # different author mutation.
+        good = _assessment(self.repo, self.head)
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce("commit_files", assessment=good)
+
+    def test_dirty_control_checkout_still_blocked_for_bootstrap(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce(BOOTSTRAP_TASK, porcelain=" M gitea_mcp_server.py\n")
+
+
+class TestCreateIssueBehaviorUnchanged(_ControlCheckoutHarness):
+    """#749 create_issue keeps its own sanctioned path."""
+
+    def test_create_issue_still_allowed_from_clean_control(self):
+        self._enforce("create_issue", assessment=None)
+
+    def test_create_issue_tool_alias_still_allowed(self):
+        self._enforce("gitea_create_issue", assessment=None)
+
+    def test_create_issue_blocked_when_control_dirty(self):
+        with self.assertRaises(workflow_scope_guard.ProductionGuardError):
+            self._enforce(
+                "create_issue",
+                porcelain=" M gitea_mcp_server.py\n",
+                assessment=None,
+            )
+
+
+class TestGuardUnitLevelWiring(unittest.TestCase):
+    """assess_root_source_mutation itself must accept and honour the evidence."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.repo, self.head = _make_control_repo(self._tmp.name)
+        patcher = mock.patch.dict(
+            os.environ,
+            {workflow_scope_guard.FORCE_PRODUCTION_GUARDS_ENV: "1"},
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def _assess(self, *, task=BOOTSTRAP_TASK, bootstrap_assessment="auto"):
+        if bootstrap_assessment == "auto":
+            bootstrap_assessment = _assessment(self.repo, self.head, task=task)
+        return workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            porcelain_status="",
+            current_branch="master",
+            role_kind="author",
+            mutation_task=task,
+            bootstrap_assessment=bootstrap_assessment,
+        )
+
+    def test_valid_evidence_unblocks(self):
+        result = self._assess()
+        self.assertFalse(result["block"])
+        self.assertIsNone(result["blocker_kind"])
+
+    def test_absent_evidence_blocks(self):
+        result = self._assess(bootstrap_assessment=None)
+        self.assertTrue(result["block"])
+        self.assertEqual(
+            result["blocker_kind"], workflow_scope_guard.BLOCKER_MISSING_WORKTREE
+        )
+
+    def test_reconciler_exemption_preserved(self):
+        result = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            porcelain_status="",
+            current_branch="master",
+            role_kind="reconciler",
+            mutation_task=BOOTSTRAP_TASK,
+        )
+        self.assertFalse(result["block"])
+
+    def test_signature_accepts_evidence_without_it_being_required(self):
+        # Callers that supply no evidence keep the pre-existing behaviour.
+        result = workflow_scope_guard.assess_root_source_mutation(
+            workspace_path=self.repo,
+            canonical_repo_root=self.repo,
+            porcelain_status="",
+            current_branch="master",
+            role_kind="author",
+            mutation_task="create_issue",
+        )
+        self.assertFalse(result["block"])
+
+
+if __name__ == "__main__":
+    unittest.main()

workflow_scope_guard.py

@@ -279,6 +279,7 @@ def assess_root_source_mutation(
     locked_issue_number: int | None = None,
     role_kind: str | None = None,
     mutation_task: str | None = None,
+    bootstrap_assessment: Any | None = None,
 ) -> dict[str, Any]:
     """Fail closed for diagnostic/source edits on the control/root checkout.
 
@@ -286,6 +287,13 @@ def assess_root_source_mutation(
     tracked source/test files on the control checkout always block, including
     temporary/diagnostic/test-only intent.
 
+    #941: ``bootstrap_author_issue_worktree`` is judged by the canonical
+    ``create_issue_bootstrap.bootstrap_permits_control_checkout`` decision over
+    *bootstrap_assessment* — the same server-derived evidence the #274 and
+    #604 guards consume — instead of a task-name allowlist local to this
+    module. Evidence that is absent, malformed, wrongly scoped, or bound to
+    another workspace leaves the ordinary block in force.
+
     #749: ``create_issue`` is a pure remote mutation with no local tree write.
     When *mutation_task* is create_issue and the control checkout has no dirty
     source/test files, the missing-worktree signal is suppressed so the
@@ -336,6 +344,19 @@ def assess_root_source_mutation(
         if _cib is not None and _cib.is_create_issue_task(mutation_task):
             # #749: clean-root create_issue is the sanctioned bootstrap path.
             create_issue_bootstrap = True
+        elif _cib is not None and _cib.bootstrap_permits_control_checkout(
+            bootstrap_assessment,
+            task=mutation_task,
+            workspace_path=workspace,
+            canonical_repo_root=root,
+        ):
+            # #941: the author issue-worktree bootstrap is authorized by the
+            # canonical shared decision over server-derived task-scope
+            # evidence, never by a task-name allowlist kept in this module.
+            # The predicate fails closed on missing, malformed, cross-scope,
+            # dirty, drifted, or wrongly bound evidence, so this arm cannot
+            # widen the waiver beyond the one sanctioned bootstrap task.
+            create_issue_bootstrap = True
         else:
             # Explicit missing-worktree signal for force-on author entrypoints.
             reasons.append(
@@ -393,8 +414,15 @@ def assess_production_mutation_guards(
     require_author_lock: bool = False,
     in_test_mode: bool = False,
     mutation_task: str | None = None,
+    bootstrap_assessment: Any | None = None,
 ) -> dict[str, Any]:
-    """Compose root + scope production guards when they must be active (#683)."""
+    """Compose root + scope production guards when they must be active (#683).
+
+    #941: *bootstrap_assessment* is the server-derived author-bootstrap
+    evidence, forwarded unchanged to :func:`assess_root_source_mutation` so
+    this guard reaches the same canonical decision as the #274 and #604
+    guards. Omitting it preserves the pre-existing behaviour.
+    """
     if not production_guards_active(in_test_mode=in_test_mode):
         return {
             "proven": True,
@@ -414,6 +442,7 @@ def assess_production_mutation_guards(
         locked_issue_number=locked_issue_number,
         role_kind=role_kind,
         mutation_task=mutation_task,
+        bootstrap_assessment=bootstrap_assessment,
     )
     if root_assess["block"]:
         return {**root_assess, "skipped": False}