Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e349839fd7 |
@@ -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",
|
||||
}
|
||||
@@ -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
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
@@ -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
@@ -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")),
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user