Compare commits

...
Author SHA1 Message Date
sysadmin e349839fd7 fix(mcp): allow create_issue from clean control checkout (#749)
Provide a narrow phase-scoped bootstrap so gitea_create_issue can run from
a clean canonical control checkout when no issue number—and therefore no
issue-backed worktree—can exist yet. Dirty roots, non-base branches, base
races, foreign workspaces, and all post-creation author mutations remain
fail-closed under the ordinary branches-only guard.

Closes #749.
2026-07-18 16:19:04 -04:00
sysadmin fdab6b6c69 Merge pull request 'feat(mcp): use a 10-minute sliding TTL for reviewer and merger PR leases (Closes #747)' (#748) from feat/issue-747-sliding-lease-ttl into master 2026-07-18 13:16:23 -05:00
7 changed files with 781 additions and 35 deletions
+227
View File
@@ -0,0 +1,227 @@
"""Sanctioned pre-issue bootstrap for ``create_issue`` (#749).
``gitea_create_issue`` is a pure remote mutation: it creates a tracking issue
and writes nothing to the local working tree. The issue-first gate forbids
creating ``branches/issue-<N>-*`` before the issue number exists, while the
#274 branches-only guard previously demanded that worktree first — a deadlock.
This module defines a **narrow, phase-scoped** exemption:
* Only tasks in :data:`CREATE_ISSUE_TASKS` may use it.
* Only the **canonical control checkout** may be used (never an arbitrary
directory, unrelated worktree, or foreign clone).
* The control checkout must be clean, on an accepted base branch, and
base-equivalent to live master when a remote tip is known.
* Every post-creation author mutation keeps the ordinary ``branches/`` rule.
The exemption cannot widen: unknown tasks, dirty roots, drifted HEADs, non-base
branches, and non-control workspaces fall through to the existing fail-closed
guards.
"""
from __future__ import annotations
import os
from typing import Any
from author_mutation_worktree import BASE_BRANCHES, is_path_under_branches
from reviewer_worktree import parse_dirty_tracked_files
CREATE_ISSUE_TASKS = frozenset({"create_issue", "gitea_create_issue"})
# Satisfiable before an issue number exists — never names issue-<N>.
EXACT_NEXT_ACTION_BOOTSTRAP = (
"Restore the canonical control checkout to a clean accepted base branch "
"(master/main/dev) that matches live master, with no tracked local edits "
"and no detached HEAD. Re-resolve the exact create_issue task, then re-run "
"gitea_create_issue from that clean control checkout. Do not create "
"branches/issue-<N>-* worktrees, dummy directories, or borrow unrelated "
"worktrees before the issue exists."
)
EXACT_NEXT_ACTION_POST_CREATE = (
"After the issue exists: create a registered worktree under "
"branches/issue-<N>-* from clean master, claim/lock the issue, set "
"GITEA_AUTHOR_WORKTREE / worktree_path to that path, then continue author "
"mutations from the issue-backed worktree only."
)
def is_create_issue_task(task: str | None) -> bool:
"""True when *task* is the create_issue mutation (or tool alias)."""
return (task or "").strip() in CREATE_ISSUE_TASKS
def assess_create_issue_bootstrap(
*,
workspace_path: str,
canonical_repo_root: str,
current_branch: str | None = None,
head_sha: str | None = None,
porcelain_status: str = "",
remote_master_sha: str | None = None,
task: str | None = None,
) -> dict[str, Any]:
"""Assess whether create_issue may proceed from the control checkout.
Returns a structured assessment:
* ``not_applicable`` — not a create_issue task, or workspace is already a
``branches/`` worktree (use ordinary guards).
* ``allowed`` — create_issue bootstrap may proceed from this control root.
* ``block`` — create_issue was attempted from control checkout but gates
failed (dirty, wrong branch, base race, etc.).
"""
reasons: list[str] = []
root = os.path.realpath(canonical_repo_root or "")
workspace = os.path.realpath(workspace_path or root or ".")
branch = (current_branch or "").strip()
dirty = parse_dirty_tracked_files(porcelain_status or "")
under_branches = is_path_under_branches(workspace, root) if root else False
if not is_create_issue_task(task):
return _result(
not_applicable=True,
allowed=False,
block=False,
reasons=["task is not create_issue"],
workspace=workspace,
root=root,
branch=branch,
dirty=dirty,
under_branches=under_branches,
)
# Registered branches/ worktrees keep the normal path (no bootstrap).
if under_branches:
return _result(
not_applicable=True,
allowed=False,
block=False,
reasons=["workspace is under branches/; ordinary #274 path applies"],
workspace=workspace,
root=root,
branch=branch,
dirty=dirty,
under_branches=True,
)
# Only the exact canonical control checkout is eligible.
if not root or workspace != root:
reasons.append(
"create_issue bootstrap requires the canonical control checkout; "
f"workspace '{workspace}' is not the repository root '{root or '(unknown)'}'"
)
return _result(
not_applicable=False,
allowed=False,
block=True,
reasons=reasons,
workspace=workspace,
root=root,
branch=branch,
dirty=dirty,
under_branches=False,
exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP,
)
if dirty:
reasons.append(
"create_issue bootstrap blocked: control checkout has tracked local "
f"edits (dirty files: {', '.join(dirty)})"
)
if not branch:
reasons.append(
"create_issue bootstrap blocked: control checkout is detached HEAD; "
"expected an accepted base branch (master/main/dev)"
)
elif branch not in BASE_BRANCHES:
reasons.append(
f"create_issue bootstrap blocked: control checkout branch '{branch}' "
f"is not an accepted base branch ({'/'.join(sorted(BASE_BRANCHES))})"
)
remote_tip = (remote_master_sha or "").strip() or None
local_tip = (head_sha or "").strip() or None
if remote_tip and local_tip and remote_tip != local_tip:
reasons.append(
"create_issue bootstrap blocked: control checkout HEAD does not match "
f"live master (HEAD {local_tip[:12]}, live master {remote_tip[:12]})"
)
if reasons:
return _result(
not_applicable=False,
allowed=False,
block=True,
reasons=reasons,
workspace=workspace,
root=root,
branch=branch or None,
dirty=dirty,
under_branches=False,
exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP,
)
return _result(
not_applicable=False,
allowed=True,
block=False,
reasons=[],
workspace=workspace,
root=root,
branch=branch or None,
dirty=dirty,
under_branches=False,
exact_next_action=EXACT_NEXT_ACTION_POST_CREATE,
bootstrap_path="clean_canonical_control_checkout",
)
def format_create_issue_bootstrap_error(assessment: dict[str, Any]) -> str:
"""RuntimeError / typed-block message for a failed bootstrap assessment."""
reasons = "; ".join(
assessment.get("reasons") or ["create_issue bootstrap failed"]
)
next_action = (
assessment.get("exact_next_action") or EXACT_NEXT_ACTION_BOOTSTRAP
)
root = assessment.get("canonical_repo_root") or "(unknown)"
workspace = assessment.get("workspace_path") or "(unknown)"
return (
f"Create-issue bootstrap guard (#749): {reasons}. "
f"canonical repository root: {root}; workspace: {workspace}. "
f"exact_next_action: {next_action}"
)
def _result(
*,
not_applicable: bool,
allowed: bool,
block: bool,
reasons: list[str],
workspace: str,
root: str,
branch: str | None,
dirty: list[str],
under_branches: bool,
exact_next_action: str | None = None,
bootstrap_path: str | None = None,
) -> dict[str, Any]:
return {
"not_applicable": not_applicable,
"allowed": allowed,
"block": block,
"proven": allowed and not block,
"reasons": list(reasons),
"workspace_path": workspace,
"canonical_repo_root": root,
"current_branch": branch,
"dirty_files": list(dirty),
"under_branches": under_branches,
"exact_next_action": exact_next_action,
"bootstrap_path": bootstrap_path,
"task_scope": "create_issue_only",
}
+10
View File
@@ -48,6 +48,16 @@ It extracts the issue-first, isolated-worktree, no-self-review, profile-safety,
merge-cleanup, fail-closed, and recovery rules into a reusable package that can
be adapted to other repositories.
### Sanctioned first mutation: `create_issue` from clean control (#749)
Creating a tracking issue has no issue number yet, so no `branches/issue-<N>-*`
worktree can exist. The sanctioned path is: clean canonical control checkout
(accepted base branch, base-equivalent to live master, no tracked dirt) →
resolve exact `create_issue``gitea_create_issue`. After the issue exists,
all further author mutations require a registered issue-backed worktree and
lock. Do not improvise with dummy directories, borrowed worktrees, or pre-issue
worktrees. See `skills/llm-project-workflow/workflows/create-issue.md` §18a.
## Principle: the profile is the role, not the LLM
```text
+58 -8
View File
@@ -817,7 +817,11 @@ def _enforce_canonical_repository_root(
)
def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
def _enforce_branches_only_author_mutation(
worktree_path: str | None = None,
*,
task: str | None = None,
) -> None:
"""#274: author file/branch mutations must run from a branches/ worktree.
Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
@@ -831,6 +835,12 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
genuine reconciler as an author and defeat this exemption. Keying off the
real profile role as well preserves the exemption without weakening author
blocking an actual author profile classifies as ``author`` in both.
#749: ``create_issue`` alone may proceed from a **clean** canonical control
checkout (pure remote mutation; no issue number exists yet for an
issue-backed worktree). Dirty roots, non-base branches, base races, and
every other author mutation keep the ordinary branches-only fail-closed
behaviour.
"""
if (
_effective_workspace_role() in nwb.NON_AUTHOR_ROLES
@@ -845,10 +855,35 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
project_root=ctx["canonical_repo_root"],
current_branch=git_state.get("current_branch"),
)
if assessment["block"]:
raise RuntimeError(
author_mutation_worktree.format_author_mutation_worktree_error(assessment)
if not assessment["block"]:
return
# #749 create_issue bootstrap: narrow phase exemption only.
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(
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"):
raise RuntimeError(_cib.format_create_issue_bootstrap_error(bootstrap))
raise RuntimeError(
author_mutation_worktree.format_author_mutation_worktree_error(assessment)
)
def _anti_stomp_in_test_mode() -> bool:
@@ -1213,7 +1248,7 @@ 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)
_enforce_branches_only_author_mutation(worktree_path, task=task)
_enforce_issue_scope_guard(
worktree_path,
task=task,
@@ -1247,7 +1282,7 @@ 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)
_enforce_branches_only_author_mutation(worktree_path, task=task)
_enforce_issue_scope_guard(
worktree_path,
task=task,
@@ -1366,10 +1401,15 @@ def _enforce_issue_scope_guard(
"mark_issue",
}
)
# #749: create_issue is pre-ownership (no issue number / lock can exist yet).
import create_issue_bootstrap as _cib
is_create_issue = _cib.is_create_issue_task(task)
require_lock = bool(require_author_lock) or (
authorish
and workflow_scope_guard.production_guards_forced()
and role == "author"
and not is_create_issue
)
assessment = workflow_scope_guard.assess_production_mutation_guards(
workspace_path=workspace,
@@ -1381,6 +1421,7 @@ def _enforce_issue_scope_guard(
role_kind=role,
require_author_lock=require_lock,
in_test_mode=_preflight_in_test_mode(),
mutation_task=task,
)
workflow_scope_guard.raise_if_blocked(assessment)
@@ -1394,7 +1435,11 @@ def _production_guard_block_from_exc(exc: BaseException, **extra) -> dict | None
kind = workflow_scope_guard.BLOCKER_PRODUCTION_GUARD
if "root_diagnostic_edit" in text or "tracked source or test edits" in text:
kind = workflow_scope_guard.BLOCKER_ROOT_DIAGNOSTIC_EDIT
elif "Branches-only mutation guard" in text or "stable control checkout" in text:
elif (
"Branches-only mutation guard" in text
or "stable control checkout" in text
or "Create-issue bootstrap guard (#749)" in text
):
kind = workflow_scope_guard.BLOCKER_MISSING_WORKTREE
elif "out-of-scope" in text or "locked to issue" in text:
kind = workflow_scope_guard.BLOCKER_OUT_OF_SCOPE_ISSUE
@@ -1405,10 +1450,15 @@ def _production_guard_block_from_exc(exc: BaseException, **extra) -> dict | None
reasons=[text],
**extra,
)
if "Branches-only mutation guard" in text:
if "Branches-only mutation guard" in text or "Create-issue bootstrap guard (#749)" in text:
# Prefer bootstrap next-action text when present (#749 AC4).
next_action = None
if "exact_next_action:" in text:
next_action = text.split("exact_next_action:", 1)[1].strip()
return workflow_scope_guard.block_response(
blocker_kind=workflow_scope_guard.BLOCKER_MISSING_WORKTREE,
reasons=[text],
exact_next_action=next_action,
**extra,
)
if "Root checkout guard" in text:
@@ -450,6 +450,35 @@ If any gate fails, do not create the issue.
Produce a recovery handoff or duplicate report.
### 18a. Sanctioned first-mutation path (#749)
`gitea_create_issue` is a **pure remote mutation** (no local tree write). The
issue-first gate forbids creating `branches/issue-<N>-*` before the issue
number exists. Therefore the **only sanctioned first mutation** is:
1. Read-only identity + capability + duplicate search from the control checkout.
2. Ensure the **canonical control checkout** is:
* the configured repository root for the requested remote/org/repo;
* on an accepted base branch (`master` / `main` / `dev`);
* base-equivalent to live master;
* clean (no tracked local edits);
* in runtime/master parity.
3. Resolve exact task `create_issue`, then call `gitea_create_issue` **from that
clean control checkout** (no `worktree_path` required for this step alone).
4. After the issue number exists: create a **registered** worktree under
`branches/issue-<N>-*`, claim/lock, and perform every subsequent author
mutation from that worktree only.
**Forbidden improvisations (fail closed):**
* `mkdir` dummy directories under `branches/` (#713)
* borrowing an unrelated pre-existing worktree
* creating a pre-issue worktree in violation of issue-first
* running create_issue from a dirty, drifted, detached, or non-canonical root
Post-creation mutations (`lock_issue`, commit, push, `create_pr`, etc.) **never**
receive this bootstrap exemption.
## 19. Issue commenting gate
Before commenting on an existing issue, verify:
+367
View File
@@ -0,0 +1,367 @@
"""Regression tests for create_issue bootstrap (#749).
TDD: these tests define the sanctioned first-mutation path for
``gitea_create_issue`` from a clean canonical control checkout, and prove
the exemption cannot widen to dirty roots, foreign clones, arbitrary
``branches/`` directories, or post-creation author mutations.
"""
from __future__ import annotations
import os
import sys
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import create_issue_bootstrap as cib # noqa: E402
import gitea_mcp_server as srv # noqa: E402
import workflow_scope_guard as wsg # 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])
class TestCreateIssueBootstrapAssessor(unittest.TestCase):
ROOT = "/repo/Gitea-Tools"
def test_non_create_issue_task_not_applicable(self):
res = cib.assess_create_issue_bootstrap(
workspace_path=self.ROOT,
canonical_repo_root=self.ROOT,
current_branch="master",
head_sha=MASTER_SHA,
porcelain_status="",
remote_master_sha=MASTER_SHA,
task="lock_issue",
)
self.assertTrue(res["not_applicable"])
self.assertFalse(res["allowed"])
self.assertFalse(res["block"])
def test_branches_worktree_not_applicable(self):
res = cib.assess_create_issue_bootstrap(
workspace_path=f"{self.ROOT}/branches/issue-1-x",
canonical_repo_root=self.ROOT,
current_branch="fix/issue-1-x",
task="create_issue",
)
self.assertTrue(res["not_applicable"])
self.assertFalse(res["allowed"])
def test_clean_control_checkout_allowed(self):
res = cib.assess_create_issue_bootstrap(
workspace_path=self.ROOT,
canonical_repo_root=self.ROOT,
current_branch="master",
head_sha=MASTER_SHA,
porcelain_status="",
remote_master_sha=MASTER_SHA,
task="create_issue",
)
self.assertFalse(res["not_applicable"])
self.assertTrue(res["allowed"])
self.assertFalse(res["block"])
self.assertEqual(res["bootstrap_path"], "clean_canonical_control_checkout")
# Post-create next action must name issue-backed worktree after N exists.
self.assertIn("branches/issue-<N>-*", res["exact_next_action"])
def test_tool_alias_gitea_create_issue_allowed(self):
res = cib.assess_create_issue_bootstrap(
workspace_path=self.ROOT,
canonical_repo_root=self.ROOT,
current_branch="main",
head_sha=MASTER_SHA,
porcelain_status="",
remote_master_sha=MASTER_SHA,
task="gitea_create_issue",
)
self.assertTrue(res["allowed"])
def test_dirty_control_checkout_blocked(self):
res = cib.assess_create_issue_bootstrap(
workspace_path=self.ROOT,
canonical_repo_root=self.ROOT,
current_branch="master",
head_sha=MASTER_SHA,
porcelain_status=" M gitea_mcp_server.py\n",
remote_master_sha=MASTER_SHA,
task="create_issue",
)
self.assertTrue(res["block"])
self.assertFalse(res["allowed"])
self.assertTrue(any("tracked local edits" in r for r in res["reasons"]))
# Pre-issue phase: next action must be satisfiable without inventing <N>.
next_a = res["exact_next_action"] or ""
self.assertIn("clean accepted base branch", next_a)
self.assertIn("before the issue exists", next_a)
# Must not prescribe "bind branches/issue-<N>" as the recovery step.
self.assertNotIn("Bind an issue-backed worktree", next_a)
def test_stale_base_blocked(self):
res = cib.assess_create_issue_bootstrap(
workspace_path=self.ROOT,
canonical_repo_root=self.ROOT,
current_branch="master",
head_sha=STALE_SHA,
porcelain_status="",
remote_master_sha=MASTER_SHA,
task="create_issue",
)
self.assertTrue(res["block"])
self.assertTrue(any("live master" in r for r in res["reasons"]))
def test_non_base_branch_blocked(self):
res = cib.assess_create_issue_bootstrap(
workspace_path=self.ROOT,
canonical_repo_root=self.ROOT,
current_branch="feat/something",
head_sha=MASTER_SHA,
porcelain_status="",
remote_master_sha=MASTER_SHA,
task="create_issue",
)
self.assertTrue(res["block"])
def test_detached_head_blocked(self):
res = cib.assess_create_issue_bootstrap(
workspace_path=self.ROOT,
canonical_repo_root=self.ROOT,
current_branch="",
head_sha=MASTER_SHA,
porcelain_status="",
remote_master_sha=MASTER_SHA,
task="create_issue",
)
self.assertTrue(res["block"])
self.assertTrue(any("detached" in r for r in res["reasons"]))
def test_foreign_workspace_blocked(self):
res = cib.assess_create_issue_bootstrap(
workspace_path="/other/clone",
canonical_repo_root=self.ROOT,
current_branch="master",
head_sha=MASTER_SHA,
porcelain_status="",
remote_master_sha=MASTER_SHA,
task="create_issue",
)
self.assertTrue(res["block"])
self.assertTrue(any("canonical control checkout" in r for r in res["reasons"]))
class TestCreateIssueBootstrapIntegration(unittest.TestCase):
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,
}
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
@patch(
"gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []),
)
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server.api_get_all", return_value=[])
@patch(
"gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha",
return_value=MASTER_SHA,
)
def test_clean_control_checkout_create_issue_succeeds(
self, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth
):
mock_api.return_value = {
"number": 99,
"html_url": "https://gitea.example.com/issues/99",
}
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch(
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=self._git_state(),
):
with patch(
"gitea_mcp_server._get_workspace_porcelain", return_value=""
):
with patch(
"gitea_mcp_server._enforce_root_checkout_guard"
):
# Anti-stomp / master parity: keep gates green.
with patch.object(
srv,
"_run_anti_stomp_preflight",
return_value=None,
):
res = srv.gitea_create_issue(
title="Bootstrap issue from clean control",
body="Body text for content gate.",
)
self.assertEqual(res.get("number"), 99)
mock_api.assert_called_once()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
@patch(
"gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []),
)
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server.api_get_all", return_value=[])
def test_dirty_control_checkout_create_issue_fails_closed(
self, _get_all, mock_api, _role, _ns, _prof, _auth
):
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch(
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=self._git_state(
porcelain=" M author_mutation_worktree.py\n"
),
):
with patch(
"gitea_mcp_server._get_workspace_porcelain",
return_value=" M author_mutation_worktree.py\n",
):
res = srv.gitea_create_issue(
title="Should fail on dirty root",
body="Body text for content gate.",
)
self.assertFalse(res.get("success", True) and res.get("number"))
if isinstance(res, dict) and res.get("success") is False:
blob = " ".join(res.get("reasons") or [])
self.assertTrue(
"tracked local edits" in blob
or "dirty" in blob.lower()
or "control checkout" in blob.lower()
or res.get("blocker_kind")
)
# Pre-issue phase must not demand issue-<N> worktree.
next_a = res.get("exact_next_action") or ""
if next_a:
self.assertNotIn("issue-<N>-*", next_a)
mock_api.assert_not_called()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
@patch(
"gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []),
)
def test_lock_issue_still_requires_branches_worktree(self, _role, _ns, _prof, _auth):
"""Existing issue-backed mutations receive no exemption (#749 AC3/AC7)."""
srv._preflight_resolved_task = "lock_issue"
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch(
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value=self._git_state(),
):
with patch(
"gitea_mcp_server._get_workspace_porcelain", return_value=""
):
with patch(
"gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha",
return_value=MASTER_SHA,
):
with patch.object(
srv, "_enforce_root_checkout_guard"
):
with self.assertRaises(RuntimeError) as ctx:
srv.verify_preflight_purity(
remote="prgs",
task="lock_issue",
)
msg = str(ctx.exception)
self.assertTrue(
"Branches-only mutation guard" in msg
or "stable control checkout" in msg,
msg,
)
self.assertIn("control checkout", msg)
def test_workflow_scope_skips_missing_worktree_for_create_issue_clean_root(self):
"""#683 root assessor must not block clean-root create_issue bootstrap."""
res = wsg.assess_root_source_mutation(
workspace_path=CONTROL_CHECKOUT_ROOT,
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
porcelain_status="",
role_kind="author",
mutation_task="create_issue",
)
self.assertFalse(res["block"], res)
self.assertTrue(res.get("create_issue_bootstrap") or res["proven"])
def test_workflow_scope_still_blocks_clean_root_for_lock_issue(self):
res = wsg.assess_root_source_mutation(
workspace_path=CONTROL_CHECKOUT_ROOT,
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
porcelain_status="",
role_kind="author",
mutation_task="lock_issue",
)
self.assertTrue(res["block"])
self.assertEqual(res["blocker_kind"], wsg.BLOCKER_MISSING_WORKTREE)
def test_arbitrary_branches_directory_not_bootstrap(self):
"""#713: mkdir fake under branches/ is not the bootstrap path."""
fake = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "fake-mkdir-only")
res = cib.assess_create_issue_bootstrap(
workspace_path=fake,
canonical_repo_root=CONTROL_CHECKOUT_ROOT,
current_branch="master",
head_sha=MASTER_SHA,
porcelain_status="",
remote_master_sha=MASTER_SHA,
task="create_issue",
)
# Under branches/ → not bootstrap; ordinary membership/registration applies.
self.assertTrue(res["not_applicable"])
self.assertFalse(res["allowed"])
class TestCreateIssueCapabilityAgreement(unittest.TestCase):
def test_map_and_alias_agree_on_create_issue(self):
import task_capability_map as tcm
self.assertEqual(
tcm.required_permission("create_issue"),
"gitea.issue.create",
)
# Tool alias must resolve to the same task contract.
alias = getattr(tcm, "TOOL_TASK_ALIASES", None) or getattr(
tcm, "TASK_ALIASES", None
)
if alias is not None:
mapped = alias.get("gitea_create_issue")
if mapped is not None:
self.assertEqual(mapped, "create_issue")
if __name__ == "__main__":
unittest.main()
+53 -20
View File
@@ -51,29 +51,62 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
"porcelain_status": "",
},
)
def test_create_issue_stable_checkout_rejected(
def test_create_issue_stable_checkout_bootstrap_allowed_when_clean(
self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth,
):
# Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
# path is the stable control checkout (not under branches/), mutation must fail.
# #749: clean canonical control checkout is the sanctioned create_issue path.
mock_api.return_value = {
"number": 77,
"html_url": "https://gitea.example.com/issues/77",
}
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
try:
res = srv.gitea_create_issue(title="Test issue", body="body text")
except RuntimeError as exc:
self.assertIn("stable control checkout", str(exc))
else:
# #683: production guards return typed blockers at entrypoints
self.assertFalse(res.get("success"))
self.assertFalse(res.get("performed"))
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.assertTrue(res.get("exact_next_action"))
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
with patch.object(srv, "_run_anti_stomp_preflight", return_value=None):
with patch.object(srv, "_enforce_root_checkout_guard"):
res = srv.gitea_create_issue(
title="Test issue", body="body text for gate"
)
self.assertEqual(res.get("number"), 77)
mock_api.assert_called_once()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
@patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server.api_get_all", return_value=[])
@patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
@patch(
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={
"current_branch": "master",
"head_sha": "a" * 40,
"porcelain_status": " M dirty.py\n",
},
)
def test_create_issue_dirty_control_checkout_rejected(
self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth,
):
# #749: dirty control checkout still fails closed (no bootstrap).
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch(
"gitea_mcp_server._get_workspace_porcelain",
return_value=" M dirty.py\n",
):
try:
res = srv.gitea_create_issue(title="Test issue", body="body text")
except RuntimeError as exc:
self.assertTrue(
"tracked local edits" in str(exc)
or "dirty" in str(exc).lower()
or "bootstrap" in str(exc).lower()
or "control checkout" in str(exc).lower()
)
else:
self.assertFalse(res.get("success", True) and res.get("number"))
blob = " ".join(res.get("reasons") or [])
self.assertTrue(blob or res.get("blocker_kind"))
mock_api.assert_not_called()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
+37 -7
View File
@@ -278,12 +278,19 @@ def assess_root_source_mutation(
current_branch: str | None = None,
locked_issue_number: int | None = None,
role_kind: str | None = None,
mutation_task: str | None = None,
) -> dict[str, Any]:
"""Fail closed for diagnostic/source edits on the control/root checkout.
Allowed only when the active workspace is under ``branches/``. Dirty
tracked source/test files on the control checkout always block, including
temporary/diagnostic/test-only intent.
#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
sanctioned bootstrap path can proceed. Dirty roots and every other task
still fail closed.
"""
role = (role_kind or "").strip().lower()
if role == "reconciler":
@@ -302,6 +309,7 @@ def assess_root_source_mutation(
dirty_src = dirty_source_files(porcelain_status)
reasons: list[str] = []
blocker_kind: str | None = None
create_issue_bootstrap = False
if not under_branches and workspace == root and dirty_src:
# Root workspace with source dirtiness is unattributed root WIP.
@@ -313,32 +321,50 @@ def assess_root_source_mutation(
)
blocker_kind = BLOCKER_ROOT_DIAGNOSTIC_EDIT
# Lazy import keeps workflow_scope_guard free of circular import at module load.
try:
import create_issue_bootstrap as _cib
except Exception: # pragma: no cover - import always available in-tree
_cib = None
if (
not under_branches
and workspace == root
and not dirty_src
and role == "author"
):
# Explicit missing-worktree signal for force-on author entrypoints.
reasons.append(
"author source/test mutation from the stable control checkout is "
"forbidden; bind an issue-backed worktree under branches/ first"
)
blocker_kind = BLOCKER_MISSING_WORKTREE
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
else:
# Explicit missing-worktree signal for force-on author entrypoints.
reasons.append(
"author source/test mutation from the stable control checkout is "
"forbidden; bind an issue-backed worktree under branches/ first"
)
blocker_kind = BLOCKER_MISSING_WORKTREE
if reasons:
kind = blocker_kind or BLOCKER_ROOT_DIAGNOSTIC_EDIT
next_action = _NEXT_ACTIONS[kind]
if (
kind == BLOCKER_ROOT_DIAGNOSTIC_EDIT
and _cib is not None
and _cib.is_create_issue_task(mutation_task)
):
next_action = _cib.EXACT_NEXT_ACTION_BOOTSTRAP
return {
"proven": False,
"block": True,
"blocker_kind": kind,
"exact_next_action": _NEXT_ACTIONS[kind],
"exact_next_action": next_action,
"reasons": reasons,
"dirty_source_files": dirty_src,
"workspace_path": workspace,
"canonical_repo_root": root,
"under_branches": under_branches,
"locked_issue_number": locked_issue_number,
"create_issue_bootstrap": False,
}
return {
"proven": True,
@@ -351,6 +377,7 @@ def assess_root_source_mutation(
"canonical_repo_root": root,
"under_branches": under_branches,
"locked_issue_number": locked_issue_number,
"create_issue_bootstrap": create_issue_bootstrap,
}
@@ -365,6 +392,7 @@ def assess_production_mutation_guards(
role_kind: str | None = None,
require_author_lock: bool = False,
in_test_mode: bool = False,
mutation_task: str | None = None,
) -> dict[str, Any]:
"""Compose root + scope production guards when they must be active (#683)."""
if not production_guards_active(in_test_mode=in_test_mode):
@@ -385,6 +413,7 @@ def assess_production_mutation_guards(
current_branch=current_branch,
locked_issue_number=locked_issue_number,
role_kind=role_kind,
mutation_task=mutation_task,
)
if root_assess["block"]:
return {**root_assess, "skipped": False}
@@ -408,6 +437,7 @@ def assess_production_mutation_guards(
"skipped": False,
"root": root_assess,
"scope": scope_assess,
"create_issue_bootstrap": bool(root_assess.get("create_issue_bootstrap")),
}