fix(mcp): require proven base equivalence for the create-issue bootstrap

Remediates review #473 (REQUEST_CHANGES) on PR #759 / issue #757 AC3/AC4.

The bootstrap compared SHAs only inside:

    if remote_tip and local_tip and remote_tip != local_tip:

so agreement was assumed whenever either tip was unknown. A missing local
HEAD, an unresolvable live master, or a resolver exception silently granted
the control-checkout exemption instead of blocking it. An unknown tip is
missing evidence, not proof of equivalence.

Changes:

* assess_create_issue_bootstrap now blocks unless BOTH tips are known and
  equal, with distinct reasons for missing local HEAD, unknown live master,
  and resolver failure. Adds a remote_master_sha_error parameter so a failed
  resolution is reported as missing evidence rather than absence of a
  constraint.
* Both normalized SHAs and a derived base_tips_verified flag are recorded on
  the assessment. normalize_sha() treats only case and surrounding whitespace
  as equivalent spellings of a commit.
* bootstrap_permits_control_checkout re-derives the comparison from the
  recorded tips instead of trusting base_tips_verified, so a hand-built or
  truncated assessment cannot assert agreement it never proved.
* _create_issue_bootstrap_assessment captures the resolver exception and
  forwards it, replacing the silent except -> None.

One shared assessment still serves both the #274 and #604 guards, and
_BOOTSTRAP_UNSET is preserved. No MCP tool signature gains a bootstrap
argument (AC6). No issue or PR number is special-cased (AC10).

Tests: 16 new assertions across two classes covering missing local, missing
remote, both missing, empty/whitespace tips, mismatch, resolver exception at
the assessment site, and resolver exception through the real
verify_preflight_purity path; plus predicate rejection of stripped tips, a
forged base_tips_verified flag, and a missing flag. All 16 fail against the
sources at adc61255 and pass after this change. The valid non-create_issue
lock_issue test is retained unchanged.

Validation: focused #757 suite 51 passed (31 subtests); affected guard and
preflight suites 217 passed (67 subtests); full suite 3637 passed, 2 failed,
6 skipped, 431 subtests. Both failures (test_issue_702 F1 worktree recovery;
test_reconciler_supersession_close org/repo forwarding) are the documented
pre-existing baseline failures on bde5c5fb and are not caused by this change.

Refs #757

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_015KRvvrFtaM5FEvQ6LvDJMH

anti_stomp_preflight.py

@@ -33,6 +33,7 @@ from __future__ import annotations
 from typing import Any
 
 import author_mutation_worktree
+import create_issue_bootstrap
 import master_parity_gate
 import remote_repo_guard
 import root_checkout_guard
@@ -354,6 +355,7 @@ def assess_anti_stomp_preflight(
     remote_master_sha: str | None = None,
     check_root_checkout: bool = True,
     check_worktree: bool = True,
+    create_issue_bootstrap_assessment: dict[str, Any] | None = None,
     # stale runtime (master parity)
     startup_head: str | None = None,
     current_code_head: str | None = None,
@@ -584,12 +586,28 @@ def assess_anti_stomp_preflight(
             project_root=project_root,
             current_branch=current_branch,
         )
+        # #757: the #274 guard consults the server-derived create_issue
+        # bootstrap before blocking the canonical control checkout. Route this
+        # guard's decision through the *same* predicate on the *same*
+        # assessment so the two cannot disagree about identical evidence.
+        # Only the wrong-worktree verdict is waived; every other check in this
+        # assessment (root checkout, repo, role, stale runtime, lease, ...) is
+        # evaluated independently and still applies.
+        bootstrap_waived = wt.get("block") and (
+            create_issue_bootstrap.bootstrap_permits_control_checkout(
+                create_issue_bootstrap_assessment,
+                task=task_name,
+                workspace_path=workspace_path,
+                canonical_repo_root=project_root,
+            )
+        )
         checks["worktree"] = {
-            "block": bool(wt.get("block")),
+            "block": bool(wt.get("block")) and not bootstrap_waived,
             "reasons": list(wt.get("reasons") or []),
             "under_branches": wt.get("under_branches"),
+            "create_issue_bootstrap_waived": bool(bootstrap_waived),
         }
-        if wt.get("block"):
+        if wt.get("block") and not bootstrap_waived:
             blockers.append(
                 _blocker(
                     BLOCKER_WRONG_WORKTREE,

create_issue_bootstrap.py

@@ -0,0 +1,359 @@
+"""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 normalize_sha(value: str | None) -> str | None:
+    """Normalize a Git object id for comparison, or ``None`` when unknown.
+
+    Whitespace and case are the only permitted variation between two spellings
+    of the same commit; anything else is a different commit. Empty and
+    whitespace-only values normalize to ``None`` so an unknown tip can never
+    compare equal to another unknown tip.
+    """
+    normalized = (value or "").strip().lower()
+    return normalized or None
+
+
+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,
+    remote_master_sha_error: 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))})"
+        )
+
+    # #757 AC3/AC4: base-equivalence must be *proven*, never assumed. An
+    # unknown tip on either side is not evidence of agreement, so a missing
+    # local HEAD, an unresolvable live master, or a resolver failure all block.
+    remote_tip = normalize_sha(remote_master_sha)
+    local_tip = normalize_sha(head_sha)
+    resolver_error = (remote_master_sha_error or "").strip() or None
+
+    if not local_tip:
+        reasons.append(
+            "create_issue bootstrap blocked: control checkout HEAD SHA is "
+            "unknown; base equivalence to live master cannot be proven "
+            "(fail closed)"
+        )
+
+    if resolver_error:
+        reasons.append(
+            "create_issue bootstrap blocked: live master tip could not be "
+            f"resolved ({resolver_error}); base equivalence cannot be proven "
+            "(fail closed)"
+        )
+    elif not remote_tip:
+        reasons.append(
+            "create_issue bootstrap blocked: live master tip is unknown; base "
+            "equivalence to live master cannot be proven (fail closed)"
+        )
+
+    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,
+            local_head_sha=local_tip,
+            remote_master_sha=remote_tip,
+        )
+
+    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",
+        local_head_sha=local_tip,
+        remote_master_sha=remote_tip,
+    )
+
+
+def bootstrap_permits_control_checkout(
+    assessment: Any,
+    *,
+    task: str | None,
+    workspace_path: str | None,
+    canonical_repo_root: str | None,
+) -> bool:
+    """Single interpretation of a bootstrap assessment (#757).
+
+    Both author-mutation guards — the #274 branches-only enforcer and the #604
+    anti-stomp preflight — route their "may this workspace mutate" decision
+    through this predicate, so the two can never reach opposite conclusions
+    about identical evidence.
+
+    Fail-closed by construction. Every proof obligation must be present and
+    affirmative in *assessment*, and the assessment must describe the very
+    workspace and canonical root being guarded. A missing, malformed, refused,
+    incomplete, or contradictory assessment returns ``False``, which leaves the
+    caller's ordinary block in force.
+
+    ``assessment`` is server-derived only: it is produced by
+    :func:`assess_create_issue_bootstrap` from inspected repository state. It is
+    never accepted from an MCP tool argument, so no caller can assert
+    eligibility it has not proven.
+    """
+    if not isinstance(assessment, dict):
+        return False
+    if not is_create_issue_task(task):
+        return False
+
+    # Positive proof: the assessment must affirmatively allow, with no
+    # competing refusal or not-applicable disposition recorded alongside it.
+    if assessment.get("allowed") is not True:
+        return False
+    if assessment.get("proven") is not True:
+        return False
+    if assessment.get("block") is not False:
+        return False
+    if assessment.get("not_applicable") is not False:
+        return False
+    if assessment.get("reasons"):
+        return False
+
+    # Scope proof: only the create_issue bootstrap, only via the clean
+    # canonical control checkout path.
+    if assessment.get("task_scope") != "create_issue_only":
+        return False
+    if assessment.get("bootstrap_path") != "clean_canonical_control_checkout":
+        return False
+
+    # State proof: clean, and not a branches/ worktree (those keep #274).
+    if assessment.get("dirty_files"):
+        return False
+    if assessment.get("under_branches") is not False:
+        return False
+
+    # Base-equivalence proof (#757 AC3/AC4): both tips must be recorded,
+    # nonempty, and equal. Re-derived here rather than trusted from the
+    # assessment's own flag, so a hand-built or truncated assessment cannot
+    # assert agreement it never proved.
+    if assessment.get("base_tips_verified") is not True:
+        return False
+    local_tip = normalize_sha(assessment.get("local_head_sha"))
+    remote_tip = normalize_sha(assessment.get("remote_master_sha"))
+    if not local_tip or not remote_tip or local_tip != remote_tip:
+        return False
+
+    # Binding proof: the assessment must describe *this* workspace and root,
+    # and that workspace must be exactly the canonical control checkout.
+    root = os.path.realpath(canonical_repo_root or "")
+    workspace = os.path.realpath(workspace_path or root or ".")
+    if not root or workspace != root:
+        return False
+    if assessment.get("canonical_repo_root") != root:
+        return False
+    if assessment.get("workspace_path") != workspace:
+        return False
+
+    return True
+
+
+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,
+    local_head_sha: str | None = None,
+    remote_master_sha: str | None = None,
+) -> dict[str, Any]:
+    # #757 AC3/AC4: equality is recorded only when BOTH tips are known, so a
+    # consumer can never read agreement out of two missing values.
+    base_tips_verified = bool(
+        local_head_sha and remote_master_sha and local_head_sha == remote_master_sha
+    )
+    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",
+        "local_head_sha": local_head_sha,
+        "remote_master_sha": remote_master_sha,
+        "base_tips_verified": base_tips_verified,
+    }

docs/gitea-execution-profiles.md

@@ -317,6 +317,44 @@ Least-privilege constraints:
   canonical names such as `gitea.pr.close` (never bare `pr.close` /
   `issue.close`, which the production normalizer rejects or drops).
 
+### Post-merge moot-lease cleanup ownership (`gitea.pr.comment`)
+
+Neutralising a reviewer lease left behind on an already-merged/closed PR is
+reconciliation work too. `task_capability_map` maps
+`cleanup_post_merge_moot_lease` — and its tool-name alias
+`gitea_cleanup_post_merge_moot_lease` — to role `reconciler` with permission
+`gitea.pr.comment` (#745). Both names carry the **same** contract.
+
+The permission alone is deliberately not sufficient: author, reviewer and
+merger profiles all hold `gitea.pr.comment` for ordinary PR discussion, so the
+role gate — not the permission gate — is what keeps the terminal lease marker
+reconciler-owned.
+
+`gitea_cleanup_post_merge_moot_lease` splits its two modes on purpose:
+
+- **`apply=false` (assessment) requires only `gitea.read`, with no role gate.**
+  This matches `gitea_cleanup_stale_review_decision_lock` and
+  `gitea_cleanup_obsolete_reviewer_comment_lease`, whose assessment paths are
+  likewise read-gated, so an operator can diagnose a stuck lease from whichever
+  namespace happens to be attached without switching roles. The dry run
+  performs no mutation and records append-only evidence in-session.
+- **`apply=true` (mutation) requires all of the following**, in order: the
+  session must have resolved exactly `cleanup_post_merge_moot_lease` (resolving
+  any other task — including a sibling reconciler task — does not authorize
+  it); the active role must be `reconciler`; the profile must hold
+  `gitea.pr.comment`; the explicit `org`/`repo` must agree with the canonical
+  repository identity, which is derived from the session binding and can never
+  be overridden by request parameters; and matching dry-run evidence must show
+  `lease_moot`, `cleanup_allowed`, and the same PR, lease session, candidate
+  head and lease marker id that are live at apply time.
+
+Everything else fails closed: a live lease on an open PR, an already-terminal
+(idempotent) lease, a lease superseded between the dry run and the apply, a
+malformed lease missing session/head/marker, and any foreign-repository target.
+The cleanup only ever appends a terminal `phase: released` marker
+(`blocker: post-merge-moot`) — it never edits or deletes another session's
+comment, and it never merges or adopts a lease.
+
 Launch a static `gitea-reconciler` MCP namespace with
 `GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
 `reconciler_profile.assess_reconciler_profile` (#304). Use the

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,63 @@ def _enforce_canonical_repository_root(
         )
 
 
-def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
+# #757: distinguishes "caller supplied no bootstrap evidence" (compute it) from
+# "caller supplied a not-applicable/refused assessment" (honour it, fail closed).
+_BOOTSTRAP_UNSET = object()
+
+
+def _create_issue_bootstrap_assessment(
+    task: str | None,
+    worktree_path: str | None = None,
+) -> dict | None:
+    """#757: the single server-derived create_issue bootstrap assessment.
+
+    Computed once per preflight from inspected repository state, then consumed
+    by BOTH the #274 branches-only guard and the #604 anti-stomp preflight so
+    the two cannot reach opposite conclusions about identical evidence.
+
+    Returns ``None`` when *task* is not a create_issue mutation, which leaves
+    every other author mutation on the ordinary branches-only path. The result
+    is derived from inspected repository state only — never from an MCP tool
+    argument.
+    """
+    import create_issue_bootstrap as _cib
+
+    if not _cib.is_create_issue_task(task):
+        return None
+
+    ctx = _resolve_namespace_mutation_context(worktree_path)
+    workspace = ctx["workspace_path"]
+    git_state = issue_lock_worktree.read_worktree_git_state(workspace)
+    # #757 AC3/AC4: a resolver failure is not "no constraint" — it is missing
+    # evidence, and must reach the assessor as such so the bootstrap fails
+    # closed instead of proceeding without base-equivalence proof.
+    remote_master_sha_error: str | None = None
+    try:
+        remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
+            ctx["canonical_repo_root"]
+        )
+    except Exception as exc:
+        remote_master_sha = None
+        remote_master_sha_error = f"{type(exc).__name__}: {exc}".strip() or "resolver failed"
+    return _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,
+        remote_master_sha_error=remote_master_sha_error,
+        task=task,
+    )
+
+
+def _enforce_branches_only_author_mutation(
+    worktree_path: str | None = None,
+    *,
+    task: str | None = None,
+    bootstrap_assessment: Any = _BOOTSTRAP_UNSET,
+) -> None:
     """#274: author file/branch mutations must run from a branches/ worktree.
 
     Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
@@ -831,6 +887,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 +907,37 @@ 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.
+    # #757: consume the caller-computed assessment when one was threaded in, so
+    # this guard and the later #604 anti-stomp guard judge identical evidence.
+    # Falling back to computing it here preserves behaviour for callers that
+    # supply none.
+    import create_issue_bootstrap as _cib
+
+    bootstrap = (
+        _create_issue_bootstrap_assessment(task, worktree_path)
+        if bootstrap_assessment is _BOOTSTRAP_UNSET
+        else bootstrap_assessment
+    )
+    if _cib.bootstrap_permits_control_checkout(
+        bootstrap,
+        task=task,
+        workspace_path=workspace,
+        canonical_repo_root=ctx["canonical_repo_root"],
+    ):
+        return
+    if (
+        isinstance(bootstrap, dict)
+        and 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:
@@ -891,6 +980,7 @@ def _run_anti_stomp_preflight(
     workflow_hash_valid: bool | None = None,
     workflow_hash_reasons: list[str] | None = None,
     raise_on_block: bool = True,
+    bootstrap_assessment: Any = _BOOTSTRAP_UNSET,
 ) -> dict | None:
     """Shared #604 anti-stomp preflight for MCP mutation entrypoints.
 
@@ -1071,6 +1161,14 @@ def _run_anti_stomp_preflight(
         source_contaminated=source_contaminated,
         contamination_reasons=contamination_reasons,
         manual_bypass_attempted=False,
+        # #757: same server-derived bootstrap decision the #274 guard consumed.
+        # Computing it here when unsupplied keeps standalone callers consistent
+        # rather than silently bootstrap-blind.
+        create_issue_bootstrap_assessment=(
+            _create_issue_bootstrap_assessment(task, worktree_path)
+            if bootstrap_assessment is _BOOTSTRAP_UNSET
+            else bootstrap_assessment
+        ),
     )
     if not assessment.get("block"):
         return None
@@ -1213,7 +1311,12 @@ 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)
+        # #757: compute the create_issue bootstrap decision ONCE and hand the
+        # same result to both the #274 and #604 guards, so they cannot disagree.
+        bootstrap_assessment = _create_issue_bootstrap_assessment(task, worktree_path)
+        _enforce_branches_only_author_mutation(
+            worktree_path, task=task, bootstrap_assessment=bootstrap_assessment
+        )
         _enforce_issue_scope_guard(
             worktree_path,
             task=task,
@@ -1223,6 +1326,7 @@ def verify_preflight_purity(
         # #604: common anti-stomp preflight after legacy + #683 enforcers.
         _run_anti_stomp_preflight(
             task,
+            bootstrap_assessment=bootstrap_assessment,
             remote=remote,
             worktree_path=worktree_path,
             org=org,
@@ -1247,7 +1351,11 @@ 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)
+        # #757: one shared bootstrap decision for both guards (see above).
+        bootstrap_assessment = _create_issue_bootstrap_assessment(task, worktree_path)
+        _enforce_branches_only_author_mutation(
+            worktree_path, task=task, bootstrap_assessment=bootstrap_assessment
+        )
         _enforce_issue_scope_guard(
             worktree_path,
             task=task,
@@ -1257,6 +1365,7 @@ def verify_preflight_purity(
         if force_anti_stomp:
             _run_anti_stomp_preflight(
                 task,
+                bootstrap_assessment=bootstrap_assessment,
                 remote=remote,
                 worktree_path=worktree_path,
                 org=org,
@@ -1366,10 +1475,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 +1495,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 +1509,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 +1524,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:
@@ -1830,6 +1954,7 @@ def _seed_session_context(
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
+import post_merge_moot_lease_gate  # noqa: E402  # #745 reconciler cleanup gate
 import merger_lease_adoption  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import worktree_cleanup_audit  # noqa: E402
@@ -2081,6 +2206,7 @@ def _assess_issue_duplicate_gate(
     auth: str,
     locked_branch: str | None = None,
     phase: str,
+    recovered_owning_pr: dict | None = None,
 ) -> dict:
     open_prs, branch_names, claim_entry = _collect_issue_duplicate_context(
         h, o, r, auth, issue_number
@@ -2092,6 +2218,7 @@ def _assess_issue_duplicate_gate(
         claim_entry=claim_entry,
         locked_branch=locked_branch,
         phase=phase,
+        recovered_owning_pr=recovered_owning_pr,
     )
 
 
@@ -3268,6 +3395,17 @@ def gitea_lock_issue(
     recovery_sanctioned = bool(
         recovery_assessment and recovery_assessment.get("recovery_sanctioned")
     )
+    # #755: a sanctioned dead-session recovery always has an owning open PR —
+    # that is what makes it a recovery rather than a fresh claim. Carry the
+    # server-derived owning-PR evidence into the duplicate-work gate below so
+    # the PR this lock already owns is not mistaken for competing duplicate
+    # work. Withheld (None) unless recovery was granted, so the ordinary
+    # duplicate blocker is untouched on every other path.
+    recovered_owning_pr = (
+        issue_lock_recovery.owning_pr_recovery_evidence(recovery_assessment)
+        if recovery_sanctioned
+        else None
+    )
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -3300,6 +3438,7 @@ def gitea_lock_issue(
         auth=auth,
         locked_branch=branch_name,
         phase=issue_work_duplicate_gate.PHASE_LOCK,
+        recovered_owning_pr=recovered_owning_pr,
     )
     if duplicate_gate.get("block"):
         raise ValueError("; ".join(duplicate_gate.get("reasons") or [
@@ -8955,13 +9094,14 @@ def gitea_review_pr(
         return out
 
 
-def _delete_branch_repository_binding_block(
+def _repository_binding_block(
     remote: str | None,
     *,
     org: str | None,
     repo: str | None,
+    required_permission: str = "gitea.branch.delete",
 ) -> dict | None:
-    """#733: validate an explicit delete target against the workspace binding.
+    """#733: validate an explicit mutation target against the workspace binding.
 
     The trusted repository identity comes only from the verified,
     workspace-aligned git remote (never ``REMOTES`` defaults, never
@@ -8994,7 +9134,7 @@ def _delete_branch_repository_binding_block(
         return {
             "success": False,
             "performed": False,
-            "required_permission": "gitea.branch.delete",
+            "required_permission": required_permission,
             "reasons": canonical_reasons,
             "blocker_kind": "repository_binding",
         }
@@ -9013,9 +9153,9 @@ def _delete_branch_repository_binding_block(
             return {
                 "success": False,
                 "performed": False,
-                "required_permission": "gitea.branch.delete",
+                "required_permission": required_permission,
                 "reasons": list(override.get("reasons") or [
-                    "delete target repository does not match the workspace "
+                    "target repository does not match the workspace "
                     "binding (fail closed)"
                 ]),
                 "blocker_kind": "repository_binding",
@@ -9025,17 +9165,51 @@ def _delete_branch_repository_binding_block(
         return {
             "success": False,
             "performed": False,
-            "required_permission": "gitea.branch.delete",
+            "required_permission": required_permission,
             "reasons": [
                 "repository binding unverified: no workspace repository "
                 "identity could be established to corroborate the explicit "
-                "delete target (fail closed)"
+                "target (fail closed)"
             ],
             "blocker_kind": "repository_binding",
         }
     return None
 
 
+def _delete_branch_repository_binding_block(
+    remote: str | None,
+    *,
+    org: str | None,
+    repo: str | None,
+) -> dict | None:
+    """Delete-branch view of :func:`_repository_binding_block` (#733).
+
+    Retained as the named entry point for the delete path so #733/#739
+    regression coverage keeps exercising the exact permission label that
+    ``gitea_delete_branch`` reports.
+    """
+    return _repository_binding_block(
+        remote, org=org, repo=repo,
+        required_permission="gitea.branch.delete",
+    )
+
+
+def _bound_repository_slug(remote: str | None) -> str | None:
+    """Canonical repository slug for the active session, or None (#745).
+
+    Same trust order as ``_repository_binding_block``: the configured canonical
+    root when the namespace declares one, otherwise the workspace-derived
+    identity. Request parameters are never consulted, so they cannot redirect a
+    mutation at a foreign repository.
+    """
+    canonical_slug, canonical_reasons = _canonical_repository_slug(
+        get_profile(), remote
+    )
+    if canonical_reasons:
+        return None
+    return canonical_slug or _workspace_repository_slug(remote)
+
+
 @mcp.tool()
 def gitea_delete_branch(
     branch: str,
@@ -12464,6 +12638,10 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
 def gitea_cleanup_post_merge_moot_lease(
     pr_number: int,
     apply: bool = False,
+    expected_session_id: str | None = None,
+    expected_candidate_head: str | None = None,
+    expected_lease_comment_id: int | None = None,
+    worktree_path: str | None = None,
     remote: str = "dadeschools",
     host: str | None = None,
     org: str | None = None,
@@ -12479,14 +12657,37 @@ def gitea_cleanup_post_merge_moot_lease(
     an append-only comment that neutralises the moot lease without deleting any
     other session's comment.
 
+    Role binding (#745). The two modes are gated differently, on purpose:
+
+    * ``apply=false`` (assessment) stays reachable under ``gitea.read`` for
+      **any** role, matching ``gitea_cleanup_stale_review_decision_lock`` and
+      ``gitea_cleanup_obsolete_reviewer_comment_lease``, so an operator can
+      diagnose a stuck lease from whichever namespace is attached. It performs
+      no mutation and records append-only dry-run evidence in-session.
+    * ``apply=true`` (mutation) additionally requires, in order: the session to
+      have resolved exactly ``cleanup_post_merge_moot_lease``; the
+      ``reconciler`` role; ``gitea.pr.comment``; a validated repository /
+      canonical-root / workspace binding; and matching dry-run evidence proving
+      ``lease_moot`` and ``cleanup_allowed`` for the same PR, lease session,
+      candidate head and lease marker. Author, reviewer and merger fail closed
+      here even though their profiles carry ``gitea.pr.comment``.
+
     Args:
         pr_number: The PR whose lingering lease to assess/clean.
         apply: When false (default) report only (read-only). When true, post the
-            terminal released marker if — and only if — cleanup is allowed.
+            terminal released marker if — and only if — cleanup is authorized.
+        expected_session_id: Optional lease session the caller expects; a
+            mismatch against the live lease fails closed.
+        expected_candidate_head: Optional leased head the caller expects; a
+            mismatch against the live lease fails closed.
+        expected_lease_comment_id: Optional lease marker id the caller expects;
+            a mismatch against the live lease fails closed.
+        worktree_path: Reconciler worktree to bind the apply-path preflight to.
         remote: Known instance — 'dadeschools' or 'prgs'.
         host: Override the Gitea host.
-        org: Override the owner/organization.
-        repo: Override the repository name.
+        org: Override the owner/organization. Validated against the canonical
+            repository identity; it can never redirect the mutation.
+        repo: Override the repository name. Validated as above.
 
     Returns:
         dict reporting PR merged/closed state, merge_commit_sha, linked-issue
@@ -12546,14 +12747,62 @@ def gitea_cleanup_post_merge_moot_lease(
         "reasons": assessment.get("reasons") or [],
     }
 
+    # The canonical repository identity comes from the session binding only —
+    # never from the org/repo request parameters (#745 requirement 10).
+    repository_slug = _bound_repository_slug(remote)
+    report["repository_slug"] = repository_slug
+    report["required_task"] = post_merge_moot_lease_gate.CLEANUP_TASK
+    report["required_role_kind"] = post_merge_moot_lease_gate.REQUIRED_ROLE
+
     if not apply:
+        # Append-only dry-run evidence. Recorded for every assessment, allowed
+        # or not, so a later apply can prove the lease it saw is the lease that
+        # is still there. Prior entries are never rewritten.
+        evidence = post_merge_moot_lease_gate.record_dry_run(
+            pr_number=pr_number,
+            repository_slug=repository_slug,
+            lease_moot=bool(assessment.get("is_moot")),
+            cleanup_allowed=bool(assessment.get("cleanup_allowed")),
+            session_id=active.get("session_id"),
+            candidate_head=active.get("candidate_head"),
+            lease_comment_id=active.get("comment_id"),
+        )
+        report["dry_run_evidence"] = evidence
         return report
+
+    # ---------------------------------------------------------------- apply --
+    # Preserve the existing dry-run safety assessment: a lease that is not moot
+    # (open PR, already-terminal, absent) is refused here exactly as before, and
+    # no mutation is attempted.
     if not assessment.get("cleanup_allowed"):
         report["cleanup_skipped_reason"] = (
             assessment.get("reasons") or ["cleanup not allowed"]
         )
         return report
 
+    # 1 + 2 + 6: exact resolved cleanup task, reconciler role, and matching
+    # append-only dry-run evidence. Permission alone is deliberately not enough.
+    authorization = post_merge_moot_lease_gate.assess_apply_authorization(
+        pr_number=pr_number,
+        repository_slug=repository_slug,
+        resolved_task=_preflight_resolved_task,
+        active_role_kind=_profile_role_kind(get_profile()),
+        assessment=assessment,
+        evidence=post_merge_moot_lease_gate.latest_dry_run(
+            pr_number=pr_number, repository_slug=repository_slug
+        ),
+        expected_session_id=expected_session_id,
+        expected_candidate_head=expected_candidate_head,
+        expected_lease_comment_id=expected_lease_comment_id,
+    )
+    report["authorization"] = authorization
+    if not authorization["allowed"]:
+        report["success"] = False
+        report["reasons"] = authorization["reasons"]
+        report["blocker_kind"] = authorization["blocker_kind"]
+        return report
+
+    # 3: the dedicated mutation permission.
     comment_block = _profile_operation_gate("gitea.pr.comment")
     if comment_block:
         report["success"] = False
@@ -12561,7 +12810,28 @@ def gitea_cleanup_post_merge_moot_lease(
         report["permission_report"] = _permission_block_report("gitea.pr.comment")
         return report
 
-    verify_preflight_purity(remote)
+    # 4: explicit target must agree with the canonical repository identity
+    # before any preflight or mutation (#733/#739).
+    repo_binding_block = _repository_binding_block(
+        remote, org=org, repo=repo,
+        required_permission="gitea.pr.comment",
+    )
+    if repo_binding_block:
+        report["success"] = False
+        report["reasons"] = repo_binding_block["reasons"]
+        report["blocker_kind"] = repo_binding_block["blocker_kind"]
+        return report
+
+    # 4 (cont.): canonical root, workspace binding and the resolved-task match
+    # are enforced by the shared preflight, with the explicit target forwarded
+    # so the #604 anti-stomp resolution validates the targeted repository.
+    verify_preflight_purity(
+        remote,
+        worktree_path=worktree_path,
+        task=post_merge_moot_lease_gate.CLEANUP_TASK,
+        org=org,
+        repo=repo,
+    )
     body = assessment["release_body"]
     comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
     with _audited(

issue_lock_recovery.py

@@ -362,6 +362,57 @@ def _result(
     }
 
 
+def owning_pr_recovery_evidence(
+    assessment: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    """Server-derived proof of the open PR a sanctioned recovery already owns (#755).
+
+    A dead-session recovery is, by construction, recovery of work that already
+    has an open PR — so the duplicate-work gate's linked-open-PR blocker would
+    otherwise discard every sanctioned recovery. This distils the completed
+    assessment into the minimum evidence that gate needs to tell "the PR this
+    lock already owns" apart from "a competing duplicate PR".
+
+    Returns ``None`` unless recovery was actually granted and the assessment's
+    own evidence names exactly one owning PR whose head agrees with the local
+    and remote heads. Nothing here is caller-supplied: every field is copied
+    from evidence the assessor built out of durable lock state plus live
+    git/Gitea observation, so a caller cannot manufacture an exemption.
+    """
+    if not isinstance(assessment, Mapping):
+        return None
+    if assessment.get("outcome") != RECOVERY_SANCTIONED:
+        return None
+    if not assessment.get("recovery_sanctioned"):
+        return None
+
+    evidence = assessment.get("evidence") or {}
+    branch_name = _text(evidence.get("locked_branch"))
+    pr_head = _text(evidence.get("pr_head"))
+    local_head = _text(evidence.get("local_head"))
+    remote_head = _text(evidence.get("remote_head"))
+    raw_pr_number = evidence.get("pr_number")
+
+    if raw_pr_number is None or not branch_name or not pr_head:
+        return None
+    # The assessor already required these to agree. Re-check, so a truncated or
+    # hand-built evidence map can never authorize an exemption.
+    if pr_head != local_head or pr_head != remote_head:
+        return None
+    try:
+        pr_number = int(raw_pr_number)
+        issue_number = int(evidence.get("issue_number"))
+    except (TypeError, ValueError):
+        return None
+
+    return {
+        "issue_number": issue_number,
+        "pr_number": pr_number,
+        "branch_name": branch_name,
+        "head_sha": pr_head,
+    }
+
+
 def build_recovery_record(
     assessment: Mapping[str, Any],
     *,

issue_work_duplicate_gate.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Mapping
 
 import issue_claim_heartbeat as claim_hb
 
@@ -27,6 +27,116 @@ def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None:
     return claim_hb._linked_open_pr(issue_number, open_prs)
 
 
+def _pr_links_issue(issue_number: int, pr: Mapping[str, Any]) -> bool:
+    """Same linkage rule ``claim_hb._linked_open_pr`` applies, per PR.
+
+    ``_linked_open_pr`` only yields the *first* match, which cannot answer
+    "is there exactly one linked PR?" — a question the owning-PR exemption
+    below must answer before it can trust any of them.
+    """
+    pattern = _issue_pattern(issue_number)
+    head = (pr.get("head") or {}).get("ref") or ""
+    text = f"{pr.get('title', '')} {pr.get('body', '')}".lower()
+    if pattern in head.lower():
+        return True
+    return (
+        f"closes #{int(issue_number)}" in text
+        or f"fixes #{int(issue_number)}" in text
+    )
+
+
+def _all_linked_open_prs(
+    issue_number: int, open_prs: list[dict]
+) -> list[Mapping[str, Any]]:
+    return [pr for pr in (open_prs or []) if _pr_links_issue(issue_number, pr)]
+
+
+def _assess_owning_pr_exemption(
+    issue_number: int,
+    *,
+    linked_open_prs: list[Mapping[str, Any]],
+    locked_branch: str | None,
+    recovered_owning_pr: Mapping[str, Any] | None,
+) -> tuple[bool, list[str]]:
+    """Is the linked open PR provably the one a sanctioned recovery owns (#755)?
+
+    ``recovered_owning_pr`` is produced by
+    ``issue_lock_recovery.owning_pr_recovery_evidence`` from a completed
+    server-side recovery assessment — it is never a caller-supplied field.
+    Every element is re-checked here against the live PR list this gate was
+    given, so a stale or partial token cannot widen the exemption.
+
+    Returns ``(exempt, diagnostic_reasons)``. Diagnostics are only emitted when
+    a token was offered and rejected, so a blocked caller can see which element
+    of ownership disagreed.
+    """
+    if not recovered_owning_pr:
+        return False, []
+
+    notes: list[str] = []
+    token_issue = recovered_owning_pr.get("issue_number")
+    token_pr = recovered_owning_pr.get("pr_number")
+    token_branch = str(recovered_owning_pr.get("branch_name") or "").strip()
+    token_head = str(recovered_owning_pr.get("head_sha") or "").strip()
+    locked = (locked_branch or "").strip()
+
+    if token_issue is not None and int(token_issue) != int(issue_number):
+        notes.append(
+            f"recovery evidence is for issue #{token_issue}, not "
+            f"#{issue_number} (no owning-PR exemption)"
+        )
+        return False, notes
+    if not locked or not token_branch or locked != token_branch:
+        notes.append(
+            f"recovery evidence branch '{token_branch or 'unknown'}' does not "
+            f"match the branch being locked '{locked or 'unknown'}' "
+            "(no owning-PR exemption)"
+        )
+        return False, notes
+    if len(linked_open_prs) != 1:
+        numbers = ", ".join(
+            f"#{pr.get('number')}" for pr in linked_open_prs
+        ) or "none"
+        notes.append(
+            f"{len(linked_open_prs)} open PRs link issue #{issue_number} "
+            f"({numbers}); recovery may only own exactly one "
+            "(no owning-PR exemption)"
+        )
+        return False, notes
+
+    only = linked_open_prs[0]
+    head_obj = only.get("head") or {}
+    only_number = only.get("number")
+    only_ref = str(head_obj.get("ref") or "").strip()
+    only_sha = str(head_obj.get("sha") or "").strip()
+
+    if token_pr is None or only_number is None or int(only_number) != int(token_pr):
+        notes.append(
+            f"linked open PR #{only_number} is not the recovered owning PR "
+            f"#{token_pr} (no owning-PR exemption)"
+        )
+        return False, notes
+    if only_ref != token_branch:
+        notes.append(
+            f"open PR #{only_number} head branch '{only_ref}' does not match "
+            f"the recovered branch '{token_branch}' (no owning-PR exemption)"
+        )
+        return False, notes
+    if not token_head or not only_sha or only_sha != token_head:
+        notes.append(
+            f"open PR #{only_number} head {only_sha or 'unknown'} does not "
+            f"match the recovered head {token_head or 'unknown'} "
+            "(no owning-PR exemption)"
+        )
+        return False, notes
+
+    return True, [
+        f"open PR #{only_number} is the exact PR already owned by the "
+        f"recovering lock for issue #{issue_number} (branch '{token_branch}', "
+        f"head {token_head}); not duplicate work"
+    ]
+
+
 def _matching_branches(
     issue_number: int,
     branch_names: list[str],
@@ -52,8 +162,15 @@ def assess_work_issue_duplicate_gate(
     claim_entry: dict | None = None,
     locked_branch: str | None = None,
     phase: str = PHASE_LOCK,
+    recovered_owning_pr: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
-    """Fail closed when duplicate work is already in flight for an issue."""
+    """Fail closed when duplicate work is already in flight for an issue.
+
+    ``recovered_owning_pr`` (#755) is server-derived evidence that a sanctioned
+    dead-session lock recovery already owns one specific open PR. It exempts
+    *only* that exact PR from the linked-open-PR blocker; every other duplicate
+    signal, and every mismatch, keeps failing closed.
+    """
     reasons: list[str] = []
     outcome = OUTCOME_DUPLICATE_WORK_NOT_PREVENTED
     prs = list(open_prs or [])
@@ -61,12 +178,23 @@ def assess_work_issue_duplicate_gate(
     pattern = _issue_pattern(issue_number)
 
     linked = _linked_open_pr(issue_number, prs)
+    linked_open_prs = _all_linked_open_prs(issue_number, prs)
+    owning_pr_exempted = False
+    exemption_notes: list[str] = []
     if linked:
-        reasons.append(
-            f"open PR #{linked.get('number')} already covers issue "
-            f"#{issue_number} (fail closed)"
+        owning_pr_exempted, exemption_notes = _assess_owning_pr_exemption(
+            issue_number,
+            linked_open_prs=linked_open_prs,
+            locked_branch=locked_branch,
+            recovered_owning_pr=recovered_owning_pr,
         )
-        outcome = OUTCOME_DUPLICATE_PR_PREVENTED
+        if not owning_pr_exempted:
+            reasons.append(
+                f"open PR #{linked.get('number')} already covers issue "
+                f"#{issue_number} (fail closed)"
+            )
+            reasons.extend(exemption_notes)
+            outcome = OUTCOME_DUPLICATE_PR_PREVENTED
 
     conflicting_branches = _matching_branches(
         issue_number, branches, locked_branch=locked_branch
@@ -122,6 +250,9 @@ def assess_work_issue_duplicate_gate(
         "phase": phase,
         "outcome": outcome,
         "linked_open_pr": linked.get("number") if linked else entry.get("linked_open_pr"),
+        "linked_open_pr_count": len(linked_open_prs),
+        "owning_pr_recovery_exempted": owning_pr_exempted,
+        "owning_pr_recovery_notes": list(exemption_notes),
         "conflicting_branches": conflicting_branches,
         "claim_status": status or None,
         "reasons": reasons,

post_merge_moot_lease_gate.py

@@ -0,0 +1,279 @@
+"""Reconciler authorization gate for post-merge moot-lease cleanup (#745).
+
+``gitea_cleanup_post_merge_moot_lease`` (#515) posts a terminal ``phase:
+released`` lease marker — a real, durable mutation of the PR lease ledger.
+Before #745 it was gated on permissions alone (``gitea.read`` to enter,
+``gitea.pr.comment`` to apply) with no canonical task and no role binding, so
+any profile carrying ``gitea.pr.comment`` reached the mutation path while the
+reconciler could not satisfy the operator-required resolve-exact-task ->
+mutation sequence.
+
+This module holds the pure half of that gate:
+
+* the canonical task name and its tool-name alias;
+* an **append-only** in-process ledger of read-only dry-run assessments;
+* ``assess_apply_authorization``, which decides whether an apply may proceed.
+
+Apply is authorized only when all of the following hold:
+
+* the session resolved exactly the cleanup task (no other task substitutes);
+* the active profile role is ``reconciler``;
+* a prior dry run in this session recorded ``lease_moot`` and
+  ``cleanup_allowed`` for the *same* repository, PR, lease session, candidate
+  head and lease marker id;
+* the live assessment still agrees with that evidence, so a lease superseded
+  between the dry run and the apply fails closed;
+* any caller-supplied expectations match the live lease exactly.
+
+Everything else fails closed. The ledger is only ever appended to — a
+superseded dry run stays visible as history instead of being rewritten — which
+keeps the cleanup audit trail append-only end to end.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+CLEANUP_TASK = "cleanup_post_merge_moot_lease"
+CLEANUP_TOOL_ALIAS = "gitea_cleanup_post_merge_moot_lease"
+REQUIRED_ROLE = "reconciler"
+REQUIRED_PERMISSION = "gitea.pr.comment"
+
+# The read-only assessment stays reachable under gitea.read for every role —
+# the convention shared with cleanup_stale_review_decision_lock and
+# cleanup_obsolete_reviewer_comment_lease — so any namespace can diagnose a
+# stuck lease. Only the apply path demands CLEANUP_TASK + REQUIRED_ROLE.
+ASSESSMENT_PERMISSION = "gitea.read"
+
+_DRY_RUN_LEDGER: list[dict[str, Any]] = []
+
+
+def _norm(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def _norm_comment_id(value: Any) -> int | None:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def record_dry_run(
+    *,
+    pr_number: int,
+    repository_slug: str | None,
+    lease_moot: bool,
+    cleanup_allowed: bool,
+    session_id: str | None,
+    candidate_head: str | None,
+    lease_comment_id: Any,
+    recorded_at: datetime | None = None,
+) -> dict[str, Any]:
+    """Append one read-only assessment to the dry-run ledger.
+
+    Never rewrites or removes a prior entry: repeated dry runs accumulate and
+    ``latest_dry_run`` returns the newest matching one.
+    """
+    entry = {
+        "task": CLEANUP_TASK,
+        "pr_number": int(pr_number),
+        "repository_slug": _norm(repository_slug) or None,
+        "lease_moot": bool(lease_moot),
+        "cleanup_allowed": bool(cleanup_allowed),
+        "session_id": _norm(session_id) or None,
+        "candidate_head": _norm(candidate_head) or None,
+        "lease_comment_id": _norm_comment_id(lease_comment_id),
+        "recorded_at": (recorded_at or datetime.now(timezone.utc)).isoformat(),
+    }
+    _DRY_RUN_LEDGER.append(entry)
+    return dict(entry)
+
+
+def dry_run_history() -> tuple[dict[str, Any], ...]:
+    """Immutable view of every recorded dry run, oldest first."""
+    return tuple(dict(entry) for entry in _DRY_RUN_LEDGER)
+
+
+def latest_dry_run(
+    *, pr_number: int, repository_slug: str | None
+) -> dict[str, Any] | None:
+    """Newest dry-run evidence for this repository + PR, or None."""
+    wanted_repo = _norm(repository_slug)
+    for entry in reversed(_DRY_RUN_LEDGER):
+        if entry["pr_number"] != int(pr_number):
+            continue
+        if _norm(entry.get("repository_slug")) != wanted_repo:
+            continue
+        return dict(entry)
+    return None
+
+
+def _reset_for_testing() -> None:
+    """Drop ledger state between tests. Never called by production paths."""
+    _DRY_RUN_LEDGER.clear()
+
+
+def assess_apply_authorization(
+    *,
+    pr_number: int,
+    repository_slug: str | None,
+    resolved_task: str | None,
+    active_role_kind: str | None,
+    assessment: dict[str, Any],
+    evidence: dict[str, Any] | None,
+    expected_session_id: str | None = None,
+    expected_candidate_head: str | None = None,
+    expected_lease_comment_id: Any = None,
+) -> dict[str, Any]:
+    """Decide whether a moot-lease cleanup apply is authorized (fail closed).
+
+    Returns ``{"allowed", "reasons", "blocker_kind", "evidence_matched", ...}``.
+    ``allowed`` is True only when every check passes; each failure contributes a
+    reason so the caller can report all of them together.
+    """
+    reasons: list[str] = []
+    blocker_kind: str | None = None
+
+    def _block(kind: str, reason: str) -> None:
+        nonlocal blocker_kind
+        reasons.append(reason)
+        if blocker_kind is None:
+            blocker_kind = kind
+
+    # 1. Exact resolved cleanup task. Resolving any other task — including a
+    #    sibling reconciler task — does not authorize this mutation.
+    if _norm(resolved_task) != CLEANUP_TASK:
+        _block(
+            "unresolved_cleanup_task",
+            "post-merge moot-lease cleanup requires the session to resolve "
+            f"task '{CLEANUP_TASK}' immediately before apply; resolved task is "
+            f"{resolved_task!r} (fail closed)",
+        )
+
+    # 2. Dedicated reconciler role, enforced independently of the permission.
+    if _norm(active_role_kind) != REQUIRED_ROLE:
+        _block(
+            "wrong_role",
+            f"profile role {active_role_kind!r} cannot apply post-merge "
+            f"moot-lease cleanup; required role is {REQUIRED_ROLE} even when "
+            f"{REQUIRED_PERMISSION} is present (fail closed)",
+        )
+
+    # 3. Canonical repository identity must be established, never inferred from
+    #    request parameters.
+    if not _norm(repository_slug):
+        _block(
+            "repository_binding",
+            "no canonical repository identity could be established for the "
+            "cleanup target (fail closed)",
+        )
+
+    # 4. The live safety assessment must still say the lease is moot/cleanable.
+    if not assessment.get("is_moot") or not assessment.get("cleanup_allowed"):
+        _block(
+            "lease_not_moot",
+            "live assessment does not report a moot, cleanable lease on PR "
+            f"#{pr_number} (lease_moot={bool(assessment.get('is_moot'))}, "
+            f"cleanup_allowed={bool(assessment.get('cleanup_allowed'))}) "
+            "(fail closed)",
+        )
+
+    live = assessment.get("active_lease") or {}
+    live_session = _norm(live.get("session_id"))
+    live_head = _norm(live.get("candidate_head"))
+    live_comment_id = _norm_comment_id(live.get("comment_id"))
+
+    # 5. A lease missing identifying fields is malformed and unsafe to act on.
+    if not live_session or not live_head or live_comment_id is None:
+        _block(
+            "malformed_lease",
+            "active lease is malformed: session_id / candidate_head / "
+            "comment_id must all be present to authorize cleanup "
+            f"(session_id={live.get('session_id')!r}, "
+            f"candidate_head={live.get('candidate_head')!r}, "
+            f"comment_id={live.get('comment_id')!r}) (fail closed)",
+        )
+
+    # 6. Caller expectations, when supplied, must match the live lease exactly.
+    if expected_session_id is not None and _norm(expected_session_id) != live_session:
+        _block(
+            "lease_mismatch",
+            f"expected lease session {expected_session_id!r} does not match the "
+            f"live lease session {live.get('session_id')!r} (fail closed)",
+        )
+    if (
+        expected_candidate_head is not None
+        and _norm(expected_candidate_head) != live_head
+    ):
+        _block(
+            "lease_mismatch",
+            f"expected candidate head {expected_candidate_head!r} does not "
+            f"match the live lease head {live.get('candidate_head')!r} "
+            "(fail closed)",
+        )
+    if expected_lease_comment_id is not None and (
+        _norm_comment_id(expected_lease_comment_id) != live_comment_id
+    ):
+        _block(
+            "lease_mismatch",
+            f"expected lease marker {expected_lease_comment_id!r} does not "
+            f"match the live lease marker {live.get('comment_id')!r} "
+            "(fail closed)",
+        )
+
+    # 7. Matching dry-run evidence recorded earlier in this session.
+    evidence_matched = False
+    if evidence is None:
+        _block(
+            "missing_dry_run_evidence",
+            "no read-only dry run recorded for this repository and PR; run the "
+            "tool with apply=false and confirm lease_moot / cleanup_allowed "
+            "before applying (fail closed)",
+        )
+    elif not evidence.get("lease_moot") or not evidence.get("cleanup_allowed"):
+        _block(
+            "dry_run_not_allowed",
+            "recorded dry run did not report an allowed cleanup "
+            f"(lease_moot={bool(evidence.get('lease_moot'))}, "
+            f"cleanup_allowed={bool(evidence.get('cleanup_allowed'))}) "
+            "(fail closed)",
+        )
+    elif int(evidence.get("pr_number") or -1) != int(pr_number) or _norm(
+        evidence.get("repository_slug")
+    ) != _norm(repository_slug):
+        _block(
+            "dry_run_mismatch",
+            "recorded dry run targets a different repository or PR "
+            f"({evidence.get('repository_slug')}#{evidence.get('pr_number')} vs "
+            f"{repository_slug}#{pr_number}) (fail closed)",
+        )
+    elif (
+        _norm(evidence.get("session_id")) != live_session
+        or _norm(evidence.get("candidate_head")) != live_head
+        or _norm_comment_id(evidence.get("lease_comment_id")) != live_comment_id
+    ):
+        _block(
+            "superseded_lease",
+            "the lease changed after the recorded dry run (dry run: "
+            f"session={evidence.get('session_id')!r}, "
+            f"head={evidence.get('candidate_head')!r}, "
+            f"marker={evidence.get('lease_comment_id')!r}; live: "
+            f"session={live.get('session_id')!r}, "
+            f"head={live.get('candidate_head')!r}, "
+            f"marker={live.get('comment_id')!r}); re-run the dry run "
+            "(fail closed)",
+        )
+    else:
+        evidence_matched = True
+
+    return {
+        "allowed": not reasons,
+        "reasons": reasons,
+        "blocker_kind": blocker_kind,
+        "evidence_matched": evidence_matched,
+        "required_task": CLEANUP_TASK,
+        "required_role_kind": REQUIRED_ROLE,
+        "required_permission": REQUIRED_PERMISSION,
+    }

role_session_router.py

@@ -87,6 +87,11 @@ RECONCILER_TASKS = frozenset({
     # only to the reconciler profile). Raw gitea_delete_branch redirects here to
     # the guarded gitea_cleanup_merged_pr_branch path (#514/#687).
     "delete_branch",
+    # #745: post-merge moot reviewer-lease cleanup is reconciler-owned; the
+    # apply path posts a terminal lease marker. Kept in step with
+    # task_capability_map so map and router cannot disagree (#723 defect A).
+    "cleanup_post_merge_moot_lease",
+    "gitea_cleanup_post_merge_moot_lease",
     "reconcile_already_landed_pr",
     "reconcile_already_landed",
     "reconcile-landed-pr",
@@ -118,6 +123,9 @@ TASK_REQUIRED_ROLE = {
     "reconcile_already_landed": "reconciler",
     "reconcile-landed-pr": "reconciler",
     "cleanup_merged_pr_branch": "reconciler",
+    # #745: post-merge moot reviewer-lease cleanup (canonical task + tool alias).
+    "cleanup_post_merge_moot_lease": "reconciler",
+    "gitea_cleanup_post_merge_moot_lease": "reconciler",
     # #309: reconciler tasks close already-landed PRs/issues only.
     "reconcile_close_landed_pr": "reconciler",
     "reconcile_close_landed_issue": "reconciler",

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:

task_capability_map.py

@@ -158,6 +158,23 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.comment",
         "role": "reviewer",
     },
+    # #745: post-merge moot reviewer-lease cleanup is reconciler-owned. The
+    # apply path posts an append-only terminal `phase: released` lease marker
+    # (gitea.pr.comment), so holding the comment permission alone must not
+    # authorize it — author, reviewer and merger fail closed on the role gate
+    # even though their profiles carry gitea.pr.comment. The read-only
+    # `apply=false` assessment deliberately stays reachable under gitea.read
+    # inside the tool (the same convention as cleanup_stale_review_decision_lock
+    # below), so any namespace can diagnose a stuck lease; only apply requires
+    # this task plus the reconciler role.
+    "cleanup_post_merge_moot_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "reconciler",
+    },
+    "gitea_cleanup_post_merge_moot_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "reconciler",
+    },
     "blind_pr_queue_review": {
         "permission": "gitea.pr.review",
         "role": "reviewer",

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_745_moot_lease_reconciler_gate.py

@@ -0,0 +1,658 @@
+"""Reconciler role binding for post-merge moot-lease cleanup (#745).
+
+``gitea_cleanup_post_merge_moot_lease`` (#515) posts a terminal ``phase:
+released`` lease marker but had no capability-map entry and no role gate: entry
+required only ``gitea.read``, apply required only ``gitea.pr.comment``, so any
+profile holding the comment permission reached the mutation while the
+reconciler could not satisfy resolve-exact-task -> mutation.
+
+These tests pin the fixed contract:
+
+- the canonical task and its tool-name alias resolve identically
+  (``gitea.pr.comment`` + ``reconciler``);
+- only a reconciler profile satisfies permission AND role;
+- ``apply=false`` assessment stays reachable under ``gitea.read`` for any role
+  and mutates nothing (documented, deliberate divergence from the apply path);
+- ``apply=true`` requires the exact resolved task, the reconciler role, the
+  comment permission, a validated repository binding and matching dry-run
+  evidence;
+- live/non-moot/superseded/mismatched/malformed leases fail closed;
+- the dry-run ledger is append-only and cleanup stays idempotent.
+
+Every fixture is synthetic. No production PR, lease session or marker is used
+anywhere in this module (see ``TestNoProductionLeaseTouched``).
+"""
+
+import sys as _sys
+from pathlib import Path as _Path
+
+_sys.path.insert(0, str(_Path(__file__).resolve().parent.parent))
+
+import os  # noqa: E402
+import unittest  # noqa: E402
+from datetime import datetime, timezone  # noqa: E402
+from unittest.mock import patch  # noqa: E402
+
+import mcp_server  # noqa: E402
+import post_merge_moot_lease_gate as gate  # noqa: E402
+import reviewer_pr_lease as leases  # noqa: E402
+from mcp_server import gitea_cleanup_post_merge_moot_lease  # noqa: E402
+from role_session_router import RECONCILER_TASKS, TASK_REQUIRED_ROLE  # noqa: E402
+from task_capability_map import required_permission, required_role  # noqa: E402
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
+PR = 487
+ISSUE = 485
+SESSION = "97274-676d20a825c4"
+HEAD_A = "a" * 40
+HEAD_B = "d" * 40
+LEASE_COMMENT_ID = 6603
+
+CANONICAL_TASK = "cleanup_post_merge_moot_lease"
+TOOL_ALIAS = "gitea_cleanup_post_merge_moot_lease"
+
+_BASE_OPS = "gitea.read,gitea.pr.comment"
+RECONCILER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-reconciler",
+    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.close,gitea.branch.delete",
+}
+AUTHOR_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-author",
+    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.create,gitea.issue.create",
+}
+REVIEWER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-reviewer",
+    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.review,gitea.pr.approve",
+}
+MERGER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-merger",
+    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.merge",
+}
+
+# Permission shape of the configured role profiles (mirrors
+# tests/test_task_capability_role_invariants.py CANONICAL_ROLE_PROFILES).
+ROLE_PROFILE_PERMISSIONS = {
+    "author": {"gitea.read", "gitea.pr.comment", "gitea.pr.create",
+               "gitea.issue.create", "gitea.issue.comment", "gitea.issue.close",
+               "gitea.branch.create", "gitea.branch.push", "gitea.repo.commit"},
+    "reviewer": {"gitea.read", "gitea.pr.comment", "gitea.pr.review",
+                 "gitea.pr.approve", "gitea.pr.request_changes",
+                 "gitea.issue.comment"},
+    "merger": {"gitea.read", "gitea.pr.comment", "gitea.pr.merge",
+               "gitea.issue.comment"},
+    "reconciler": {"gitea.read", "gitea.pr.comment", "gitea.pr.close",
+                   "gitea.issue.close", "gitea.branch.delete"},
+}
+
+
+def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed",
+                   candidate_head=HEAD_A, comment_id=LEASE_COMMENT_ID):
+    body = leases.format_lease_body(
+        repo=SLUG,
+        pr_number=pr_number,
+        issue_number=ISSUE,
+        reviewer_identity="sysadmin",
+        profile="prgs-reviewer",
+        session_id=session_id,
+        worktree="branches/review-pr487",
+        phase=phase,
+        candidate_head=candidate_head,
+        target_branch="master",
+        target_branch_sha="b" * 40,
+        last_activity=datetime.now(timezone.utc),
+    )
+    return {"id": comment_id, "body": body, "user": {"login": "sysadmin"}}
+
+
+def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999):
+    """api_request side effect keyed on method + url; records POSTs."""
+    calls = {"post": []}
+
+    def _side(method, url, auth=None, payload=None, *a, **k):
+        if (method or "").upper() == "POST":
+            calls["post"].append({"url": url, "payload": payload})
+            return {"id": posted_id}
+        if "/comments" in url:
+            return list(comments)
+        if "/pulls/" in url:
+            pr = {"state": pr_state, "number": PR, "merge_commit_sha": "c" * 40}
+            if pr_merged:
+                pr["merged"] = True
+                pr["merged_at"] = "2026-07-08T07:46:04Z"
+            return pr
+        if "/issues/" in url:
+            return {"state": "closed" if pr_merged else "open", "number": ISSUE}
+        return {}
+
+    return _side, calls
+
+
+def _assessment(comments, *, pr_merged=True, pr_state="closed"):
+    return leases.assess_post_merge_moot_lease(
+        comments, pr_number=PR, pr_merged=pr_merged, pr_state=pr_state,
+        merge_commit_sha="c" * 40,
+    )
+
+
+# --------------------------------------------------------------------------- #
+# 1-5. Capability map / router contract
+# --------------------------------------------------------------------------- #
+class TestCleanupTaskContract(unittest.TestCase):
+    def test_canonical_task_is_reconciler_owned(self):
+        """1. The reconciler is the role that can resolve the cleanup task."""
+        self.assertEqual(required_permission(CANONICAL_TASK), "gitea.pr.comment")
+        self.assertEqual(required_role(CANONICAL_TASK), "reconciler")
+
+    def test_tool_alias_resolves_to_identical_contract(self):
+        """5. Alias and canonical task must not diverge."""
+        self.assertEqual(
+            (required_permission(TOOL_ALIAS), required_role(TOOL_ALIAS)),
+            (required_permission(CANONICAL_TASK), required_role(CANONICAL_TASK)),
+        )
+
+    def test_author_reviewer_merger_cannot_resolve_the_task(self):
+        """2-4. No non-reconciler role satisfies permission AND role."""
+        for role in ("author", "reviewer", "merger"):
+            with self.subTest(role=role):
+                self.assertNotEqual(required_role(CANONICAL_TASK), role)
+                # They hold the permission — which is exactly why the role gate
+                # is required rather than optional.
+                self.assertIn(
+                    "gitea.pr.comment", ROLE_PROFILE_PERMISSIONS[role],
+                    "test premise: non-reconciler roles do hold pr.comment",
+                )
+
+    def test_reconciler_profile_satisfies_permission_and_role(self):
+        self.assertIn(
+            required_permission(CANONICAL_TASK),
+            ROLE_PROFILE_PERMISSIONS[required_role(CANONICAL_TASK)],
+        )
+
+    def test_router_agrees_with_capability_map(self):
+        for task in (CANONICAL_TASK, TOOL_ALIAS):
+            with self.subTest(task=task):
+                self.assertIn(task, RECONCILER_TASKS)
+                self.assertEqual(TASK_REQUIRED_ROLE[task], required_role(task))
+
+    def test_unknown_alias_still_rejected(self):
+        for bogus in ("cleanup_post_merge_moot_leases", "cleanup_moot_lease", ""):
+            with self.subTest(task=bogus):
+                with self.assertRaises(KeyError):
+                    required_role(bogus)
+                with self.assertRaises(KeyError):
+                    required_permission(bogus)
+
+
+# --------------------------------------------------------------------------- #
+# Authorization gate unit tests
+# --------------------------------------------------------------------------- #
+class TestApplyAuthorizationGate(unittest.TestCase):
+    def setUp(self):
+        gate._reset_for_testing()
+        self.addCleanup(gate._reset_for_testing)
+        self.assessment = _assessment([_lease_comment()])
+
+    def _evidence(self, **over):
+        base = dict(
+            pr_number=PR, repository_slug=SLUG, lease_moot=True,
+            cleanup_allowed=True, session_id=SESSION, candidate_head=HEAD_A,
+            lease_comment_id=LEASE_COMMENT_ID,
+        )
+        base.update(over)
+        return gate.record_dry_run(**base)
+
+    def _assess(self, **over):
+        kwargs = dict(
+            pr_number=PR, repository_slug=SLUG, resolved_task=CANONICAL_TASK,
+            active_role_kind="reconciler", assessment=self.assessment,
+            evidence=self._evidence(),
+        )
+        kwargs.update(over)
+        return gate.assess_apply_authorization(**kwargs)
+
+    def test_reconciler_with_matching_evidence_is_authorized(self):
+        result = self._assess()
+        self.assertTrue(result["allowed"], result["reasons"])
+        self.assertTrue(result["evidence_matched"])
+
+    def test_apply_without_exact_task_resolution_fails(self):
+        """8. Apply without exact task resolution fails preflight."""
+        result = self._assess(resolved_task=None)
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task")
+
+    def test_resolving_another_task_does_not_authorize_cleanup(self):
+        """9. A sibling reconciler task is not a substitute."""
+        for other in ("delete_branch", "reconcile_already_landed_pr",
+                      "cleanup_merged_pr_branch", "comment_pr"):
+            with self.subTest(task=other):
+                result = self._assess(resolved_task=other)
+                self.assertFalse(result["allowed"])
+                self.assertEqual(
+                    result["blocker_kind"], "unresolved_cleanup_task")
+
+    def test_non_reconciler_roles_fail_closed(self):
+        """10. Author/reviewer/merger cannot apply despite pr.comment."""
+        for role in ("author", "reviewer", "merger", None, ""):
+            with self.subTest(role=role):
+                result = self._assess(active_role_kind=role)
+                self.assertFalse(result["allowed"])
+                self.assertEqual(result["blocker_kind"], "wrong_role")
+
+    def test_missing_repository_identity_fails_closed(self):
+        result = self._assess(repository_slug=None)
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["blocker_kind"], "repository_binding")
+
+    def test_missing_dry_run_evidence_fails_closed(self):
+        result = self._assess(evidence=None)
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["blocker_kind"], "missing_dry_run_evidence")
+
+    def test_dry_run_that_disallowed_cleanup_fails_closed(self):
+        result = self._assess(evidence=self._evidence(cleanup_allowed=False))
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["blocker_kind"], "dry_run_not_allowed")
+
+    def test_dry_run_for_another_pr_or_repo_fails_closed(self):
+        """11. Wrong PR / repository fails closed."""
+        for over in ({"pr_number": PR + 1}, {"repository_slug": "Other/Repo"}):
+            with self.subTest(**over):
+                result = self._assess(evidence=self._evidence(**over))
+                self.assertFalse(result["allowed"])
+                self.assertEqual(result["blocker_kind"], "dry_run_mismatch")
+
+    def test_superseded_lease_fails_closed(self):
+        """12. Head/session/marker drift since the dry run fails closed."""
+        for over in ({"session_id": "other-session"},
+                     {"candidate_head": HEAD_B},
+                     {"lease_comment_id": 7777}):
+            with self.subTest(**over):
+                result = self._assess(evidence=self._evidence(**over))
+                self.assertFalse(result["allowed"])
+                self.assertEqual(result["blocker_kind"], "superseded_lease")
+
+    def test_expectation_mismatch_fails_closed(self):
+        """11. Wrong session / head / marker expectations fail closed."""
+        for over in ({"expected_session_id": "nope"},
+                     {"expected_candidate_head": HEAD_B},
+                     {"expected_lease_comment_id": 7777}):
+            with self.subTest(**over):
+                result = self._assess(**over)
+                self.assertFalse(result["allowed"])
+                self.assertEqual(result["blocker_kind"], "lease_mismatch")
+
+    def test_matching_expectations_are_authorized(self):
+        result = self._assess(
+            expected_session_id=SESSION,
+            expected_candidate_head=HEAD_A,
+            expected_lease_comment_id=LEASE_COMMENT_ID,
+        )
+        self.assertTrue(result["allowed"], result["reasons"])
+
+    def test_non_moot_lease_fails_closed(self):
+        """12. A live lease on an open PR is never cleanable."""
+        open_pr = _assessment(
+            [_lease_comment()], pr_merged=False, pr_state="open")
+        result = self._assess(assessment=open_pr)
+        self.assertFalse(result["allowed"])
+        self.assertIn(
+            result["blocker_kind"], ("lease_not_moot", "malformed_lease"))
+
+    def test_malformed_lease_fails_closed(self):
+        malformed = dict(self.assessment)
+        malformed["active_lease"] = {
+            "session_id": "", "candidate_head": None, "comment_id": None}
+        result = self._assess(assessment=malformed)
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["blocker_kind"], "malformed_lease")
+
+    def test_ledger_is_append_only(self):
+        """14. Recording never rewrites or drops prior entries."""
+        first = self._evidence()
+        second = self._evidence(candidate_head=HEAD_B, lease_comment_id=7777)
+        history = gate.dry_run_history()
+        self.assertEqual(len(history), 2)
+        self.assertEqual(history[0]["candidate_head"], first["candidate_head"])
+        self.assertEqual(history[1]["candidate_head"], HEAD_B)
+        # Newest-wins for lookup, but the older entry survives in history.
+        latest = gate.latest_dry_run(pr_number=PR, repository_slug=SLUG)
+        self.assertEqual(latest["lease_comment_id"], second["lease_comment_id"])
+        self.assertEqual(gate.dry_run_history()[0]["lease_comment_id"],
+                         LEASE_COMMENT_ID)
+
+    def test_history_view_cannot_mutate_the_ledger(self):
+        self._evidence()
+        snapshot = gate.dry_run_history()
+        snapshot[0]["pr_number"] = 999999
+        self.assertEqual(
+            gate.dry_run_history()[0]["pr_number"], PR,
+            "dry_run_history must hand out copies, not live rows")
+
+
+# --------------------------------------------------------------------------- #
+# Tool-level behavior
+# --------------------------------------------------------------------------- #
+class _ToolCase(unittest.TestCase):
+    def setUp(self):
+        leases.clear_session_lease()
+        gate._reset_for_testing()
+        self.addCleanup(gate._reset_for_testing)
+
+    def _run(self, env, *, apply, comments, pr_state="closed", pr_merged=True,
+             resolved_task=CANONICAL_TASK, slug=SLUG, **kwargs):
+        side, calls = _api_side_effect(
+            pr_state=pr_state, pr_merged=pr_merged, comments=comments)
+        with patch("mcp_server.api_request", side_effect=side), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", return_value=None), \
+                patch("mcp_server._bound_repository_slug", return_value=slug), \
+                patch("mcp_server._repository_binding_block", return_value=None), \
+                patch.object(mcp_server, "_preflight_resolved_task",
+                             resolved_task), \
+                patch.dict(os.environ, env, clear=True):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=apply, remote="prgs", **kwargs)
+        return result, calls
+
+
+class TestDryRunOpenToEveryRole(_ToolCase):
+    """7. Dry run performs no mutation and stays under the read capability."""
+
+    def test_dry_run_reports_moot_and_mutates_nothing_for_every_role(self):
+        for name, env in (("reconciler", RECONCILER_ENV), ("author", AUTHOR_ENV),
+                          ("reviewer", REVIEWER_ENV), ("merger", MERGER_ENV)):
+            with self.subTest(role=name):
+                gate._reset_for_testing()
+                result, calls = self._run(
+                    env, apply=False, comments=[_lease_comment()],
+                    resolved_task=None)
+                self.assertTrue(result["success"])
+                self.assertTrue(result["lease_moot"])
+                self.assertTrue(result["cleanup_allowed"])
+                self.assertFalse(result["cleanup_performed"])
+                self.assertEqual(result["mode"], "read_only")
+                self.assertEqual(calls["post"], [], "dry run must not mutate")
+
+    def test_dry_run_records_evidence(self):
+        result, _ = self._run(
+            RECONCILER_ENV, apply=False, comments=[_lease_comment()])
+        evidence = result["dry_run_evidence"]
+        self.assertEqual(evidence["pr_number"], PR)
+        self.assertEqual(evidence["repository_slug"], SLUG)
+        self.assertEqual(evidence["session_id"], SESSION)
+        self.assertEqual(evidence["candidate_head"], HEAD_A)
+        self.assertEqual(evidence["lease_comment_id"], LEASE_COMMENT_ID)
+        self.assertTrue(evidence["lease_moot"])
+        self.assertTrue(evidence["cleanup_allowed"])
+
+
+class TestApplyRequiresReconciler(_ToolCase):
+    def test_reconciler_apply_succeeds_after_matching_dry_run(self):
+        """7 (apply). Allowed dry run then apply posts exactly one marker."""
+        comments = [_lease_comment()]
+        dry, dry_calls = self._run(
+            RECONCILER_ENV, apply=False, comments=comments)
+        self.assertTrue(dry["cleanup_allowed"])
+        self.assertEqual(dry_calls["post"], [])
+
+        result, calls = self._run(
+            RECONCILER_ENV, apply=True, comments=comments)
+        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertTrue(result["cleanup_performed"])
+        self.assertEqual(result["released_comment_id"], 9999)
+        self.assertEqual(len(calls["post"]), 1)
+        body = calls["post"][0]["payload"]["body"]
+        self.assertIn("phase: released", body)
+        self.assertIn("post-merge-moot", body)
+
+    def test_apply_without_dry_run_fails_closed(self):
+        """6. Apply must be preceded by a matching dry run."""
+        result, calls = self._run(
+            RECONCILER_ENV, apply=True, comments=[_lease_comment()])
+        self.assertFalse(result["success"])
+        self.assertFalse(result["cleanup_performed"])
+        self.assertEqual(result["blocker_kind"], "missing_dry_run_evidence")
+        self.assertEqual(calls["post"], [])
+
+    def test_apply_without_exact_task_resolution_fails_closed(self):
+        """8. No resolved cleanup task -> no mutation."""
+        comments = [_lease_comment()]
+        self._run(RECONCILER_ENV, apply=False, comments=comments)
+        result, calls = self._run(
+            RECONCILER_ENV, apply=True, comments=comments, resolved_task=None)
+        self.assertFalse(result["success"])
+        self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task")
+        self.assertEqual(calls["post"], [])
+
+    def test_resolving_a_different_task_does_not_authorize_apply(self):
+        """9. Another resolved task is not a substitute."""
+        comments = [_lease_comment()]
+        self._run(RECONCILER_ENV, apply=False, comments=comments)
+        result, calls = self._run(
+            RECONCILER_ENV, apply=True, comments=comments,
+            resolved_task="delete_branch")
+        self.assertFalse(result["success"])
+        self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task")
+        self.assertEqual(calls["post"], [])
+
+    def test_author_reviewer_merger_cannot_apply(self):
+        """10. Permission-only roles are refused at the role gate."""
+        for name, env in (("author", AUTHOR_ENV), ("reviewer", REVIEWER_ENV),
+                          ("merger", MERGER_ENV)):
+            with self.subTest(role=name):
+                gate._reset_for_testing()
+                comments = [_lease_comment()]
+                self._run(env, apply=False, comments=comments)
+                result, calls = self._run(env, apply=True, comments=comments)
+                self.assertFalse(result["success"])
+                self.assertFalse(result["cleanup_performed"])
+                self.assertEqual(result["blocker_kind"], "wrong_role")
+                self.assertEqual(
+                    calls["post"], [],
+                    f"{name} must not post a terminal lease marker")
+
+
+class TestApplyFailsClosedOnLeaseState(_ToolCase):
+    def test_open_pr_lease_is_never_force_cleaned(self):
+        """12. Non-moot: an active lease on an open PR stays untouched."""
+        comments = [_lease_comment()]
+        result, calls = self._run(
+            RECONCILER_ENV, apply=True, comments=comments,
+            pr_state="open", pr_merged=False)
+        self.assertFalse(result["cleanup_performed"])
+        self.assertFalse(result["pr_merged_or_closed"])
+        self.assertEqual(calls["post"], [], "never force-clean an open PR lease")
+        self.assertTrue(any(
+            "still open" in r for r in result.get("cleanup_skipped_reason", [])))
+
+    def test_superseded_lease_between_dry_run_and_apply_fails_closed(self):
+        """12. The lease moved on after the dry run -> refuse."""
+        self._run(RECONCILER_ENV, apply=False, comments=[_lease_comment()])
+        moved = [_lease_comment(session_id="fresh-session",
+                                candidate_head=HEAD_B, comment_id=7777)]
+        result, calls = self._run(RECONCILER_ENV, apply=True, comments=moved)
+        self.assertFalse(result["success"])
+        self.assertEqual(result["blocker_kind"], "superseded_lease")
+        self.assertEqual(calls["post"], [])
+
+    def test_expectation_mismatch_fails_closed(self):
+        """11. Wrong session / head / marker expectations refuse the apply."""
+        comments = [_lease_comment()]
+        for kwargs in ({"expected_session_id": "wrong-session"},
+                       {"expected_candidate_head": HEAD_B},
+                       {"expected_lease_comment_id": 7777}):
+            with self.subTest(**kwargs):
+                gate._reset_for_testing()
+                self._run(RECONCILER_ENV, apply=False, comments=comments)
+                result, calls = self._run(
+                    RECONCILER_ENV, apply=True, comments=comments, **kwargs)
+                self.assertFalse(result["success"])
+                self.assertEqual(result["blocker_kind"], "lease_mismatch")
+                self.assertEqual(calls["post"], [])
+
+    def test_already_terminal_cleanup_is_idempotent(self):
+        """13. A released lease reports nothing to clean and posts nothing."""
+        first = _assessment([_lease_comment()])
+        released = {"id": 7000, "body": first["release_body"],
+                    "user": {"login": "sysadmin"}}
+        comments = [_lease_comment(), released]
+        self._run(RECONCILER_ENV, apply=False, comments=comments)
+        result, calls = self._run(RECONCILER_ENV, apply=True, comments=comments)
+        self.assertFalse(result["cleanup_performed"])
+        self.assertFalse(result["lease_moot"])
+        self.assertEqual(calls["post"], [], "no second terminal marker")
+        self.assertTrue(any(
+            "already released/terminal" in r
+            for r in result.get("cleanup_skipped_reason", [])))
+
+    def test_apply_is_append_only_never_deletes(self):
+        """14. The only write is a POST; nothing is edited or deleted."""
+        comments = [_lease_comment()]
+        self._run(RECONCILER_ENV, apply=False, comments=comments)
+        side, _calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=comments)
+        seen = []
+
+        def _recording(method, url, auth=None, payload=None, *a, **k):
+            seen.append((method or "").upper())
+            return side(method, url, auth, payload, *a, **k)
+
+        with patch("mcp_server.api_request", side_effect=_recording), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", return_value=None), \
+                patch("mcp_server._bound_repository_slug", return_value=SLUG), \
+                patch("mcp_server._repository_binding_block", return_value=None), \
+                patch.object(mcp_server, "_preflight_resolved_task",
+                             CANONICAL_TASK), \
+                patch.dict(os.environ, RECONCILER_ENV, clear=True):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs")
+        self.assertTrue(result["cleanup_performed"])
+        self.assertNotIn("DELETE", seen)
+        self.assertNotIn("PATCH", seen)
+        self.assertNotIn("PUT", seen)
+        self.assertEqual(seen.count("POST"), 1)
+
+
+class TestRepositoryBinding(_ToolCase):
+    """11. Foreign-repository targets fail closed before any mutation."""
+
+    def test_explicit_foreign_repository_is_rejected(self):
+        comments = [_lease_comment()]
+        side, calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=comments)
+        with patch("mcp_server.api_request", side_effect=side), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", return_value=None), \
+                patch("mcp_server._canonical_repository_slug",
+                      return_value=(None, [])), \
+                patch("mcp_server._workspace_repository_slug",
+                      return_value=SLUG), \
+                patch.object(mcp_server, "_preflight_resolved_task",
+                             CANONICAL_TASK), \
+                patch.dict(os.environ, RECONCILER_ENV, clear=True):
+            gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs",
+                org="Some-Other-Org", repo="Some-Other-Repo")
+        self.assertFalse(result["success"])
+        self.assertEqual(result["blocker_kind"], "repository_binding")
+        self.assertEqual(calls["post"], [])
+
+    def test_unresolvable_canonical_root_fails_closed(self):
+        comments = [_lease_comment()]
+        side, calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=comments)
+        with patch("mcp_server.api_request", side_effect=side), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", return_value=None), \
+                patch("mcp_server._canonical_repository_slug",
+                      return_value=(None, ["canonical root unresolvable"])), \
+                patch.object(mcp_server, "_preflight_resolved_task",
+                             CANONICAL_TASK), \
+                patch.dict(os.environ, RECONCILER_ENV, clear=True):
+            gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs")
+        self.assertFalse(result["success"])
+        self.assertEqual(result["blocker_kind"], "repository_binding")
+        self.assertEqual(calls["post"], [])
+
+
+class TestPreflightBinding(_ToolCase):
+    """4. The apply path binds the shared preflight to the exact task."""
+
+    def test_apply_forwards_task_and_target_to_preflight(self):
+        comments = [_lease_comment()]
+        self._run(RECONCILER_ENV, apply=False, comments=comments)
+        side, _calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=comments)
+        seen = {}
+
+        def _purity(remote=None, worktree_path=None, task=None, **kw):
+            seen.update({"remote": remote, "worktree_path": worktree_path,
+                         "task": task, **kw})
+            return None
+
+        with patch("mcp_server.api_request", side_effect=side), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", _purity), \
+                patch("mcp_server._bound_repository_slug", return_value=SLUG), \
+                patch("mcp_server._repository_binding_block", return_value=None), \
+                patch.object(mcp_server, "_preflight_resolved_task",
+                             CANONICAL_TASK), \
+                patch.dict(os.environ, RECONCILER_ENV, clear=True):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs",
+                worktree_path="/tmp/branches/reconciler-745",
+                org="Scaled-Tech-Consulting", repo="Gitea-Tools")
+        self.assertTrue(result["cleanup_performed"])
+        self.assertEqual(seen["task"], CANONICAL_TASK)
+        self.assertEqual(seen["worktree_path"], "/tmp/branches/reconciler-745")
+        self.assertEqual(seen["org"], "Scaled-Tech-Consulting")
+        self.assertEqual(seen["repo"], "Gitea-Tools")
+
+    def test_dry_run_does_not_require_preflight(self):
+        def _boom(*a, **k):
+            raise AssertionError("dry run must not run mutation preflight")
+
+        side, calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=[_lease_comment()])
+        with patch("mcp_server.api_request", side_effect=side), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", _boom), \
+                patch("mcp_server._bound_repository_slug", return_value=SLUG), \
+                patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
+        self.assertTrue(result["success"])
+        self.assertEqual(calls["post"], [])
+
+
+class TestNoProductionLeaseTouched(unittest.TestCase):
+    """16. No real production lease or PR is referenced by these tests."""
+
+    PRODUCTION_PR = 744
+    PRODUCTION_SESSION = "33673-1d54887a0415"
+    PRODUCTION_MARKER = 12452
+
+    def test_fixtures_are_synthetic(self):
+        self.assertNotEqual(PR, self.PRODUCTION_PR)
+        self.assertNotEqual(SESSION, self.PRODUCTION_SESSION)
+        self.assertNotEqual(LEASE_COMMENT_ID, self.PRODUCTION_MARKER)
+
+    def test_module_source_never_names_the_production_lease(self):
+        source = _Path(__file__).read_text()
+        for token in (self.PRODUCTION_SESSION, str(self.PRODUCTION_MARKER)):
+            self.assertEqual(
+                source.count(token), 1,
+                f"{token!r} must appear only in this guard's own constants",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_755_owning_pr_recovery.py

@@ -0,0 +1,609 @@
+import sys as _sys
+from pathlib import Path as _Path
+_sys.path.insert(0, str(_Path(__file__).resolve().parent))
+from mutation_profile_fixture import shared_mutation_env  # noqa: E402
+"""Dead-session lock recovery when the issue already owns an open PR (#755).
+
+#753 added the recovery *assessor*, but the production ``gitea_lock_issue``
+path still rejected every sanctioned recovery: a dead-session lock is by
+construction a lock for work that already has an open PR, and the #400
+duplicate-work gate blocked unconditionally on any linked open PR. These tests
+drive the real MCP handler, not just the pure assessor, so that gap cannot
+reopen.
+"""
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_lock_provenance  # noqa: E402
+import issue_lock_recovery  # noqa: E402
+import issue_lock_store  # noqa: E402
+import mcp_server  # noqa: E402
+from issue_work_duplicate_gate import (  # noqa: E402
+    OUTCOME_DUPLICATE_PR_PREVENTED,
+    OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
+    PHASE_LOCK,
+    assess_work_issue_duplicate_gate,
+)
+
+ISSUE = 4755
+BRANCH = f"fix/issue-{ISSUE}-owning-pr"
+OTHER_BRANCH = f"fix/issue-{ISSUE}-competing"
+HEAD = "c" * 40
+OTHER_HEAD = "d" * 40
+OWNING_PR = 4756
+OTHER_PR = 4757
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+
+def dead_pid() -> int:
+    """A PID that has certainly exited (spawned, then reaped)."""
+    proc = subprocess.Popen([sys.executable, "-c", "pass"])
+    proc.wait()
+    return proc.pid
+
+
+def shifted_ts(hours: int = 4) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
+    return {
+        "number": number,
+        "title": f"fix: something (Closes #{issue})",
+        "body": f"Closes #{issue}.",
+        "head": {"ref": ref, "sha": sha},
+    }
+
+
+def sanctioned_token(
+    issue_number=ISSUE, pr_number=OWNING_PR, branch=BRANCH, head=HEAD
+):
+    """The evidence shape the server derives from a granted recovery."""
+    return {
+        "issue_number": issue_number,
+        "pr_number": pr_number,
+        "branch_name": branch,
+        "head_sha": head,
+    }
+
+
+# ───────────────────────── duplicate gate: exemption ─────────────────────────
+
+
+class TestOwningPrExemptionGranted(unittest.TestCase):
+    def test_exact_owning_pr_is_not_duplicate_work(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
+        self.assertEqual(result["linked_open_pr"], OWNING_PR)
+        self.assertEqual(result["linked_open_pr_count"], 1)
+
+    def test_unrelated_open_pr_alongside_owning_pr_is_ignored(self):
+        unrelated = {
+            "number": 999,
+            "title": "chore: unrelated",
+            "body": "no linkage",
+            "head": {"ref": "chore/unrelated", "sha": OTHER_HEAD},
+        }
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[unrelated, owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["linked_open_pr_count"], 1)
+
+
+class TestOwningPrExemptionRefused(unittest.TestCase):
+    def assert_blocked(self, result):
+        self.assertTrue(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED)
+
+    def test_no_recovery_evidence_keeps_ordinary_blocker(self):
+        self.assert_blocked(
+            assess_work_issue_duplicate_gate(
+                ISSUE,
+                open_prs=[owning_pr()],
+                branch_names=[BRANCH],
+                claim_entry={"status": "not_claimed"},
+                locked_branch=BRANCH,
+                phase=PHASE_LOCK,
+            )
+        )
+
+    def test_competing_pr_number_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(number=OTHER_PR)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_multiple_linked_open_prs_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+        self.assertEqual(result["linked_open_pr_count"], 2)
+
+    def test_different_branch_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(ref=OTHER_BRANCH)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_locked_branch_differing_from_evidence_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=OTHER_BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_different_head_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(sha=OTHER_HEAD)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_evidence_for_another_issue_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(issue_number=ISSUE + 1),
+        )
+        self.assert_blocked(result)
+
+    def test_missing_head_in_live_pr_refused(self):
+        pr = owning_pr()
+        pr["head"] = {"ref": BRANCH}
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[pr],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+
+class TestOrdinaryDuplicateBehaviorUnchanged(unittest.TestCase):
+    def test_clean_issue_still_passes(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[],
+            branch_names=["feat/other-issue-99"],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+        )
+        self.assertFalse(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+
+    def test_competing_branch_still_blocks_even_with_owning_pr_evidence(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH, OTHER_BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        # The owning PR is exempt, but the competing branch is not.
+        self.assertTrue(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertIn(OTHER_BRANCH, result["conflicting_branches"])
+
+
+# ─────────────────── server-derived evidence cannot be forged ───────────────────
+
+
+class TestOwningPrEvidenceDerivation(unittest.TestCase):
+    def granted(self, **evidence_overrides):
+        evidence = {
+            "issue_number": ISSUE,
+            "locked_branch": BRANCH,
+            "local_head": HEAD,
+            "remote_head": HEAD,
+            "pr_head": HEAD,
+            "pr_number": OWNING_PR,
+        }
+        evidence.update(evidence_overrides)
+        return {
+            "outcome": issue_lock_recovery.RECOVERY_SANCTIONED,
+            "recovery_sanctioned": True,
+            "is_candidate": True,
+            "reasons": [],
+            "evidence": evidence,
+        }
+
+    def test_granted_recovery_yields_evidence(self):
+        token = issue_lock_recovery.owning_pr_recovery_evidence(self.granted())
+        self.assertEqual(token, sanctioned_token())
+
+    def test_none_assessment_yields_nothing(self):
+        self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(None))
+
+    def test_refused_assessment_yields_nothing(self):
+        refused = self.granted()
+        refused["outcome"] = issue_lock_recovery.REFUSED
+        refused["recovery_sanctioned"] = False
+        self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(refused))
+
+    def test_sanctioned_flag_without_outcome_yields_nothing(self):
+        forged = self.granted()
+        forged["outcome"] = "SOMETHING_ELSE"
+        self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(forged))
+
+    def test_missing_pr_number_yields_nothing(self):
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(
+                self.granted(pr_number=None)
+            )
+        )
+
+    def test_head_disagreement_in_evidence_yields_nothing(self):
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(
+                self.granted(remote_head=OTHER_HEAD)
+            )
+        )
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(
+                self.granted(local_head=OTHER_HEAD)
+            )
+        )
+
+    def test_real_refused_assessment_yields_nothing(self):
+        """End-to-end against the real assessor, not a hand-built dict."""
+        lock = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "worktree_path": "/scratch/wt",
+            "remote": "prgs",
+            "org": "Example-Org",
+            "repo": "Example-Repo",
+            "session_pid": os.getpid(),  # alive → must refuse
+            "work_lease": {
+                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+                "issue_number": ISSUE,
+                "branch": BRANCH,
+                "worktree_path": "/scratch/wt",
+                "claimant": {"username": IDENTITY, "profile": PROFILE},
+                "expires_at": shifted_ts(),
+            },
+        }
+        assessment = issue_lock_recovery.assess_dead_session_lock_recovery(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path="/scratch/wt",
+            remote="prgs",
+            org="Example-Org",
+            repo="Example-Repo",
+            identity=IDENTITY,
+            profile=PROFILE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            head_sha=HEAD,
+            remote_head_sha=HEAD,
+            pr_head_sha=HEAD,
+            pr_number=OWNING_PR,
+            competing_live_locks=[],
+            candidate_branches=[BRANCH],
+            current_pid=os.getpid(),
+        )
+        self.assertFalse(assessment["recovery_sanctioned"])
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(assessment)
+        )
+
+
+# ──────────────────── end-to-end: the real gitea_lock_issue ────────────────────
+
+
+class LockIssueEndToEndBase(unittest.TestCase):
+    """Drives ``mcp_server.gitea_lock_issue`` with live git/Gitea observation
+    stubbed at the module boundary — the production gate chain itself runs."""
+
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.worktree = os.path.realpath(os.getcwd())
+        # Bind host/org/repo to what the ``test-author-prgs`` fixture profile is
+        # pinned to, so the session-context gate under test is the real one and
+        # not a cross-host denial. The issue number and lock dir stay synthetic.
+        self.remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.prgs.cc",
+                "org": ORG,
+                "repo": REPO,
+            },
+        })
+        self.remotes.start()
+        self.addCleanup(patch.stopall)
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def write_durable_lock(self, *, pid, branch=BRANCH, worktree=None):
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        claimant = {"username": IDENTITY, "profile": PROFILE}
+        data = {
+            "issue_number": ISSUE,
+            "branch_name": branch,
+            "remote": "prgs",
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": worktree or self.worktree,
+            "session_pid": pid,
+            "pid": pid,
+            "work_lease": {
+                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+                "issue_number": ISSUE,
+                "branch": branch,
+                "worktree_path": worktree or self.worktree,
+                "claimant": claimant,
+                "created_at": shifted_ts(-1),
+                "last_heartbeat_at": shifted_ts(-1),
+                "expires_at": shifted_ts(),
+            },
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue",
+                claimant=claimant,
+            ),
+        }
+        issue_lock_store.save_lock_file(path, data)
+        return path
+
+    def run_lock(
+        self,
+        *,
+        open_prs,
+        porcelain="",
+        current_branch=BRANCH,
+        base_equivalent=False,
+        branch_names=None,
+        head_sha=HEAD,
+        remote_head=HEAD,
+    ):
+        branch_names = branch_names if branch_names is not None else [BRANCH]
+        branch_entries = [
+            {"name": name, "commit": {"id": remote_head}} for name in branch_names
+        ]
+        env = shared_mutation_env(
+            "test-author-prgs",
+            include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+        with patch(
+            "mcp_server.api_get_all", return_value=branch_entries
+        ), patch(
+            "mcp_server._list_open_pulls", return_value=list(open_prs)
+        ), patch(
+            "mcp_server.get_auth_header", return_value="token x"
+        ), patch(
+            "mcp_server._work_lease_claimant",
+            return_value={"username": IDENTITY, "profile": PROFILE},
+        ), patch(
+            "mcp_server.issue_lock_worktree.read_worktree_git_state",
+            return_value={
+                "current_branch": current_branch,
+                "porcelain_status": porcelain,
+                "base_equivalent": base_equivalent,
+                "head_sha": head_sha,
+                "inspected_git_root": self.worktree,
+                "base_branch": "master",
+            },
+        ), patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            side_effect=lambda h, o, r, auth, issue_number: (
+                list(open_prs), list(branch_names), {"status": "not_claimed"}
+            ),
+        ):
+            with patch.dict(os.environ, env, clear=True):
+                os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+                return mcp_server.gitea_lock_issue(
+                    issue_number=ISSUE,
+                    branch_name=BRANCH,
+                    remote="prgs",
+                    worktree_path=self.worktree,
+                )
+
+
+class TestRecoveryWithOwningPrSucceeds(LockIssueEndToEndBase):
+    def test_dead_session_recovery_with_owning_pr_relocks(self):
+        self.write_durable_lock(pid=dead_pid())
+        result = self.run_lock(open_prs=[owning_pr()])
+
+        self.assertTrue(result["success"])
+        self.assertEqual(result["issue_number"], ISSUE)
+        self.assertEqual(result["branch_name"], BRANCH)
+        self.assertTrue(result["lock_freshness"]["live"])
+        self.assertTrue(result["lock_freshness"]["pid_alive"])
+
+    def test_recovered_lock_records_truthful_provenance(self):
+        prior = dead_pid()
+        self.write_durable_lock(pid=prior)
+        self.run_lock(open_prs=[owning_pr()])
+
+        lock = issue_lock_store.load_issue_lock(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        record = lock.get("dead_session_recovery") or {}
+        self.assertTrue(record.get("recovered"))
+        self.assertEqual(record.get("prior_session_pid"), prior)
+        self.assertEqual(record.get("replacement_session_pid"), os.getpid())
+        self.assertFalse(record.get("prior_pid_alive"))
+        self.assertEqual(record.get("pr_number"), OWNING_PR)
+        self.assertEqual(record.get("branch_name"), BRANCH)
+        self.assertEqual(record.get("identity"), IDENTITY)
+
+    def test_recovered_lock_is_live_and_proves_pr_ownership(self):
+        """AC6: the persisted lock satisfies update-by-merge's ownership prover."""
+        self.write_durable_lock(pid=dead_pid())
+        self.run_lock(open_prs=[owning_pr()])
+
+        lock = issue_lock_store.load_issue_lock(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        self.assertTrue(issue_lock_store.is_lease_live(lock))
+
+        env = shared_mutation_env(
+            "test-author-prgs",
+            include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+        with patch.dict(os.environ, env, clear=True):
+            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+            ownership = mcp_server._prove_author_ownership_for_pr(
+                pr_number=OWNING_PR,
+                pr_title=f"fix: something (Closes #{ISSUE})",
+                pr_body=f"Closes #{ISSUE}.",
+                source_branch=BRANCH,
+                remote="prgs",
+                host="gitea.prgs.cc",
+                org=ORG,
+                repo=REPO,
+                worktree_path=self.worktree,
+            )
+        self.assertTrue(ownership["proven"], ownership["reasons"])
+        self.assertTrue(ownership["has_author_lock"])
+        self.assertEqual(ownership["matched_issue"], ISSUE)
+
+
+class TestRecoveryRejectionsEndToEnd(LockIssueEndToEndBase):
+    def assert_lock_refused(self, **kwargs):
+        with self.assertRaises((ValueError, RuntimeError)) as ctx:
+            self.run_lock(**kwargs)
+        return str(ctx.exception)
+
+    def test_competing_pr_still_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        message = self.assert_lock_refused(
+            open_prs=[owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+            branch_names=[BRANCH],
+        )
+        self.assertIn("already covers issue", message)
+
+    def test_multiple_linked_open_prs_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        message = self.assert_lock_refused(
+            open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+        )
+        self.assertIn("already covers issue", message)
+
+    def test_owning_pr_on_a_different_head_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(open_prs=[owning_pr(sha=OTHER_HEAD)])
+
+    def test_lock_registered_to_a_different_worktree_blocked(self):
+        self.write_durable_lock(
+            pid=dead_pid(), worktree=os.path.join(self.worktree, "elsewhere")
+        )
+        self.assert_lock_refused(open_prs=[owning_pr()])
+
+    def test_dirty_worktree_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(
+            open_prs=[owning_pr()], porcelain=" M gitea_mcp_server.py"
+        )
+
+    def test_worktree_parked_on_another_branch_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(
+            open_prs=[owning_pr()], current_branch="master"
+        )
+
+    def test_local_head_differing_from_remote_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(
+            open_prs=[owning_pr()], head_sha=OTHER_HEAD
+        )
+
+    def test_live_prior_pid_blocked(self):
+        self.write_durable_lock(pid=os.getpid())
+        self.assert_lock_refused(open_prs=[owning_pr()])
+
+    def test_new_claim_without_prior_lock_still_requires_base_equivalence(self):
+        """AC10: no durable lock → no recovery → base-equivalence still rules."""
+        self.assert_lock_refused(open_prs=[], branch_names=[])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_757_bootstrap_guard_agreement.py

@@ -0,0 +1,809 @@
+"""Regression tests for #757: #274 and #604 must not disagree.
+
+The sanctioned ``create_issue`` bootstrap (#749/#750) was unreachable in
+production: the #274 branches-only guard consulted the bootstrap and permitted
+a clean canonical control checkout, then the bootstrap-blind #604 anti-stomp
+preflight rejected the same checkout as ``wrong_worktree``.
+
+These tests pin the fix:
+
+* one server-derived assessment, interpreted by one shared predicate;
+* the waiver is narrow (only the wrong-worktree verdict, only create_issue,
+  only from the exact clean canonical control checkout);
+* every other guard and rejection reason keeps its fail-closed behaviour;
+* eligibility cannot be forged through a public tool signature.
+"""
+
+from __future__ import annotations
+
+import inspect
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import anti_stomp_preflight as asp  # noqa: E402
+import create_issue_bootstrap as cib  # noqa: E402
+import gitea_mcp_server as srv  # 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])
+
+BRANCHES_WORKTREE = str(Path(CONTROL_CHECKOUT_ROOT) / "branches" / "issue-1-x")
+
+
+def proven_bootstrap(
+    *,
+    workspace=CONTROL_CHECKOUT_ROOT,
+    root=CONTROL_CHECKOUT_ROOT,
+    task="create_issue",
+    branch="master",
+    porcelain="",
+    head=MASTER_SHA,
+    remote_sha=MASTER_SHA,
+):
+    """Build a real assessment via the production assessor (never hand-rolled)."""
+    return cib.assess_create_issue_bootstrap(
+        workspace_path=workspace,
+        canonical_repo_root=root,
+        current_branch=branch,
+        head_sha=head,
+        porcelain_status=porcelain,
+        remote_master_sha=remote_sha,
+        task=task,
+    )
+
+
+class TestSharedPredicate(unittest.TestCase):
+    """The one interpretation both guards consume (AC6, AC7)."""
+
+    def _permits(self, assessment, task="create_issue", workspace=None, root=None):
+        return cib.bootstrap_permits_control_checkout(
+            assessment,
+            task=task,
+            workspace_path=workspace or CONTROL_CHECKOUT_ROOT,
+            canonical_repo_root=root or CONTROL_CHECKOUT_ROOT,
+        )
+
+    def test_proven_bootstrap_permits(self):
+        self.assertTrue(self._permits(proven_bootstrap()))
+
+    def test_tool_alias_permits(self):
+        assessment = proven_bootstrap(task="gitea_create_issue")
+        self.assertTrue(self._permits(assessment, task="gitea_create_issue"))
+
+    def test_missing_assessment_fails_closed(self):
+        self.assertFalse(self._permits(None))
+
+    def test_malformed_assessment_fails_closed(self):
+        for bogus in ("allowed", 1, True, [], ["allowed"], object()):
+            with self.subTest(bogus=bogus):
+                self.assertFalse(self._permits(bogus))
+
+    def test_refused_assessment_fails_closed(self):
+        # Dirty control checkout -> assessor blocks -> predicate must refuse.
+        self.assertFalse(
+            self._permits(proven_bootstrap(porcelain=" M gitea_mcp_server.py\n"))
+        )
+
+    def test_not_applicable_assessment_fails_closed(self):
+        # branches/ worktree -> not_applicable -> no waiver.
+        self.assertFalse(
+            self._permits(proven_bootstrap(workspace=BRANCHES_WORKTREE))
+        )
+
+    def test_incomplete_assessment_fails_closed(self):
+        incomplete = dict(proven_bootstrap())
+        del incomplete["bootstrap_path"]
+        self.assertFalse(self._permits(incomplete))
+
+    def test_contradictory_block_and_allow_fails_closed(self):
+        contradictory = dict(proven_bootstrap(), block=True)
+        self.assertFalse(self._permits(contradictory))
+
+    def test_contradictory_dirty_but_allowed_fails_closed(self):
+        contradictory = dict(proven_bootstrap(), dirty_files=["gitea_mcp_server.py"])
+        self.assertFalse(self._permits(contradictory))
+
+    def test_contradictory_under_branches_but_allowed_fails_closed(self):
+        contradictory = dict(proven_bootstrap(), under_branches=True)
+        self.assertFalse(self._permits(contradictory))
+
+    def test_contradictory_reasons_present_fails_closed(self):
+        contradictory = dict(proven_bootstrap(), reasons=["something refused"])
+        self.assertFalse(self._permits(contradictory))
+
+    def test_truthy_non_true_values_fail_closed(self):
+        """Strict identity: no truthy smuggling (1, 'yes') can assert eligibility."""
+        for value in (1, "yes", "true", [1]):
+            with self.subTest(value=value):
+                self.assertFalse(self._permits(dict(proven_bootstrap(), allowed=value)))
+
+    def test_wrong_task_fails_closed(self):
+        """An otherwise-proven assessment cannot license a different task."""
+        self.assertFalse(self._permits(proven_bootstrap(), task="lock_issue"))
+        self.assertFalse(self._permits(proven_bootstrap(), task="create_pr"))
+        self.assertFalse(self._permits(proven_bootstrap(), task=None))
+
+    def test_foreign_workspace_binding_fails_closed(self):
+        """Assessment must describe the workspace actually being guarded."""
+        other = proven_bootstrap(workspace="/other/clone", root="/other/clone")
+        self.assertFalse(self._permits(other))
+
+    def test_scope_and_path_tampering_fails_closed(self):
+        self.assertFalse(
+            self._permits(dict(proven_bootstrap(), task_scope="all_tasks"))
+        )
+        self.assertFalse(
+            self._permits(dict(proven_bootstrap(), bootstrap_path="anything_goes"))
+        )
+
+
+class TestAntiStompHonorsBootstrap(unittest.TestCase):
+    """#604 consumes the same decision, and waives only wrong_worktree."""
+
+    def _assess(self, *, task="create_issue", bootstrap=None, workspace=None, **kw):
+        params = dict(
+            task=task,
+            profile_role="author",
+            required_role="author",
+            workspace_path=workspace or CONTROL_CHECKOUT_ROOT,
+            project_root=CONTROL_CHECKOUT_ROOT,
+            current_branch="master",
+            root_head_sha=MASTER_SHA,
+            root_porcelain="",
+            remote_master_sha=MASTER_SHA,
+            check_repo=False,
+            create_issue_bootstrap_assessment=bootstrap,
+        )
+        params.update(kw)
+        return asp.assess_anti_stomp_preflight(**params)
+
+    def _blocker_kinds(self, result):
+        return {b["kind"] for b in result.get("blockers") or []}
+
+    def test_control_checkout_without_bootstrap_still_blocked(self):
+        """Baseline: the exact production failure, unwaived."""
+        res = self._assess(bootstrap=None)
+        self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
+
+    def test_control_checkout_with_proven_bootstrap_permitted(self):
+        """AC1/AC2: the #757 fix — same inputs, bootstrap honoured."""
+        res = self._assess(bootstrap=proven_bootstrap())
+        self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
+        self.assertTrue(res["checks"]["worktree"]["create_issue_bootstrap_waived"])
+        self.assertFalse(res["checks"]["worktree"]["block"])
+
+    def test_tool_alias_permitted(self):
+        res = self._assess(
+            task="gitea_create_issue",
+            bootstrap=proven_bootstrap(task="gitea_create_issue"),
+        )
+        self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
+
+    def test_non_create_issue_task_never_waived(self):
+        """AC5: other author mutations keep the branches-only requirement."""
+        for task in ("lock_issue", "create_pr", "commit_files", "mark_issue"):
+            with self.subTest(task=task):
+                res = self._assess(task=task, bootstrap=proven_bootstrap())
+                self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
+
+    def test_dirty_control_checkout_still_blocked(self):
+        """AC4: dirty root refuses the bootstrap, so no waiver."""
+        dirty = " M gitea_mcp_server.py\n"
+        res = self._assess(
+            bootstrap=proven_bootstrap(porcelain=dirty), root_porcelain=dirty
+        )
+        self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
+
+    def test_detached_control_checkout_still_blocked(self):
+        res = self._assess(
+            bootstrap=proven_bootstrap(branch=""), current_branch=""
+        )
+        self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
+
+    def test_base_divergence_still_blocked(self):
+        res = self._assess(
+            bootstrap=proven_bootstrap(head=STALE_SHA), root_head_sha=STALE_SHA
+        )
+        self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
+
+    def test_waiver_does_not_suppress_stale_runtime(self):
+        """AC: waive only wrong_worktree — stale runtime still fails closed."""
+        res = self._assess(
+            bootstrap=proven_bootstrap(),
+            startup_head=STALE_SHA,
+            current_code_head=MASTER_SHA,
+        )
+        self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
+        self.assertIn(asp.BLOCKER_STALE_RUNTIME, self._blocker_kinds(res))
+        self.assertTrue(res["block"])
+
+    def test_waiver_does_not_suppress_wrong_repo(self):
+        res = self._assess(
+            bootstrap=proven_bootstrap(),
+            check_repo=True,
+            remote="prgs",
+            resolved_org="Scaled-Tech-Consulting",
+            resolved_repo="Timesheet",
+            # Not both-explicit, so the #530 repo guard actually evaluates the
+            # mismatch against the local remote instead of trusting the caller.
+            org_explicit=False,
+            repo_explicit=False,
+            local_remote_url=(
+                "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+            ),
+        )
+        self.assertNotIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
+        self.assertIn(asp.BLOCKER_WRONG_REPO, self._blocker_kinds(res))
+
+    def test_waiver_does_not_suppress_wrong_role(self):
+        res = self._assess(
+            bootstrap=proven_bootstrap(),
+            profile_role="reviewer",
+            required_role="author",
+        )
+        self.assertIn(asp.BLOCKER_WRONG_ROLE, self._blocker_kinds(res))
+
+    def test_branches_worktree_unaffected(self):
+        """AC: ordinary branches/ worktrees keep working, waived or not."""
+        for bootstrap in (None, proven_bootstrap(workspace=BRANCHES_WORKTREE)):
+            with self.subTest(bootstrap=bool(bootstrap)):
+                res = self._assess(
+                    workspace=BRANCHES_WORKTREE,
+                    bootstrap=bootstrap,
+                    current_branch="fix/issue-1-x",
+                )
+                self.assertNotIn(
+                    asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res)
+                )
+
+    def test_forged_assessment_rejected(self):
+        """AC6: a hand-built 'allowed' dict cannot unlock the waiver."""
+        forged = {"allowed": True, "block": False}
+        res = self._assess(bootstrap=forged)
+        self.assertIn(asp.BLOCKER_WRONG_WORKTREE, self._blocker_kinds(res))
+
+
+class TestGuardAgreement(unittest.TestCase):
+    """AC7: the regression that would have caught the #757 defect.
+
+    For identical evidence, the #274 guard and the #604 guard must return the
+    same wrong-worktree verdict across the full workspace-state matrix.
+    """
+
+    MATRIX = [
+        (
+            "clean control + create_issue",
+            CONTROL_CHECKOUT_ROOT, "create_issue", "master", "", MASTER_SHA,
+        ),
+        (
+            "clean control + alias",
+            CONTROL_CHECKOUT_ROOT, "gitea_create_issue", "master", "", MASTER_SHA,
+        ),
+        (
+            "clean control + lock_issue",
+            CONTROL_CHECKOUT_ROOT, "lock_issue", "master", "", MASTER_SHA,
+        ),
+        (
+            "clean control + create_pr",
+            CONTROL_CHECKOUT_ROOT, "create_pr", "master", "", MASTER_SHA,
+        ),
+        (
+            "dirty control + create_issue",
+            CONTROL_CHECKOUT_ROOT, "create_issue", "master",
+            " M gitea_mcp_server.py\n", MASTER_SHA,
+        ),
+        (
+            "detached control + create_issue",
+            CONTROL_CHECKOUT_ROOT, "create_issue", "", "", MASTER_SHA,
+        ),
+        (
+            "non-base control + create_issue",
+            CONTROL_CHECKOUT_ROOT, "create_issue", "feat/x", "", MASTER_SHA,
+        ),
+        (
+            "diverged control + create_issue",
+            CONTROL_CHECKOUT_ROOT, "create_issue", "master", "", STALE_SHA,
+        ),
+        (
+            "branches wt + create_issue",
+            BRANCHES_WORKTREE, "create_issue", "fix/issue-1-x", "", MASTER_SHA,
+        ),
+        (
+            "branches wt + lock_issue",
+            BRANCHES_WORKTREE, "lock_issue", "fix/issue-1-x", "", MASTER_SHA,
+        ),
+    ]
+
+    def _guard_274_blocks(self, *, workspace, task, branch, porcelain, head, bootstrap):
+        git_state = {
+            "current_branch": branch,
+            "head_sha": head,
+            "porcelain_status": porcelain,
+        }
+        ctx = {
+            "workspace_path": workspace,
+            "canonical_repo_root": CONTROL_CHECKOUT_ROOT,
+        }
+        with patch.object(srv, "_effective_workspace_role", return_value="author"), \
+             patch.object(srv, "_actual_profile_role", return_value="author"), \
+             patch.object(
+                 srv, "_resolve_namespace_mutation_context", return_value=ctx
+             ), \
+             patch.object(
+                 srv.issue_lock_worktree,
+                 "read_worktree_git_state",
+                 return_value=git_state,
+             ):
+            try:
+                srv._enforce_branches_only_author_mutation(
+                    workspace, task=task, bootstrap_assessment=bootstrap
+                )
+                return False
+            except RuntimeError:
+                return True
+
+    def _guard_604_blocks(self, *, workspace, task, branch, porcelain, head, bootstrap):
+        res = asp.assess_anti_stomp_preflight(
+            task=task,
+            profile_role="author",
+            required_role="author",
+            workspace_path=workspace,
+            project_root=CONTROL_CHECKOUT_ROOT,
+            current_branch=branch,
+            root_head_sha=head,
+            root_porcelain=porcelain,
+            remote_master_sha=MASTER_SHA,
+            check_repo=False,
+            create_issue_bootstrap_assessment=bootstrap,
+        )
+        return asp.BLOCKER_WRONG_WORKTREE in {
+            b["kind"] for b in res.get("blockers") or []
+        }
+
+    def test_guards_agree_across_matrix(self):
+        for label, workspace, task, branch, porcelain, head in self.MATRIX:
+            with self.subTest(case=label):
+                # ONE server-derived assessment, exactly as production computes it.
+                bootstrap = cib.assess_create_issue_bootstrap(
+                    workspace_path=workspace,
+                    canonical_repo_root=CONTROL_CHECKOUT_ROOT,
+                    current_branch=branch,
+                    head_sha=head,
+                    porcelain_status=porcelain,
+                    remote_master_sha=MASTER_SHA,
+                    task=task,
+                )
+                kwargs = dict(
+                    workspace=workspace,
+                    task=task,
+                    branch=branch,
+                    porcelain=porcelain,
+                    head=head,
+                    bootstrap=bootstrap,
+                )
+                blocked_274 = self._guard_274_blocks(**kwargs)
+                blocked_604 = self._guard_604_blocks(**kwargs)
+                self.assertEqual(
+                    blocked_274,
+                    blocked_604,
+                    f"{label}: #274 blocked={blocked_274} "
+                    f"but #604 blocked={blocked_604}",
+                )
+
+    def test_clean_control_create_issue_permitted_by_both(self):
+        """The specific case that was broken: both guards must permit."""
+        kwargs = dict(
+            workspace=CONTROL_CHECKOUT_ROOT,
+            task="create_issue",
+            branch="master",
+            porcelain="",
+            head=MASTER_SHA,
+            bootstrap=proven_bootstrap(),
+        )
+        self.assertFalse(self._guard_274_blocks(**kwargs))
+        self.assertFalse(self._guard_604_blocks(**kwargs))
+
+
+class TestNoCallerForgeableEligibility(unittest.TestCase):
+    """AC6: eligibility is never reachable through a public tool signature."""
+
+    def test_create_issue_tool_exposes_no_bootstrap_argument(self):
+        params = set(inspect.signature(srv.gitea_create_issue).parameters)
+        for forbidden in (
+            "bootstrap",
+            "bootstrap_assessment",
+            "create_issue_bootstrap",
+            "create_issue_bootstrap_assessment",
+            "allow_control_checkout",
+            "bootstrap_allowed",
+        ):
+            self.assertNotIn(forbidden, params)
+
+    def test_no_mcp_tool_exposes_bootstrap_argument(self):
+        """No public gitea_* tool may take bootstrap evidence from the caller."""
+        for name in dir(srv):
+            if not name.startswith("gitea_"):
+                continue
+            fn = getattr(srv, name)
+            if not callable(fn):
+                continue
+            try:
+                params = set(inspect.signature(fn).parameters)
+            except (TypeError, ValueError):
+                continue
+            leaked = {p for p in params if "bootstrap" in p.lower()}
+            self.assertFalse(leaked, f"{name} exposes bootstrap args: {leaked}")
+
+    def test_internal_helper_yields_nothing_for_other_tasks(self):
+        self.assertTrue(hasattr(srv, "_create_issue_bootstrap_assessment"))
+        self.assertIsNone(
+            srv._create_issue_bootstrap_assessment("lock_issue"),
+            "non-create_issue tasks must yield no bootstrap evidence",
+        )
+
+
+class TestNativeCreateIssueEndToEnd(unittest.TestCase):
+    """AC8: the production handler, with the #604 gate LIVE (not patched out).
+
+    The pre-existing #749 e2e test patched ``_run_anti_stomp_preflight`` to a
+    no-op, which is exactly why this defect reached production. These tests
+    leave it running.
+    """
+
+    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,
+        }
+
+    def _run_create_issue(
+        self,
+        *,
+        git_state,
+        remote_sha=MASTER_SHA,
+        parity=None,
+        title="Bootstrap issue from clean control",
+    ):
+        """Invoke the native handler with the anti-stomp gate live."""
+        parity = parity or {
+            "startup_head": MASTER_SHA,
+            "current_head": MASTER_SHA,
+        }
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \
+             patch.object(srv, "_auth", return_value=FAKE_AUTH), \
+             patch.object(srv, "_profile_permission_block", return_value=None), \
+             patch.object(srv, "_namespace_mutation_block", return_value=None), \
+             patch.object(
+                 srv.role_session_router,
+                 "check_author_mutation_after_reviewer_stop",
+                 return_value=(True, []),
+             ), \
+             patch.object(srv, "api_get_all", return_value=[]), \
+             patch.object(srv, "api_request") as mock_api, \
+             patch.object(
+                 srv.root_checkout_guard,
+                 "resolve_remote_master_sha",
+                 return_value=remote_sha,
+             ), \
+             patch.object(srv, "_current_master_parity", return_value=parity), \
+             patch.object(
+                 srv,
+                 "get_profile",
+                 return_value={
+                     "profile_name": "prgs-author",
+                     "allowed_operations": [
+                         "gitea.issue.create",
+                         "gitea.issue.comment",
+                         "gitea.pr.create",
+                         "gitea.read",
+                     ],
+                 },
+             ), \
+             patch.object(srv, "_actual_profile_role", return_value="author"), \
+             patch.object(srv, "_effective_workspace_role", return_value="author"), \
+             patch.object(
+                 srv.issue_lock_worktree,
+                 "read_worktree_git_state",
+                 return_value=git_state,
+             ), \
+             patch.object(
+                 srv,
+                 "_get_workspace_porcelain",
+                 return_value=git_state["porcelain_status"],
+             ), \
+             patch.object(srv, "_enforce_root_checkout_guard"):
+            mock_api.return_value = {
+                "number": 99,
+                "html_url": "https://gitea.example.com/issues/99",
+            }
+            try:
+                result = srv.gitea_create_issue(
+                    title=title,
+                    body="Body text for content gate.",
+                )
+            except RuntimeError as exc:
+                return {"raised": str(exc)}, mock_api
+            return result, mock_api
+
+    def test_clean_control_checkout_reaches_api(self):
+        """The exact production failure that blocked filing #757 and #758."""
+        res, mock_api = self._run_create_issue(git_state=self._git_state())
+        self.assertNotIn("raised", res, f"guard still blocks: {res.get('raised')}")
+        self.assertEqual(res.get("number"), 99)
+        mock_api.assert_called_once()
+
+    def test_dirty_control_checkout_blocked(self):
+        dirty = " M gitea_mcp_server.py\n"
+        res, mock_api = self._run_create_issue(
+            git_state=self._git_state(porcelain=dirty)
+        )
+        self.assertFalse(isinstance(res, dict) and res.get("number") == 99)
+        mock_api.assert_not_called()
+
+    def test_detached_control_checkout_blocked(self):
+        res, mock_api = self._run_create_issue(git_state=self._git_state(branch=""))
+        self.assertFalse(isinstance(res, dict) and res.get("number") == 99)
+        mock_api.assert_not_called()
+
+    def test_base_mismatch_blocked(self):
+        res, mock_api = self._run_create_issue(
+            git_state=self._git_state(head=STALE_SHA)
+        )
+        self.assertFalse(isinstance(res, dict) and res.get("number") == 99)
+        mock_api.assert_not_called()
+
+    def test_stale_runtime_blocked(self):
+        res, mock_api = self._run_create_issue(
+            git_state=self._git_state(),
+            parity={"startup_head": STALE_SHA, "current_head": MASTER_SHA},
+        )
+        self.assertFalse(isinstance(res, dict) and res.get("number") == 99)
+        mock_api.assert_not_called()
+
+    def test_non_create_issue_mutation_still_blocked_from_control(self):
+        """AC5 through the real preflight, with anti-stomp live."""
+        srv._preflight_resolved_task = "lock_issue"
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \
+             patch.object(srv, "_actual_profile_role", return_value="author"), \
+             patch.object(srv, "_effective_workspace_role", return_value="author"), \
+             patch.object(
+                 srv.issue_lock_worktree,
+                 "read_worktree_git_state",
+                 return_value=self._git_state(),
+             ), \
+             patch.object(srv, "_get_workspace_porcelain", return_value=""), \
+             patch.object(
+                 srv.root_checkout_guard,
+                 "resolve_remote_master_sha",
+                 return_value=MASTER_SHA,
+             ), \
+             patch.object(srv, "_enforce_root_checkout_guard"):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.verify_preflight_purity(remote="prgs", task="lock_issue")
+        self.assertIn("control checkout", str(ctx.exception))
+
+
+class TestBaseEquivalenceProofRequired(unittest.TestCase):
+    """#757 AC3/AC4: an unknown tip is not evidence of agreement.
+
+    The original fix compared SHAs only inside
+    ``if remote_tip and local_tip and remote_tip != local_tip``, so a missing
+    local HEAD, an unresolvable live master, or a resolver failure all fell
+    through and *granted* the bootstrap exemption. Base equivalence must be
+    proven, and anything less must fail closed.
+    """
+
+    def _permits(self, assessment):
+        return cib.bootstrap_permits_control_checkout(
+            assessment,
+            task="create_issue",
+            workspace_path=CONTROL_CHECKOUT_ROOT,
+            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
+        )
+
+    def _assert_fails_closed(self, assessment, *, expect_reason):
+        self.assertTrue(assessment["block"], "assessor must block")
+        self.assertFalse(assessment["allowed"])
+        self.assertFalse(assessment["proven"])
+        self.assertFalse(assessment["base_tips_verified"])
+        self.assertFalse(self._permits(assessment), "predicate must refuse")
+        joined = " ".join(assessment["reasons"]).lower()
+        self.assertIn(expect_reason, joined)
+
+    def test_missing_local_head_fails_closed(self):
+        self._assert_fails_closed(
+            proven_bootstrap(head=None),
+            expect_reason="control checkout head sha is unknown",
+        )
+
+    def test_missing_remote_master_fails_closed(self):
+        self._assert_fails_closed(
+            proven_bootstrap(remote_sha=None),
+            expect_reason="live master tip is unknown",
+        )
+
+    def test_both_tips_missing_fails_closed(self):
+        assessment = proven_bootstrap(head=None, remote_sha=None)
+        self._assert_fails_closed(
+            assessment, expect_reason="control checkout head sha is unknown"
+        )
+        self.assertIn("live master tip is unknown", " ".join(assessment["reasons"]))
+
+    def test_empty_and_whitespace_tips_fail_closed(self):
+        for head, remote in (("", MASTER_SHA), (MASTER_SHA, ""), ("   ", "  ")):
+            with self.subTest(head=repr(head), remote=repr(remote)):
+                assessment = proven_bootstrap(head=head, remote_sha=remote)
+                self.assertTrue(assessment["block"])
+                self.assertFalse(self._permits(assessment))
+
+    def test_mismatched_tips_fail_closed(self):
+        self._assert_fails_closed(
+            proven_bootstrap(head=STALE_SHA),
+            expect_reason="does not match",
+        )
+
+    def test_resolver_failure_fails_closed(self):
+        assessment = cib.assess_create_issue_bootstrap(
+            workspace_path=CONTROL_CHECKOUT_ROOT,
+            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
+            current_branch="master",
+            head_sha=MASTER_SHA,
+            porcelain_status="",
+            remote_master_sha=None,
+            remote_master_sha_error="TimeoutError: remote unreachable",
+            task="create_issue",
+        )
+        self._assert_fails_closed(
+            assessment, expect_reason="could not be resolved"
+        )
+        # The operator must be able to see *why*, not just that it blocked.
+        self.assertIn("remote unreachable", " ".join(assessment["reasons"]))
+
+    def test_proven_bootstrap_records_both_normalized_shas(self):
+        assessment = proven_bootstrap()
+        self.assertEqual(assessment["local_head_sha"], MASTER_SHA)
+        self.assertEqual(assessment["remote_master_sha"], MASTER_SHA)
+        self.assertTrue(assessment["base_tips_verified"])
+        self.assertTrue(self._permits(assessment))
+
+    def test_tips_are_normalized_before_comparison(self):
+        """Case and surrounding whitespace are not a different commit."""
+        assessment = proven_bootstrap(
+            head=f"  {MASTER_SHA.upper()}  ", remote_sha=MASTER_SHA
+        )
+        self.assertTrue(assessment["allowed"])
+        self.assertEqual(assessment["local_head_sha"], MASTER_SHA)
+        self.assertTrue(self._permits(assessment))
+
+    def test_predicate_rejects_assessment_with_tips_stripped(self):
+        """A recorded proof that is later removed cannot still permit."""
+        for field in ("local_head_sha", "remote_master_sha"):
+            with self.subTest(field=field):
+                assessment = dict(proven_bootstrap())
+                assessment[field] = None
+                self.assertFalse(self._permits(assessment))
+
+    def test_predicate_rejects_forged_verified_flag(self):
+        """base_tips_verified is re-derived, never trusted on its own."""
+        assessment = dict(proven_bootstrap())
+        assessment["local_head_sha"] = MASTER_SHA
+        assessment["remote_master_sha"] = STALE_SHA
+        assessment["base_tips_verified"] = True
+        self.assertFalse(self._permits(assessment))
+
+    def test_predicate_rejects_missing_verified_flag(self):
+        assessment = dict(proven_bootstrap())
+        assessment.pop("base_tips_verified")
+        self.assertFalse(self._permits(assessment))
+
+
+class TestServerAssessmentFailsClosedOnResolverError(unittest.TestCase):
+    """AC3/AC4 at the single server-derived computation site."""
+
+    def setUp(self):
+        # verify_preflight_purity short-circuits under pytest; the production
+        # path only runs with test mode disabled (same setup the #757 e2e uses).
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_resolved_role = "author"
+        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):
+        return {
+            "current_branch": "master",
+            "head_sha": MASTER_SHA,
+            "porcelain_status": "",
+        }
+
+    def test_resolver_exception_produces_blocking_assessment(self):
+        """A raising resolver must not become a silent, permissive None."""
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \
+             patch.object(
+                 srv.issue_lock_worktree,
+                 "read_worktree_git_state",
+                 return_value=self._git_state(),
+             ), \
+             patch.object(
+                 srv.root_checkout_guard,
+                 "resolve_remote_master_sha",
+                 side_effect=TimeoutError("remote unreachable"),
+             ):
+            assessment = srv._create_issue_bootstrap_assessment("create_issue")
+
+        self.assertIsNotNone(assessment)
+        self.assertTrue(assessment["block"])
+        self.assertFalse(assessment["allowed"])
+        self.assertFalse(assessment["base_tips_verified"])
+        self.assertIsNone(assessment["remote_master_sha"])
+        self.assertIn("could not be resolved", " ".join(assessment["reasons"]))
+        self.assertFalse(
+            cib.bootstrap_permits_control_checkout(
+                assessment,
+                task="create_issue",
+                workspace_path=CONTROL_CHECKOUT_ROOT,
+                canonical_repo_root=CONTROL_CHECKOUT_ROOT,
+            )
+        )
+
+    def test_resolver_exception_blocks_real_preflight(self):
+        """Production path: verify_preflight_purity must fail closed."""
+        srv._preflight_resolved_task = "create_issue"
+        try:
+            with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT), \
+                 patch.object(srv, "_actual_profile_role", return_value="author"), \
+                 patch.object(
+                     srv, "_effective_workspace_role", return_value="author"
+                 ), \
+                 patch.object(
+                     srv.issue_lock_worktree,
+                     "read_worktree_git_state",
+                     return_value=self._git_state(),
+                 ), \
+                 patch.object(srv, "_get_workspace_porcelain", return_value=""), \
+                 patch.object(
+                     srv.root_checkout_guard,
+                     "resolve_remote_master_sha",
+                     side_effect=TimeoutError("remote unreachable"),
+                 ), \
+                 patch.object(srv, "_enforce_root_checkout_guard"):
+                with self.assertRaises(RuntimeError):
+                    srv.verify_preflight_purity(remote="prgs", task="create_issue")
+        finally:
+            srv._preflight_resolved_task = None
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_post_merge_moot_lease.py

@@ -19,6 +19,8 @@ from unittest.mock import patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
+import mcp_server  # noqa: E402
+import post_merge_moot_lease_gate as moot_gate  # noqa: E402
 import reviewer_pr_lease as leases  # noqa: E402
 from mcp_server import (  # noqa: E402
     gitea_acquire_reviewer_pr_lease,
@@ -30,6 +32,13 @@ MERGER_ENV = {
     "GITEA_PROFILE_NAME": "prgs-merger",
     "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment",
 }
+# #745: applying the terminal marker is reconciler-owned.
+RECONCILER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-reconciler",
+    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment,gitea.pr.close",
+}
+CLEANUP_TASK = moot_gate.CLEANUP_TASK
+REPO_SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
 PR = 487
 ISSUE = 485
 SESSION = "97274-676d20a825c4"
@@ -204,6 +213,8 @@ class TestAcquireToolRefusesMergedPR(unittest.TestCase):
 class TestCleanupTool(unittest.TestCase):
     def setUp(self):
         leases.clear_session_lease()
+        moot_gate._reset_for_testing()
+        self.addCleanup(moot_gate._reset_for_testing)
 
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.api_request")
@@ -223,23 +234,53 @@ class TestCleanupTool(unittest.TestCase):
         self.assertEqual(calls["post"], [])
 
     @patch("mcp_server.verify_preflight_purity", return_value=None)
+    @patch("mcp_server._repository_binding_block", return_value=None)
+    @patch("mcp_server._bound_repository_slug", return_value=REPO_SLUG)
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.api_request")
     def test_apply_posts_released_marker_on_merged_pr(
-            self, mock_api, _auth, _purity):
+            self, mock_api, _auth, _slug, _binding, _purity):
+        """#745: apply is reconciler-only and needs a matching dry run first."""
         side, calls = _api_side_effect(
             pr_state="closed", pr_merged=True, comments=[_lease_comment()])
         mock_api.side_effect = side
-        with patch.dict(os.environ, MERGER_ENV, clear=True):
+        with patch.object(mcp_server, "_preflight_resolved_task",
+                          CLEANUP_TASK), \
+                patch.dict(os.environ, RECONCILER_ENV, clear=True):
+            gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
             result = gitea_cleanup_post_merge_moot_lease(
                 pr_number=PR, apply=True, remote="prgs")
-        self.assertTrue(result["success"])
+        self.assertTrue(result["success"], result.get("reasons"))
         self.assertTrue(result["cleanup_performed"])
         self.assertEqual(result["released_comment_id"], 9999)
         self.assertEqual(len(calls["post"]), 1)
         self.assertIn("phase: released", calls["post"][0]["payload"]["body"])
         self.assertIn("post-merge-moot", calls["post"][0]["payload"]["body"])
 
+    @patch("mcp_server.verify_preflight_purity", return_value=None)
+    @patch("mcp_server._repository_binding_block", return_value=None)
+    @patch("mcp_server._bound_repository_slug", return_value=REPO_SLUG)
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    @patch("mcp_server.api_request")
+    def test_merger_can_no_longer_apply(
+            self, mock_api, _auth, _slug, _binding, _purity):
+        """#745: holding gitea.pr.comment is no longer sufficient to apply."""
+        side, calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=[_lease_comment()])
+        mock_api.side_effect = side
+        with patch.object(mcp_server, "_preflight_resolved_task",
+                          CLEANUP_TASK), \
+                patch.dict(os.environ, MERGER_ENV, clear=True):
+            gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs")
+        self.assertFalse(result["success"])
+        self.assertFalse(result["cleanup_performed"])
+        self.assertEqual(result["blocker_kind"], "wrong_role")
+        self.assertEqual(calls["post"], [])
+
     @patch("mcp_server.verify_preflight_purity", return_value=None)
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.api_request")

tests/test_reconciler_close_workspace_guard.py

@@ -38,11 +38,17 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
         srv._preflight_capability_violation = False
         self._orig_in_test = srv._preflight_in_test_mode
         srv._preflight_in_test_mode = lambda: False
+        self._orig_resolved_task = srv._preflight_resolved_task
+        self._orig_resolved_role = srv._preflight_resolved_role
         self._env_patch = patch.dict(os.environ, {"GITEA_MCP_DISABLE_PARITY_GATE": "1"}, clear=False)
         self._env_patch.start()
 
     def tearDown(self):
         srv._preflight_in_test_mode = self._orig_in_test
+        # Preflight task/role are module-level; restore so test order cannot
+        # leak a resolved task into sibling cases.
+        srv._preflight_resolved_task = self._orig_resolved_task
+        srv._preflight_resolved_role = self._orig_resolved_role
         self._env_patch.stop()
 
     @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@@ -81,26 +87,28 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
         "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
         return_value={"current_branch": "master"},
     )
-    def test_author_create_issue_still_blocked_on_control_checkout(
+    def test_author_non_create_issue_still_blocked_on_control_checkout(
         self, _git, _get_all, _role, _ns, _prof, _auth
     ):
+        """Author mutations other than create_issue keep the branches-only rule.
+
+        #749/#750 sanctioned ``create_issue`` from a clean control checkout, and
+        #757 made the #604 anti-stomp guard honour that same decision — so
+        ``create_issue`` is no longer a valid probe for this boundary. This case
+        previously asserted create_issue stayed blocked, which only held because
+        the bootstrap-blind #604 guard was overriding #750; that is precisely
+        the defect #757 fixed. ``lock_issue`` is issue-backed and post-ownership,
+        so it still requires a ``branches/`` worktree.
+        """
         srv._preflight_resolved_role = "author"
+        srv._preflight_resolved_task = "lock_issue"
         with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
             try:
-                res = srv.gitea_create_issue(title="Test", body="body")
+                srv.verify_preflight_purity(remote="prgs", task="lock_issue")
             except RuntimeError as exc:
-                self.assertIn("stable control checkout", str(exc))
+                self.assertIn("control checkout", str(exc).lower())
             else:
-                # #683 typed blocker at mutation entrypoint
-                self.assertFalse(res.get("success"))
-                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.fail("lock_issue must stay blocked on the control checkout")
 
 
 if __name__ == "__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")),
     }