From a09c485fc0eba75b35986a1833dba986be6c2101 Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Sun, 26 Jul 2026 05:00:09 -0500 Subject: [PATCH] fix(scope): wire author bootstrap scope into workflow scope guard (Closes #941) PR #926 (#892) taught create_issue_bootstrap.bootstrap_permits_control_checkout to 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 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. The real call path gitea_bootstrap_author_issue_worktree -> verify_preflight_purity -> _enforce_issue_scope_guard -> workflow_scope_guard.assess_production_mutation_guards therefore raised ProductionGuardError(missing_issue_worktree) before assess_author_issue_bootstrap was ever consulted, leaving the #892 deadlock partially present and blocking issue #931. Changes: * workflow_scope_guard.assess_root_source_mutation accepts the server-derived bootstrap_assessment and, for the clean-root author case, consults the canonical bootstrap_permits_control_checkout decision instead of a local task-name allowlist. The create_issue arm is unchanged. * workflow_scope_guard.assess_production_mutation_guards forwards the evidence. * _enforce_issue_scope_guard accepts and threads the same assessment the #274 and #604 guards already consume, so all three judge identical evidence, and waives the pre-ownership issue-lock requirement for the bootstrap task via that same canonical decision rather than a second task-name allowlist. * verify_preflight_purity passes the once-computed assessment at both sites. The waiver cannot widen: the predicate fails closed on missing, malformed, cross-scope, dirty, drifted, or wrongly bound evidence, so ordinary author source and test mutation from the control checkout stays forbidden and the #274/#604/#618/#683 protections are unchanged. Regression: tests/test_issue_941_scope_guard_bootstrap_wiring.py drives the real enforcer rather than the authorization helper in isolation, which is why #892's own predicate tests passed while the live bootstrap stayed blocked. Closes #941 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --- gitea_mcp_server.py | 33 +- ..._issue_941_scope_guard_bootstrap_wiring.py | 346 ++++++++++++++++++ workflow_scope_guard.py | 31 +- 3 files changed, 408 insertions(+), 2 deletions(-) create mode 100644 tests/test_issue_941_scope_guard_bootstrap_wiring.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index ed25bfd..f3f6edc 100644 --- a/gitea_mcp_server.py +++ b/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) diff --git a/tests/test_issue_941_scope_guard_bootstrap_wiring.py b/tests/test_issue_941_scope_guard_bootstrap_wiring.py new file mode 100644 index 0000000..f4b7351 --- /dev/null +++ b/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() diff --git a/workflow_scope_guard.py b/workflow_scope_guard.py index 3996102..91ad4e6 100644 --- a/workflow_scope_guard.py +++ b/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} -- 2.43.7