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.

create_issue_bootstrap.py

@@ -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",
+    }

docs/llm-workflow-runbooks.md

@@ -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

gitea_mcp_server.py

@@ -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:
@@ -15367,151 +15417,46 @@ def _count_commits_behind(
     return None
 
 
-def _matching_protection_rule(protections: Any, branch: str) -> dict | None:
-    """Select the branch-protection rule governing ``branch`` (#751)."""
-    if not isinstance(protections, list):
-        return None
-    for rule in protections:
-        if not isinstance(rule, dict):
-            continue
-        name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
-        # Exact match or glob-ish contains for common patterns.
-        if name == branch or name in ("*", f"{branch}"):
-            return rule
-    # Fallback: any protection that mentions the base branch.
-    for rule in protections:
-        if not isinstance(rule, dict):
-            continue
-        name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
-        if name and (branch in name or name.endswith(branch)):
-            return rule
-    return None
-
-
-def _branch_protection_policy(
-    base_url: str,
-    auth: dict,
-    *,
-    base_branch: str,
-) -> dict:
-    """Read the live branch-protection policy for ``base_branch`` (#751).
-
-    Exposes both the current-base rule (``block_on_outdated_branch``) and the
-    status-check requirement (``enable_status_check`` / ``status_check_contexts``)
-    from the same payload, so the checks assessor can tell "no checks required"
-    apart from "required checks pending".
-
-    ``determinable`` is False only when the policy genuinely could not be read
-    (missing branch or API failure) — never merely because no rule exists. A
-    successful read that finds no protection for the branch is authoritative
-    evidence that status checks are not required.
-    """
-    policy: dict[str, Any] = {
-        "determinable": False,
-        "protection_found": False,
-        "requires_current_base": None,
-        "checks_enabled": None,
-        "required_contexts": [],
-        "base_branch": (base_branch or "").strip() or None,
-    }
-    branch = (base_branch or "").strip()
-    if not branch:
-        return policy
-
-    def _apply(rule: dict) -> None:
-        policy["protection_found"] = True
-        if "block_on_outdated_branch" in rule:
-            policy["requires_current_base"] = bool(
-                rule.get("block_on_outdated_branch")
-            )
-        if "enable_status_check" in rule:
-            policy["checks_enabled"] = bool(rule.get("enable_status_check"))
-        contexts = rule.get("status_check_contexts")
-        if isinstance(contexts, list):
-            policy["required_contexts"] = [
-                str(ctx).strip() for ctx in contexts if str(ctx or "").strip()
-            ]
-
-    try:
-        protections = api_request(
-            "GET", f"{base_url}/branch_protections", auth
-        )
-        rule = _matching_protection_rule(protections, branch)
-        if rule is not None:
-            _apply(rule)
-        if not policy["protection_found"]:
-            # Branch payload may embed effective protection.
-            br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {}
-            prot = br.get("protection") or br.get("effective_branch_protection") or {}
-            if isinstance(prot, dict) and prot:
-                _apply(prot)
-        policy["determinable"] = True
-    except Exception:
-        # Genuine read failure — leave determinable False so callers fail closed.
-        return policy
-
-    if not policy["protection_found"]:
-        # Authoritative absence: no protection governs the branch, so no status
-        # check is required by policy.
-        policy["checks_enabled"] = False
-    elif policy["checks_enabled"] is None:
-        # Protection exists but omits the status-check field entirely: Gitea
-        # only enforces contexts when the flag is set, so absence means off.
-        policy["checks_enabled"] = False
-
-    return policy
-
-
 def _branch_protection_requires_current_base(
     base_url: str,
     auth: dict,
     *,
     base_branch: str,
 ) -> bool | None:
-    """Read Gitea branch protection ``block_on_outdated_branch`` when present.
-
-    Thin accessor over :func:`_branch_protection_policy`; return semantics are
-    unchanged (``None`` when the rule is absent or unreadable).
-    """
-    policy = _branch_protection_policy(base_url, auth, base_branch=base_branch)
-    return policy.get("requires_current_base")
-
-
-def _commit_checks_snapshot(
-    base_url: str,
-    auth: dict,
-    *,
-    sha: str,
-) -> dict:
-    """Read the combined commit status *and its context collection* (#751).
-
-    The combined ``state`` alone is not decisive: Gitea reports ``pending`` for
-    a commit with an empty status-context collection, which is indistinguishable
-    from executing CI unless the collection itself is inspected.
-    """
-    snapshot: dict[str, Any] = {
-        "determinable": False,
-        "combined_state": None,
-        "statuses": [],
-    }
-    head = (sha or "").strip()
-    if not head:
-        return snapshot
+    """Read Gitea branch protection ``block_on_outdated_branch`` when present."""
+    branch = (base_branch or "").strip()
+    if not branch:
+        return None
     try:
-        payload = api_request(
-            "GET", f"{base_url}/commits/{head}/status", auth
-        )
+        # Prefer the named protection rule matching the base branch.
+        protections = api_request(
+            "GET", f"{base_url}/branch_protections", auth
+        ) or []
+        if isinstance(protections, list):
+            for rule in protections:
+                if not isinstance(rule, dict):
+                    continue
+                name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
+                # Exact match or glob-ish contains for common patterns.
+                if name == branch or name in ("*", f"{branch}"):
+                    if "block_on_outdated_branch" in rule:
+                        return bool(rule.get("block_on_outdated_branch"))
+            # Fallback: any protection that mentions the base branch.
+            for rule in protections:
+                if not isinstance(rule, dict):
+                    continue
+                name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
+                if branch in name or name.endswith(branch):
+                    if "block_on_outdated_branch" in rule:
+                        return bool(rule.get("block_on_outdated_branch"))
+        # Branch payload may embed effective protection.
+        br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {}
+        prot = br.get("protection") or br.get("effective_branch_protection") or {}
+        if isinstance(prot, dict) and "block_on_outdated_branch" in prot:
+            return bool(prot.get("block_on_outdated_branch"))
     except Exception:
-        return snapshot
-    if payload is None:
-        return snapshot
-    snapshot["determinable"] = True
-    if not isinstance(payload, dict):
-        return snapshot
-    snapshot["combined_state"] = (payload.get("state") or "").strip().lower() or None
-    statuses = payload.get("statuses")
-    snapshot["statuses"] = statuses if isinstance(statuses, list) else []
-    return snapshot
+        return None
+    return None
 
 
 def _prove_author_ownership_for_pr(
@@ -15756,14 +15701,11 @@ def gitea_assess_pr_sync_status(
             # Fail closed on unknown behind-count only when SHAs differ.
             commits_behind = 0 if pr_head_sha == base_head_sha else None
 
-    # Single live read of the base-branch protection policy; it carries both
-    # the current-base rule and the status-check requirement (#751).
-    protection_policy = _branch_protection_policy(
-        base, auth, base_branch=base_branch
-    )
     if branch_protection_requires_current_base is None:
-        branch_protection_requires_current_base = protection_policy.get(
-            "requires_current_base"
+        branch_protection_requires_current_base = (
+            _branch_protection_requires_current_base(
+                base, auth, base_branch=base_branch
+            )
         )
     # When protection cannot be read, fail closed by requiring current base
     # whenever the PR is behind (safer for protected repos). Callers may pass
@@ -15830,30 +15772,23 @@ def gitea_assess_pr_sync_status(
     except Exception:
         pass
 
-    # ── Live checks derivation (#751) ────────────────────────────────────
-    # Classify from the actual status-context collection plus the live
-    # protection policy. The combined ``state`` is never treated as proof that
-    # CI is executing, because Gitea reports ``pending`` for an empty
-    # collection. ``checks_required`` is always derived from live evidence and
-    # is deliberately not a caller-supplied input, so no session can declare
-    # checks optional without proof.
-    checks_snapshot = _commit_checks_snapshot(base, auth, sha=pr_head_sha or "")
-    checks_classification = pr_sync_status.classify_commit_checks(
-        combined_state=checks_snapshot.get("combined_state"),
-        statuses=checks_snapshot.get("statuses"),
-        checks_enabled=protection_policy.get("checks_enabled"),
-        required_contexts=protection_policy.get("required_contexts"),
-        policy_determinable=bool(protection_policy.get("determinable")),
-        status_determinable=bool(checks_snapshot.get("determinable")),
-    )
-    checks_required = bool(checks_classification.get("checks_required", True))
-    caller_supplied_checks_status = checks_status
     if checks_status is None:
-        checks_status = checks_classification.get("checks_status")
-    elif checks_classification.get("checks_status") != pr_sync_status.CHECKS_UNKNOWN:
-        # Live evidence outranks a caller-supplied value; the override only
-        # applies when live status could not be classified at all.
-        checks_status = checks_classification.get("checks_status")
+        checks_status = "unknown"
+        # Combined status on PR head when Actions/status API is available.
+        if pr_head_sha:
+            try:
+                st = api_request(
+                    "GET",
+                    f"{base}/commits/{pr_head_sha}/status",
+                    auth,
+                ) or {}
+                state = (st.get("state") or "").strip().lower()
+                if state:
+                    checks_status = state
+                elif not st:
+                    checks_status = "none"
+            except Exception:
+                checks_status = "unknown"
 
     assessment = pr_sync_status.assess_pr_sync_status(
         host=h,
@@ -15874,26 +15809,9 @@ def gitea_assess_pr_sync_status(
         active_reviewer_lease=active_reviewer_lease,
         active_merger_lease=active_merger_lease,
         prepared_verdict_head_sha=prepared_verdict_head_sha,
-        checks_required=checks_required,
     )
     assessment["remote"] = remote if remote in REMOTES else None
     assessment["base_branch"] = base_branch
-    # Evidence for the checks decision (#751) — no secrets, read-only.
-    assessment["checks_evidence"] = {
-        "combined_state": checks_classification.get("combined_state"),
-        "context_count": checks_classification.get("context_count"),
-        "observed_contexts": checks_classification.get("observed_contexts"),
-        "required_contexts": checks_classification.get("required_contexts"),
-        "missing_required_contexts": checks_classification.get(
-            "missing_required_contexts"
-        ),
-        "policy_determinable": checks_classification.get("policy_determinable"),
-        "status_determinable": checks_classification.get("status_determinable"),
-        "protection_found": protection_policy.get("protection_found"),
-        "checks_enabled": protection_policy.get("checks_enabled"),
-        "caller_supplied_checks_status": caller_supplied_checks_status,
-        "reasons": checks_classification.get("reasons"),
-    }
     assessment["success"] = True
     assessment["performed"] = False
     return assessment

pr_sync_status.py

@@ -41,186 +41,6 @@ _DENIED_UPDATE_ROLES = frozenset({"reviewer", "merger", "reconciler", "mixed", "
 UPDATE_STYLE_MERGE = "merge"
 _FORBIDDEN_UPDATE_STYLES = frozenset({"rebase", "rebase-merge", "squash", "force"})
 
-# ── Commit check classifications (#751) ──────────────────────────────────
-# Gitea's *combined* commit status reports ``state: pending`` both when a real
-# check is executing and when the status-context collection is empty. Reading
-# ``state`` alone therefore cannot distinguish "CI is running" from "no CI
-# exists", which permanently blocks a merge-ready PR that no check will ever
-# report on. These classifications are derived from the actual context
-# collection plus the live branch-protection policy.
-CHECKS_SUCCESS = "success"
-CHECKS_FAILURE = "failure"
-CHECKS_PENDING = "pending"
-CHECKS_NONE = "none"                          # configured/produced nothing
-CHECKS_NOT_REQUIRED = "not_required"          # protection does not require checks
-CHECKS_MISSING_REQUIRED = "missing_required"  # required contexts have no result
-CHECKS_UNKNOWN = "unknown"                    # indeterminable — fail closed
-
-# Values that permit merge_now when checks are required.
-_CHECKS_OK = frozenset({"success", "passed", "ok", "skipped", "not_required"})
-
-# Raw per-context state vocabularies reported by Gitea.
-_CTX_SUCCESS = frozenset({"success", "passed", "ok"})
-_CTX_FAILURE = frozenset({"failure", "failed", "error", "cancelled", "canceled"})
-_CTX_PENDING = frozenset({"pending", "running", "queued", "expected"})
-_CTX_SKIPPED = frozenset({"skipped", "neutral"})
-
-
-def _normalize_context_rows(statuses: Any) -> list[dict[str, str]]:
-    """Reduce a raw status collection to newest-wins ``{context, state}`` rows.
-
-    Gitea returns the status collection newest-first, so the first row seen for
-    a context wins. Rows without a usable state are discarded rather than being
-    silently treated as passing.
-    """
-    rows: list[dict[str, str]] = []
-    seen: set[str] = set()
-    if not isinstance(statuses, list):
-        return rows
-    for raw in statuses:
-        if not isinstance(raw, dict):
-            continue
-        context = (raw.get("context") or raw.get("name") or "").strip()
-        state = (raw.get("status") or raw.get("state") or "").strip().lower()
-        if not state:
-            continue
-        key = context or f"__unnamed__{len(rows)}"
-        if key in seen:
-            continue
-        seen.add(key)
-        rows.append({"context": context, "state": state})
-    return rows
-
-
-def _aggregate_context_states(rows: list[dict[str, str]]) -> str:
-    """Fail-closed aggregate: failure > pending > unknown-state > success."""
-    states = {row["state"] for row in rows}
-    if states & _CTX_FAILURE:
-        return CHECKS_FAILURE
-    if states & _CTX_PENDING:
-        return CHECKS_PENDING
-    unresolved = states - _CTX_SUCCESS - _CTX_SKIPPED
-    if unresolved:
-        # An unrecognized context state must never read as success.
-        return CHECKS_UNKNOWN
-    return CHECKS_SUCCESS
-
-
-def classify_commit_checks(
-    *,
-    combined_state: str | None = None,
-    statuses: Any = None,
-    checks_enabled: bool | None = None,
-    required_contexts: Any = None,
-    policy_determinable: bool = True,
-    status_determinable: bool = True,
-) -> dict[str, Any]:
-    """Classify head checks from live evidence (#751).
-
-    ``combined_state`` is deliberately **not** authoritative: it is recorded for
-    observability but never used to infer that CI is executing. The context
-    collection and the live protection policy decide.
-
-    Returns ``checks_status`` (one of the ``CHECKS_*`` values), the derived
-    ``checks_required`` flag, and structured ``reasons``.
-    """
-    reasons: list[str] = []
-    rows = _normalize_context_rows(statuses)
-    required = [
-        str(ctx).strip()
-        for ctx in (required_contexts or [])
-        if str(ctx or "").strip()
-    ]
-    observed_combined = (combined_state or "").strip().lower() or None
-
-    result: dict[str, Any] = {
-        "checks_status": CHECKS_UNKNOWN,
-        "checks_required": True,
-        "combined_state": observed_combined,
-        "context_count": len(rows),
-        "observed_contexts": [row["context"] for row in rows],
-        "required_contexts": required,
-        "missing_required_contexts": [],
-        "policy_determinable": bool(policy_determinable),
-        "status_determinable": bool(status_determinable),
-        "reasons": reasons,
-    }
-
-    # Policy unreadable → never assume checks are optional.
-    if not policy_determinable:
-        reasons.append(
-            "branch-protection check policy could not be read; cannot prove "
-            "whether status checks are required (fail closed)"
-        )
-        return result
-
-    if checks_enabled is False:
-        result["checks_required"] = False
-        result["checks_status"] = CHECKS_NOT_REQUIRED
-        reasons.append(
-            "live branch protection does not require status checks for the base "
-            "branch; head status contexts do not gate merge"
-        )
-        return result
-
-    if checks_enabled is None:
-        reasons.append(
-            "branch-protection status-check requirement is indeterminate "
-            "(fail closed)"
-        )
-        return result
-
-    # Checks are required from here on.
-    if not status_determinable:
-        reasons.append(
-            "head commit status collection could not be read while branch "
-            "protection requires status checks (fail closed)"
-        )
-        return result
-
-    if required:
-        by_context = {row["context"]: row["state"] for row in rows if row["context"]}
-        missing = [ctx for ctx in required if ctx not in by_context]
-        if missing:
-            result["missing_required_contexts"] = missing
-            result["checks_status"] = CHECKS_MISSING_REQUIRED
-            reasons.append(
-                "branch protection requires status context(s) "
-                f"{', '.join(missing)} but no matching status result exists at "
-                "the head commit (fail closed)"
-            )
-            return result
-        matched = [
-            {"context": ctx, "state": by_context[ctx]} for ctx in required
-        ]
-        result["checks_status"] = _aggregate_context_states(matched)
-        reasons.append(
-            f"evaluated {len(matched)} required status context(s) from live "
-            "branch protection; unrelated contexts were ignored"
-        )
-        return result
-
-    # Status checks enabled with no specific required contexts configured.
-    if not rows:
-        result["checks_status"] = CHECKS_NONE
-        reasons.append(
-            "branch protection enables status checks but no status context was "
-            "produced for the head commit"
-        )
-        if observed_combined in _CTX_PENDING:
-            reasons.append(
-                f"combined commit state '{observed_combined}' does not indicate "
-                "executing CI because the status-context collection is empty"
-            )
-        return result
-
-    result["checks_status"] = _aggregate_context_states(rows)
-    reasons.append(
-        f"aggregated {len(rows)} reported status context(s); branch protection "
-        "configures no explicit required-context list"
-    )
-    return result
-
 
 def _normalize_sha(value: str | None) -> str | None:
     text = (value or "").strip().lower()
@@ -300,7 +120,6 @@ def assess_pr_sync_status(
         "branch_protection_requires_current_base": requires_current,
         "approval_at_current_head": approval_ok if approval_at_current_head is not None else None,
         "checks_status": checks,
-        "checks_required": bool(checks_required),
         "active_locks_and_leases": {
             "author_lock": bool(active_author_lock) if active_author_lock is not None else None,
             "reviewer_lease": bool(active_reviewer_lease) if active_reviewer_lease is not None else None,
@@ -439,41 +258,21 @@ def assess_pr_sync_status(
         result["recommended_next_action"] = ACTION_BLOCKED
         return result
 
-    # ── Checks gate for merge_now (#751) ─────────────────────────────────
-    # ``checks_required`` is derived from the live branch-protection policy by
-    # the production caller. When protection does not require status checks,
-    # head contexts cannot gate the merge and this whole gate is skipped.
-    if not checks_required:
-        reasons.append(
-            "live branch protection does not require status checks; head check "
-            f"state ({checks}) does not gate merge"
-        )
-    elif checks not in _CHECKS_OK:
-        if checks in _CTX_PENDING:
+    # ── Checks gate for merge_now ────────────────────────────────────────
+    if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"):
+        if checks in ("pending", "running", "queued"):
             reasons.append(f"required checks are not finished (status={checks})")
-        elif checks in _CTX_FAILURE:
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+        if checks in ("failure", "failed", "error", "cancelled"):
             reasons.append(f"required checks failed (status={checks})")
-        elif checks == CHECKS_MISSING_REQUIRED:
-            reasons.append(
-                "branch protection configures required status context(s) but no "
-                "matching status result exists at the current head (fail closed)"
-            )
-        elif checks == CHECKS_NONE:
-            reasons.append(
-                "branch protection requires status checks but no status context "
-                "was produced for the current head (fail closed); an empty "
-                "status collection is not executing CI"
-            )
-        elif checks == CHECKS_UNKNOWN:
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
+        # unknown — fail closed when checks_required
+        if checks == "unknown":
             reasons.append("checks status unknown (fail closed)")
-        else:
-            # Unrecognized vocabulary must never fall through to merge_now.
-            reasons.append(
-                f"unrecognized checks status '{checks}' cannot prove required "
-                "checks passed (fail closed)"
-            )
-        result["recommended_next_action"] = ACTION_BLOCKED
-        return result
+            result["recommended_next_action"] = ACTION_BLOCKED
+            return result
 
     # ── Ready to merge without update ────────────────────────────────────
     # Includes: current with approval; outdated when update is NOT required.

skills/llm-project-workflow/workflows/create-issue.md

@@ -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:

tests/test_create_issue_bootstrap.py

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

tests/test_create_issue_workspace_guard.py

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

tests/test_issue_751_checks_assessor.py

@@ -1,602 +0,0 @@
-"""Regression coverage for the PR checks assessor defect (#751).
-
-Gitea's *combined* commit status reports ``state: pending`` both when a check is
-executing and when the status-context collection is empty. The assessor read
-``state`` alone and defaulted ``checks_required`` to ``True``, so a PR whose head
-had no status contexts — and never would — was routed to ``blocked`` forever.
-
-These tests pin the corrected semantics end to end: live branch protection
-decides whether checks are required, and the actual context collection decides
-what the checks say.
-"""
-
-from __future__ import annotations
-
-import sys
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-import pr_sync_status  # noqa: E402
-from pr_sync_status import (  # noqa: E402
-    ACTION_BLOCKED,
-    ACTION_MERGE_NOW,
-    ACTION_UPDATE_BRANCH_BY_MERGE,
-    CHECKS_FAILURE,
-    CHECKS_MISSING_REQUIRED,
-    CHECKS_NONE,
-    CHECKS_NOT_REQUIRED,
-    CHECKS_PENDING,
-    CHECKS_SUCCESS,
-    CHECKS_UNKNOWN,
-    assess_pr_sync_status,
-    classify_commit_checks,
-)
-
-
-def _sha(prefix: str) -> str:
-    return (prefix + "0" * 40)[:40]
-
-
-PR_HEAD = _sha("aaaaaaaa")
-BASE_HEAD = _sha("bbbbbbbb")
-
-
-def _base_kwargs(**overrides):
-    data = {
-        "host": "gitea.prgs.cc",
-        "org": "Scaled-Tech-Consulting",
-        "repo": "Gitea-Tools",
-        "pr_number": 751,
-        "pr_state": "open",
-        "source_branch": "fix/issue-751-checks-assessor",
-        "pr_head_sha": PR_HEAD,
-        "base_head_sha": BASE_HEAD,
-        "commits_behind": 0,
-        "mergeable": True,
-        "has_conflicts": False,
-        "branch_protection_requires_current_base": False,
-        "approval_at_current_head": True,
-        "checks_status": "success",
-        "active_author_lock": True,
-        "active_reviewer_lease": False,
-        "active_merger_lease": False,
-    }
-    data.update(overrides)
-    return data
-
-
-def _reasons(result) -> str:
-    return " | ".join(result["reasons"]).lower()
-
-
-class TestClassifyCommitChecks(unittest.TestCase):
-    """Pure classification from live evidence."""
-
-    def test_empty_collection_with_combined_pending_is_not_executing_ci(self):
-        # The exact PR #750 failure mode at the classification layer.
-        result = classify_commit_checks(
-            combined_state="pending",
-            statuses=[],
-            checks_enabled=True,
-            required_contexts=[],
-        )
-        self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
-        self.assertEqual(result["checks_status"], CHECKS_NONE)
-        self.assertEqual(result["context_count"], 0)
-        joined = " ".join(result["reasons"]).lower()
-        self.assertIn("empty", joined)
-        self.assertIn("does not indicate", joined)
-
-    def test_protection_disables_status_checks(self):
-        result = classify_commit_checks(
-            combined_state="pending",
-            statuses=[],
-            checks_enabled=False,
-            required_contexts=[],
-        )
-        self.assertFalse(result["checks_required"])
-        self.assertEqual(result["checks_status"], CHECKS_NOT_REQUIRED)
-
-    def test_no_required_contexts_aggregates_reported_contexts(self):
-        result = classify_commit_checks(
-            combined_state="success",
-            statuses=[{"context": "build", "status": "success"}],
-            checks_enabled=True,
-            required_contexts=[],
-        )
-        self.assertTrue(result["checks_required"])
-        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
-
-    def test_required_checks_pending(self):
-        result = classify_commit_checks(
-            combined_state="pending",
-            statuses=[{"context": "build", "status": "pending"}],
-            checks_enabled=True,
-            required_contexts=["build"],
-        )
-        self.assertEqual(result["checks_status"], CHECKS_PENDING)
-
-    def test_required_checks_failed(self):
-        result = classify_commit_checks(
-            combined_state="failure",
-            statuses=[{"context": "build", "status": "failure"}],
-            checks_enabled=True,
-            required_contexts=["build"],
-        )
-        self.assertEqual(result["checks_status"], CHECKS_FAILURE)
-
-    def test_required_checks_successful(self):
-        result = classify_commit_checks(
-            combined_state="success",
-            statuses=[{"context": "build", "status": "success"}],
-            checks_enabled=True,
-            required_contexts=["build"],
-        )
-        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
-        self.assertTrue(result["checks_required"])
-
-    def test_required_context_configured_with_no_matching_result(self):
-        result = classify_commit_checks(
-            combined_state="success",
-            statuses=[{"context": "lint", "status": "success"}],
-            checks_enabled=True,
-            required_contexts=["build"],
-        )
-        self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
-        self.assertEqual(result["missing_required_contexts"], ["build"])
-
-    def test_mixed_required_and_unrelated_contexts_ignores_unrelated(self):
-        # An unrelated failing context must not fail a satisfied required set.
-        result = classify_commit_checks(
-            combined_state="failure",
-            statuses=[
-                {"context": "build", "status": "success"},
-                {"context": "optional-scan", "status": "failure"},
-            ],
-            checks_enabled=True,
-            required_contexts=["build"],
-        )
-        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
-        # ...and a failing *required* context still fails despite passing extras.
-        failing = classify_commit_checks(
-            combined_state="success",
-            statuses=[
-                {"context": "build", "status": "failure"},
-                {"context": "optional-scan", "status": "success"},
-            ],
-            checks_enabled=True,
-            required_contexts=["build"],
-        )
-        self.assertEqual(failing["checks_status"], CHECKS_FAILURE)
-
-    def test_policy_unreadable_fails_closed(self):
-        result = classify_commit_checks(
-            combined_state=None,
-            statuses=[],
-            checks_enabled=None,
-            required_contexts=[],
-            policy_determinable=False,
-        )
-        self.assertTrue(result["checks_required"])
-        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
-        self.assertIn("fail closed", " ".join(result["reasons"]).lower())
-
-    def test_malformed_policy_indeterminate_flag_fails_closed(self):
-        result = classify_commit_checks(
-            combined_state="success",
-            statuses=[{"context": "build", "status": "success"}],
-            checks_enabled=None,
-            required_contexts=[],
-            policy_determinable=True,
-        )
-        self.assertTrue(result["checks_required"])
-        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
-
-    def test_status_collection_unreadable_fails_closed_when_required(self):
-        result = classify_commit_checks(
-            checks_enabled=True,
-            required_contexts=["build"],
-            status_determinable=False,
-        )
-        self.assertTrue(result["checks_required"])
-        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
-
-    def test_unrecognized_context_state_never_reads_as_success(self):
-        result = classify_commit_checks(
-            combined_state="success",
-            statuses=[{"context": "build", "status": "banana"}],
-            checks_enabled=True,
-            required_contexts=["build"],
-        )
-        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
-
-    def test_newest_wins_per_context(self):
-        # Gitea returns newest-first; the stale failure must not win.
-        result = classify_commit_checks(
-            combined_state="success",
-            statuses=[
-                {"context": "build", "status": "success"},
-                {"context": "build", "status": "failure"},
-            ],
-            checks_enabled=True,
-            required_contexts=["build"],
-        )
-        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
-
-    def test_malformed_status_rows_are_discarded_not_treated_as_passing(self):
-        result = classify_commit_checks(
-            combined_state="pending",
-            statuses=[{"context": "build"}, "not-a-dict", None],
-            checks_enabled=True,
-            required_contexts=["build"],
-        )
-        self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
-
-
-class TestChecksGateSemantics(unittest.TestCase):
-    """``assess_pr_sync_status`` routing and blocker reasons."""
-
-    def test_not_required_allows_merge_now_with_empty_checks(self):
-        result = assess_pr_sync_status(
-            **_base_kwargs(checks_status=CHECKS_NOT_REQUIRED),
-            checks_required=False,
-        )
-        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
-        self.assertTrue(result["approval_valid_for_merge"])
-        self.assertFalse(result["checks_required"])
-        self.assertIn("does not require status checks", _reasons(result))
-
-    def test_none_blocks_when_checks_required(self):
-        result = assess_pr_sync_status(
-            **_base_kwargs(checks_status=CHECKS_NONE),
-            checks_required=True,
-        )
-        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
-        self.assertIn("no status context", _reasons(result))
-        self.assertIn("not executing ci", _reasons(result))
-
-    def test_missing_required_blocks(self):
-        result = assess_pr_sync_status(
-            **_base_kwargs(checks_status=CHECKS_MISSING_REQUIRED),
-            checks_required=True,
-        )
-        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
-        self.assertIn("no matching status result", _reasons(result))
-
-    def test_required_pending_blocks(self):
-        result = assess_pr_sync_status(
-            **_base_kwargs(checks_status=CHECKS_PENDING), checks_required=True
-        )
-        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
-        self.assertIn("not finished", _reasons(result))
-
-    def test_required_failure_blocks(self):
-        result = assess_pr_sync_status(
-            **_base_kwargs(checks_status=CHECKS_FAILURE), checks_required=True
-        )
-        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
-        self.assertIn("failed", _reasons(result))
-
-    def test_required_success_merges(self):
-        result = assess_pr_sync_status(
-            **_base_kwargs(checks_status=CHECKS_SUCCESS), checks_required=True
-        )
-        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
-
-    def test_unknown_blocks_when_required(self):
-        result = assess_pr_sync_status(
-            **_base_kwargs(checks_status=CHECKS_UNKNOWN), checks_required=True
-        )
-        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
-        self.assertIn("unknown", _reasons(result))
-
-    def test_unrecognized_status_does_not_fall_through_to_merge(self):
-        # Regression: the previous gate only handled a fixed vocabulary and let
-        # anything else reach merge_now.
-        result = assess_pr_sync_status(
-            **_base_kwargs(checks_status="totally-unexpected"),
-            checks_required=True,
-        )
-        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
-        self.assertIn("unrecognized checks status", _reasons(result))
-
-    def test_checks_gate_does_not_bypass_approval_requirement(self):
-        result = assess_pr_sync_status(
-            **_base_kwargs(
-                checks_status=CHECKS_NOT_REQUIRED, approval_at_current_head=False
-            ),
-            checks_required=False,
-        )
-        self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
-        self.assertFalse(result["approval_valid_for_merge"])
-
-    def test_checks_gate_does_not_bypass_current_base_protection(self):
-        # Existing current-base behavior stays intact (#727).
-        result = assess_pr_sync_status(
-            **_base_kwargs(
-                checks_status=CHECKS_NOT_REQUIRED,
-                commits_behind=3,
-                branch_protection_requires_current_base=True,
-            ),
-            checks_required=False,
-        )
-        self.assertEqual(
-            result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
-        )
-
-    def test_checks_gate_does_not_bypass_conflicts(self):
-        result = assess_pr_sync_status(
-            **_base_kwargs(checks_status=CHECKS_NOT_REQUIRED, mergeable=False),
-            checks_required=False,
-        )
-        self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
-
-
-class TestBranchProtectionPolicy(unittest.TestCase):
-    """Live protection reader derives the status-check requirement."""
-
-    def setUp(self):
-        import gitea_mcp_server as gms
-
-        self.gms = gms
-
-    def _policy(self, responses):
-        def fake_api_request(method, url, auth, *args, **kwargs):
-            for fragment, payload in responses.items():
-                if fragment in url:
-                    if isinstance(payload, Exception):
-                        raise payload
-                    return payload
-            return None
-
-        with patch.object(self.gms, "api_request", side_effect=fake_api_request):
-            return self.gms._branch_protection_policy(
-                "https://example/api/v1/repos/o/r", {"h": "1"}, base_branch="master"
-            )
-
-    def test_status_checks_enabled_with_contexts(self):
-        policy = self._policy({
-            "branch_protections": [{
-                "branch_name": "master",
-                "block_on_outdated_branch": True,
-                "enable_status_check": True,
-                "status_check_contexts": ["ci/build", " "],
-            }],
-        })
-        self.assertTrue(policy["determinable"])
-        self.assertTrue(policy["checks_enabled"])
-        self.assertTrue(policy["requires_current_base"])
-        self.assertEqual(policy["required_contexts"], ["ci/build"])
-
-    def test_status_checks_disabled(self):
-        policy = self._policy({
-            "branch_protections": [{
-                "branch_name": "master",
-                "enable_status_check": False,
-            }],
-        })
-        self.assertTrue(policy["determinable"])
-        self.assertFalse(policy["checks_enabled"])
-
-    def test_no_protection_rule_means_checks_not_required(self):
-        # PR #750's repository shape: protection list readable but empty.
-        policy = self._policy({"branch_protections": [], "branches/master": {}})
-        self.assertTrue(policy["determinable"])
-        self.assertFalse(policy["protection_found"])
-        self.assertFalse(policy["checks_enabled"])
-        self.assertIsNone(policy["requires_current_base"])
-
-    def test_protection_without_status_field_means_not_enabled(self):
-        policy = self._policy({
-            "branch_protections": [{
-                "branch_name": "master",
-                "block_on_outdated_branch": True,
-            }],
-        })
-        self.assertTrue(policy["determinable"])
-        self.assertFalse(policy["checks_enabled"])
-        self.assertTrue(policy["requires_current_base"])
-
-    def test_api_failure_is_not_determinable(self):
-        policy = self._policy({
-            "branch_protections": RuntimeError("boom"),
-        })
-        self.assertFalse(policy["determinable"])
-        self.assertIsNone(policy["checks_enabled"])
-
-    def test_missing_branch_is_not_determinable(self):
-        with patch.object(self.gms, "api_request", return_value=[]):
-            policy = self.gms._branch_protection_policy(
-                "https://example/api/v1/repos/o/r", {}, base_branch="  "
-            )
-        self.assertFalse(policy["determinable"])
-
-    def test_requires_current_base_accessor_preserved(self):
-        def fake(method, url, auth, *a, **k):
-            if "branch_protections" in url:
-                return [{"branch_name": "master",
-                         "block_on_outdated_branch": True}]
-            return None
-
-        with patch.object(self.gms, "api_request", side_effect=fake):
-            value = self.gms._branch_protection_requires_current_base(
-                "https://example/api/v1/repos/o/r", {}, base_branch="master"
-            )
-        self.assertTrue(value)
-
-
-class TestCommitChecksSnapshot(unittest.TestCase):
-    def setUp(self):
-        import gitea_mcp_server as gms
-
-        self.gms = gms
-
-    def test_reads_state_and_context_collection(self):
-        payload = {"state": "pending", "statuses": [{"context": "b",
-                                                     "status": "pending"}]}
-        with patch.object(self.gms, "api_request", return_value=payload):
-            snap = self.gms._commit_checks_snapshot(
-                "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
-            )
-        self.assertTrue(snap["determinable"])
-        self.assertEqual(snap["combined_state"], "pending")
-        self.assertEqual(len(snap["statuses"]), 1)
-
-    def test_empty_collection_recorded_as_determinable(self):
-        with patch.object(
-            self.gms, "api_request", return_value={"state": "pending", "statuses": []}
-        ):
-            snap = self.gms._commit_checks_snapshot(
-                "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
-            )
-        self.assertTrue(snap["determinable"])
-        self.assertEqual(snap["statuses"], [])
-
-    def test_api_failure_is_not_determinable(self):
-        with patch.object(self.gms, "api_request", side_effect=RuntimeError("x")):
-            snap = self.gms._commit_checks_snapshot(
-                "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
-            )
-        self.assertFalse(snap["determinable"])
-
-
-class TestMcpWrapperForwarding(unittest.TestCase):
-    """The production MCP path must reach the derived ``checks_required``."""
-
-    def setUp(self):
-        import gitea_mcp_server as gms
-
-        self.gms = gms
-        self.base = (
-            "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools"
-        )
-
-    def _patches(self, fake_api_request, spy):
-        return [
-            patch.object(self.gms, "_profile_operation_gate", return_value=None),
-            patch.object(self.gms, "_resolve", return_value=(
-                "gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")),
-            patch.object(self.gms, "_auth", return_value={"Authorization": "x"}),
-            patch.object(self.gms, "repo_api_url", return_value=self.base),
-            patch.object(self.gms, "api_request", side_effect=fake_api_request),
-            patch.object(self.gms, "gitea_get_pr_review_feedback", return_value={
-                "success": True, "approval_at_current_head": True}),
-            patch.object(self.gms, "_prove_author_ownership_for_pr", return_value={
-                "has_author_lock": True}),
-            patch.object(pr_sync_status, "assess_pr_sync_status", side_effect=spy),
-        ]
-
-    def _run(self, *, protections, status_payload, pr_number=750,
-             caller_checks_status=None):
-        captured = {}
-        real_assess = pr_sync_status.assess_pr_sync_status
-
-        def spy(**kwargs):
-            captured.update(kwargs)
-            return real_assess(**kwargs)
-
-        def fake_api_request(method, url, auth, *args, **kwargs):
-            if f"/pulls/{pr_number}" in url:
-                return {
-                    "number": pr_number,
-                    "state": "open",
-                    "title": "t",
-                    "body": "b",
-                    "mergeable": True,
-                    "head": {"sha": PR_HEAD, "ref": "fix/x"},
-                    "base": {"sha": BASE_HEAD, "ref": "master"},
-                }
-            if "/branch_protections" in url:
-                if isinstance(protections, Exception):
-                    raise protections
-                return protections
-            if "/branches/master" in url:
-                return {"commit": {"id": BASE_HEAD}}
-            if "/status" in url:
-                if isinstance(status_payload, Exception):
-                    raise status_payload
-                return status_payload
-            if "/compare/" in url:
-                return {"total_commits": 0}
-            if "/comments" in url:
-                return []
-            return None
-
-        stack = self._patches(fake_api_request, spy)
-        for p in stack:
-            p.start()
-        try:
-            result = self.gms.gitea_assess_pr_sync_status(
-                pr_number=pr_number, remote="prgs",
-                checks_status=caller_checks_status,
-            )
-        finally:
-            for p in reversed(stack):
-                p.stop()
-        return result, captured
-
-    def test_derived_checks_required_is_forwarded(self):
-        result, captured = self._run(
-            protections=[{"branch_name": "master", "enable_status_check": True,
-                          "status_check_contexts": ["ci/build"]}],
-            status_payload={"state": "pending", "statuses": [
-                {"context": "ci/build", "status": "pending"}]},
-        )
-        self.assertIn("checks_required", captured)
-        self.assertTrue(captured["checks_required"])
-        self.assertEqual(captured["checks_status"], CHECKS_PENDING)
-        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
-
-    def test_pr750_empty_status_with_no_protection_is_merge_ready(self):
-        # The exact reported failure: combined pending, zero contexts, and no
-        # branch protection requiring checks.
-        result, captured = self._run(
-            protections=[],
-            status_payload={"state": "pending", "statuses": []},
-        )
-        self.assertFalse(captured["checks_required"])
-        self.assertEqual(captured["checks_status"], CHECKS_NOT_REQUIRED)
-        self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
-        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
-
-    def test_protection_read_failure_blocks(self):
-        result, captured = self._run(
-            protections=RuntimeError("protection unavailable"),
-            status_payload={"state": "success", "statuses": []},
-        )
-        self.assertTrue(captured["checks_required"])
-        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
-
-    def test_caller_supplied_status_cannot_mask_live_required_failure(self):
-        # A caller-declared "success" must not override live evidence.
-        result, captured = self._run(
-            protections=[{"branch_name": "master", "enable_status_check": True,
-                          "status_check_contexts": ["ci/build"]}],
-            status_payload={"state": "failure", "statuses": [
-                {"context": "ci/build", "status": "failure"}]},
-            caller_checks_status="success",
-        )
-        self.assertEqual(captured["checks_status"], CHECKS_FAILURE)
-        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
-        self.assertEqual(
-            result["checks_evidence"]["caller_supplied_checks_status"], "success"
-        )
-
-    def test_checks_evidence_is_reported_without_secrets(self):
-        result, _ = self._run(
-            protections=[],
-            status_payload={"state": "pending", "statuses": []},
-        )
-        evidence = result["checks_evidence"]
-        self.assertEqual(evidence["context_count"], 0)
-        self.assertEqual(evidence["combined_state"], "pending")
-        self.assertFalse(evidence["protection_found"])
-        blob = repr(result).lower()
-        self.assertNotIn("authorization", blob)
-        self.assertNotIn("token", blob)
-
-
-if __name__ == "__main__":
-    unittest.main()

workflow_scope_guard.py

@@ -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")),
     }