Merge pull request 'feat: add workflow dashboard for queue, leases, and next safe action (Closes #605)' (#762) from feat/issue-605-workflow-dashboard into master

Read-only MCP tool gitea_workflow_dashboard plus mcp-menu entry so operators
and LLMs can see PR/issue queues, leases, terminal locks, blockers, and exact
next-safe prompts without reconstructing state from comments. Never assigns
work or presents blocked/terminal-locked items as safe.

allocator_dependencies.py

@@ -0,0 +1,150 @@
+"""Canonical dependency parsing and live-state resolution for the allocator (#758).
+
+The work allocator previously inferred dependency state from two lowercase
+substrings (``"blocked on #"`` / ``"downstream of #"``). The repository's
+canonical declaration form is a ``Depends:`` field inside the issue body's
+linkage line, for example::
+
+    * Parent: #631 · Depends: #633, #634 · Related: #630, #434
+
+That form matched neither substring, so dependency-blocked issues were emitted
+as eligible candidates. This module replaces substring inference with:
+
+1. structured parsing of ``Depends:`` declarations into issue references, and
+2. resolution of each reference against **live issue state**, never body text.
+
+Both halves fail closed: a reference whose state cannot be established makes
+the owning candidate ineligible rather than assignable.
+
+No issue number is special-cased here (#758 AC4/AC14); the parser is driven
+entirely by the declaration syntax.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Callable, Iterable
+
+# Live issue states, as reported by Gitea.
+DEP_STATE_OPEN = "open"
+DEP_STATE_CLOSED = "closed"
+
+# "Depends:" / "Depends on:" introduces the declaration. "Dependencies" does
+# not match: after "depend" it continues with "e", not "s".
+_DEPENDS_KEYWORD = re.compile(r"depends(?:\s+on)?\s*:?\s*", re.IGNORECASE)
+
+# Immediately after the keyword, consume only the contiguous run of issue
+# references. Anchoring the run this way means the declaration ends naturally
+# at the next separator ("·", newline) or sibling field ("Related:"), without
+# needing to enumerate separators.
+_DEP_RUN = re.compile(
+    r"\s*(#\d+(?:\s*(?:,|and|&)\s*#\d+)*)",
+    re.IGNORECASE,
+)
+
+# Legacy marker retained so previously-recognized bodies keep working.
+_LEGACY_BLOCKED = re.compile(r"blocked\s+on\s+#(\d+)", re.IGNORECASE)
+
+_ISSUE_REF = re.compile(r"#(\d+)")
+
+
+def parse_dependency_refs(body: str | None) -> tuple[int, ...]:
+    """Extract declared dependency issue numbers from an issue *body*.
+
+    Recognizes the canonical ``Depends: #N, #N`` field (including the
+    ``Depends on`` spelling) plus the legacy ``blocked on #N`` marker.
+    Returns references in first-seen order with duplicates removed. Malformed
+    or absent declarations yield an empty tuple rather than raising.
+    """
+    if not body:
+        return ()
+
+    refs: list[int] = []
+
+    def _add(value: str) -> None:
+        number = int(value)
+        if number > 0 and number not in refs:
+            refs.append(number)
+
+    for match in _DEPENDS_KEYWORD.finditer(body):
+        run = _DEP_RUN.match(body, match.end())
+        if not run:
+            continue
+        for ref in _ISSUE_REF.findall(run.group(1)):
+            _add(ref)
+
+    for ref in _LEGACY_BLOCKED.findall(body):
+        _add(ref)
+
+    return tuple(refs)
+
+
+def resolve_dependency_state(
+    refs: Iterable[int],
+    state_lookup: Callable[[int], str | None],
+    *,
+    subject: str = "candidate",
+) -> dict:
+    """Resolve declared *refs* against live issue state.
+
+    *state_lookup* maps an issue number to its live state string, or to
+    ``None`` when that evidence could not be obtained. A reference is:
+
+    * **met** when live state is ``closed``;
+    * **unmet** when live state is any other live value (``open``, etc.);
+    * **unavailable** when state is ``None`` or the lookup raises.
+
+    Unmet *and* unavailable both mark the candidate ineligible (#758 AC6/AC7):
+    allocation must never assume a dependency is satisfied.
+    """
+    unmet: list[int] = []
+    unavailable: list[int] = []
+    met: list[int] = []
+
+    for ref in refs:
+        try:
+            number = int(ref)
+        except (TypeError, ValueError):
+            continue
+        try:
+            state = state_lookup(number)
+        except Exception:  # noqa: BLE001 — unavailable evidence fails closed
+            state = None
+        normalized = (str(state).strip().lower() if state is not None else "") or None
+        if normalized is None:
+            unavailable.append(number)
+        elif normalized == DEP_STATE_CLOSED:
+            met.append(number)
+        else:
+            unmet.append(number)
+
+    reason: str | None = None
+    if unmet and unavailable:
+        reason = (
+            f"{subject} has unresolved dependencies "
+            f"{_fmt(unmet)} and unverifiable dependencies {_fmt(unavailable)} "
+            "(fail closed)"
+        )
+    elif unmet:
+        reason = (
+            f"{subject} depends on unresolved issue(s) {_fmt(unmet)}; "
+            "they are not closed"
+        )
+    elif unavailable:
+        reason = (
+            f"{subject} dependency evidence unavailable for {_fmt(unavailable)} "
+            "(fail closed)"
+        )
+
+    return {
+        "refs": tuple(int(x) for x in refs),
+        "met": tuple(met),
+        "unmet": tuple(unmet),
+        "unavailable": tuple(unavailable),
+        "dependency_unmet": bool(unmet or unavailable),
+        "reason": reason,
+    }
+
+
+def _fmt(numbers: Iterable[int]) -> str:
+    return ", ".join(f"#{n}" for n in numbers)

allocator_service.py

@@ -41,6 +41,15 @@ OUTCOME_NO_SAFE = "no_safe_work"
 OUTCOME_ROLE_INELIGIBLE = "role_ineligible"
 OUTCOME_PREVIEW = "preview"  # dry-run only (apply=false)
 
+# Human-readable statement of how a winner is chosen (#758 AC10). Reported
+# alongside allocator results so the flat status:ready tier and its
+# oldest-number tie-break are explicit rather than incidental.
+SELECTION_POLICY = (
+    "rank complete inventory by (priority desc, PRs before issues, "
+    "number asc); status:ready issues share priority 20, so the oldest "
+    "eligible number wins ties; result limits never affect selection"
+)
+
 ROLE_AUTHOR = "author"
 ROLE_REVIEWER = "reviewer"
 ROLE_MERGER = "merger"
@@ -242,7 +251,23 @@ def classify_skip(
 
 
 def sort_candidates(candidates: Sequence[WorkCandidate]) -> list[WorkCandidate]:
-    """Higher priority first; then lower number (older issues) for stability."""
+    """Rank candidates deterministically (#758 AC10).
+
+    Ordering key, in precedence order:
+
+    1. ``priority`` descending — the loader scores ``status:ready`` issues at
+       20 and everything else at 1, so the ready queue ties at a single value
+       by design;
+    2. PRs before issues — in-flight review work drains before new authoring;
+    3. ``number`` ascending — oldest first, which is what actually breaks the
+       flat ``status:ready`` tie.
+
+    Because the ready tier is intentionally flat, rule 3 decides most real
+    selections. That is only safe when ranking sees the *complete* candidate
+    inventory: truncating before this call silently redefines "oldest" as
+    "oldest among whatever survived the slice", which is the defect #758
+    fixed. Callers must rank everything and bound reporting afterwards.
+    """
     return sorted(
         candidates,
         key=lambda c: (-int(c.priority), c.kind != "pr", int(c.number)),

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

@@ -52,6 +52,18 @@ def is_create_issue_task(task: str | None) -> bool:
     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,
@@ -60,6 +72,7 @@ def assess_create_issue_bootstrap(
     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.
@@ -142,8 +155,32 @@ def assess_create_issue_bootstrap(
             f"is not an accepted base branch ({'/'.join(sorted(BASE_BRANCHES))})"
         )
 
-    remote_tip = (remote_master_sha or "").strip() or None
+    # #757 AC3/AC4: base-equivalence must be *proven*, never assumed. An
-    local_tip = (head_sha or "").strip() or None
+    # 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 "
@@ -162,6 +199,8 @@ def assess_create_issue_bootstrap(
             dirty=dirty,
             under_branches=False,
             exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP,
+            local_head_sha=local_tip,
+            remote_master_sha=remote_tip,
         )
 
     return _result(
@@ -176,9 +215,92 @@ def assess_create_issue_bootstrap(
         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(
@@ -209,7 +331,14 @@ def _result(
     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,
@@ -224,4 +353,7 @@ def _result(
         "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/mcp-menu.md

@@ -40,6 +40,7 @@ The script must be executable (`chmod +x mcp-menu.sh`). It uses bash with
 | Option | Description |
 |--------|-------------|
 | Project status / root checkout health | Shows cwd, branch, `git status --short --branch`, HEAD SHA, `prgs/master` SHA, and warnings when the root checkout is dirty or off `master`. |
+| Workflow dashboard (queue, leases, next safe action) | Documents the read-only `gitea_workflow_dashboard` MCP tool (#605): live PR/issue queues, leases by role, terminal review lock, blocked items, and exact next-safe prompts. **Does not assign work** — assignment still uses `gitea_allocate_next_work`. Never presents blocked/terminal-locked items as safe. The shell entry is documentation only (no Gitea mutation). |
 | Author workflow prompts | Ready-to-copy prompts for issue work, conflict-fix sessions, and root checkout recovery. |
 | Reviewer workflow prompts | Standard PR review prompt, and a skip-already-reviewed-stale-`REQUEST_CHANGES` prompt that hands off to the author without a duplicate terminal mutation (review-only; no merge). |
 | Merger workflow prompts | PR merge prompt (merge gates and explicit approval). |
@@ -50,6 +51,22 @@ The script must be executable (`chmod +x mcp-menu.sh`). It uses bash with
 | Run tests | Runs `./run-tests.sh` when present; otherwise `venv/bin/python -m pytest`; otherwise fails closed with a clear error. |
 | Exit | Quit the menu. |
 
+### Workflow dashboard MCP tool (#605)
+
+From any healthy Gitea MCP namespace with `gitea.read`:
+
+```text
+gitea_workflow_dashboard(
+    remote="prgs",
+    org="Scaled-Tech-Consulting",
+    repo="Gitea-Tools",
+)
+```
+
+Response includes `human_summary` plus structured queues, `active_leases_by_role`,
+`terminal_review_lock`, `blocked_items`, `next_safe_by_role`, and
+`primary_next_safe_action`. Incomplete inventory fails closed.
+
 ## Placeholder-only entries
 
 **Proxmox deployment** and **Create Proxmox LXC** are placeholders until

gitea_mcp_server.py

@@ -701,6 +701,18 @@ def _clear_preflight_capability_state() -> None:
     _preflight_reviewer_violation_files = []
 
 
+def _invalidate_preflight_identity_state() -> None:
+    """Fail closed when whoami cannot prove the configured identity."""
+    global _preflight_whoami_called, _preflight_whoami_violation
+    global _preflight_whoami_baseline_porcelain, _preflight_whoami_violation_files
+
+    _preflight_whoami_called = False
+    _preflight_whoami_violation = False
+    _preflight_whoami_baseline_porcelain = None
+    _preflight_whoami_violation_files = []
+    _clear_preflight_capability_state()
+
+
 def record_preflight_check(
     type_name: str,
     resolved_role: str | None = None,
@@ -817,10 +829,62 @@ def _enforce_canonical_repository_root(
         )
 
 
+# #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.
 
@@ -859,27 +923,29 @@ def _enforce_branches_only_author_mutation(
         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
 
-    remote_master_sha = None
+    bootstrap = (
-    try:
+        _create_issue_bootstrap_assessment(task, worktree_path)
-        remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
+        if bootstrap_assessment is _BOOTSTRAP_UNSET
-            ctx["canonical_repo_root"]
+        else bootstrap_assessment
-        )
+    )
-    except Exception:
+    if _cib.bootstrap_permits_control_checkout(
-        remote_master_sha = None
+        bootstrap,
-    bootstrap = _cib.assess_create_issue_bootstrap(
+        task=task,
         workspace_path=workspace,
         canonical_repo_root=ctx["canonical_repo_root"],
-        current_branch=git_state.get("current_branch"),
+    ):
-        head_sha=git_state.get("head_sha"),
-        porcelain_status=git_state.get("porcelain_status") or "",
-        remote_master_sha=remote_master_sha,
-        task=task,
-    )
-    if bootstrap.get("allowed"):
         return
-    if bootstrap.get("block") and not bootstrap.get("not_applicable"):
+    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)
@@ -926,6 +992,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.
 
@@ -1046,6 +1113,8 @@ def _run_anti_stomp_preflight(
 
     # Workflow-hash facts when the task is review/merge oriented.
     if workflow_hash_valid is None and task in {
+        "acquire_reviewer_pr_lease",
+        "gitea_acquire_reviewer_pr_lease",
         "review_pr",
         "submit_pr_review",
         "approve_pr",
@@ -1106,6 +1175,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
@@ -1173,7 +1250,9 @@ def verify_preflight_purity(
         if (
             task is not None
             and _preflight_resolved_task is not None
-            and task != _preflight_resolved_task
+            and not task_capability_map.preflight_task_matches(
+                _preflight_resolved_task, task
+            )
         ):
             raise _PreflightOrderError(
                 "Pre-flight task mismatch: "
@@ -1248,7 +1327,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, task=task)
+        # #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,
@@ -1258,6 +1342,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,
@@ -1282,7 +1367,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, task=task)
+        # #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,
@@ -1292,6 +1381,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,
@@ -1691,8 +1781,10 @@ import review_workflow_load  # noqa: E402
 import mcp_session_state  # noqa: E402
 import stale_review_decision_lock  # noqa: E402
 import allocator_service  # noqa: E402
+import allocator_dependencies  # noqa: E402
 import control_plane_db  # noqa: E402
 import lease_lifecycle  # noqa: E402
+import workflow_dashboard  # noqa: E402  # #605 live queue/lease dashboard
 import incident_bridge  # noqa: E402
 import sentry_observability  # noqa: E402  (#606 optional Sentry observability)
 import agent_temp_artifacts
@@ -1880,6 +1972,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
@@ -9019,13 +9112,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
@@ -9058,7 +9152,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",
         }
@@ -9077,9 +9171,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",
@@ -9089,17 +9183,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,
@@ -12528,6 +12656,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,
@@ -12543,14 +12675,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.
+        org: Override the owner/organization. Validated against the canonical
-        repo: Override the repository name.
+            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
@@ -12610,14 +12765,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
@@ -12625,7 +12828,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(
@@ -13955,6 +14179,8 @@ def gitea_whoami(
         "identity_match": not id_match.get("block"),
         "identity_match_reasons": id_match.get("reasons") or [],
     }
+    if id_match.get("block"):
+        _invalidate_preflight_identity_state()
     if _reveal_endpoints():
         result["server"] = gitea_url(h, "").rstrip("/")
     return result
@@ -17523,7 +17749,12 @@ def gitea_resolve_task_capability(
         result["stale_binding_recovery"] = stale_binding
     if reason_msg:
         result["reason"] = reason_msg
-    if task in ("review_pr", "merge_pr"):
+    if task in (
+        "acquire_reviewer_pr_lease",
+        "gitea_acquire_reviewer_pr_lease",
+        "review_pr",
+        "merge_pr",
+    ):
         result["workflow_load_proof"] = review_workflow_load.workflow_load_status(
             PROJECT_ROOT)
         if not result["workflow_load_proof"].get("workflow_load_valid"):
@@ -17546,6 +17777,10 @@ def gitea_resolve_task_capability(
         )
     elif was_terminal and not (stop_required or restart_required):
         result["cleared_stale_denial"] = True
+    if stop_required or restart_required:
+        # A denied or stale resolver result is diagnostic evidence, never a
+        # consumable capability token for a later mutation (#763).
+        _clear_preflight_capability_state()
     return result
 
 
@@ -17588,16 +17823,24 @@ def _allocator_candidates_from_gitea(
     repo: str,
     include_issues: bool = True,
     include_prs: bool = True,
-    limit: int = 50,
+) -> tuple[list[Any], list[str], bool]:
-) -> tuple[list[Any], list[str]]:
+    """Build allocator candidates from live Gitea open issues/PRs.
-    """Build allocator candidates from live Gitea open issues/PRs."""
+
+    Returns ``(candidates, reasons, inventory_complete)``. The inventory is
+    assembled in full and is never truncated before ranking (#758 AC1/AC3):
+    any result/display bound belongs to the caller's reporting, not to
+    candidate construction. ``inventory_complete`` is False when a required
+    listing failed, so the caller can fail closed instead of ranking a
+    silently short candidate set.
+    """
     reasons: list[str] = []
     candidates: list[Any] = []
+    inventory_complete = True
     try:
         h, o, r = _resolve(remote, host, org, repo)
         auth = _auth(h)
     except Exception as exc:  # noqa: BLE001
-        return [], [f"failed to resolve Gitea target: {_redact(str(exc))}"]
+        return [], [f"failed to resolve Gitea target: {_redact(str(exc))}"], False
 
     if include_prs:
         try:
@@ -17607,7 +17850,8 @@ def _allocator_candidates_from_gitea(
         except Exception as exc:  # noqa: BLE001
             reasons.append(f"failed to list open PRs: {_redact(str(exc))}")
             prs = []
-        for pr in prs[: max(1, int(limit))]:
+            inventory_complete = False
+        for pr in prs:
             if not isinstance(pr, dict):
                 continue
             number = pr.get("number")
@@ -17679,7 +17923,39 @@ def _allocator_candidates_from_gitea(
                     f"failed to list open issues: {_redact(str(exc2))}"
                 )
                 issues = []
-        for issue in issues[: max(1, int(limit))]:
+                inventory_complete = False
+
+        # Live dependency evidence (#758 AC5). The complete open-issue listing
+        # already proves which references are still open; anything absent from
+        # it is confirmed by a targeted lookup rather than assumed closed, so a
+        # deleted or unreachable reference fails closed instead of passing.
+        open_issue_numbers = {
+            int(i["number"])
+            for i in issues
+            if isinstance(i, dict)
+            and i.get("number") is not None
+            and i.get("pull_request") is None
+        }
+        dep_state_cache: dict[int, str | None] = {}
+
+        def _issue_state(number: int) -> str | None:
+            if number in open_issue_numbers:
+                return allocator_dependencies.DEP_STATE_OPEN
+            if number in dep_state_cache:
+                return dep_state_cache[number]
+            state: str | None = None
+            try:
+                data = api_request(
+                    "GET", f"{repo_api_url(h, o, r)}/issues/{number}", auth
+                )
+                if isinstance(data, dict):
+                    state = str(data.get("state") or "").strip().lower() or None
+            except Exception:  # noqa: BLE001 — unavailable evidence fails closed
+                state = None
+            dep_state_cache[number] = state
+            return state
+
+        for issue in issues:
             if not isinstance(issue, dict):
                 continue
             # Pull requests also appear in /issues on Gitea — skip them.
@@ -17697,21 +17973,15 @@ def _allocator_candidates_from_gitea(
             body = str(issue.get("body") or "")
             title = str(issue.get("title") or "")
             blocked = "status:blocked" in labels
-            # Explicit downstream dependency: #612 waits on #600 allocator.
+            # #758: parse the canonical "Depends: #N, #N" declaration into
-            dep_unmet = False
+            # structured references and resolve each against live issue state.
-            dep_reason = None
+            # Body substrings no longer decide dependency status.
-            # Generic body markers for blocked-on unfinished deps.
+            dep_refs = allocator_dependencies.parse_dependency_refs(body)
-            # (Hard-coded #612→#600 block removed after #600 merged — #612.)
+            dep_result = allocator_dependencies.resolve_dependency_state(
-            lower_body = body.lower()
+                dep_refs, _issue_state, subject=f"issue#{number}"
-            if "blocked on #" in lower_body or "downstream of #" in lower_body:
+            )
-                # Only treat as unmet when body still marks a live open dependency
+            dep_unmet = dep_result["dependency_unmet"]
-                # pattern; callers may clear labels when deps complete.
+            dep_reason = dep_result["reason"]
-                dep_unmet = "blocked on #" in lower_body
-                dep_reason = (
-                    f"issue#{number} body marks an unmet dependency"
-                    if dep_unmet
-                    else None
-                )
             try:
                 candidates.append(
                     allocator_service.WorkCandidate(
@@ -17731,7 +18001,7 @@ def _allocator_candidates_from_gitea(
                     f"skipped invalid issue candidate #{number}: {_redact(str(exc))}"
                 )
 
-    return candidates, reasons
+    return candidates, reasons, inventory_complete
 
 
 @mcp.tool()
@@ -18047,15 +18317,32 @@ def gitea_allocate_next_work(
                 "substrate": "control_plane_db",
             }
     else:
-        candidates, inv_reasons = _allocator_candidates_from_gitea(
+        candidates, inv_reasons, inventory_complete = _allocator_candidates_from_gitea(
             remote=remote,
             host=host,
             org=o,
             repo=r,
             include_issues=include_issues,
             include_prs=include_prs,
-            limit=limit,
         )
+        # #758 AC3: ranking a silently short inventory could select the wrong
+        # candidate, so an incomplete listing is a fail-closed stop.
+        if not inventory_complete:
+            return {
+                "success": False,
+                "outcome": allocator_service.OUTCOME_NO_SAFE,
+                "apply": bool(apply),
+                "reasons": inv_reasons
+                + [
+                    "candidate inventory is incomplete; refusing to rank a "
+                    "partial candidate set (fail closed, #758)"
+                ],
+                "inventory_complete": False,
+                "assignment": None,
+                "substrate": "control_plane_db",
+                "file_lock_only": False,
+                "comment_lease_only": False,
+            }
 
     sid = (session_id or "").strip() or (
         f"{profile_name or 'session'}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
@@ -18078,6 +18365,24 @@ def gitea_allocate_next_work(
     result["inventory_source"] = (
         "candidates_json" if candidates_json else "gitea_live"
     )
+    result["inventory_complete"] = True
+    result["selection_policy"] = allocator_service.SELECTION_POLICY
+    # #758 AC2/AC10: `limit` bounds the reported skip list only — selection has
+    # already happened over the complete inventory above. Any truncation here is
+    # stated explicitly so a shortened report is never read as full coverage.
+    reported_limit = max(1, int(limit))
+    skipped_all = result.get("skipped") or []
+    result["skipped_total"] = len(skipped_all)
+    result["limit_applies_to"] = "reported_skip_list_only"
+    if len(skipped_all) > reported_limit:
+        result["skipped"] = skipped_all[:reported_limit]
+        result["skipped_report_truncated"] = True
+        result.setdefault("reasons", []).append(
+            f"skip list truncated for reporting: showing {reported_limit} of "
+            f"{len(skipped_all)} skipped candidates (selection unaffected)"
+        )
+    else:
+        result["skipped_report_truncated"] = False
     # #606: watchdog check-ins for the recurring jobs this allocator run
     # performs — global stale-lease expiry, terminal-lock lookup, and the
     # allocator itself. Best-effort; a failed selection reports "error".
@@ -18143,6 +18448,141 @@ def gitea_list_workflow_leases(
     return result
 
 
+@mcp.tool()
+def gitea_workflow_dashboard(
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    include_issues: bool = True,
+    include_prs: bool = True,
+    include_non_active_leases: bool = True,
+    candidates_json: str | None = None,
+    limit: int = 100,
+) -> dict:
+    """Read-only MCP workflow dashboard: queues, leases, blockers, next safe action (#605).
+
+    Machine-readable and human-readable. Never assigns work — exclusive work
+    still requires ``gitea_allocate_next_work``. Never presents blocked or
+    terminal-locked items as safe next work.
+
+    *candidates_json* may inject a JSON list of candidate dicts (tests /
+    offline fixtures). When omitted, open issues/PRs are loaded from Gitea.
+    Incomplete live inventory fails closed (no safe suggestions).
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "read_only": True,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+            "primary_next_safe_action": None,
+        }
+
+    try:
+        h, o, r = _resolve(remote, host, org, repo)
+    except ValueError as exc:
+        return {
+            "success": False,
+            "read_only": True,
+            "reasons": [str(exc)],
+            "primary_next_safe_action": None,
+        }
+
+    inv_reasons: list[str] = []
+    candidates: list[Any] = []
+    inventory_complete = True
+    if candidates_json:
+        try:
+            raw = json.loads(candidates_json)
+            if not isinstance(raw, list):
+                raise ValueError("candidates_json must be a JSON list")
+            for item in raw:
+                if not isinstance(item, dict):
+                    continue
+                candidates.append(allocator_service.candidate_from_dict(item))
+        except Exception as exc:  # noqa: BLE001
+            return {
+                "success": False,
+                "read_only": True,
+                "inventory_complete": False,
+                "reasons": [
+                    f"invalid candidates_json: {_redact(str(exc))} (fail closed)"
+                ],
+                "primary_next_safe_action": None,
+            }
+    else:
+        candidates, inv_reasons, inventory_complete = _allocator_candidates_from_gitea(
+            remote=remote,
+            host=host,
+            org=o,
+            repo=r,
+            include_issues=include_issues,
+            include_prs=include_prs,
+        )
+
+    leases: list[dict] = []
+    terminal_pr: int | None = None
+    terminal_lock: dict | None = None
+    db, db_errs = _control_plane_db_or_error()
+    if db is None:
+        inv_reasons.extend(db_errs or ["control-plane DB unavailable"])
+        # Leases/terminal lock unavailable — still render queues fail-closed
+        # for safe suggestions when inventory is incomplete; when inventory is
+        # complete, surface lease gap in reasons without inventing leases.
+        inv_reasons.append(
+            "control-plane DB unavailable; lease/terminal sections empty"
+        )
+    else:
+        try:
+            lease_result = lease_lifecycle.list_active_leases(
+                db,
+                remote=remote if remote in REMOTES else remote,
+                org=o,
+                repo=r,
+                role=None,
+                include_non_active=bool(include_non_active_leases),
+                limit=max(1, int(limit)),
+            )
+            leases = list(lease_result.get("leases") or [])
+        except Exception as exc:  # noqa: BLE001
+            inv_reasons.append(
+                f"lease listing failed: {_redact(str(exc))}"
+            )
+        try:
+            terminal = db.get_active_terminal_lock(
+                remote=remote if remote in REMOTES else remote,
+                org=o,
+                repo=r,
+            )
+            if terminal:
+                terminal_lock = dict(terminal)
+                terminal_pr = int(terminal["terminal_pr"])
+        except Exception as exc:  # noqa: BLE001
+            inv_reasons.append(
+                f"terminal lock lookup failed: {_redact(str(exc))}"
+            )
+
+    snap = workflow_dashboard.build_workflow_dashboard(
+        candidates=candidates,
+        remote=remote if remote in REMOTES else remote,
+        org=o,
+        repo=r,
+        leases=leases,
+        terminal_pr=terminal_pr,
+        terminal_lock=terminal_lock,
+        inventory_complete=inventory_complete,
+        inventory_reasons=inv_reasons,
+    )
+    payload = snap.as_dict()
+    payload["inventory_source"] = (
+        "candidates_json" if candidates_json else "gitea_live"
+    )
+    payload["limit_applies_to"] = "lease_listing_only"
+    return payload
+
+
 @mcp.tool()
 def gitea_inspect_workflow_lease(
     lease_id: str,

mcp-menu.sh

@@ -19,6 +19,32 @@ print_banner() {
   printf 'Safe by default — destructive actions require explicit confirmation.\n\n'
 }
 
+show_workflow_dashboard_help() {
+  printf '\n--- Workflow dashboard (queue, leases, next safe action) ---\n\n'
+  printf 'Read-only operational view (#605). Does NOT assign work.\n'
+  printf 'Exclusive assignment still requires gitea_allocate_next_work.\n\n'
+  printf 'Canonical MCP tool (any healthy Gitea namespace with gitea.read):\n\n'
+  printf '  gitea_workflow_dashboard(\n'
+  printf '      remote=\"prgs\",\n'
+  printf '      org=\"Scaled-Tech-Consulting\",\n'
+  printf '      repo=\"Gitea-Tools\",\n'
+  printf '  )\n\n'
+  printf 'Returns machine-readable sections:\n'
+  printf '  - open_pr_queue / open_issue_queue\n'
+  printf '  - active_leases_by_role / stale_or_expired_leases\n'
+  printf '  - terminal_review_lock\n'
+  printf '  - blocked_items (never presented as safe)\n'
+  printf '  - review_ready_prs / merge_ready_prs / author_remediation\n'
+  printf '  - discussion_issues / controller_needed\n'
+  printf '  - next_safe_by_role + primary_next_safe_action with exact prompts\n'
+  printf '  - human_summary (copy-friendly multi-line text)\n\n'
+  printf 'Safety:\n'
+  printf '  - Never suggests blocked or terminal-locked items as safe.\n'
+  printf '  - Incomplete inventory fails closed (no safe suggestions).\n'
+  printf '  - This menu entry is documentation only; it does not call Gitea.\n'
+  pause
+}
+
 show_root_checkout_health() {
   printf '\n--- Project status / root checkout health ---\n\n'
   printf 'Current directory: %s\n' "$(pwd)"
@@ -241,25 +267,27 @@ main_menu() {
   while true; do
     print_banner
     printf '  1) Project status / root checkout health\n'
-    printf '  2) Author workflow prompts\n'
+    printf '  2) Workflow dashboard (queue, leases, next safe action)\n'
-    printf '  3) Reviewer workflow prompts\n'
+    printf '  3) Author workflow prompts\n'
-    printf '  4) Merger workflow prompts\n'
+    printf '  4) Reviewer workflow prompts\n'
-    printf '  5) Reconciler workflow prompts\n'
+    printf '  5) Merger workflow prompts\n'
-    printf '  6) Onboarding new project to this MCP workflow\n'
+    printf '  6) Reconciler workflow prompts\n'
-    printf '  7) Proxmox deployment menu placeholder\n'
+    printf '  7) Onboarding new project to this MCP workflow\n'
-    printf '  8) Create Proxmox LXC placeholder\n'
+    printf '  8) Proxmox deployment menu placeholder\n'
-    printf '  9) Run tests\n'
+    printf '  9) Create Proxmox LXC placeholder\n'
+    printf '  t) Run tests\n'
     printf '  0) Exit\n'
     read -r -p 'Choice: ' choice
     case "$choice" in
       1) show_root_checkout_health ;;
-      2) show_author_prompts ;;
+      2) show_workflow_dashboard_help ;;
-      3) show_reviewer_prompts ;;
+      3) show_author_prompts ;;
-      4) show_merger_prompts ;;
+      4) show_reviewer_prompts ;;
-      5) show_reconciler_prompts ;;
+      5) show_merger_prompts ;;
-      6) show_onboarding_prompt ;;
+      6) show_reconciler_prompts ;;
-      7|8) show_proxmox_placeholder ;;
+      7) show_onboarding_prompt ;;
-      9) run_tests ;;
+      8|9) show_proxmox_placeholder ;;
+      t|T|tests) run_tests ;;
       0) printf 'Goodbye.\n'; exit 0 ;;
       *) printf 'Invalid choice.\n'; pause ;;
     esac

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",

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",
@@ -415,6 +432,36 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
     },
 }
 
+
+# A reviewer lease is the first mutation in the canonical ``review_pr``
+# workflow, so the already-resolved review capability is valid for that one
+# narrower transition. Keep this directed and explicit: lease acquisition
+# does not authorize a review verdict, and reviewer proof never authorizes a
+# merger lease (#763).
+_PREFLIGHT_TASK_TRANSITIONS = frozenset({
+    ("review_pr", "acquire_reviewer_pr_lease"),
+})
+
+
+def _canonical_preflight_task(task: str | None) -> str:
+    """Normalize only declared ``gitea_`` aliases for preflight comparison."""
+    value = (task or "").strip()
+    if value.startswith("gitea_") and value[6:] in TASK_CAPABILITY_MAP:
+        return value[6:]
+    return value
+
+
+def preflight_task_matches(
+    resolved_task: str | None,
+    mutation_task: str | None,
+) -> bool:
+    """Return whether capability proof authorizes this mutation transition."""
+    resolved = _canonical_preflight_task(resolved_task)
+    mutation = _canonical_preflight_task(mutation_task)
+    if not resolved or not mutation:
+        return False
+    return resolved == mutation or (resolved, mutation) in _PREFLIGHT_TASK_TRANSITIONS
+
 # Issue-mutating MCP tools and their resolver task keys.
 ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
     "gitea_create_issue": "create_issue",

tests/test_allocator_dependencies.py

@@ -0,0 +1,266 @@
+"""Dependency parsing/resolution and allocator completeness tests (#758).
+
+Covers the two defects behind #758:
+
+* Defect 1 — candidate truncation before ranking, which let a result-size
+  parameter change the winner.
+* Defect 2 — dependency state inferred from body substrings, which emitted
+  canonical ``Depends:`` blocked issues as eligible.
+
+No production behavior is special-cased for any issue number (#758 AC14), so
+these tests use synthetic issue numbers throughout.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+
+import allocator_dependencies
+from allocator_service import (
+    OUTCOME_PREVIEW,
+    SELECTION_POLICY,
+    WorkCandidate,
+    allocate_next_work,
+    classify_skip,
+    sort_candidates,
+)
+from control_plane_db import ControlPlaneDB
+
+# The canonical linkage line this repository writes into issue bodies.
+CANONICAL_BODY = (
+    "## Dependencies and linkage\n\n"
+    "* Parent: #900 · Depends: #901, #902 · Related: #903, #904\n"
+)
+
+
+class ParseDependencyRefsTest(unittest.TestCase):
+    def test_parses_canonical_depends_field(self) -> None:
+        self.assertEqual(
+            allocator_dependencies.parse_dependency_refs(CANONICAL_BODY),
+            (901, 902),
+        )
+
+    def test_stops_at_sibling_field_and_ignores_related(self) -> None:
+        """"Related:" refs must never be treated as dependencies."""
+        refs = allocator_dependencies.parse_dependency_refs(CANONICAL_BODY)
+        self.assertNotIn(903, refs)
+        self.assertNotIn(904, refs)
+        self.assertNotIn(900, refs)  # Parent is not a dependency
+
+    def test_single_reference(self) -> None:
+        body = "* Parent: #10 · Depends: #11 · Related: #12"
+        self.assertEqual(allocator_dependencies.parse_dependency_refs(body), (11,))
+
+    def test_depends_on_spelling_and_and_separator(self) -> None:
+        body = "Depends on #21 and #22\n"
+        self.assertEqual(
+            allocator_dependencies.parse_dependency_refs(body), (21, 22)
+        )
+
+    def test_newline_terminates_declaration(self) -> None:
+        body = "Depends: #31, #32\nRelated: #33\n"
+        self.assertEqual(
+            allocator_dependencies.parse_dependency_refs(body), (31, 32)
+        )
+
+    def test_legacy_blocked_on_marker_still_recognized(self) -> None:
+        body = "This work is blocked on #41 until that lands.\n"
+        self.assertEqual(allocator_dependencies.parse_dependency_refs(body), (41,))
+
+    def test_dependencies_heading_alone_is_not_a_declaration(self) -> None:
+        """"Dependencies and linkage" must not parse as "Depends"."""
+        body = "## Dependencies and linkage\n\n* Related: #51\n"
+        self.assertEqual(allocator_dependencies.parse_dependency_refs(body), ())
+
+    def test_deduplicates_and_preserves_order(self) -> None:
+        body = "Depends: #61, #62, #61\n"
+        self.assertEqual(
+            allocator_dependencies.parse_dependency_refs(body), (61, 62)
+        )
+
+    def test_malformed_and_empty_inputs(self) -> None:
+        for body in ("", None, "Depends:", "Depends: none", "Depends: TBD\n"):
+            self.assertEqual(allocator_dependencies.parse_dependency_refs(body), ())
+
+
+class ResolveDependencyStateTest(unittest.TestCase):
+    def test_open_dependency_is_unmet(self) -> None:
+        result = allocator_dependencies.resolve_dependency_state(
+            (901, 902), lambda n: "open", subject="issue#644"
+        )
+        self.assertTrue(result["dependency_unmet"])
+        self.assertEqual(result["unmet"], (901, 902))
+        self.assertIn("#901", result["reason"])
+
+    def test_closed_dependencies_are_met(self) -> None:
+        result = allocator_dependencies.resolve_dependency_state(
+            (901, 902), lambda n: "closed"
+        )
+        self.assertFalse(result["dependency_unmet"])
+        self.assertEqual(result["met"], (901, 902))
+        self.assertIsNone(result["reason"])
+
+    def test_mixed_open_and_closed_is_unmet(self) -> None:
+        states = {901: "closed", 902: "open"}
+        result = allocator_dependencies.resolve_dependency_state(
+            (901, 902), states.get
+        )
+        self.assertTrue(result["dependency_unmet"])
+        self.assertEqual(result["unmet"], (902,))
+        self.assertEqual(result["met"], (901,))
+
+    def test_unavailable_evidence_fails_closed(self) -> None:
+        """AC7: unknown state must block, never pass."""
+        result = allocator_dependencies.resolve_dependency_state(
+            (901,), lambda n: None
+        )
+        self.assertTrue(result["dependency_unmet"])
+        self.assertEqual(result["unavailable"], (901,))
+        self.assertIn("fail closed", result["reason"])
+
+    def test_raising_lookup_fails_closed(self) -> None:
+        def boom(_n: int) -> str:
+            raise RuntimeError("lookup exploded")
+
+        result = allocator_dependencies.resolve_dependency_state((901,), boom)
+        self.assertTrue(result["dependency_unmet"])
+        self.assertEqual(result["unavailable"], (901,))
+
+    def test_no_refs_is_eligible(self) -> None:
+        result = allocator_dependencies.resolve_dependency_state((), lambda n: None)
+        self.assertFalse(result["dependency_unmet"])
+        self.assertIsNone(result["reason"])
+
+
+class DependencyBlockedCandidateTest(unittest.TestCase):
+    """A dependency-blocked candidate must be skipped, not selected."""
+
+    def test_classify_skip_rejects_unmet_dependency(self) -> None:
+        candidate = WorkCandidate(
+            kind="issue",
+            number=644,
+            labels=("status:ready",),
+            priority=20,
+            dependency_unmet=True,
+            dependency_reason="issue#644 depends on unresolved issue(s) #633",
+        )
+        reason = classify_skip(candidate, role="author", terminal_pr=None)
+        self.assertIsNotNone(reason)
+        self.assertIn("#633", reason)
+
+
+class SelectionInvarianceTest(unittest.TestCase):
+    """AC1/AC2/AC11: ranking sees everything; result bounds cannot move the winner."""
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    @staticmethod
+    def _ready_issue(number: int, **kw) -> WorkCandidate:
+        return WorkCandidate(
+            kind="issue",
+            number=number,
+            labels=("status:ready",),
+            priority=20,
+            title=f"issue {number}",
+            **kw,
+        )
+
+    def _preview(self, candidates):
+        return allocate_next_work(
+            self.db,
+            session_id="s-758",
+            role="author",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            candidates=candidates,
+            apply=False,
+        )
+
+    def test_more_than_fifty_candidates_lowest_number_wins(self) -> None:
+        """Winner is the oldest eligible issue across a >50 inventory."""
+        candidates = [self._ready_issue(n) for n in range(600, 700)]  # 100 items
+        result = self._preview(candidates)
+        self.assertEqual(result["outcome"], OUTCOME_PREVIEW)
+        self.assertEqual(result["selected"]["number"], 600)
+
+    def test_selection_is_invariant_to_candidate_ordering(self) -> None:
+        """Ranking must not depend on the order the inventory arrived in."""
+        forward = [self._ready_issue(n) for n in range(600, 700)]
+        reverse = list(reversed(forward))
+        self.assertEqual(
+            self._preview(forward)["selected"]["number"],
+            self._preview(reverse)["selected"]["number"],
+        )
+
+    def test_truncating_inventory_changes_winner(self) -> None:
+        """Regression guard: this is exactly what pre-ranking slicing did.
+
+        A 50-item slice of a 100-item inventory yields a different winner, so
+        any future reintroduction of pre-ranking truncation is detectable.
+        """
+        full = [self._ready_issue(n) for n in range(600, 700)]
+        sliced = sorted(full, key=lambda c: -c.number)[:50]
+        self.assertNotEqual(
+            self._preview(full)["selected"]["number"],
+            self._preview(sliced)["selected"]["number"],
+        )
+
+    def test_blocked_first_candidate_falls_through_to_next(self) -> None:
+        """AC8: a blocked winner must not end the iteration."""
+        blocked = self._ready_issue(
+            600,
+            dependency_unmet=True,
+            dependency_reason="issue#600 depends on unresolved issue(s) #599",
+        )
+        result = self._preview([blocked, self._ready_issue(601)])
+        self.assertEqual(result["selected"]["number"], 601)
+        skipped = {s["number"] for s in result["skipped"]}
+        self.assertIn(600, skipped)
+
+    def test_all_blocked_yields_no_safe_work(self) -> None:
+        candidates = [
+            self._ready_issue(
+                n, dependency_unmet=True, dependency_reason=f"issue#{n} blocked"
+            )
+            for n in range(600, 605)
+        ]
+        result = self._preview(candidates)
+        self.assertIsNone(result["selected"])
+        self.assertEqual(len(result["skipped"]), 5)
+
+    def test_dry_run_and_apply_select_identically(self) -> None:
+        """AC9: apply mode must not re-rank differently from preview."""
+        candidates = [self._ready_issue(n) for n in range(600, 700)]
+        preview = self._preview(candidates)
+        applied = allocate_next_work(
+            self.db,
+            session_id="s-758-apply",
+            role="author",
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            candidates=candidates,
+            apply=True,
+        )
+        self.assertEqual(
+            preview["selected"]["number"], applied["selected"]["number"]
+        )
+
+    def test_sort_is_stable_and_documented(self) -> None:
+        ordered = sort_candidates(
+            [self._ready_issue(603), self._ready_issue(601), self._ready_issue(602)]
+        )
+        self.assertEqual([c.number for c in ordered], [601, 602, 603])
+        self.assertIn("never affect selection", SELECTION_POLICY)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_allocator_inventory_mcp.py

@@ -0,0 +1,227 @@
+"""MCP-level allocator inventory and dependency regressions (#758).
+
+Exercises ``_allocator_candidates_from_gitea`` and the ``gitea_allocate_next_work``
+tool end to end against a faked Gitea API, proving:
+
+* the complete open-issue inventory is ranked (no pre-ranking truncation);
+* ``limit`` cannot change which candidate wins;
+* canonical ``Depends:`` declarations are resolved from live issue state;
+* unavailable dependency evidence fails closed;
+* an incomplete listing fails closed instead of ranking a partial set.
+
+Issue numbers here are synthetic; no production number is special-cased.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import gitea_mcp_server as srv
+from control_plane_db import ControlPlaneDB
+
+FAKE_AUTH = "token REDACTED"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+
+def _issue(number: int, *, body: str = "", labels=("status:ready",)) -> dict:
+    return {
+        "number": number,
+        "title": f"issue {number}",
+        "body": body,
+        "labels": [{"name": name} for name in labels],
+        "state": "open",
+    }
+
+
+def _depends_body(*refs: int) -> str:
+    joined = ", ".join(f"#{r}" for r in refs)
+    return f"## Dependencies and linkage\n\n* Parent: #999 · Depends: {joined}\n"
+
+
+class _FakeGitea:
+    """Minimal stand-in for the two Gitea list endpoints plus issue lookups."""
+
+    def __init__(self, issues, *, closed=(), unavailable=(), fail_issue_list=False):
+        self.issues = list(issues)
+        self.closed = set(closed)
+        self.unavailable = set(unavailable)
+        self.fail_issue_list = fail_issue_list
+        self.lookups: list[int] = []
+
+    def api_get_all(self, url, _auth, **_kw):
+        if "/pulls" in url:
+            return []
+        if self.fail_issue_list:
+            raise RuntimeError("issue listing failed")
+        return list(self.issues)
+
+    def api_request(self, _method, url, _auth, **_kw):
+        number = int(url.rsplit("/", 1)[-1])
+        self.lookups.append(number)
+        if number in self.unavailable:
+            raise RuntimeError("lookup failed")
+        if number in self.closed:
+            return {"number": number, "state": "closed"}
+        return {"number": number, "state": "open"}
+
+
+class AllocatorInventoryTest(unittest.TestCase):
+    """Direct tests of the candidate loader."""
+
+    def _load(self, fake, **kwargs):
+        with patch("gitea_mcp_server._resolve", return_value=("h", ORG, REPO)), patch(
+            "gitea_mcp_server._auth", return_value=FAKE_AUTH
+        ), patch("gitea_mcp_server.api_get_all", side_effect=fake.api_get_all), patch(
+            "gitea_mcp_server.api_request", side_effect=fake.api_request
+        ):
+            return srv._allocator_candidates_from_gitea(
+                remote="prgs", host=None, org=ORG, repo=REPO, **kwargs
+            )
+
+    def test_full_inventory_above_fifty_is_ranked(self) -> None:
+        """AC1: all 73 open issues become candidates, not the first 50."""
+        fake = _FakeGitea([_issue(n) for n in range(600, 673)])
+        candidates, _reasons, complete = self._load(fake)
+        self.assertTrue(complete)
+        self.assertEqual(len(candidates), 73)
+        self.assertEqual(min(c.number for c in candidates), 600)
+        self.assertEqual(max(c.number for c in candidates), 672)
+
+    def test_open_dependency_marks_candidate_unmet(self) -> None:
+        """AC4/AC5/AC6: canonical Depends on an open issue blocks the candidate."""
+        fake = _FakeGitea(
+            [_issue(600, body=_depends_body(601, 602)), _issue(601), _issue(602)]
+        )
+        candidates, _reasons, _complete = self._load(fake)
+        blocked = next(c for c in candidates if c.number == 600)
+        self.assertTrue(blocked.dependency_unmet)
+        self.assertIn("#601", blocked.dependency_reason)
+
+    def test_closed_dependency_is_eligible(self) -> None:
+        """A dependency absent from the open list is confirmed closed, not assumed."""
+        fake = _FakeGitea([_issue(600, body=_depends_body(500))], closed={500})
+        candidates, _reasons, _complete = self._load(fake)
+        candidate = next(c for c in candidates if c.number == 600)
+        self.assertFalse(candidate.dependency_unmet)
+        self.assertIn(500, fake.lookups)  # proved live, not inferred
+
+    def test_unavailable_dependency_evidence_fails_closed(self) -> None:
+        """AC7: an unreachable dependency must block, never pass."""
+        fake = _FakeGitea([_issue(600, body=_depends_body(500))], unavailable={500})
+        candidates, _reasons, _complete = self._load(fake)
+        candidate = next(c for c in candidates if c.number == 600)
+        self.assertTrue(candidate.dependency_unmet)
+        self.assertIn("fail closed", candidate.dependency_reason)
+
+    def test_dependency_state_lookups_are_cached(self) -> None:
+        """Repeated references resolve with a single live lookup."""
+        fake = _FakeGitea(
+            [_issue(n, body=_depends_body(500)) for n in range(600, 610)],
+            closed={500},
+        )
+        self._load(fake)
+        self.assertEqual(fake.lookups.count(500), 1)
+
+    def test_open_dependency_needs_no_lookup(self) -> None:
+        """The complete open listing already proves openness."""
+        fake = _FakeGitea([_issue(600, body=_depends_body(601)), _issue(601)])
+        self._load(fake)
+        self.assertNotIn(601, fake.lookups)
+
+    def test_failed_issue_listing_reports_incomplete(self) -> None:
+        """AC3: a failed listing must not silently yield a short inventory."""
+        fake = _FakeGitea([], fail_issue_list=True)
+        _candidates, reasons, complete = self._load(fake)
+        self.assertFalse(complete)
+        self.assertTrue(any("failed to list open issues" in r for r in reasons))
+
+
+class AllocateNextWorkToolTest(unittest.TestCase):
+    """End-to-end tests of the gitea_allocate_next_work MCP tool."""
+
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def _allocate(self, fake, **kwargs):
+        with patch("gitea_mcp_server._profile_operation_gate", return_value=None), patch(
+            "gitea_mcp_server._resolve", return_value=("h", ORG, REPO)
+        ), patch("gitea_mcp_server._auth", return_value=FAKE_AUTH), patch(
+            "gitea_mcp_server.get_profile",
+            return_value={"profile_name": "prgs-author", "role": "author"},
+        ), patch(
+            "gitea_mcp_server._authenticated_username", return_value="jcwalker3"
+        ), patch(
+            "gitea_mcp_server._control_plane_db_or_error", return_value=(self.db, [])
+        ), patch(
+            "gitea_mcp_server.api_get_all", side_effect=fake.api_get_all
+        ), patch(
+            "gitea_mcp_server.api_request", side_effect=fake.api_request
+        ), patch(
+            "gitea_mcp_server.sentry_observability.monitor_checkin", return_value=None
+        ):
+            return srv.gitea_allocate_next_work(
+                remote="prgs", org=ORG, repo=REPO, role="author", **kwargs
+            )
+
+    def test_limit_does_not_change_selection(self) -> None:
+        """AC2/AC11: the winner is identical at limit=1 and limit=300."""
+        issues = [_issue(n) for n in range(600, 673)]  # 73 candidates
+        low = self._allocate(_FakeGitea(issues), limit=1)
+        high = self._allocate(_FakeGitea(issues), limit=300)
+        self.assertEqual(low["selected"]["number"], high["selected"]["number"])
+        self.assertEqual(low["selected"]["number"], 600)
+        self.assertEqual(low["candidate_count"], 73)
+        self.assertEqual(high["candidate_count"], 73)
+
+    def test_dependency_blocked_winner_falls_through(self) -> None:
+        """AC8: a blocked highest-ranked issue yields the next eligible one."""
+        issues = [
+            _issue(600, body=_depends_body(601)),
+            _issue(601),
+            _issue(602),
+        ]
+        result = self._allocate(_FakeGitea(issues))
+        # 600 is blocked by open 601; 601 itself is a valid candidate.
+        self.assertEqual(result["selected"]["number"], 601)
+        skipped = {s["number"] for s in result["skipped"]}
+        self.assertIn(600, skipped)
+
+    def test_limit_truncates_only_the_reported_skip_list(self) -> None:
+        """A shortened report is labelled, never presented as full coverage."""
+        issues = [_issue(n, body=_depends_body(999)) for n in range(600, 640)]
+        issues.append(_issue(999))  # open dependency blocks all of the above
+        issues.append(_issue(700))  # the one eligible candidate
+        result = self._allocate(_FakeGitea(issues), limit=5)
+        self.assertTrue(result["skipped_report_truncated"])
+        self.assertEqual(len(result["skipped"]), 5)
+        self.assertGreater(result["skipped_total"], 5)
+        self.assertEqual(result["limit_applies_to"], "reported_skip_list_only")
+
+    def test_incomplete_inventory_fails_closed(self) -> None:
+        """AC3: no selection is made from a partial candidate set."""
+        result = self._allocate(_FakeGitea([], fail_issue_list=True))
+        self.assertFalse(result["success"])
+        self.assertFalse(result["inventory_complete"])
+        self.assertIsNone(result["assignment"])
+        self.assertTrue(
+            any("fail closed" in r for r in result["reasons"]),
+            result["reasons"],
+        )
+
+    def test_selection_policy_is_reported(self) -> None:
+        """AC10: tie-breaking is stated in the result, not left implicit."""
+        result = self._allocate(_FakeGitea([_issue(600)]))
+        self.assertIn("selection_policy", result)
+        self.assertIn("number asc", result["selection_policy"])
+
+
+if __name__ == "__main__":
+    unittest.main()

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_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_issue_763_reviewer_lease_preflight_order.py

@@ -0,0 +1,395 @@
+"""Regression coverage for reviewer-lease preflight ordering (#763)."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+import pytest
+
+import anti_stomp_preflight
+import gitea_mcp_server as server
+import merger_lease_adoption
+import reviewer_pr_lease
+import task_capability_map
+
+
+def _prime_clean_reviewer_preflight(monkeypatch, resolved_task: str) -> None:
+    """Install a clean reviewer preflight without bypassing task matching."""
+    monkeypatch.setenv("GITEA_TEST_PORCELAIN", "")
+    monkeypatch.delenv("GITEA_TEST_FORCE_DIRTY", raising=False)
+    monkeypatch.setattr(server, "_preflight_in_test_mode", lambda: False)
+    monkeypatch.setattr(server, "_process_start_porcelain", "")
+    monkeypatch.setattr(server, "_preflight_whoami_called", False)
+    monkeypatch.setattr(server, "_preflight_capability_called", False)
+    monkeypatch.setattr(server, "_preflight_whoami_violation", False)
+    monkeypatch.setattr(server, "_preflight_capability_violation", False)
+    monkeypatch.setattr(server, "_preflight_resolved_role", None)
+    monkeypatch.setattr(server, "_preflight_resolved_task", None)
+    monkeypatch.setattr(server, "_preflight_whoami_baseline_porcelain", None)
+    monkeypatch.setattr(server, "_preflight_capability_baseline_porcelain", None)
+    monkeypatch.setattr(server, "_preflight_whoami_violation_files", [])
+    monkeypatch.setattr(server, "_preflight_capability_violation_files", [])
+    monkeypatch.setattr(server, "_preflight_reviewer_violation_files", [])
+    monkeypatch.setattr(
+        server,
+        "_resolve_namespace_mutation_context",
+        lambda _worktree=None: {
+            "workspace_path": server.PROJECT_ROOT,
+            "canonical_repo_root": server.PROJECT_ROOT,
+            "process_project_root": server.PROJECT_ROOT,
+            "workspace_role_kind": "reviewer",
+            "workspace_binding_source": "test reviewer binding",
+            "ignored_bindings": [],
+        },
+    )
+    monkeypatch.setattr(server, "_enforce_stable_branch_contamination_gate", lambda *_a: None)
+    monkeypatch.setattr(server, "_enforce_canonical_repository_root", lambda *_a, **_k: None)
+    monkeypatch.setattr(server, "_enforce_root_checkout_guard", lambda *_a: None)
+    monkeypatch.setattr(server, "_enforce_branches_only_author_mutation", lambda *_a, **_k: None)
+    monkeypatch.setattr(server, "_enforce_issue_scope_guard", lambda *_a, **_k: None)
+    monkeypatch.setattr(server, "_create_issue_bootstrap_assessment", lambda *_a: None)
+    monkeypatch.setattr(server, "_run_anti_stomp_preflight", lambda *_a, **_k: None)
+
+    server.record_preflight_check("whoami")
+    server.record_preflight_check(
+        "capability", resolved_role="reviewer", resolved_task=resolved_task
+    )
+
+
+def test_documented_review_capability_allows_reviewer_lease_acquire(monkeypatch):
+    """whoami -> resolve(review_pr) -> acquire reviewer lease is canonical."""
+    _prime_clean_reviewer_preflight(monkeypatch, "review_pr")
+
+    server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
+
+    assert server._preflight_capability_called is False
+
+
+def test_exact_lease_capability_without_intervening_call_still_succeeds(monkeypatch):
+    _prime_clean_reviewer_preflight(monkeypatch, "acquire_reviewer_pr_lease")
+
+    server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
+
+    assert server._preflight_capability_called is False
+
+
+def test_missing_wrong_and_consumed_capability_fail_closed(monkeypatch):
+    _prime_clean_reviewer_preflight(monkeypatch, "create_issue")
+    with pytest.raises(RuntimeError, match="task mismatch"):
+        server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
+
+    _prime_clean_reviewer_preflight(monkeypatch, "acquire_reviewer_pr_lease")
+    server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
+    with pytest.raises(RuntimeError, match="has not been resolved"):
+        server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
+
+
+def test_documented_intervening_whoami_read_preserves_capability(monkeypatch):
+    _prime_clean_reviewer_preflight(monkeypatch, "review_pr")
+    with patch.object(server, "_get_workspace_porcelain", return_value=""):
+        server.record_preflight_check("whoami")
+
+    server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
+
+
+def test_reviewer_transition_is_narrow_alias_aware_and_one_way():
+    assert task_capability_map.preflight_task_matches(
+        "review_pr", "gitea_acquire_reviewer_pr_lease"
+    )
+    assert task_capability_map.preflight_task_matches(
+        "gitea_acquire_reviewer_pr_lease", "acquire_reviewer_pr_lease"
+    )
+    assert not task_capability_map.preflight_task_matches(
+        "acquire_reviewer_pr_lease", "review_pr"
+    )
+    assert not task_capability_map.preflight_task_matches(
+        "review_pr", "acquire_merger_pr_lease"
+    )
+    assert not task_capability_map.preflight_task_matches(
+        "merge_pr", "acquire_reviewer_pr_lease"
+    )
+
+
+def test_dirty_reviewer_workspace_still_fails_closed(monkeypatch):
+    _prime_clean_reviewer_preflight(monkeypatch, "review_pr")
+    monkeypatch.setenv("GITEA_TEST_PORCELAIN", " M gitea_mcp_server.py\n")
+
+    with pytest.raises(RuntimeError, match="Reviewer role violation"):
+        server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
+
+
+def test_mismatched_reviewer_workspace_still_fails_closed(monkeypatch):
+    _prime_clean_reviewer_preflight(monkeypatch, "review_pr")
+    monkeypatch.setattr(
+        server,
+        "_resolve_namespace_mutation_context",
+        lambda _worktree=None: {
+            "workspace_path": "/outside/review-pr-762",
+            "canonical_repo_root": "/repo",
+            "process_project_root": "/repo",
+            "workspace_role_kind": "reviewer",
+            "workspace_binding_source": "test reviewer binding",
+            "ignored_bindings": [],
+        },
+    )
+    monkeypatch.setattr(
+        server.author_mutation_worktree,
+        "assess_workspace_repo_membership",
+        lambda **_kwargs: {"block": True, "reasons": ["workspace mismatch"]},
+    )
+    monkeypatch.setattr(
+        server.author_mutation_worktree,
+        "format_workspace_repo_membership_error",
+        lambda _assessment: "workspace mismatch (fail closed)",
+    )
+
+    with pytest.raises(RuntimeError, match="workspace mismatch"):
+        server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
+
+
+def test_reviewer_lease_acquire_requires_workflow_load_proof(monkeypatch):
+    sha = "a" * 40
+    monkeypatch.setattr(server, "_anti_stomp_in_test_mode", lambda: False)
+    monkeypatch.setattr(
+        server,
+        "get_profile",
+        lambda: {
+            "profile_name": "prgs-reviewer",
+            "role": "reviewer",
+            "allowed_operations": [
+                "gitea.read",
+                "gitea.pr.comment",
+                "gitea.pr.review",
+            ],
+        },
+    )
+    monkeypatch.setattr(server, "_actual_profile_role", lambda: "reviewer")
+    monkeypatch.setattr(
+        server,
+        "_resolve_namespace_mutation_context",
+        lambda _worktree=None: {
+            "workspace_path": "/repo/branches/review-pr-762",
+            "canonical_repo_root": "/repo",
+            "process_project_root": "/repo",
+        },
+    )
+    monkeypatch.setattr(
+        server.issue_lock_worktree,
+        "read_worktree_git_state",
+        lambda _path: {
+            "current_branch": "master",
+            "head_sha": sha,
+            "porcelain_status": "",
+        },
+    )
+    monkeypatch.setattr(
+        server.root_checkout_guard,
+        "resolve_remote_master_sha",
+        lambda _path: sha,
+    )
+    monkeypatch.setattr(
+        server,
+        "_current_master_parity",
+        lambda: {"startup_head": sha, "current_head": sha},
+    )
+    monkeypatch.setattr(server, "_local_git_remote_url", lambda _remote: None)
+    monkeypatch.setattr(server, "_load_stable_contamination_marker", lambda _remote: None)
+    monkeypatch.setattr(
+        server,
+        "_review_workflow_load_gate_reasons",
+        lambda: ["canonical review workflow proof missing"],
+    )
+
+    with pytest.raises(RuntimeError, match="workflow"):
+        server._run_anti_stomp_preflight(
+            "acquire_reviewer_pr_lease",
+            remote="prgs",
+            worktree_path="/repo/branches/review-pr-762",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+
+
+def test_whoami_identity_mismatch_invalidates_preflight(monkeypatch):
+    monkeypatch.setenv("GITEA_TEST_PORCELAIN", "")
+    monkeypatch.setattr(server, "_process_start_porcelain", "")
+    monkeypatch.setattr(server, "_preflight_whoami_called", False)
+    monkeypatch.setattr(server, "_preflight_capability_called", True)
+    monkeypatch.setattr(server, "_auth", lambda _host: "redacted")
+    monkeypatch.setattr(
+        server,
+        "api_request",
+        lambda *_args, **_kwargs: {"login": "wrong-reviewer", "id": 7},
+    )
+    monkeypatch.setattr(
+        server,
+        "get_profile",
+        lambda: {
+            "profile_name": "prgs-reviewer",
+            "role": "reviewer",
+            "username": "sysadmin",
+            "allowed_operations": ["gitea.read", "gitea.pr.review"],
+            "forbidden_operations": [],
+        },
+    )
+    monkeypatch.setattr(server, "_seed_session_context", lambda **_kwargs: None)
+    monkeypatch.setattr(server.session_ctx, "mutation_context_audit_fields", lambda: {})
+    monkeypatch.setattr(server, "_reveal_endpoints", lambda: False)
+
+    result = server.gitea_whoami(remote="prgs")
+
+    assert result["identity_match"] is False
+    assert server._preflight_whoami_called is False
+    assert server._preflight_capability_called is False
+
+
+def test_denied_reviewer_profile_does_not_leave_capability_proof(monkeypatch):
+    profile = {
+        "profile_name": "prgs-author",
+        "role": "author",
+        "username": "jcwalker3",
+        "allowed_operations": [
+            "gitea.read",
+            "gitea.pr.comment",
+            "gitea.pr.review",
+        ],
+        "forbidden_operations": [],
+    }
+    monkeypatch.setenv("GITEA_TEST_PORCELAIN", "")
+    monkeypatch.setattr(server, "_process_start_porcelain", "")
+    monkeypatch.setattr(server, "get_profile", lambda: profile)
+    monkeypatch.setattr(
+        server.gitea_config,
+        "load_config",
+        lambda: {"profiles": {"prgs-author": profile}},
+    )
+    monkeypatch.setattr(server.gitea_config, "is_runtime_switching_enabled", lambda: False)
+    monkeypatch.setattr(server, "_authenticated_username", lambda _host: "jcwalker3")
+    monkeypatch.setattr(server, "_seed_session_context", lambda **_kwargs: None)
+    monkeypatch.setattr(
+        server.session_ctx,
+        "assess_session_context",
+        lambda **_kwargs: {"block": False, "reasons": []},
+    )
+    monkeypatch.setattr(
+        server.session_ctx,
+        "assess_identity_match",
+        lambda **_kwargs: {"block": False, "reasons": []},
+    )
+    monkeypatch.setattr(
+        server.session_ctx,
+        "profile_allowed_for_remote",
+        lambda *_args, **_kwargs: {"block": False, "reasons": []},
+    )
+    monkeypatch.setattr(server.session_ctx, "mutation_context_audit_fields", lambda: {})
+    monkeypatch.setattr(
+        server.role_session_router,
+        "assess_infra_stop",
+        lambda _root: {"infra_stop": False, "infra_stop_reasons": []},
+    )
+    monkeypatch.setattr(server, "_check_mcp_runtimes_diagnostics", lambda *_a: [])
+    monkeypatch.setattr(
+        server,
+        "_assess_stale_active_binding",
+        lambda **_kwargs: {"classification": "unbound"},
+    )
+    monkeypatch.setattr(server, "record_mutation_authority", lambda *_args: None)
+    monkeypatch.setattr(server, "init_review_decision_lock", lambda *_a, **_k: None)
+    monkeypatch.setattr(server.capability_stop_terminal, "is_active", lambda: False)
+    monkeypatch.setattr(
+        server.capability_stop_terminal,
+        "sync_from_capability_result",
+        lambda _result: False,
+    )
+
+    result = server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+
+    assert result["allowed_in_current_session"] is False
+    assert result["required_role_kind"] == "reviewer"
+    assert server._preflight_capability_called is False
+
+
+def test_head_and_foreign_lease_protections_remain_enforced():
+    now = datetime.now(timezone.utc)
+    head = "a" * 40
+    moved_head = "b" * 40
+    body = reviewer_pr_lease.format_lease_body(
+        repo="Scaled-Tech-Consulting/Gitea-Tools",
+        pr_number=762,
+        issue_number=605,
+        reviewer_identity="other-reviewer",
+        profile="prgs-reviewer",
+        session_id="foreign-session",
+        worktree="/repo/branches/review-pr-762",
+        phase="claimed",
+        candidate_head=head,
+        target_branch="master",
+        target_branch_sha="c" * 40,
+        last_activity=now,
+    )
+    comments = [{"id": 10, "author": "other-reviewer", "body": body}]
+
+    acquire = reviewer_pr_lease.assess_acquire_lease(
+        comments,
+        pr_number=762,
+        reviewer_identity="sysadmin",
+        profile="prgs-reviewer",
+        session_id="my-session",
+        repo="Scaled-Tech-Consulting/Gitea-Tools",
+        issue_number=605,
+        worktree="/repo/branches/review-pr-762-mine",
+        candidate_head=head,
+        target_branch="master",
+        target_branch_sha="c" * 40,
+        now=now,
+    )
+    assert acquire["acquire_allowed"] is False
+
+    reviewer_pr_lease.clear_session_lease()
+    reviewer_pr_lease.record_session_lease(
+        {
+            "pr_number": 762,
+            "session_id": "foreign-session",
+            "candidate_head": head,
+            "comment_id": 10,
+        },
+        lease_provenance=merger_lease_adoption.build_lease_provenance(
+            source=merger_lease_adoption.SOURCE_ACQUIRE,
+            comment_id=10,
+        ),
+    )
+    try:
+        gate = reviewer_pr_lease.assess_mutation_lease_gate(
+            pr_number=762,
+            comments=comments,
+            reviewer_identity="other-reviewer",
+            session_id="foreign-session",
+            mutation="approve",
+            live_head_sha=moved_head,
+            pinned_head_sha=head,
+            now=now,
+        )
+    finally:
+        reviewer_pr_lease.clear_session_lease()
+
+    assert gate["block"] is True
+    assert any("head changed" in reason for reason in gate["reasons"])
+
+
+def test_reviewer_lease_role_gate_is_not_weakened():
+    result = anti_stomp_preflight.assess_anti_stomp_preflight(
+        task="acquire_reviewer_pr_lease",
+        profile_name="prgs-author",
+        profile_role="author",
+        required_role="reviewer",
+        required_permission="gitea.pr.comment",
+        allowed_operations=["gitea.read"],
+        check_repo=False,
+        check_root_checkout=False,
+        check_worktree=False,
+        check_stale_runtime=False,
+    )
+
+    assert result["block"] is True
+    assert result["blocker_kind"] == anti_stomp_preflight.BLOCKER_WRONG_ROLE

tests/test_mcp_menu_script.py

@@ -10,6 +10,7 @@ DOCS = REPO_ROOT / "docs" / "mcp-menu.md"
 
 REQUIRED_MENU_LABELS = (
     "Project status / root checkout health",
+    "Workflow dashboard (queue, leases, next safe action)",
     "Author workflow prompts",
     "Reviewer workflow prompts",
     "Merger workflow prompts",
@@ -105,6 +106,23 @@ class TestMcpMenuScript(unittest.TestCase):
         self.assertIn("./mcp-menu.sh", docs_text)
         self.assertIn("placeholder", docs_text.lower())
 
+    def test_workflow_dashboard_menu_entry_is_read_only(self):
+        # #605: dashboard entry documents gitea_workflow_dashboard and never
+        # mutates Gitea / assigns work from the shell menu.
+        label = "Workflow dashboard (queue, leases, next safe action)"
+        self.assertIn(label, self.content)
+        dash_fn = self._extract_function("show_workflow_dashboard_help")
+        self.assertIn("gitea_workflow_dashboard", dash_fn)
+        self.assertIn("gitea_allocate_next_work", dash_fn)
+        self.assertIn("Read-only", dash_fn)
+        self.assertIn("never presented as safe", dash_fn.lower())
+        for bad in ("gitea_merge_pr", "gitea_submit_pr_review", "git push"):
+            with self.subTest(bad=bad):
+                self.assertNotIn(bad, dash_fn)
+        docs_text = DOCS.read_text(encoding="utf-8")
+        self.assertIn("gitea_workflow_dashboard", docs_text)
+        self.assertIn("Workflow dashboard", docs_text)
+
     def test_reviewer_skip_stale_request_changes_prompt_discoverable(self):
         # #482: the skip-already-reviewed-stale-REQUEST_CHANGES reviewer prompt
         # must be reachable from the reviewer menu and documented.

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.fail("lock_issue must stay blocked on the control checkout")
-                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()
-                )
 
 
 if __name__ == "__main__":

tests/test_workflow_dashboard.py

@@ -0,0 +1,274 @@
+"""Hermetic tests for workflow dashboard (#605).
+
+Covers terminal-blocked queue shapes in the spirit of #593/#592/#587 where an
+active terminal-review lock must suppress other PRs as safe review/merge work.
+"""
+
+from __future__ import annotations
+
+import unittest
+
+from allocator_service import WorkCandidate
+from workflow_dashboard import (
+    DASHBOARD_VERSION,
+    build_workflow_dashboard,
+    format_human_summary,
+)
+
+
+def _issue(
+    number: int,
+    *,
+    title: str = "",
+    labels: tuple[str, ...] = ("status:ready",),
+    priority: int = 20,
+    blocked: bool = False,
+    dependency_unmet: bool = False,
+    dependency_reason: str | None = None,
+    claimed: bool = False,
+) -> WorkCandidate:
+    return WorkCandidate(
+        kind="issue",
+        number=number,
+        title=title or f"issue {number}",
+        labels=labels,
+        priority=priority,
+        blocked=blocked,
+        dependency_unmet=dependency_unmet,
+        dependency_reason=dependency_reason,
+        already_claimed_elsewhere=claimed,
+    )
+
+
+def _pr(
+    number: int,
+    *,
+    title: str = "",
+    head_sha: str = "abc123",
+    request_changes: bool = False,
+    approved: bool = False,
+    mergeable: bool = False,
+    contaminated: bool = False,
+    approval_stale: bool = False,
+    priority: int = 5,
+) -> WorkCandidate:
+    return WorkCandidate(
+        kind="pr",
+        number=number,
+        title=title or f"pr {number}",
+        head_sha=head_sha,
+        request_changes_current_head=request_changes,
+        approval_on_current_head=approved,
+        mergeable=mergeable,
+        approval_contaminated=contaminated,
+        approval_stale=approval_stale,
+        priority=priority,
+    )
+
+
+class TestWorkflowDashboard(unittest.TestCase):
+    def test_version_and_read_only_payload(self):
+        snap = build_workflow_dashboard(
+            candidates=[_issue(605)],
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        payload = snap.as_dict()
+        self.assertTrue(payload["read_only"])
+        self.assertEqual(payload["dashboard_version"], DASHBOARD_VERSION)
+        self.assertTrue(payload["success"])
+        self.assertTrue(payload["inventory_complete"])
+        self.assertIn("human_summary", payload)
+
+    def test_never_marks_blocked_as_safe(self):
+        candidates = [
+            _issue(10, blocked=True, labels=("status:blocked",)),
+            _issue(11, dependency_unmet=True, dependency_reason="depends on #9"),
+            _issue(12, claimed=True),
+            _issue(605, labels=("status:ready",)),
+        ]
+        snap = build_workflow_dashboard(candidates=candidates)
+        blocked_numbers = {e.number for e in snap.blocked_items}
+        self.assertIn(10, blocked_numbers)
+        self.assertIn(11, blocked_numbers)
+        self.assertIn(12, blocked_numbers)
+        for entry in snap.blocked_items:
+            self.assertFalse(entry.as_dict()["is_safe"])
+            self.assertEqual(entry.safe_for_roles, ())
+            self.assertIsNotNone(entry.block_reason)
+
+        author = snap.next_safe_by_role["author"]
+        self.assertEqual(author.status, "safe")
+        self.assertEqual(author.target_number, 605)
+        self.assertNotIn(author.target_number, blocked_numbers)
+        summary = format_human_summary(snap)
+        self.assertIn("NOT safe", summary)
+        self.assertIn("issue#10", summary.replace(" ", ""))
+
+    def test_author_prefers_oldest_ready_issue(self):
+        candidates = [
+            _issue(620, labels=("status:ready",)),
+            _issue(605, labels=("status:ready",)),
+            _issue(610, labels=("status:ready",)),
+        ]
+        snap = build_workflow_dashboard(candidates=candidates)
+        author = snap.next_safe_by_role["author"]
+        self.assertEqual(author.status, "safe")
+        self.assertEqual(author.target_number, 605)
+        self.assertIn("gitea_allocate_next_work", author.prompt)
+        self.assertIn("role='author'", author.prompt)
+
+    def test_review_and_merge_ready_buckets(self):
+        candidates = [
+            _pr(100, head_sha="r1"),  # review-ready
+            _pr(101, approved=True, mergeable=True, head_sha="m1", priority=8),
+            _pr(102, request_changes=True, head_sha="a1", priority=10),
+        ]
+        snap = build_workflow_dashboard(candidates=candidates)
+        self.assertEqual([e.number for e in snap.review_ready_prs], [100])
+        self.assertEqual([e.number for e in snap.merge_ready_prs], [101])
+        self.assertEqual([e.number for e in snap.author_remediation], [102])
+
+        reviewer = snap.next_safe_by_role["reviewer"]
+        self.assertEqual(reviewer.status, "safe")
+        self.assertEqual(reviewer.target_number, 100)
+        self.assertEqual(reviewer.head_sha, "r1")
+
+        merger = snap.next_safe_by_role["merger"]
+        self.assertEqual(merger.status, "safe")
+        self.assertEqual(merger.target_number, 101)
+        self.assertEqual(merger.head_sha, "m1")
+
+        author = snap.next_safe_by_role["author"]
+        self.assertEqual(author.status, "safe")
+        self.assertEqual(author.target_number, 102)
+
+    def test_terminal_lock_blocks_other_prs_as_safe(
+        self,
+    ):
+        """#593/#592/#587-style: terminal lock ⇒ other PRs are not safe."""
+        candidates = [
+            _pr(587, head_sha="deadbeef", priority=5),
+            _pr(592, approved=True, mergeable=True, head_sha="cafebabe", priority=8),
+            _pr(593, head_sha="terminalhead", priority=9),
+            _issue(605, labels=("status:ready",)),
+        ]
+        snap = build_workflow_dashboard(
+            candidates=candidates,
+            terminal_pr=593,
+            terminal_lock={"terminal_pr": 593, "active": True, "state": "locked"},
+        )
+
+        # Non-terminal PRs must appear blocked, never in safe buckets.
+        blocked_prs = {
+            e.number for e in snap.blocked_items if e.kind == "pr"
+        }
+        self.assertIn(587, blocked_prs)
+        self.assertIn(592, blocked_prs)
+        self.assertNotIn(593, blocked_prs)  # terminal PR itself may still be routeable
+
+        # Terminal PR itself may remain review-ready; others must not.
+        self.assertEqual([e.number for e in snap.review_ready_prs], [593])
+        self.assertEqual(snap.merge_ready_prs, [])
+        self.assertNotIn(587, [e.number for e in snap.review_ready_prs])
+        self.assertNotIn(592, [e.number for e in snap.merge_ready_prs])
+
+        for entry in snap.blocked_items:
+            if entry.number in (587, 592):
+                self.assertIn("terminal-review lock", entry.block_reason or "")
+                self.assertEqual(entry.safe_for_roles, ())
+                self.assertFalse(entry.as_dict()["is_safe"])
+
+        reviewer = snap.next_safe_by_role["reviewer"]
+        # Reviewer may only target the terminal PR — never 587/592.
+        self.assertEqual(reviewer.status, "safe")
+        self.assertEqual(reviewer.target_number, 593)
+        self.assertEqual(reviewer.head_sha, "terminalhead")
+        self.assertNotEqual(reviewer.target_number, 587)
+        self.assertNotEqual(reviewer.target_number, 592)
+
+        merger = snap.next_safe_by_role["merger"]
+        # Merge-ready #592 is NOT safe while terminal lock is on #593.
+        self.assertNotEqual(merger.target_number, 592)
+        self.assertIn("593", merger.prompt)
+        self.assertIn(
+            merger.status,
+            ("blocked_terminal", "idle", "safe"),
+        )
+        if merger.status == "safe":
+            self.assertEqual(merger.target_number, 593)
+
+        # Author issue work remains visible (issues are not terminal-blocked).
+        author = snap.next_safe_by_role["author"]
+        self.assertEqual(author.status, "safe")
+        self.assertEqual(author.target_number, 605)
+
+        summary = format_human_summary(snap)
+        self.assertIn("Terminal review lock: ACTIVE on PR #593", summary)
+        self.assertIn("Do not treat other open PRs as safe", summary)
+
+    def test_incomplete_inventory_fails_closed(self):
+        snap = build_workflow_dashboard(
+            candidates=[_issue(605)],
+            inventory_complete=False,
+            inventory_reasons=["page truncated"],
+        )
+        payload = snap.as_dict()
+        self.assertFalse(payload["inventory_complete"])
+        self.assertEqual(payload["review_ready_prs"], [])
+        self.assertEqual(payload["merge_ready_prs"], [])
+        for action in snap.next_safe_by_role.values():
+            self.assertEqual(action.status, "none")
+            self.assertIsNone(action.target_number)
+            self.assertIn("inventory incomplete", action.prompt.lower())
+            self.assertFalse(action.as_dict()["is_safe"])
+
+    def test_leases_partition_active_vs_stale(self):
+        leases = [
+            {"lease_id": "L1", "role": "author", "status": "active", "work_number": 605},
+            {"lease_id": "L2", "role": "reviewer", "status": "expired", "work_number": 99},
+            {"lease_id": "L3", "role": "merger", "stale": True, "work_number": 88},
+        ]
+        snap = build_workflow_dashboard(candidates=[], leases=leases)
+        self.assertEqual(len(snap.active_leases_by_role["author"]), 1)
+        self.assertEqual(len(snap.stale_or_expired_leases), 2)
+
+    def test_discussion_and_controller_needed(self):
+        candidates = [
+            _issue(1, labels=("discussion", "type:discussion")),
+            _pr(2, contaminated=True, head_sha="x"),
+        ]
+        snap = build_workflow_dashboard(candidates=candidates)
+        self.assertEqual([e.number for e in snap.discussion_issues], [1])
+        self.assertTrue(any(e.number == 2 for e in snap.controller_needed))
+        recon = snap.next_safe_by_role["reconciler"]
+        self.assertEqual(recon.status, "safe")
+        self.assertEqual(recon.target_number, 2)
+
+    def test_human_summary_includes_exact_prompts(self):
+        snap = build_workflow_dashboard(
+            candidates=[_issue(605)],
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+        )
+        text = format_human_summary(snap)
+        self.assertIn("gitea_allocate_next_work", text)
+        self.assertIn("prgs/Scaled-Tech-Consulting/Gitea-Tools", text)
+        self.assertIn("never self-selects", text.lower())
+        self.assertIn("Primary next:", text)
+
+    def test_missing_pr_head_sha_is_blocked(self):
+        candidates = [_pr(50, head_sha="")]
+        # WorkCandidate allows empty head; dashboard must block it.
+        c = candidates[0]
+        c.head_sha = ""
+        snap = build_workflow_dashboard(candidates=[c])
+        self.assertEqual(len(snap.blocked_items), 1)
+        self.assertIn("head_sha", snap.blocked_items[0].block_reason or "")
+        self.assertEqual(snap.review_ready_prs, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

workflow_dashboard.py

@@ -0,0 +1,680 @@
+"""Read-only workflow dashboard for live queue / lease / next-safe-action (#605).
+
+Builds a machine-readable + human-readable operational view so humans and LLMs
+can see what is safe to work on without reconstructing state from comments.
+
+Design rules:
+* Read-only: never assigns work. Assignment still goes through
+  ``gitea_allocate_next_work`` (#600).
+* Never present blocked / terminal-locked / dependency-unmet items as safe.
+* Prefer pure classification so unit tests can inject inventory (including
+  terminal-blocked queues from #593/#592/#587-style scenarios).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Iterable, Sequence
+
+from allocator_service import (
+    ROLE_AUTHOR,
+    ROLE_CONTROLLER,
+    ROLE_MERGER,
+    ROLE_RECONCILER,
+    ROLE_REVIEWER,
+    WorkCandidate,
+    classify_skip,
+    expected_role_for_candidate,
+    sort_candidates,
+)
+
+DASHBOARD_VERSION = "1.0.0-issue-605"
+
+# Roles the dashboard surfaces next-safe prompts for.
+DASHBOARD_ROLES: tuple[str, ...] = (
+    ROLE_AUTHOR,
+    ROLE_REVIEWER,
+    ROLE_MERGER,
+    ROLE_RECONCILER,
+    ROLE_CONTROLLER,
+)
+
+# Exact operator prompts (fill-in tokens only — no self-selection).
+PROMPT_AUTHOR = (
+    "AUTHOR session: call gitea_allocate_next_work(apply=true, role='author') "
+    "for {remote}/{org}/{repo}, then implement only the assigned issue under "
+    "branches/ and open/update its PR. Do not self-select outside the allocator."
+)
+PROMPT_REVIEWER = (
+    "REVIEWER session: call gitea_allocate_next_work(apply=true, role='reviewer') "
+    "for {remote}/{org}/{repo}, pin the assigned PR head SHA, submit exactly one "
+    "formal review verdict for that head. Do not merge."
+)
+PROMPT_MERGER = (
+    "MERGER session: call gitea_allocate_next_work(apply=true, role='merger') "
+    "for {remote}/{org}/{repo}, reassess the assigned approved head, and merge "
+    "only that exact head via gitea_merge_pr. Do not review."
+)
+PROMPT_RECONCILER = (
+    "RECONCILER session: call gitea_allocate_next_work(apply=true, role='reconciler') "
+    "for {remote}/{org}/{repo}, then perform only the assigned terminal "
+    "reconciliation (already-landed / post-merge cleanup). Do not approve or merge."
+)
+PROMPT_CONTROLLER = (
+    "CONTROLLER session: inspect gitea_workflow_dashboard + control-plane leases, "
+    "diagnose blocked/terminal-locked items for {remote}/{org}/{repo}, and schedule "
+    "exactly one fresh role-scoped cycle. Do not implement, review, or merge in-band."
+)
+PROMPT_IDLE = (
+    "IDLE: no safe assignable work for role '{role}' on {remote}/{org}/{repo}. "
+    "Do not self-select. Re-run gitea_workflow_dashboard on the next cycle."
+)
+PROMPT_TERMINAL_BLOCK = (
+    "BLOCKED by terminal-review lock on PR #{terminal_pr} for {remote}/{org}/{repo}. "
+    "Resolve the terminal path for that exact PR before any other review/merge work. "
+    "Do not treat other open PRs as safe."
+)
+PROMPT_BLOCKED_ITEM = (
+    "NOT SAFE: {kind}#{number} is blocked ({reason}). Never present as next safe work."
+)
+
+
+@dataclass(frozen=True)
+class QueueEntry:
+    kind: str
+    number: int
+    title: str
+    expected_role: str
+    safe_for_roles: tuple[str, ...]
+    badges: tuple[str, ...]
+    block_reason: str | None = None
+    head_sha: str | None = None
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "kind": self.kind,
+            "number": self.number,
+            "title": self.title,
+            "expected_role": self.expected_role,
+            "safe_for_roles": list(self.safe_for_roles),
+            "badges": list(self.badges),
+            "block_reason": self.block_reason,
+            "head_sha": self.head_sha,
+            "is_safe": self.block_reason is None and bool(self.safe_for_roles),
+        }
+
+
+@dataclass(frozen=True)
+class RoleNextAction:
+    role: str
+    status: str  # safe | idle | blocked_terminal | none
+    target_kind: str | None
+    target_number: int | None
+    head_sha: str | None
+    prompt: str
+    reasons: tuple[str, ...] = ()
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "role": self.role,
+            "status": self.status,
+            "target_kind": self.target_kind,
+            "target_number": self.target_number,
+            "head_sha": self.head_sha,
+            "prompt": self.prompt,
+            "reasons": list(self.reasons),
+            "is_safe": self.status == "safe",
+        }
+
+
+@dataclass
+class DashboardSnapshot:
+    remote: str
+    org: str
+    repo: str
+    inventory_complete: bool
+    candidate_count: int
+    open_prs: list[QueueEntry] = field(default_factory=list)
+    open_issues: list[QueueEntry] = field(default_factory=list)
+    review_ready_prs: list[QueueEntry] = field(default_factory=list)
+    merge_ready_prs: list[QueueEntry] = field(default_factory=list)
+    author_remediation: list[QueueEntry] = field(default_factory=list)
+    discussion_issues: list[QueueEntry] = field(default_factory=list)
+    blocked_items: list[QueueEntry] = field(default_factory=list)
+    controller_needed: list[QueueEntry] = field(default_factory=list)
+    active_leases_by_role: dict[str, list[dict[str, Any]]] = field(default_factory=dict)
+    stale_or_expired_leases: list[dict[str, Any]] = field(default_factory=list)
+    terminal_review_lock: dict[str, Any] | None = None
+    next_safe_by_role: dict[str, RoleNextAction] = field(default_factory=dict)
+    primary_next_safe_action: RoleNextAction | None = None
+    reasons: list[str] = field(default_factory=list)
+    dashboard_version: str = DASHBOARD_VERSION
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "success": self.inventory_complete and not any(
+                r.startswith("inventory incomplete") for r in self.reasons
+            ),
+            "read_only": True,
+            "dashboard_version": self.dashboard_version,
+            "remote": self.remote,
+            "org": self.org,
+            "repo": self.repo,
+            "inventory_complete": self.inventory_complete,
+            "candidate_count": self.candidate_count,
+            "open_pr_queue": [e.as_dict() for e in self.open_prs],
+            "open_issue_queue": [e.as_dict() for e in self.open_issues],
+            "review_ready_prs": [e.as_dict() for e in self.review_ready_prs],
+            "merge_ready_prs": [e.as_dict() for e in self.merge_ready_prs],
+            "author_remediation": [e.as_dict() for e in self.author_remediation],
+            "discussion_issues": [e.as_dict() for e in self.discussion_issues],
+            "blocked_items": [e.as_dict() for e in self.blocked_items],
+            "controller_needed": [e.as_dict() for e in self.controller_needed],
+            "active_leases_by_role": {
+                role: list(items) for role, items in self.active_leases_by_role.items()
+            },
+            "stale_or_expired_leases": list(self.stale_or_expired_leases),
+            "terminal_review_lock": self.terminal_review_lock,
+            "next_safe_by_role": {
+                role: action.as_dict() for role, action in self.next_safe_by_role.items()
+            },
+            "primary_next_safe_action": (
+                self.primary_next_safe_action.as_dict()
+                if self.primary_next_safe_action
+                else None
+            ),
+            "reasons": list(self.reasons),
+            "human_summary": format_human_summary(self),
+        }
+
+
+def _scope_tokens(remote: str, org: str, repo: str) -> dict[str, str]:
+    return {"remote": remote, "org": org, "repo": repo}
+
+
+def _prompt_for_role(
+    role: str,
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    terminal_pr: int | None = None,
+    idle: bool = False,
+) -> str:
+    scope = _scope_tokens(remote, org, repo)
+    if terminal_pr is not None and role in (ROLE_REVIEWER, ROLE_MERGER):
+        return PROMPT_TERMINAL_BLOCK.format(terminal_pr=terminal_pr, **scope)
+    if idle:
+        return PROMPT_IDLE.format(role=role, **scope)
+    templates = {
+        ROLE_AUTHOR: PROMPT_AUTHOR,
+        ROLE_REVIEWER: PROMPT_REVIEWER,
+        ROLE_MERGER: PROMPT_MERGER,
+        ROLE_RECONCILER: PROMPT_RECONCILER,
+        ROLE_CONTROLLER: PROMPT_CONTROLLER,
+    }
+    return templates.get(role, PROMPT_CONTROLLER).format(**scope)
+
+
+def _badges_for_candidate(c: WorkCandidate, *, terminal_pr: int | None) -> tuple[str, ...]:
+    badges: list[str] = []
+    if c.kind == "pr":
+        if c.request_changes_current_head:
+            badges.append("request-changes")
+        if c.approval_on_current_head and c.mergeable:
+            badges.append("merge-ready")
+        elif c.approval_on_current_head and not c.mergeable:
+            badges.append("approved-not-mergeable")
+        if c.approval_stale:
+            badges.append("approval-stale")
+        if c.approval_contaminated:
+            badges.append("contaminated")
+        if not c.approval_on_current_head and not c.request_changes_current_head:
+            badges.append("review-ready")
+        if terminal_pr is not None and c.number == terminal_pr:
+            badges.append("terminal-lock")
+        if terminal_pr is not None and c.number != terminal_pr:
+            badges.append("blocked-by-terminal")
+    else:
+        labels = set(c.labels)
+        if "status:ready" in labels:
+            badges.append("ready")
+        if "status:in-progress" in labels:
+            badges.append("in-progress")
+        if "status:blocked" in labels or c.blocked:
+            badges.append("blocked")
+        if "discussion" in labels or "type:discussion" in labels:
+            badges.append("discussion")
+    if c.dependency_unmet:
+        badges.append("dependency-unmet")
+    if c.already_claimed_elsewhere:
+        badges.append("claimed")
+    if c.blocked:
+        badges.append("blocked")
+    # de-dupe preserve order
+    seen: set[str] = set()
+    out: list[str] = []
+    for b in badges:
+        if b not in seen:
+            seen.add(b)
+            out.append(b)
+    return tuple(out)
+
+
+def _entry_for_candidate(
+    c: WorkCandidate,
+    *,
+    terminal_pr: int | None,
+) -> QueueEntry:
+    expected = expected_role_for_candidate(c)
+    badges = _badges_for_candidate(c, terminal_pr=terminal_pr)
+    safe_roles: list[str] = []
+    block_reason: str | None = None
+
+    # Global hard blocks (never safe for any worker role).
+    if c.blocked or "status:blocked" in c.labels:
+        block_reason = "status blocked"
+    elif c.dependency_unmet:
+        block_reason = c.dependency_reason or "unmet dependency"
+    elif c.already_claimed_elsewhere:
+        block_reason = "already claimed elsewhere"
+    elif c.kind == "pr" and not (c.head_sha or "").strip():
+        block_reason = "missing head_sha pin"
+    elif (
+        terminal_pr is not None
+        and c.kind == "pr"
+        and c.number != terminal_pr
+    ):
+        # Other PRs remain visible but are not safe for review/merge while a
+        # terminal lock is active (#593/#592/#587-style queue).
+        block_reason = f"active terminal-review lock on PR #{terminal_pr}"
+
+    if block_reason is None:
+        # Safe only for the expected role, and only when classify_skip agrees.
+        skip = classify_skip(c, role=expected, terminal_pr=terminal_pr)
+        if skip is None:
+            safe_roles.append(expected)
+        else:
+            block_reason = skip
+
+    return QueueEntry(
+        kind=c.kind,
+        number=c.number,
+        title=c.title or "",
+        expected_role=expected,
+        safe_for_roles=tuple(safe_roles),
+        badges=badges,
+        block_reason=block_reason,
+        head_sha=c.head_sha,
+    )
+
+
+def _partition_leases(
+    leases: Sequence[dict[str, Any]] | None,
+) -> tuple[dict[str, list[dict[str, Any]]], list[dict[str, Any]]]:
+    by_role: dict[str, list[dict[str, Any]]] = {r: [] for r in DASHBOARD_ROLES}
+    stale: list[dict[str, Any]] = []
+    for raw in leases or ():
+        if not isinstance(raw, dict):
+            continue
+        role = str(raw.get("role") or raw.get("owner_role") or "unknown").strip().lower()
+        status = str(raw.get("status") or raw.get("lease_status") or "active").strip().lower()
+        entry = dict(raw)
+        if status in ("expired", "stale", "released", "moot") or raw.get("stale") or raw.get(
+            "expired"
+        ):
+            stale.append(entry)
+            continue
+        if role in by_role:
+            by_role[role].append(entry)
+        else:
+            by_role.setdefault(role, []).append(entry)
+    return by_role, stale
+
+
+def _first_safe_for_role(
+    entries: Iterable[QueueEntry],
+    role: str,
+) -> QueueEntry | None:
+    for entry in entries:
+        if role in entry.safe_for_roles and entry.block_reason is None:
+            return entry
+    return None
+
+
+def _role_next_action(
+    role: str,
+    *,
+    entries: Sequence[QueueEntry],
+    remote: str,
+    org: str,
+    repo: str,
+    terminal_pr: int | None,
+) -> RoleNextAction:
+    # Terminal lock blocks reviewer/merger from non-terminal work.
+    if terminal_pr is not None and role in (ROLE_REVIEWER, ROLE_MERGER):
+        terminal_entry = next(
+            (
+                e
+                for e in entries
+                if e.kind == "pr" and e.number == terminal_pr and role in e.safe_for_roles
+            ),
+            None,
+        )
+        if terminal_entry is None:
+            return RoleNextAction(
+                role=role,
+                status="blocked_terminal",
+                target_kind="pr",
+                target_number=terminal_pr,
+                head_sha=None,
+                prompt=_prompt_for_role(
+                    role,
+                    remote=remote,
+                    org=org,
+                    repo=repo,
+                    terminal_pr=terminal_pr,
+                ),
+                reasons=(
+                    f"active terminal-review lock on PR #{terminal_pr}; "
+                    "no other review/merge target is safe",
+                ),
+            )
+        return RoleNextAction(
+            role=role,
+            status="safe",
+            target_kind="pr",
+            target_number=terminal_pr,
+            head_sha=terminal_entry.head_sha,
+            prompt=_prompt_for_role(role, remote=remote, org=org, repo=repo),
+            reasons=(f"terminal-path PR #{terminal_pr} is the only safe target",),
+        )
+
+    if role == ROLE_CONTROLLER:
+        needed = [e for e in entries if e.expected_role == ROLE_CONTROLLER or e.block_reason]
+        if not needed:
+            return RoleNextAction(
+                role=role,
+                status="idle",
+                target_kind=None,
+                target_number=None,
+                head_sha=None,
+                prompt=_prompt_for_role(
+                    role, remote=remote, org=org, repo=repo, idle=True
+                ),
+                reasons=("no controller-needed items",),
+            )
+        target = needed[0]
+        return RoleNextAction(
+            role=role,
+            status="safe",
+            target_kind=target.kind,
+            target_number=target.number,
+            head_sha=target.head_sha,
+            prompt=_prompt_for_role(role, remote=remote, org=org, repo=repo),
+            reasons=(target.block_reason or "controller diagnosis required",),
+        )
+
+    hit = _first_safe_for_role(entries, role)
+    if hit is None:
+        return RoleNextAction(
+            role=role,
+            status="idle",
+            target_kind=None,
+            target_number=None,
+            head_sha=None,
+            prompt=_prompt_for_role(
+                role, remote=remote, org=org, repo=repo, idle=True
+            ),
+            reasons=(f"no safe assignable work for role '{role}'",),
+        )
+    return RoleNextAction(
+        role=role,
+        status="safe",
+        target_kind=hit.kind,
+        target_number=hit.number,
+        head_sha=hit.head_sha,
+        prompt=_prompt_for_role(role, remote=remote, org=org, repo=repo),
+        reasons=(f"highest-ranked safe candidate for role '{role}'",),
+    )
+
+
+def build_workflow_dashboard(
+    *,
+    candidates: Sequence[WorkCandidate],
+    remote: str = "prgs",
+    org: str = "Scaled-Tech-Consulting",
+    repo: str = "Gitea-Tools",
+    leases: Sequence[dict[str, Any]] | None = None,
+    terminal_pr: int | None = None,
+    terminal_lock: dict[str, Any] | None = None,
+    inventory_complete: bool = True,
+    inventory_reasons: Sequence[str] | None = None,
+) -> DashboardSnapshot:
+    """Build a full dashboard snapshot from injected inventory (pure)."""
+    reasons = [str(r) for r in (inventory_reasons or ()) if str(r).strip()]
+    if not inventory_complete:
+        reasons.append(
+            "inventory incomplete: refuse to present partial queues as complete "
+            "(fail closed, #605/#758)"
+        )
+
+    ranked = sort_candidates(list(candidates))
+    entries = [
+        _entry_for_candidate(c, terminal_pr=terminal_pr) for c in ranked
+    ]
+
+    open_prs = [e for e in entries if e.kind == "pr"]
+    open_issues = [e for e in entries if e.kind == "issue"]
+    review_ready = [
+        e
+        for e in open_prs
+        if "review-ready" in e.badges
+        and e.block_reason is None
+        and ROLE_REVIEWER in e.safe_for_roles
+    ]
+    merge_ready = [
+        e
+        for e in open_prs
+        if "merge-ready" in e.badges
+        and e.block_reason is None
+        and ROLE_MERGER in e.safe_for_roles
+    ]
+    author_remediation = [
+        e
+        for e in open_prs
+        if "request-changes" in e.badges
+        and e.block_reason is None
+        and ROLE_AUTHOR in e.safe_for_roles
+    ]
+    discussion = [
+        e
+        for e in open_issues
+        if "discussion" in e.badges
+    ]
+    blocked = [e for e in entries if e.block_reason is not None]
+    controller_needed = [
+        e
+        for e in entries
+        if e.expected_role == ROLE_CONTROLLER
+        or (e.block_reason and "contaminated" in (e.badges or ()))
+        or "contaminated" in e.badges
+    ]
+
+    leases_by_role, stale_leases = _partition_leases(leases)
+
+    term_payload = None
+    if terminal_lock is not None:
+        term_payload = dict(terminal_lock)
+    elif terminal_pr is not None:
+        term_payload = {
+            "active": True,
+            "terminal_pr": terminal_pr,
+            "state": "locked",
+        }
+
+    next_by_role: dict[str, RoleNextAction] = {}
+    for role in DASHBOARD_ROLES:
+        next_by_role[role] = _role_next_action(
+            role,
+            entries=entries,
+            remote=remote,
+            org=org,
+            repo=repo,
+            terminal_pr=terminal_pr,
+        )
+
+    # Primary next action prefers in-flight PR work, then author issues.
+    primary: RoleNextAction | None = None
+    for role in (ROLE_REVIEWER, ROLE_MERGER, ROLE_AUTHOR, ROLE_RECONCILER, ROLE_CONTROLLER):
+        action = next_by_role[role]
+        if action.status == "safe":
+            primary = action
+            break
+    if primary is None:
+        # Prefer an explicit terminal block signal over generic idle.
+        for role in (ROLE_REVIEWER, ROLE_MERGER):
+            if next_by_role[role].status == "blocked_terminal":
+                primary = next_by_role[role]
+                break
+    if primary is None:
+        primary = next_by_role[ROLE_AUTHOR]
+
+    # Incomplete inventory: strip all safe flags / never suggest work.
+    if not inventory_complete:
+        for role, action in list(next_by_role.items()):
+            next_by_role[role] = RoleNextAction(
+                role=role,
+                status="none",
+                target_kind=None,
+                target_number=None,
+                head_sha=None,
+                prompt=(
+                    f"BLOCKED: inventory incomplete for {remote}/{org}/{repo}; "
+                    "do not select work. Re-run after a complete listing."
+                ),
+                reasons=tuple(reasons) or ("inventory incomplete",),
+            )
+        primary = next_by_role[ROLE_CONTROLLER]
+        review_ready = []
+        merge_ready = []
+        author_remediation = []
+
+    return DashboardSnapshot(
+        remote=remote,
+        org=org,
+        repo=repo,
+        inventory_complete=inventory_complete,
+        candidate_count=len(ranked),
+        open_prs=open_prs,
+        open_issues=open_issues,
+        review_ready_prs=review_ready,
+        merge_ready_prs=merge_ready,
+        author_remediation=author_remediation,
+        discussion_issues=discussion,
+        blocked_items=blocked,
+        controller_needed=controller_needed,
+        active_leases_by_role=leases_by_role,
+        stale_or_expired_leases=stale_leases,
+        terminal_review_lock=term_payload,
+        next_safe_by_role=next_by_role,
+        primary_next_safe_action=primary,
+        reasons=reasons,
+    )
+
+
+def format_human_summary(snapshot: DashboardSnapshot) -> str:
+    """Compact human-readable multi-line summary for menus and operators."""
+    lines: list[str] = []
+    lines.append(
+        f"Workflow dashboard v{snapshot.dashboard_version} — "
+        f"{snapshot.remote}/{snapshot.org}/{snapshot.repo}"
+    )
+    lines.append(
+        f"Inventory: complete={snapshot.inventory_complete} "
+        f"candidates={snapshot.candidate_count}"
+    )
+    if snapshot.terminal_review_lock:
+        tpr = snapshot.terminal_review_lock.get("terminal_pr")
+        lines.append(f"Terminal review lock: ACTIVE on PR #{tpr}")
+    else:
+        lines.append("Terminal review lock: none")
+
+    lines.append(
+        f"Open PRs: {len(snapshot.open_prs)} | Open issues: {len(snapshot.open_issues)}"
+    )
+    lines.append(
+        f"Review-ready: {len(snapshot.review_ready_prs)} | "
+        f"Merge-ready: {len(snapshot.merge_ready_prs)} | "
+        f"Author remediation: {len(snapshot.author_remediation)}"
+    )
+    lines.append(
+        f"Blocked: {len(snapshot.blocked_items)} | "
+        f"Controller-needed: {len(snapshot.controller_needed)} | "
+        f"Discussion: {len(snapshot.discussion_issues)}"
+    )
+
+    active_counts = {
+        role: len(items)
+        for role, items in snapshot.active_leases_by_role.items()
+        if items
+    }
+    if active_counts:
+        parts = [f"{role}={n}" for role, n in sorted(active_counts.items())]
+        lines.append("Active leases by role: " + ", ".join(parts))
+    else:
+        lines.append("Active leases by role: none")
+    lines.append(
+        f"Stale/expired leases: {len(snapshot.stale_or_expired_leases)}"
+    )
+
+    # Never list blocked items as safe.
+    if snapshot.blocked_items:
+        lines.append("Blocked (NOT safe):")
+        for entry in snapshot.blocked_items[:12]:
+            lines.append(
+                f"  - {entry.kind}#{entry.number}: {entry.block_reason}"
+            )
+            lines.append(
+                "    "
+                + PROMPT_BLOCKED_ITEM.format(
+                    kind=entry.kind,
+                    number=entry.number,
+                    reason=entry.block_reason or "blocked",
+                )
+            )
+
+    lines.append("Next safe action by role:")
+    for role in DASHBOARD_ROLES:
+        action = snapshot.next_safe_by_role.get(role)
+        if action is None:
+            continue
+        target = (
+            f"{action.target_kind}#{action.target_number}"
+            if action.target_number is not None
+            else "none"
+        )
+        lines.append(
+            f"  - {role}: status={action.status} target={target} "
+            f"safe={action.status == 'safe'}"
+        )
+        lines.append(f"    prompt: {action.prompt}")
+
+    if snapshot.primary_next_safe_action:
+        p = snapshot.primary_next_safe_action
+        lines.append(
+            f"Primary next: role={p.role} status={p.status} "
+            f"target={p.target_kind}#{p.target_number if p.target_number else 'none'}"
+        )
+        lines.append(f"  prompt: {p.prompt}")
+
+    if snapshot.reasons:
+        lines.append("Notes:")
+        for r in snapshot.reasons:
+            lines.append(f"  - {r}")
+
+    lines.append(
+        "Assignment still requires gitea_allocate_next_work; "
+        "this dashboard never self-selects exclusive work."
+    )
+    return "\n".join(lines)