docs/bootstrap-review-path.md

@@ -0,0 +1,190 @@
+# Bootstrap Review Path for Self-Hosted MCP Workflow Fixes (#557)
+
+## Purpose
+
+Gitea-Tools MCP workflow fixes can create a **bootstrap deadlock**: the live
+MCP daemons still run the broken code from `master`, so canonical reviewer /
+merger tools cannot complete the review that would land the fix.
+
+This document defines a **narrow, controller-authorized bootstrap path** so
+such fixes can land without weakening normal review/merge gates.
+
+It is **not** a general bypass. Normal PRs must still use the full
+reviewer → merger MCP workflow.
+
+## When this path applies
+
+All of the following must be true:
+
+1. The PR changes **Gitea-Tools MCP workflow / preflight / lease / gate** code
+   that the live daemon must load to review itself (self-hosted control plane).
+2. Canonical review or merge tools **fail closed** because of that defect
+   (documented tool errors, not “inconvenience”).
+3. A **controller** records an explicit bootstrap authorization on the PR or
+   linked issue (see below).
+4. The PR source is clean, testable, and free of unrelated scope.
+
+Example (historical): PR #553 / Issue #546 (reviewer preflight capability/lease
+deadlock). Live daemons on unpatched `master` could not complete canonical
+review of the fix PR.
+
+## Durable authorization (required)
+
+**No manual remote merge, force-merge, or API merge of a bootstrap PR is
+allowed** unless a controller has posted a durable authorization record on
+Gitea.
+
+### Authorization record (issue or PR comment)
+
+The controller comment **must** include a machine-readable marker and the
+fields below:
+
+```text
+## BOOTSTRAP REVIEW AUTHORIZATION (#557)
+
+Status: APPROVED
+PR: <number>
+Issue: <number>
+Controller: <gitea username>
+Reason: live MCP runtime cannot canonically review this self-hosted workflow fix
+Scope: narrow bootstrap only — does not weaken normal PR gates
+
+Validation evidence:
+- git diff --check: clean
+- targeted pytest: <command> → pass
+- full suite attribution (if non-zero): baseline compared to prgs/master
+
+Duplicate-PR audit: <none | list and disposition>
+Mutation ledger audit: <summary or path to proof>
+Contamination audit: <clean | list contaminated comment/lease IDs and disposition>
+
+Allowed verification actions used: <list>
+Forbidden actions NOT used: root checkout edits; raw curl mutation; direct
+module import of gitea_mcp_server for mutations; in-memory gate restoration
+
+Post-land plan:
+- restart MCP daemons for author/reviewer/merger/reconciler profiles
+- re-verify with canonical tools (see Post-bootstrap steps)
+```
+
+Without this record, bootstrap merge is **forbidden**. Agents must stop with
+`BLOCKED + DIAGNOSE` rather than improvise a merge.
+
+## Review gates and proof (still required)
+
+Bootstrap does **not** skip technical review. It only allows verification and
+landing when the **live MCP mutation path** cannot complete the review.
+
+Before authorization, the controller (or a delegated read-only verifier in an
+isolated worktree) must have:
+
+| Gate | Requirement |
+|------|-------------|
+| Diff hygiene | `git diff --check` clean on the PR tip vs `prgs/master` |
+| Tests | Targeted tests for the fix pass; full suite either green or failures attributed to baseline `prgs/master` |
+| Duplicate PR | Open-PR inventory shows no competing open PR for the same issue (or supersession is documented) |
+| Mutation ledger | Any claimed mutations are ledger-consistent; no silent root edits |
+| Contamination audit | PR/issue comments and leases classified as clean vs workflow-contaminated |
+| Scope | Diff limited to the bootstrap issue; no drive-by refactors |
+
+### Contamination audit (comments / leases)
+
+Comments or leases created via **raw curl**, **direct Python import of MCP
+modules**, or **token extraction** are **workflow-contaminated**. They must be
+listed and must **not** be treated as canonical review proof.
+
+Only comments posted via **sanctioned MCP reviewer tools** (after the fix is
+landed and daemons restarted, or during a clean path that did not bypass gates)
+count as canonical review state.
+
+Historical example on PR #553:
+
+| Comment | Disposition |
+|---------|-------------|
+| #7247 test / raw API | contaminated |
+| #7251 lease metadata / raw API | contaminated |
+| #7252 lease metadata / direct import | contaminated |
+| #7265 lease metadata / MCP reviewer tools | workflow-clean |
+
+## Allowed verification actions
+
+These may run **outside** the broken live mutation path to establish facts:
+
+- Read-only MCP tools (`gitea_view_pr`, `gitea_list_prs`, `gitea_whoami`,
+  `gitea_get_profile`, inventory, eligibility **reads**).
+- Isolated `branches/` worktree checkout of the PR head (never root).
+- `git fetch`, `git log`, `git diff`, `git merge-tree` (read-only analysis).
+- Targeted `pytest` / `py_compile` inside that worktree.
+- Controller posting of the bootstrap authorization comment via a healthy
+  MCP profile **if available**; if comment mutation is also deadlocked, a
+  **human operator** posts the authorization in the Gitea UI (still durable on
+  the thread).
+
+## Forbidden actions (always)
+
+Even under bootstrap:
+
+- Editing or committing in the **project root / control checkout**.
+- Treating root as a worktree for implementation or review mutations.
+- Raw `curl` / REST mutation with extracted tokens as a normal workflow.
+- `import gitea_mcp_server` (or sibling modules) from a shell to call mutation
+  helpers and bypass preflight.
+- In-memory restoration of capability / lease / decision state to “finish”
+  a blocked chain.
+- Force-push, history rewrite, or deleting evidence comments.
+- Weakening normal gates in code “temporarily” without a tracked issue/PR.
+- Self-review or self-merge by the PR author identity.
+
+## Landing procedure (after APPROVED authorization)
+
+1. Confirm the authorization record is present and complete on the PR or issue.
+2. Merge **only** with an independent merger identity when possible.
+   - Preferred: restart/replace the MCP merger daemon with a build that includes
+     the fix (or a one-shot process started from the PR worktree binary) so
+     `gitea_merge_pr` can run under normal gates.
+   - If the live merger daemon still cannot load the fix, a **human operator**
+     may merge via the Gitea UI **only** after the authorization record exists.
+3. Never merge from contaminated proof alone.
+
+## Post-bootstrap steps
+
+### 1. Restart MCP daemons
+
+After the fix lands on `prgs/master`:
+
+1. Stop author / reviewer / merger / reconciler MCP server processes for this
+   repo (IDE MCP pool + any long-lived terminals).
+2. Confirm no stale PID still serves the old code.
+3. Restart each profile so it loads the updated `master` tree (or installed
+   package path).
+4. Call `gitea_whoami` / `gitea_get_profile` on each namespace and record
+   profile name + identity.
+
+### 2. Re-verify with canonical tools
+
+Example pattern after PR #553-class fixes (lease release / #550-style follow-up):
+
+1. From a **reviewer** MCP session: acquire/release or inspect the relevant
+   lease with canonical tools only.
+2. Confirm no direct-import or raw-API path is required.
+3. Post a short controller or reconciler note: bootstrap complete; normal gates
+   restored.
+
+If verification still requires a bypass, open a new issue — do **not** extend
+this bootstrap authorization silently.
+
+## Explicit non-goals
+
+- This path does **not** authorize skipping tests, duplicate audits, or
+  contamination audits.
+- This path does **not** authorize root checkout implementation work.
+- This path does **not** replace BLOCKED + DIAGNOSE when a non-bootstrap
+  failure occurs (#552).
+- This path does **not** grant permanent “operator exception” culture.
+
+## Related
+
+- Root checkout policy: `docs/llm-workflow-runbooks.md` (Global LLM Worktree Rule, #475)
+- BLOCKED + DIAGNOSE: issue #552 / skill workflows
+- Reviewer lease / preflight deadlock class: issues #546, #548, #550
+- Durable session proofs across daemon pools: issue #559

docs/llm-workflow-runbooks.md

@@ -408,6 +408,28 @@ The guard blocks when the **control checkout** (repository root) is:
 `branches/...` directories are disposable role worktrees; the root checkout is
 the stable orchestration surface only.
 
+## Bootstrap Review Path for self-hosted MCP fixes (#557)
+
+When a PR fixes Gitea-Tools MCP workflow code that the **live daemon** still
+runs from broken `master`, canonical review can deadlock on itself.
+
+Do **not** improvise raw API, direct imports, root edits, or gate bypasses.
+
+Use the narrow controller-authorized path documented in:
+
+- [`docs/bootstrap-review-path.md`](bootstrap-review-path.md)
+
+Hard rules:
+
+- No merge without a durable `BOOTSTRAP REVIEW AUTHORIZATION (#557)` record on
+  the PR or linked issue.
+- Validation, duplicate-PR, mutation-ledger, and contamination audits remain
+  mandatory.
+- Root checkout stays read-only orchestration only; verification runs in
+  `branches/` worktrees.
+- After land: restart MCP daemons and re-verify with canonical tools only.
+- This never weakens normal reviewer/merger gates for ordinary PRs.
+
 ## Shell Spawn Hard-Stop Rule
 
 Symptom: a shell tool call returns `exit_code: -1` with empty stdout/stderr.
@@ -939,6 +961,21 @@ When posting a Canonical Thread Handoff after a binding blocker:
 - [`credential-isolation.md`](credential-isolation.md) — credential handling.
 - [`release-workflows.md`](release-workflows.md) — release/merge workflow.
 - [`../README.md`](../README.md) — canonical config, thin launchers, the menu.
+- [`state-handoff-ledger.md`](state-handoff-ledger.md) — canonical state comments and next-action handoff (#494).
+
+## Canonical state handoff ledger (#494)
+
+Gitea comments and final reports must make continuation obvious without chat
+history. See [`state-handoff-ledger.md`](state-handoff-ledger.md) for:
+
+- discussion → issue → PR → review → merge → reconcile lifecycle
+- templates for discussion, issue, PR, and queue-controller state comments
+- final-report requirements: Current status, Next actor, Next action, Next prompt
+- queue-controller priority order and discussion ≥5-comment rule (urgent/trivial
+  exceptions)
+
+Helpers live in `state_handoff_ledger.py`; final-report enforcement is wired
+through `assess_final_report_validator`.
 
 ## PR-only queue cleanup mode (#390)
 

docs/state-handoff-ledger.md

@@ -0,0 +1,86 @@
+# Canonical state handoff ledger (#494)
+
+Gitea is the system of record for continuation. Every discussion, issue, and PR
+should answer: current state, last proof, who acts next, and the exact prompt for
+the next role.
+
+## Lifecycle
+
+1. Controller opens a discussion.
+2. Discussion accumulates substantive comments (default minimum: five).
+3. Controller posts a discussion summary when ready.
+4. Controller creates linked issues from the summary.
+5. Author locks issue, implements, opens PR.
+6. Reviewer reviews at current head.
+7. Merger merges after formal approval.
+8. Reconciler closes superseded or already-landed PRs.
+9. Issue/PR/discussion state comments make the next action obvious at every step.
+
+## Discussion rules
+
+- Do **not** convert a discussion into issues until it has at least **five
+  substantive comments**, unless the discussion state comment marks
+  `URGENCY: urgent` or `URGENCY: trivial`.
+- Substantive comment types: proposal, risk/concern, acceptance criteria,
+  implementation approach, dependency/sequence, summary.
+- Before issue creation, post a **discussion summary** with decision, issues to
+  create, non-goals, unresolved questions, and next prompt.
+- Created issues must link back to the discussion; the discussion must link
+  forward to created issues.
+
+## State comment templates
+
+Use `state_handoff_ledger.py` helpers or copy the canonical blocks:
+
+- `render_discussion_state_comment(...)`
+- `render_discussion_summary_comment(...)`
+- `render_issue_state_comment(...)`
+- `render_pr_state_comment(...)`
+- `render_queue_controller_report(...)`
+
+Post state comments as the **latest canonical update** on the object. Do not
+bury state inside PR bodies only.
+
+## Final report requirements
+
+Every final report must include a `Controller Handoff` section with:
+
+- **Current status** — live state after this session
+- **Next actor** — `author`, `reviewer`, `merger`, `reconciler`, or `controller`
+- **Next action** — one imperative step
+- **Next prompt** — ready-to-paste prompt for the next role
+
+`assess_final_report_next_action_handoff` and
+`assess_contradictory_state_handoff` enforce these fields and reject
+contradictory claims (for example, ready-to-merge without approval, issue done
+without PR proof, discussion complete without summary).
+
+## Queue controller selection (priority order)
+
+1. Merge clean approved PRs at current head.
+2. Review PRs needing review.
+3. Reconcile superseded or already-landed duplicates.
+4. Continue blocked-but-now-unblocked issues.
+5. Create issues from completed discussions.
+6. Start new author work only when higher-priority queue items are clear.
+
+For each candidate object, read the **latest canonical state comment** before
+choosing an action. Output the exact next-role prompt in the controller report.
+
+## Example workflow states
+
+| State | Next actor | Typical next action |
+|-------|------------|---------------------|
+| Discussion needs more comments | controller | Facilitate discussion until five substantive comments or urgent/trivial exception |
+| Issue ready for author | author | Lock issue and implement in `branches/` worktree |
+| PR needs review | reviewer | Review at pinned head in reviewer worktree |
+| PR approved at head | merger | Merge with merger profile after eligibility proof |
+| PR superseded | reconciler | Close duplicate/already-landed PR with proof |
+| Issue blocked | controller | Post issue state comment with blockers and next prompt |
+
+## Non-goals
+
+- Chat history is not the source of truth.
+- Do not weaken author/reviewer/merger/reconciler separation.
+- Do not replace Gitea issues, PRs, or canonical workflow files under
+  `skills/llm-project-workflow/`.

final_report_validator.py

@@ -25,6 +25,7 @@ from review_proofs import (
     assess_review_mutation_final_report,
     assess_validation_report,
 )
+from state_handoff_ledger import assess_state_handoff_ledger_report
 from validation_status_vocabulary import assess_validation_status_vocabulary
 
 FINAL_REPORT_TASK_KINDS = frozenset({
@@ -336,6 +337,38 @@ def _rule_shared_controller_handoff(
     )
 
 
+def _rule_shared_state_handoff_next_action(report_text: str) -> list[dict[str, str]]:
+    result = assess_state_handoff_ledger_report(report_text)
+    if result.get("complete"):
+        return []
+    findings: list[dict[str, str]] = []
+    next_action = result.get("next_action") or {}
+    contradictions = result.get("contradictions") or {}
+    if next_action.get("block"):
+        findings.extend(
+            _findings_from_reasons(
+                "shared.state_handoff_next_action",
+                next_action.get("reasons") or [],
+                field="Next action handoff",
+                severity="block",
+                safe_next_action=next_action.get("safe_next_action")
+                or "add next-action handoff fields to Controller Handoff",
+            )
+        )
+    if contradictions.get("block"):
+        findings.extend(
+            _findings_from_reasons(
+                "shared.state_handoff_contradiction",
+                contradictions.get("reasons") or [],
+                field="State handoff",
+                severity="block",
+                safe_next_action=contradictions.get("safe_next_action")
+                or "resolve contradictory state claims before submission",
+            )
+        )
+    return findings
+
+
 def _rule_shared_email_disclosure(report_text: str) -> list[dict[str, str]]:
     result = assess_email_disclosure(report_text)
     if result.get("proven"):
@@ -716,6 +749,25 @@ def _rule_reviewer_main_checkout_baseline(report_text: str) -> list[dict[str, st
     ]
 
 
+def _rule_reviewer_premerge_baseline_proof(report_text: str) -> list[dict[str, str]]:
+    from premerge_baseline_proof import assess_premerge_baseline_proof
+
+    result = assess_premerge_baseline_proof(report_text)
+    if not result.get("block"):
+        return []
+    return _findings_from_reasons(
+        "reviewer.premerge_baseline_proof",
+        result.get("reasons") or [],
+        field="Pre-merge baseline proof",
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or (
+            "prove the failure on the PR pre-merge base commit or a known-failure "
+            "record predating the PR; current-master reproduction is not baseline proof"
+        ),
+    )
+
+
 def _rule_reviewer_validation_status_vocabulary(
     report_text: str,
     *,
@@ -1295,6 +1347,7 @@ _SHARED_CLEANUP_PROOF_RULES = (
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "review_pr": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_legacy_workspace_mutations,
@@ -1307,6 +1360,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_validation_structured,
         _rule_reviewer_linked_issue,
         _rule_reviewer_baseline_on_failure,
+        _rule_reviewer_premerge_baseline_proof,
         _rule_reviewer_validation_status_vocabulary,
         _rule_reviewer_main_checkout_baseline,
         _rule_reviewer_main_checkout_path,
@@ -1323,6 +1377,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     ],
     "reconcile_already_landed": [
         _rule_reconcile_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         *_SHARED_CLEANUP_PROOF_RULES,
@@ -1331,6 +1386,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reconcile_linked_issue_live,
         _rule_reconcile_close_proof,
         _rule_reconcile_pagination_proof,
+        _rule_reviewer_premerge_baseline_proof,
         _rule_reviewer_git_fetch_readonly,
         _rule_reviewer_legacy_workspace_mutations,
         _rule_reviewer_vague_mutations_none,
@@ -1338,12 +1394,14 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     ],
     "author_issue": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
     ],
     "work_issue": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_shared_issue_acceptance_gate,
@@ -1353,17 +1411,20 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     ],
     "issue_filing": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
     "inventory": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_pagination_proof,
     ],
     "issue_selection": [
         _rule_shared_controller_handoff,
+        _rule_shared_state_handoff_next_action,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
     ],

gitea_mcp_server.py

@@ -4881,6 +4881,34 @@ def gitea_reconcile_merged_cleanups(
         )
 
     delete_capability_allowed = not _profile_operation_gate("gitea.branch.delete")
+
+    # #534: discover reviewer scratch trees and active leases for those PRs.
+    scratch_candidates = merged_cleanup_reconcile.discover_reviewer_scratch_worktrees(
+        PROJECT_ROOT
+    )
+    active_reviewer_leases: dict[int, bool] = {}
+    pr_states: dict[int, dict] = {}
+    for scratch in scratch_candidates:
+        pr_num = int(scratch["pr_number"])
+        if pr_num in active_reviewer_leases:
+            continue
+        try:
+            comments = api_get_all(f"{base}/issues/{pr_num}/comments", auth)
+        except Exception:
+            comments = []
+        lease = reviewer_pr_lease.find_active_reviewer_lease(
+            comments, pr_number=pr_num
+        )
+        active_reviewer_leases[pr_num] = bool(lease)
+        try:
+            pr_live = api_request("GET", f"{base}/pulls/{pr_num}", auth)
+        except Exception:
+            pr_live = {}
+        pr_states[pr_num] = {
+            "merged": bool((pr_live or {}).get("merged") or (pr_live or {}).get("merged_at")),
+            "closed": (pr_live or {}).get("state") == "closed",
+        }
+
     report = merged_cleanup_reconcile.build_reconciliation_report(
         project_root=PROJECT_ROOT,
         closed_prs=merged_closed,
@@ -4888,6 +4916,8 @@ def gitea_reconcile_merged_cleanups(
         remote_branch_exists=remote_branch_exists,
         head_on_master=head_on_master,
         delete_capability_allowed=delete_capability_allowed,
+        active_reviewer_leases=active_reviewer_leases,
+        pr_states=pr_states,
     )
 
     if dry_run:
@@ -4931,6 +4961,14 @@ def gitea_reconcile_merged_cleanups(
             )
             actions.append({"action": "remove_local_worktree", **result})
 
+    for scratch in report.get("reviewer_scratch_entries") or []:
+        if not scratch.get("safe_to_remove_worktree"):
+            continue
+        result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree(
+            PROJECT_ROOT, scratch.get("worktree_path") or ""
+        )
+        actions.append({"action": "remove_reviewer_scratch_worktree", **result})
+
     report["dry_run"] = False
     report["executed"] = True
     report["actions"] = actions

merged_cleanup_reconcile.py

@@ -17,6 +17,11 @@ import issue_lock_store
 
 PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
 CLOSES_FIXES_RE = re.compile(r"\b(?:closes|fixes)\s+#(\d+)\b", re.IGNORECASE)
+# Reviewer scratch folders: review-pr<N> or review-pr<N>-submit (#534).
+REVIEWER_SCRATCH_NAME_RE = re.compile(
+    r"^review-pr(?P<pr>\d+)(?:-submit)?$",
+    re.IGNORECASE,
+)
 
 
 def extract_linked_issue(title: str, body: str) -> int | None:
@@ -36,6 +41,148 @@ def resolve_worktree_path(project_root: str, branch: str) -> str:
     return os.path.join(project_root, "branches", branch_worktree_folder(branch))
 
 
+def _git_worktree_porcelain(project_root: str) -> list[dict[str, str | None]]:
+    """Parse ``git worktree list --porcelain`` into structured records."""
+    res = subprocess.run(
+        ["git", "-C", project_root, "worktree", "list", "--porcelain"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if res.returncode != 0:
+        return []
+
+    records: list[dict[str, str | None]] = []
+    current: dict[str, str | None] = {}
+    for line in res.stdout.splitlines():
+        line = line.strip()
+        if not line:
+            if current.get("worktree"):
+                records.append(current)
+            current = {}
+            continue
+        if line.startswith("worktree "):
+            if current.get("worktree"):
+                records.append(current)
+            current = {"worktree": line[9:].strip(), "branch": None, "head": None}
+        elif line.startswith("branch ") and current:
+            ref = line[7:].strip()
+            current["branch"] = (
+                ref[11:] if ref.startswith("refs/heads/") else ref
+            )
+        elif line.startswith("HEAD ") and current:
+            current["head"] = line[5:].strip()
+        elif line == "detached" and current:
+            current["branch"] = None
+    if current.get("worktree"):
+        records.append(current)
+    return records
+
+
+def discover_local_worktrees(project_root: str) -> dict[str, str]:
+    """Return a mapping from branch name to absolute worktree path (#528)."""
+    mapping: dict[str, str] = {}
+    for record in _git_worktree_porcelain(project_root):
+        branch_name = record.get("branch")
+        path = record.get("worktree")
+        if branch_name and path:
+            mapping[str(branch_name)] = str(path)
+    return mapping
+
+
+def parse_reviewer_scratch_folder(folder_name: str) -> int | None:
+    """Return PR number for a reviewer scratch folder name, else None (#534)."""
+    match = REVIEWER_SCRATCH_NAME_RE.match((folder_name or "").strip())
+    if not match:
+        return None
+    return int(match.group("pr"))
+
+
+def discover_reviewer_scratch_worktrees(project_root: str) -> list[dict[str, Any]]:
+    """Discover detached reviewer scratch worktrees under ``branches/`` (#534).
+
+    Matches folder names ``review-pr<N>`` and ``review-pr<N>-submit`` only.
+    These are never treated as author worktree paths.
+    """
+    root = os.path.realpath(project_root)
+    branches_root = os.path.realpath(os.path.join(root, "branches"))
+    found: list[dict[str, Any]] = []
+    for record in _git_worktree_porcelain(project_root):
+        path = record.get("worktree")
+        if not path:
+            continue
+        real = os.path.realpath(path)
+        parent = os.path.dirname(real)
+        if parent != branches_root:
+            continue
+        folder = os.path.basename(real)
+        pr_number = parse_reviewer_scratch_folder(folder)
+        if pr_number is None:
+            continue
+        found.append(
+            {
+                "pr_number": pr_number,
+                "worktree_path": real,
+                "folder_name": folder,
+                "current_branch": record.get("branch"),
+                "head_sha": record.get("head"),
+                "worktree_kind": "reviewer_scratch",
+            }
+        )
+    return found
+
+
+def assess_reviewer_scratch_cleanup(
+    *,
+    pr_number: int,
+    worktree_path: str,
+    folder_name: str,
+    pr_merged: bool,
+    pr_closed: bool,
+    worktree_state: dict[str, Any],
+    active_reviewer_lease: bool,
+) -> dict[str, Any]:
+    """Assess whether a reviewer scratch worktree is safe to remove (#534)."""
+    reasons: list[str] = []
+    exists = bool(worktree_state.get("exists"))
+    if not (pr_merged or pr_closed):
+        reasons.append("associated PR is still open")
+    if not exists:
+        reasons.append("reviewer scratch worktree not present")
+    if exists and worktree_state.get("clean") is False:
+        dirty = worktree_state.get("dirty_files") or []
+        reasons.append(
+            "reviewer scratch worktree has tracked edits"
+            + (f" ({', '.join(dirty)})" if dirty else "")
+        )
+    if active_reviewer_lease:
+        reasons.append("active reviewer lease still requires this worktree")
+    current_branch = worktree_state.get("current_branch")
+    if current_branch:
+        # Scratch trees are expected detached; a named branch is ambiguous ownership.
+        reasons.append(
+            f"reviewer scratch worktree is on branch '{current_branch}' "
+            "(expected detached HEAD for review-pr scratch trees)"
+        )
+
+    safe = exists and not reasons
+    return {
+        "pr_number": pr_number,
+        "worktree_path": worktree_path,
+        "folder_name": folder_name,
+        "worktree_kind": "reviewer_scratch",
+        "worktree_exists": exists,
+        "worktree_clean": worktree_state.get("clean"),
+        "safe_to_remove_worktree": safe,
+        "block_reasons": reasons,
+        "recommended_action": (
+            "remove_reviewer_scratch_worktree" if safe else "keep_reviewer_scratch_worktree"
+        ),
+        # Never treat as author worktree cleanup.
+        "author_worktree_cleanup": False,
+    }
+
+
 def read_issue_lock(path: str | None = None) -> dict[str, Any] | None:
     if path:
         return issue_lock_store.read_lock_file(path.strip())
@@ -216,6 +363,7 @@ def build_pr_cleanup_entry(
     delete_capability_allowed: bool,
     issue_lock_path: str | None = None,
     protected_branches: frozenset[str] | None = None,
+    worktree_map: dict[str, str] | None = None,
 ) -> dict[str, Any]:
     pr_number = int(pr["number"])
     head_branch = pr.get("head") or ""
@@ -224,7 +372,16 @@ def build_pr_cleanup_entry(
     title = pr.get("title") or ""
     body = pr.get("body") or ""
     merged = bool(pr.get("merged_at"))
-    worktree_path = resolve_worktree_path(project_root, head_branch)
+
+    discovered_path = worktree_map.get(head_branch) if worktree_map else None
+    derived_path = resolve_worktree_path(project_root, head_branch)
+    if discovered_path:
+        worktree_path = discovered_path
+        match_type = "branch"
+    else:
+        worktree_path = derived_path
+        match_type = "path"
+
     worktree_state = read_local_worktree_state(worktree_path)
     worktree_state["worktree_path"] = worktree_path
     active_lock = has_active_issue_lock(head_branch, issue_lock_path)
@@ -247,6 +404,8 @@ def build_pr_cleanup_entry(
         worktree_state=worktree_state,
         active_lock=active_lock,
     )
+    local["match_type"] = match_type
+
     return {
         "pr_number": pr_number,
         "issue_number": extract_linked_issue(title, body),
@@ -270,8 +429,11 @@ def build_reconciliation_report(
     delete_capability_allowed: bool,
     issue_lock_path: str | None = None,
     protected_branches: frozenset[str] | None = None,
+    active_reviewer_leases: dict[int, bool] | None = None,
+    pr_states: dict[int, dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
     open_heads = collect_open_pr_heads(open_prs)
+    worktree_map = discover_local_worktrees(project_root)
     entries: list[dict[str, Any]] = []
     for pr in closed_prs or []:
         if not pr.get("merged_at"):
@@ -290,19 +452,63 @@ def build_reconciliation_report(
                 delete_capability_allowed=delete_capability_allowed,
                 issue_lock_path=issue_lock_path,
                 protected_branches=protected_branches,
+                worktree_map=worktree_map,
             )
         )
+
+    # #534: reviewer scratch worktrees are reported separately from author cleanup.
+    lease_map = active_reviewer_leases or {}
+    state_map = pr_states or {}
+    open_pr_numbers = {
+        int(pr["number"]) for pr in (open_prs or []) if pr.get("number") is not None
+    }
+    closed_by_number = {
+        int(pr["number"]): pr for pr in (closed_prs or []) if pr.get("number") is not None
+    }
+    scratch_entries: list[dict[str, Any]] = []
+    for scratch in discover_reviewer_scratch_worktrees(project_root):
+        pr_number = int(scratch["pr_number"])
+        state_info = state_map.get(pr_number) or {}
+        closed_pr = closed_by_number.get(pr_number)
+        if closed_pr is not None:
+            pr_merged = bool(closed_pr.get("merged_at") or closed_pr.get("merged"))
+            pr_closed = True
+        elif pr_number in open_pr_numbers:
+            pr_merged = False
+            pr_closed = False
+        else:
+            pr_merged = bool(state_info.get("merged"))
+            pr_closed = bool(state_info.get("closed") or state_info.get("merged"))
+        worktree_path = scratch["worktree_path"]
+        worktree_state = read_local_worktree_state(worktree_path)
+        worktree_state["worktree_path"] = worktree_path
+        # Prefer live branch from state (git) over porcelain branch field.
+        assessment = assess_reviewer_scratch_cleanup(
+            pr_number=pr_number,
+            worktree_path=worktree_path,
+            folder_name=scratch["folder_name"],
+            pr_merged=pr_merged,
+            pr_closed=pr_closed,
+            worktree_state=worktree_state,
+            active_reviewer_lease=bool(lease_map.get(pr_number)),
+        )
+        scratch_entries.append(assessment)
+
     return {
         "project_root": os.path.realpath(project_root),
         "merged_pr_count": len(entries),
         "entries": entries,
+        "reviewer_scratch_entries": scratch_entries,
+        "reviewer_scratch_count": len(scratch_entries),
         "dry_run": True,
         "executed": False,
     }
 
 
 def remove_local_worktree(project_root: str, branch: str) -> dict[str, Any]:
-    worktree_path = resolve_worktree_path(project_root, branch)
+    worktree_map = discover_local_worktrees(project_root)
+    discovered_path = worktree_map.get(branch)
+    worktree_path = discovered_path or resolve_worktree_path(project_root, branch)
     if not os.path.isdir(worktree_path):
         return {
             "success": False,
@@ -326,4 +532,65 @@ def remove_local_worktree(project_root: str, branch: str) -> dict[str, Any]:
         "performed": True,
         "message": f"removed worktree {worktree_path}",
         "worktree_path": worktree_path,
+    }
+
+
+def remove_reviewer_scratch_worktree(
+    project_root: str,
+    worktree_path: str,
+) -> dict[str, Any]:
+    """Remove a reviewer scratch worktree by absolute path (#534).
+
+    Fail closed unless the path is a direct ``branches/review-pr*`` child of
+    *project_root*. Never routes through author branch-name derivation.
+    """
+    root = os.path.realpath(project_root)
+    branches_root = os.path.realpath(os.path.join(root, "branches"))
+    real = os.path.realpath((worktree_path or "").strip())
+    folder = os.path.basename(real)
+    if os.path.dirname(real) != branches_root:
+        return {
+            "success": False,
+            "performed": False,
+            "worktree_kind": "reviewer_scratch",
+            "message": (
+                "reviewer scratch path must be a direct child of "
+                f"{branches_root}; got {real}"
+            ),
+        }
+    if parse_reviewer_scratch_folder(folder) is None:
+        return {
+            "success": False,
+            "performed": False,
+            "worktree_kind": "reviewer_scratch",
+            "message": f"path is not a reviewer scratch folder: {folder}",
+        }
+    if not os.path.isdir(real):
+        return {
+            "success": False,
+            "performed": False,
+            "worktree_kind": "reviewer_scratch",
+            "message": f"worktree not found: {real}",
+        }
+    res = subprocess.run(
+        ["git", "-C", project_root, "worktree", "remove", real],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if res.returncode != 0:
+        return {
+            "success": False,
+            "performed": False,
+            "worktree_kind": "reviewer_scratch",
+            "message": (res.stderr or res.stdout or "worktree remove failed").strip(),
+            "worktree_path": real,
+        }
+    return {
+        "success": True,
+        "performed": True,
+        "worktree_kind": "reviewer_scratch",
+        "message": f"removed reviewer scratch worktree {real}",
+        "worktree_path": real,
+        "author_worktree_cleanup": False,
     }

premerge_baseline_proof.py

@@ -0,0 +1,176 @@
+"""Pre-merge baseline proof verifier for non-zero validation exits (#533).
+
+A controller/reviewer may reproduce a failing test on *current* master and
+describe it as proof of a "known baseline failure". Reproducing a failure on
+post-merge master only proves the failure exists on current master — it does
+NOT prove the failure pre-existed the PR under review. A PR can introduce or
+preserve a regression, then once merged the failure appears on master and gets
+mislabeled "baseline", weakening validation gates.
+
+This module distinguishes four canonical validation outcomes:
+
+- ``CLEAN_PASS`` — the validation command exited zero.
+- ``CURRENT_MASTER_FAILURE_REPRODUCED`` — the same failure reproduces on
+  current (post-merge) master. Not sufficient as baseline proof.
+- ``PREMERGE_BASELINE_PROVEN_FAILURE`` — the failure is proven on the PR's
+  pre-merge base commit, or by a documented known-failure record that predates
+  the PR.
+- ``UNRESOLVED_REGRESSION_RISK`` — a non-zero exit that is neither a clean pass
+  nor proven pre-existing.
+
+A report may only call a non-zero validation exit a "baseline"/"pre-existing"
+failure when it carries pre-merge proof.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+CLEAN_PASS = "clean pass"
+CURRENT_MASTER_FAILURE_REPRODUCED = "current-master failure reproduced"
+PREMERGE_BASELINE_PROVEN_FAILURE = "pre-merge baseline-proven failure"
+UNRESOLVED_REGRESSION_RISK = "unresolved regression risk"
+
+# A non-zero validation exit is claimed somewhere in the report.
+_NONZERO_EXIT_RE = re.compile(
+    r"(?:exit(?:\s*(?:status|code))?\s*[:=]?\s*(?:[1-9]\d*)"
+    r"|\b\d+\s+failed\b"
+    r"|\bnon[- ]zero\s+(?:exit|validation))",
+    re.IGNORECASE,
+)
+# The report claims the failure is pre-existing / a baseline failure.
+_BASELINE_CLAIM_RE = re.compile(
+    r"(?:pre[- ]?existing(?:\s+(?:baseline|failure))?"
+    r"|baseline(?:[- ]proven)?\s+failure"
+    r"|known[- ]baseline"
+    r"|failure\s+is\s+baseline"
+    r"|already\s+fail(?:s|ing)\s+on\s+master)",
+    re.IGNORECASE,
+)
+# Only-current-master reproduction language (insufficient on its own).
+_CURRENT_MASTER_RE = re.compile(
+    r"(?:current[- ]master\s+(?:reproduc|failure)"
+    r"|reproduc\w*\s+on\s+(?:current\s+)?master"
+    r"|post[- ]merge\s+master"
+    r"|fails\s+on\s+(?:current\s+)?master\s+(?:now|too|as\s+well))",
+    re.IGNORECASE,
+)
+# Pre-merge base proof: a base commit tested before the PR landed.
+_PREMERGE_BASE_RE = re.compile(
+    r"(?:pre[- ]merge\s+base(?:\s+commit)?"
+    r"|pr\s+base\s+commit"
+    r"|merge[- ]base(?:\s+commit)?"
+    r"|base\s+commit\s+before\s+(?:the\s+)?pr)",
+    re.IGNORECASE,
+)
+# Documented known-failure record predating the PR.
+_KNOWN_FAILURE_RECORD_RE = re.compile(
+    r"known[- ]failure\s+(?:record|reference|ledger)\b.{0,80}?"
+    r"(?:predat|before\s+(?:the\s+)?pr|pre[- ]existing|prior\s+to\s+(?:the\s+)?pr)",
+    re.IGNORECASE | re.DOTALL,
+)
+
+# Required proof fields for a pre-merge baseline claim.
+_FIELD_PATTERNS: dict[str, re.Pattern[str]] = {
+    "base commit": re.compile(
+        r"(?:pre[- ]merge\s+)?base\s+commit\s*[:=]\s*([0-9a-f]{7,40})", re.IGNORECASE
+    ),
+    "tested commit": re.compile(
+        r"tested\s+commit\s*[:=]\s*([0-9a-f]{7,40})", re.IGNORECASE
+    ),
+    "command": re.compile(r"command\s*[:=]\s*\S", re.IGNORECASE),
+    "exit status": re.compile(r"exit\s*(?:status|code)?\s*[:=]\s*\d+", re.IGNORECASE),
+    "failure signature": re.compile(
+        r"failure\s+signature\s*[:=]\s*\S", re.IGNORECASE
+    ),
+}
+
+
+def assess_premerge_baseline_proof(report_text: str) -> dict[str, Any]:
+    """Assess whether a claimed baseline failure carries pre-merge proof.
+
+    Returns a dict with ``proven``, ``block``, ``label``, ``reasons``,
+    ``skipped`` and ``safe_next_action``. Only blocks when the report claims a
+    non-zero validation exit is baseline/pre-existing without valid pre-merge
+    proof.
+    """
+    text = report_text or ""
+
+    has_failure = bool(_NONZERO_EXIT_RE.search(text))
+    claims_baseline = bool(_BASELINE_CLAIM_RE.search(text))
+
+    # No failure claimed, or no baseline assertion — nothing to enforce here.
+    if not has_failure and not claims_baseline:
+        return {
+            "proven": True,
+            "block": False,
+            "label": CLEAN_PASS,
+            "reasons": [],
+            "skipped": True,
+            "safe_next_action": "",
+        }
+
+    if not claims_baseline:
+        # A non-zero exit not asserted as baseline — surface as regression risk,
+        # but do not block (the report never claimed a clean/baseline pass).
+        return {
+            "proven": True,
+            "block": False,
+            "label": UNRESOLVED_REGRESSION_RISK,
+            "reasons": [],
+            "skipped": False,
+            "safe_next_action": "",
+        }
+
+    has_premerge_base = bool(_PREMERGE_BASE_RE.search(text))
+    has_known_record = bool(_KNOWN_FAILURE_RECORD_RE.search(text))
+    only_current_master = bool(_CURRENT_MASTER_RE.search(text))
+
+    reasons: list[str] = []
+
+    if not (has_premerge_base or has_known_record):
+        if only_current_master:
+            reasons.append(
+                "baseline failure claimed from current-master reproduction only; "
+                "that proves the failure on current master, not that it pre-existed "
+                "the PR — provide pre-merge base proof or a known-failure record"
+            )
+        else:
+            reasons.append(
+                "baseline/pre-existing failure claimed without pre-merge proof "
+                "(no pre-merge base commit and no documented known-failure record "
+                "predating the PR)"
+            )
+
+    # Require the concrete proof fields regardless of proof source.
+    missing = [name for name, pat in _FIELD_PATTERNS.items() if not pat.search(text)]
+    if missing:
+        reasons.append(
+            "pre-merge baseline claim missing required proof field(s): "
+            + ", ".join(missing)
+        )
+
+    if reasons:
+        return {
+            "proven": False,
+            "block": True,
+            "label": UNRESOLVED_REGRESSION_RISK,
+            "reasons": reasons,
+            "skipped": False,
+            "safe_next_action": (
+                "prove the failure on the PR pre-merge base commit (or cite a "
+                "known-failure record predating the PR) and state base commit, "
+                "tested commit, command, exit status, and failure signature; "
+                "current-master reproduction alone is not baseline proof"
+            ),
+        }
+
+    return {
+        "proven": True,
+        "block": False,
+        "label": PREMERGE_BASELINE_PROVEN_FAILURE,
+        "reasons": [],
+        "skipped": False,
+        "safe_next_action": "",
+    }

skills/llm-project-workflow/SKILL.md

@@ -182,4 +182,17 @@ Ready-to-copy task prompts live in [`templates/`](templates/):
 
 Releases follow SemVer from remote `master` only, after full test suite passes.
 See [`templates/release-tag.md`](templates/release-tag.md) and
-`scripts/release-tag`.
+`scripts/release-tag`.
+
+## Bootstrap Review Path (#557)
+
+Self-hosted MCP workflow fixes can deadlock live review daemons. Do not bypass
+gates with raw API, direct imports, or root checkout edits.
+
+If and only if a controller posts a durable `BOOTSTRAP REVIEW AUTHORIZATION
+(#557)` record, follow:
+
+`docs/bootstrap-review-path.md`
+
+Otherwise stop with BLOCKED + DIAGNOSE. Bootstrap never weakens normal PR gates.
+

skills/llm-project-workflow/templates/pr-queue-cleanup.md

@@ -16,7 +16,7 @@ Rules:
   reviewing a second PR after a terminal mutation in this run.
 - After REQUEST_CHANGES: stop. After APPROVED: merge only this PR and only if
   operator explicitly authorized merge for this PR in this run.
-- Report Next suggested PR without continuing to it.
+- Report Next suggested PR without continuing to it. If the PR queue is empty, look at approvals, or issues next.
 
 Operator PR list (optional): <pr numbers or "oldest eligible from inventory">
 Merge authorized for selected PR in this run: <true|false>

skills/llm-project-workflow/templates/review-pr.md

@@ -28,7 +28,8 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
   `pr_inventory_trust_gate.status`, trust-gate reasons, corroboration,
   remote/owner/repo/state filter, and the inventory MCP profile. A recent merge
   commit is not valid corroboration. Author-bound sessions must not present
-  reviewer queue inventory as a reviewer decision.
+  reviewer queue inventory as a reviewer decision. If the PR queue is empty,
+  look at approvals, or issues next.
 
 Load the canonical workflow first:
 `skills/llm-project-workflow/workflows/review-merge-pr.md` (task mode: review-merge-pr).
@@ -126,4 +127,14 @@ including the review/merge role fields: Selected PR, Reviewer eligibility,
 Pinned reviewed head, Review decision, Merge result, Linked issue status,
 Cleanup status. If you could not merge, name the exact gate. Reports missing
 the handoff are downgraded (review_proofs.assess_controller_handoff).
+
+Baseline failure proof (#533): if a validation command exits non-zero, do NOT
+call it a clean pass. Only label a failure "baseline"/"pre-existing" with
+pre-merge proof — the failure reproduced on the PR's pre-merge base commit, or a
+documented known-failure record predating the PR. Reproducing on current
+(post-merge) master is "current-master failure reproduced", NOT baseline proof.
+State: base commit, tested commit, command, exit status, failure signature.
+Use one label: clean pass / current-master failure reproduced / pre-merge
+baseline-proven failure / unresolved regression risk
+(final_report_validator: reviewer.premerge_baseline_proof).
 ```

skills/llm-project-workflow/workflows/pr-queue-cleanup.md

@@ -55,6 +55,10 @@ claim. Select exactly one PR according to project queue ordering rules
 PRs only with live per-PR proof. No multi-PR validation and no batch report
 may substitute for per-PR proof.
 
+If the open PR queue is empty:
+* First, look at **Approvals** next: check if there are open PRs with pending/completed approvals requiring attention or merge.
+* Next, look at **Issues** next: check if there are unresolved open issues requiring action/fixes.
+
 ## 4. Terminal mutation chain
 
 `pr_queue_cleanup.resolve_cleanup_run_state` is the authority:

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -272,6 +272,10 @@ If queue ordering cannot be proven, stop and produce a recovery handoff.
 
 ## 10. Select the next actionable PR using project rules
 
+If the open PR queue is empty:
+* First, look at approvals next to see if there are pending approvals or approved PRs that need attention/merge.
+* Next, look at issues next to see if there are open issues requiring action/fixes.
+
 Do not review your own PR.
 
 Do not review stale, draft, blocked, duplicate, already-owned, dependency-blocked, already-landed, already-requested-changes work unless the rules explicitly allow it.

state_handoff_ledger.py

@@ -0,0 +1,374 @@
+"""Canonical state handoff ledger for discussions, issues, and PRs (#494)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+HANDOFF_HEADING = "Controller Handoff"
+
+ISSUE_STATE_FIELDS = (
+    "STATE",
+    "NEXT_ACTOR",
+    "NEXT_ACTION",
+    "NEXT_PROMPT",
+    "STATUS",
+    "RELATED_PRS",
+    "BRANCH",
+    "HEAD_SHA",
+    "VALIDATION",
+    "BLOCKERS",
+    "DUPLICATES_OR_SUPERSEDES",
+    "LAST_UPDATED_BY",
+)
+
+PR_STATE_FIELDS = (
+    "STATE",
+    "NEXT_ACTOR",
+    "NEXT_ACTION",
+    "NEXT_PROMPT",
+    "ISSUE",
+    "BASE",
+    "HEAD",
+    "HEAD_SHA",
+    "REVIEW_STATUS",
+    "VALIDATION",
+    "BLOCKERS",
+    "SUPERSEDES",
+    "SUPERSEDED_BY",
+    "MERGE_READY",
+    "LAST_UPDATED_BY",
+)
+
+DISCUSSION_STATE_FIELDS = (
+    "STATE",
+    "NEXT_ACTOR",
+    "NEXT_ACTION",
+    "NEXT_PROMPT",
+    "COMMENT_COUNT",
+    "SUBSTANTIVE_COMMENTS",
+    "URGENCY",
+    "BLOCKERS",
+    "LAST_UPDATED_BY",
+)
+
+DISCUSSION_SUMMARY_FIELDS = (
+    "DECISION",
+    "ISSUES_TO_CREATE",
+    "NON_GOALS",
+    "UNRESOLVED_QUESTIONS",
+    "NEXT_PROMPT",
+    "LAST_UPDATED_BY",
+)
+
+QUEUE_CONTROLLER_FIELDS = (
+    "QUEUE_STATE",
+    "SELECTED_OBJECT",
+    "SELECTED_OBJECT_TYPE",
+    "NEXT_ACTOR",
+    "NEXT_ACTION",
+    "NEXT_PROMPT",
+    "EVIDENCE",
+    "LAST_UPDATED_BY",
+)
+
+FINAL_REPORT_NEXT_ACTION_FIELDS = (
+    ("Current status", ("current status",)),
+    ("Next actor", ("next actor", "next_actor")),
+    ("Next action", ("next action", "next_action")),
+    ("Next prompt", ("next prompt", "next_prompt")),
+)
+
+STATE_COMMENT_KINDS = frozenset({
+    "issue_state",
+    "pr_state",
+    "discussion_state",
+    "discussion_summary",
+    "queue_controller",
+})
+
+_FIELDS_BY_KIND = {
+    "issue_state": ISSUE_STATE_FIELDS,
+    "pr_state": PR_STATE_FIELDS,
+    "discussion_state": DISCUSSION_STATE_FIELDS,
+    "discussion_summary": DISCUSSION_SUMMARY_FIELDS,
+    "queue_controller": QUEUE_CONTROLLER_FIELDS,
+}
+
+_FIELD_LINE_RE = re.compile(
+    r"^([A-Z][A-Z0-9_]+)\s*:\s*(.*)$",
+    re.MULTILINE,
+)
+_HANDOFF_SECTION_RE = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M)
+_READY_TO_MERGE_RE = re.compile(
+    r"\b(?:ready[_ ]to[_ ]merge|merge[_ ]ready\s*:\s*(?:yes|true))\b",
+    re.IGNORECASE,
+)
+_APPROVAL_PROOF_RE = re.compile(
+    r"\b(?:review decision\s*:\s*approve|approved at|approval at current head|"
+    r"formal approval|review_status\s*:\s*approved)\b",
+    re.IGNORECASE,
+)
+_EXTERNAL_NONE_RE = re.compile(
+    r"^\s*[-*]?\s*external[- ]state mutations\s*:\s*none\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_LOCK_MUTATION_RE = re.compile(
+    r"\b(?:gitea_lock_issue|issue-locks/|lock file edited)\b",
+    re.IGNORECASE,
+)
+_ISSUE_DONE_RE = re.compile(
+    r"\b(?:issue (?:done|closed)\b|current status\s*:\s*(?:done|closed)\b|"
+    r"current status\s*:\s*issue complete\b)",
+    re.IGNORECASE,
+)
+_PR_MERGE_PROOF_RE = re.compile(
+    r"\b(?:pr (?:merged|number)|merge result\s*:\s*merged|pr mutations\s*:\s*created)\b",
+    re.IGNORECASE,
+)
+_DISCUSSION_COMPLETE_RE = re.compile(
+    r"\b(?:discussion complete|discussion ready for issue creation)\b",
+    re.IGNORECASE,
+)
+_DISCUSSION_SUMMARY_PROOF_RE = re.compile(
+    r"\b(?:discussion summary|issues to create|issues created|linked issues)\b",
+    re.IGNORECASE,
+)
+
+MIN_DISCUSSION_COMMENTS_DEFAULT = 5
+
+
+def _render_fields(fields: tuple[str, ...], values: dict[str, Any]) -> str:
+    lines = ["```text"]
+    for name in fields:
+        raw = values.get(name, values.get(name.lower(), "none"))
+        text = str(raw).strip() if raw is not None else "none"
+        lines.append(f"{name}: {text or 'none'}")
+    lines.append("```")
+    return "\n".join(lines)
+
+
+def render_issue_state_comment(**values: Any) -> str:
+    """Render canonical issue state comment body."""
+    return _render_fields(ISSUE_STATE_FIELDS, values)
+
+
+def render_pr_state_comment(**values: Any) -> str:
+    """Render canonical PR state comment body."""
+    return _render_fields(PR_STATE_FIELDS, values)
+
+
+def render_discussion_state_comment(**values: Any) -> str:
+    """Render canonical discussion state comment body."""
+    return _render_fields(DISCUSSION_STATE_FIELDS, values)
+
+
+def render_discussion_summary_comment(**values: Any) -> str:
+    """Render discussion-to-issue conversion summary comment."""
+    return _render_fields(DISCUSSION_SUMMARY_FIELDS, values)
+
+
+def render_queue_controller_report(**values: Any) -> str:
+    """Render queue-controller selection report."""
+    return _render_fields(QUEUE_CONTROLLER_FIELDS, values)
+
+
+def parse_state_comment(text: str) -> dict[str, str]:
+    """Parse KEY: value lines from a canonical state comment."""
+    parsed: dict[str, str] = {}
+    for match in _FIELD_LINE_RE.finditer(text or ""):
+        parsed[match.group(1)] = match.group(2).strip()
+    return parsed
+
+
+def _handoff_field_map(report_text: str) -> dict[str, str]:
+    section_match = _HANDOFF_SECTION_RE.search(report_text or "")
+    if not section_match:
+        return {}
+    section = (report_text or "")[section_match.end():]
+    fields: dict[str, str] = {}
+    for line in section.splitlines():
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" not in stripped:
+            continue
+        key, value = stripped.split(":", 1)
+        fields[key.strip().lower()] = value.strip()
+    return fields
+
+
+def assess_state_comment(text: str, *, kind: str) -> dict[str, Any]:
+    """Validate a canonical Gitea state comment for *kind*."""
+    normalized = (kind or "").strip().lower()
+    if normalized not in STATE_COMMENT_KINDS:
+        return {
+            "complete": False,
+            "block": True,
+            "missing_fields": [],
+            "reasons": [f"unknown state comment kind '{kind}'"],
+            "safe_next_action": "use a supported state comment kind",
+        }
+
+    required = _FIELDS_BY_KIND[normalized]
+    parsed = parse_state_comment(text)
+    missing = [name for name in required if name not in parsed or not parsed[name].strip()]
+    reasons = [f"state comment missing field: {name}" for name in missing]
+
+    if normalized == "discussion_state":
+        urgency = parsed.get("URGENCY", "").strip().lower()
+        count_raw = parsed.get("COMMENT_COUNT", "").strip()
+        try:
+            count = int(count_raw)
+        except (TypeError, ValueError):
+            count = -1
+        if (
+            urgency not in {"urgent", "trivial"}
+            and count >= 0
+            and count < MIN_DISCUSSION_COMMENTS_DEFAULT
+        ):
+            reasons.append(
+                f"discussion has {count} comments; need at least "
+                f"{MIN_DISCUSSION_COMMENTS_DEFAULT} substantive comments "
+                "or URGENCY: urgent|trivial"
+            )
+
+    block = bool(missing or reasons)
+    return {
+        "complete": not block,
+        "block": block,
+        "missing_fields": missing,
+        "parsed_fields": parsed,
+        "reasons": reasons,
+        "safe_next_action": (
+            "fill all required state comment fields before posting"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+def assess_final_report_next_action_handoff(report_text: str) -> dict[str, Any]:
+    """Require current status plus next actor/action/prompt in Controller Handoff."""
+    fields = _handoff_field_map(report_text)
+    if not fields:
+        return {
+            "complete": False,
+            "block": True,
+            "missing_fields": [name for name, _ in FINAL_REPORT_NEXT_ACTION_FIELDS],
+            "reasons": [f"final report has no section titled exactly '{HANDOFF_HEADING}'"],
+            "safe_next_action": f"add a complete {HANDOFF_HEADING} section",
+        }
+
+    missing = []
+    for name, aliases in FINAL_REPORT_NEXT_ACTION_FIELDS:
+        value = ""
+        for alias in aliases:
+            if alias in fields:
+                value = fields[alias]
+                break
+        if not value or value.lower() in {"none", "n/a", "not verified in this session", ""}:
+            missing.append(name)
+
+    reasons = [f"handoff missing next-action field: {name}" for name in missing]
+    block = bool(missing)
+    return {
+        "complete": not block,
+        "block": block,
+        "missing_fields": missing,
+        "reasons": reasons,
+        "safe_next_action": (
+            "add Current status, Next actor, Next action, and Next prompt to Controller Handoff"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+def assess_contradictory_state_handoff(report_text: str) -> dict[str, Any]:
+    """Fail closed on contradictory queue/state claims in final reports."""
+    text = report_text or ""
+    fields = _handoff_field_map(text)
+    reasons: list[str] = []
+
+    review_decision = fields.get("review decision", fields.get("review status", ""))
+    has_approval = (
+        review_decision.lower() in {"approve", "approved"}
+        or bool(_APPROVAL_PROOF_RE.search(text))
+    )
+    if _READY_TO_MERGE_RE.search(text) and not has_approval:
+        reasons.append(
+            "report claims ready to merge without approval-at-current-head proof"
+        )
+
+    lock_val = fields.get("external-state mutations", "")
+    mutation_lines = " ".join(
+        fields.get(key, "")
+        for key in (
+            "issue mutations",
+            "mcp/gitea mutations",
+            "external-state mutations",
+            "claim/lock state",
+        )
+    )
+    if (
+        (_EXTERNAL_NONE_RE.search(text) or lock_val.lower() == "none")
+        and _LOCK_MUTATION_RE.search(mutation_lines)
+    ):
+        reasons.append(
+            "report claims External-state mutations: none while lock "
+            "activity is documented in mutation fields"
+        )
+
+    if _ISSUE_DONE_RE.search(text) and not _PR_MERGE_PROOF_RE.search(text):
+        reasons.append(
+            "report claims issue done/complete without PR or merge proof"
+        )
+
+    if _DISCUSSION_COMPLETE_RE.search(text) and not _DISCUSSION_SUMMARY_PROOF_RE.search(text):
+        reasons.append(
+            "report claims discussion complete without summary or issue-creation proof"
+        )
+
+    review_status = fields.get("review status", fields.get("review decision", ""))
+    merge_ready = fields.get("merge ready", "")
+    if (
+        merge_ready.lower() in {"yes", "true", "ready"}
+        and review_status.lower() not in {"approve", "approved"}
+        and not _APPROVAL_PROOF_RE.search(text)
+    ):
+        reasons.append(
+            "MERGE_READY or merge-ready claim without approved review status"
+        )
+
+    block = bool(reasons)
+    return {
+        "complete": not block,
+        "block": block,
+        "reasons": reasons,
+        "safe_next_action": (
+            "correct contradictory state claims or add missing proof"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+def assess_state_handoff_ledger_report(report_text: str) -> dict[str, Any]:
+    """Composite #494 assessment for final reports."""
+    next_action = assess_final_report_next_action_handoff(report_text)
+    contradictions = assess_contradictory_state_handoff(report_text)
+    reasons = list(next_action.get("reasons") or []) + list(contradictions.get("reasons") or [])
+    block = bool(next_action.get("block") or contradictions.get("block"))
+    return {
+        "complete": not block,
+        "block": block,
+        "next_action": next_action,
+        "contradictions": contradictions,
+        "reasons": reasons,
+        "safe_next_action": (
+            next_action.get("safe_next_action")
+            if next_action.get("block")
+            else contradictions.get("safe_next_action")
+            if contradictions.get("block")
+            else "proceed"
+        ),
+    }

tests/test_bootstrap_review_path_docs.py

@@ -0,0 +1,81 @@
+"""Doc-contract checks for the bootstrap review path (#557).
+
+Self-hosted MCP workflow fixes can deadlock live review daemons. These tests
+pin the narrow controller-authorized bootstrap path so it cannot silently
+regress into a general gate bypass.
+"""
+
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+
+DOC = REPO_ROOT / "docs" / "bootstrap-review-path.md"
+RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
+SKILL = REPO_ROOT / "skills" / "llm-project-workflow" / "SKILL.md"
+
+
+def _normalize(text: str) -> str:
+    return " ".join(text.split())
+
+
+def test_bootstrap_doc_exists_and_names_authorization_marker():
+    text = DOC.read_text(encoding="utf-8")
+    assert "BOOTSTRAP REVIEW AUTHORIZATION (#557)" in text
+    assert "bootstrap deadlock" in text.lower() or "Bootstrap deadlock" in text
+
+
+def test_bootstrap_doc_requires_validation_and_audits():
+    text = _normalize(DOC.read_text(encoding="utf-8"))
+    for phrase in (
+        "git diff --check",
+        "Duplicate-PR audit",
+        "Mutation ledger audit",
+        "Contamination audit",
+        "workflow-contaminated",
+    ):
+        assert phrase in text, f"bootstrap doc missing: {phrase!r}"
+
+
+def test_bootstrap_doc_lists_allowed_and_forbidden_actions():
+    text = _normalize(DOC.read_text(encoding="utf-8"))
+    lower = text.lower()
+    for phrase in (
+        "Allowed verification actions",
+        "Forbidden actions",
+        "project root",
+        "direct Python import",
+        "branches/",
+    ):
+        assert phrase in text, f"bootstrap doc missing: {phrase!r}"
+    assert "curl" in lower
+    assert "raw" in lower
+
+
+def test_bootstrap_doc_forbids_general_gate_weakening():
+    text = _normalize(DOC.read_text(encoding="utf-8")).lower()
+    assert "never weakens normal" in text or "does not weaken normal" in text
+    assert "general bypass" in text or "general gate" in text
+
+
+def test_bootstrap_doc_post_land_restart_and_verify():
+    text = _normalize(DOC.read_text(encoding="utf-8"))
+    for phrase in (
+        "Restart MCP daemons",
+        "canonical tools",
+        "gitea_whoami",
+    ):
+        assert phrase in text, f"post-bootstrap guidance missing: {phrase!r}"
+
+
+def test_runbook_links_bootstrap_path():
+    text = _normalize(RUNBOOK.read_text(encoding="utf-8"))
+    assert "Bootstrap Review Path for self-hosted MCP fixes (#557)" in text
+    assert "bootstrap-review-path.md" in text
+    assert "BOOTSTRAP REVIEW AUTHORIZATION (#557)" in text
+
+
+def test_skill_links_bootstrap_path():
+    text = _normalize(SKILL.read_text(encoding="utf-8"))
+    assert "Bootstrap Review Path (#557)" in text
+    assert "bootstrap-review-path.md" in text
+    assert "BLOCKED + DIAGNOSE" in text or "BLOCKED" in text

tests/test_final_report_validator.py

@@ -35,6 +35,9 @@ def _review_handoff(**overrides):
         "- External-state mutations: none",
         "- Read-only diagnostics: issue and PR metadata inspected",
         "- Current status: PR open",
+        "- Next actor: author",
+        "- Next action: address review feedback",
+        "- Next prompt: Fix PR #203 per reviewer request_changes and push updated head",
         "- Blockers: none",
         "- Next: await author",
         "- Safety: no self-review; no self-merge",
@@ -92,6 +95,9 @@ def _reconcile_handoff(drop=(), **overrides):
         "- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
         "- Reconciliation mutations: PR comment posted",
         "- Current status: reconciliation complete",
+        "- Next actor: reconciler",
+        "- Next action: close PR #99 after capability proof",
+        "- Next prompt: Reconcile already-landed PR #99 with prgs-reconciler profile",
         "- Blocker: missing gitea.pr.close capability",
         "- Safe next action: hand off to a profile with gitea.pr.close",
         "- No review/merge confirmation: confirmed",

tests/test_merged_cleanup_reconcile.py

@@ -4,7 +4,7 @@ import os
 import sys
 import tempfile
 import unittest
-from unittest.mock import patch
+from unittest.mock import patch, MagicMock
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
@@ -154,6 +154,290 @@ class TestMergedCleanupReport(unittest.TestCase):
         finally:
             os.remove(lock_path)
 
+    @patch("subprocess.run")
+    def test_discover_local_worktrees(self, mock_run):
+        mock_output = (
+            "worktree /repo/Gitea-Tools\n"
+            "bare\n\n"
+            "worktree /repo/Gitea-Tools/branches/custom-name\n"
+            "branch refs/heads/feat/issue-11-x\n"
+            "HEAD cafebabe\n\n"
+        )
+        mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
+        res = mcr.discover_local_worktrees("/repo/Gitea-Tools")
+        self.assertEqual(res, {"feat/issue-11-x": "/repo/Gitea-Tools/branches/custom-name"})
+
+    @patch("subprocess.run")
+    @patch("merged_cleanup_reconcile.read_local_worktree_state")
+    def test_build_report_discovers_non_canonical_worktree_by_branch(self, mock_state, mock_run):
+        mock_output = (
+            "worktree /repo/Gitea-Tools\n"
+            "bare\n\n"
+            "worktree /repo/Gitea-Tools/branches/custom-name\n"
+            "branch refs/heads/feat/issue-11-x\n"
+            "HEAD cafebabe\n\n"
+        )
+        mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
+        mock_state.return_value = {
+            "exists": True,
+            "clean": True,
+            "current_branch": "feat/issue-11-x",
+            "head_sha": "cafebabe",
+            "dirty_files": [],
+        }
+
+        report = mcr.build_reconciliation_report(
+            project_root="/repo/Gitea-Tools",
+            closed_prs=[
+                {
+                    "number": 11,
+                    "title": "merged",
+                    "body": "Closes #11",
+                    "head": {"ref": "feat/issue-11-x"},
+                    "merged_at": "2026-07-06T12:00:00Z",
+                    "merge_commit_sha": "abc123",
+                }
+            ],
+            open_prs=[],
+            remote_branch_exists={"feat/issue-11-x": True},
+            head_on_master={11: True},
+            delete_capability_allowed=True,
+        )
+        self.assertEqual(report["merged_pr_count"], 1)
+        entry = report["entries"][0]
+        local = entry["local_worktree"]
+        self.assertEqual(local["worktree_path"], "/repo/Gitea-Tools/branches/custom-name")
+        self.assertEqual(local["match_type"], "branch")
+
+    @patch("subprocess.run")
+    def test_remove_local_worktree_uses_discovered_path(self, mock_run):
+        mock_output = (
+            "worktree /repo/Gitea-Tools\n"
+            "bare\n\n"
+            "worktree /repo/Gitea-Tools/branches/custom-name\n"
+            "branch refs/heads/feat/issue-11-x\n"
+            "HEAD cafebabe\n\n"
+        )
+        # First call is discover_local_worktrees, second is git worktree remove
+        mock_run.side_effect = [
+            MagicMock(returncode=0, stdout=mock_output),
+            MagicMock(returncode=0, stdout="", stderr=""),
+        ]
+
+        with patch("os.path.isdir", return_value=True):
+            result = mcr.remove_local_worktree("/repo/Gitea-Tools", "feat/issue-11-x")
+        self.assertTrue(result["success"])
+        self.assertTrue(result["performed"])
+        self.assertEqual(result["worktree_path"], "/repo/Gitea-Tools/branches/custom-name")
+        # Assert git worktree remove was called with the custom path
+        mock_run.assert_any_call(
+            ["git", "-C", "/repo/Gitea-Tools", "worktree", "remove", "/repo/Gitea-Tools/branches/custom-name"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
+
+class TestReviewerScratchCleanup(unittest.TestCase):
+    """#534: authorized cleanup path for reviewer scratch worktrees."""
+
+    def test_parse_reviewer_scratch_folder(self):
+        self.assertEqual(mcr.parse_reviewer_scratch_folder("review-pr512-submit"), 512)
+        self.assertEqual(mcr.parse_reviewer_scratch_folder("review-pr99"), 99)
+        self.assertIsNone(mcr.parse_reviewer_scratch_folder("feat-issue-11-x"))
+        self.assertIsNone(mcr.parse_reviewer_scratch_folder("review-feat-issue-11"))
+        self.assertIsNone(mcr.parse_reviewer_scratch_folder("review-pr512-extra-tail"))
+
+    def test_assess_safe_clean_detached_after_merge(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=True,
+            pr_closed=True,
+            worktree_state={
+                "exists": True,
+                "clean": True,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": [],
+            },
+            active_reviewer_lease=False,
+        )
+        self.assertTrue(assessment["safe_to_remove_worktree"])
+        self.assertFalse(assessment["author_worktree_cleanup"])
+        self.assertEqual(
+            assessment["recommended_action"], "remove_reviewer_scratch_worktree"
+        )
+        self.assertEqual(assessment["worktree_kind"], "reviewer_scratch")
+
+    def test_assess_blocks_dirty_scratch(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=True,
+            pr_closed=True,
+            worktree_state={
+                "exists": True,
+                "clean": False,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": ["foo.py"],
+            },
+            active_reviewer_lease=False,
+        )
+        self.assertFalse(assessment["safe_to_remove_worktree"])
+        self.assertTrue(
+            any("tracked edits" in r for r in assessment["block_reasons"])
+        )
+
+    def test_assess_blocks_active_lease(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=True,
+            pr_closed=True,
+            worktree_state={
+                "exists": True,
+                "clean": True,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": [],
+            },
+            active_reviewer_lease=True,
+        )
+        self.assertFalse(assessment["safe_to_remove_worktree"])
+        self.assertTrue(
+            any("active reviewer lease" in r for r in assessment["block_reasons"])
+        )
+
+    def test_assess_blocks_open_pr(self):
+        assessment = mcr.assess_reviewer_scratch_cleanup(
+            pr_number=512,
+            worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
+            folder_name="review-pr512-submit",
+            pr_merged=False,
+            pr_closed=False,
+            worktree_state={
+                "exists": True,
+                "clean": True,
+                "current_branch": None,
+                "head_sha": "abc",
+                "dirty_files": [],
+            },
+            active_reviewer_lease=False,
+        )
+        self.assertFalse(assessment["safe_to_remove_worktree"])
+        self.assertTrue(any("still open" in r for r in assessment["block_reasons"]))
+
+    @patch("subprocess.run")
+    @patch("merged_cleanup_reconcile.read_local_worktree_state")
+    def test_report_lists_scratch_separately_from_author(self, mock_state, mock_run):
+        mock_output = (
+            "worktree /repo/Gitea-Tools\n"
+            "bare\n\n"
+            "worktree /repo/Gitea-Tools/branches/feat-issue-11-x\n"
+            "branch refs/heads/feat/issue-11-x\n"
+            "HEAD deadbeef\n\n"
+            "worktree /repo/Gitea-Tools/branches/review-pr512-submit\n"
+            "HEAD abcdef0\n"
+            "detached\n\n"
+        )
+        mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
+
+        def _state(path):
+            if "review-pr512" in path:
+                return {
+                    "exists": True,
+                    "clean": True,
+                    "current_branch": None,
+                    "head_sha": "abcdef0",
+                    "dirty_files": [],
+                }
+            return {
+                "exists": True,
+                "clean": True,
+                "current_branch": "feat/issue-11-x",
+                "head_sha": "deadbeef",
+                "dirty_files": [],
+            }
+
+        mock_state.side_effect = _state
+        report = mcr.build_reconciliation_report(
+            project_root="/repo/Gitea-Tools",
+            closed_prs=[
+                {
+                    "number": 11,
+                    "title": "merged author",
+                    "body": "Closes #11",
+                    "head": {"ref": "feat/issue-11-x"},
+                    "merged_at": "2026-07-06T12:00:00Z",
+                },
+                {
+                    "number": 512,
+                    "title": "merged reviewed",
+                    "body": "Closes #512",
+                    "head": {"ref": "feat/issue-512-x"},
+                    "merged_at": "2026-07-06T12:00:00Z",
+                },
+            ],
+            open_prs=[],
+            remote_branch_exists={
+                "feat/issue-11-x": False,
+                "feat/issue-512-x": False,
+            },
+            head_on_master={11: True, 512: True},
+            delete_capability_allowed=True,
+            active_reviewer_leases={512: False},
+        )
+        self.assertEqual(report["merged_pr_count"], 2)
+        self.assertEqual(report["reviewer_scratch_count"], 1)
+        scratch = report["reviewer_scratch_entries"][0]
+        self.assertEqual(scratch["pr_number"], 512)
+        self.assertEqual(scratch["worktree_kind"], "reviewer_scratch")
+        self.assertFalse(scratch["author_worktree_cleanup"])
+        self.assertTrue(scratch["safe_to_remove_worktree"])
+        # Author entries must not swallow the scratch path as local_worktree.
+        author_paths = {
+            (e.get("local_worktree") or {}).get("worktree_path")
+            for e in report["entries"]
+        }
+        self.assertNotIn(
+            "/repo/Gitea-Tools/branches/review-pr512-submit", author_paths
+        )
+
+    @patch("subprocess.run")
+    def test_remove_reviewer_scratch_worktree_path_guard(self, mock_run):
+        mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+        with patch("os.path.isdir", return_value=True):
+            bad = mcr.remove_reviewer_scratch_worktree(
+                "/repo/Gitea-Tools",
+                "/repo/Gitea-Tools/branches/feat-issue-11-x",
+            )
+            self.assertFalse(bad["performed"])
+            good = mcr.remove_reviewer_scratch_worktree(
+                "/repo/Gitea-Tools",
+                "/repo/Gitea-Tools/branches/review-pr512-submit",
+            )
+        self.assertTrue(good["success"])
+        self.assertTrue(good["performed"])
+        self.assertFalse(good["author_worktree_cleanup"])
+        mock_run.assert_any_call(
+            [
+                "git",
+                "-C",
+                "/repo/Gitea-Tools",
+                "worktree",
+                "remove",
+                "/repo/Gitea-Tools/branches/review-pr512-submit",
+            ],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_premerge_baseline_proof.py

@@ -0,0 +1,112 @@
+import unittest
+
+from premerge_baseline_proof import (
+    CLEAN_PASS,
+    PREMERGE_BASELINE_PROVEN_FAILURE,
+    UNRESOLVED_REGRESSION_RISK,
+    assess_premerge_baseline_proof,
+)
+from final_report_validator import assess_final_report_validator
+
+_FIELDS = (
+    "Pre-merge base commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
+    "Tested commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
+    "Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
+    "Exit status: 1\n"
+    "Failure signature: AssertionError: None is not true\n"
+)
+
+# Current-master-only reproduction carries no pre-merge base proof (that is the
+# whole defect #533 guards against): a command/exit/signature but no base commit.
+_CURRENT_ONLY_FIELDS = (
+    "Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
+    "Exit status: 1\n"
+    "Failure signature: AssertionError: None is not true\n"
+)
+
+
+class TestPremergeBaselineProof(unittest.TestCase):
+    def test_clean_pass_not_blocked(self):
+        result = assess_premerge_baseline_proof(
+            "Validation: full suite passed, exit status 0. Clean pass."
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["skipped"])
+        self.assertEqual(result["label"], CLEAN_PASS)
+
+    def test_nonzero_exit_not_claimed_baseline_not_blocked(self):
+        result = assess_premerge_baseline_proof(
+            "Full suite: 1 failed. Investigating the regression; not yet classified."
+        )
+        self.assertFalse(result["block"])
+        self.assertEqual(result["label"], UNRESOLVED_REGRESSION_RISK)
+
+    def test_current_master_only_reproduction_rejected(self):
+        report = (
+            "Full suite: 1 failed. This is a pre-existing baseline failure — "
+            "reproduced on current master after the PR merged.\n" + _CURRENT_ONLY_FIELDS
+        )
+        result = assess_premerge_baseline_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("current-master reproduction only" in r for r in result["reasons"])
+        )
+
+    def test_baseline_claim_missing_fields_blocked(self):
+        report = (
+            "Full suite: 1 failed. This failure is baseline / pre-existing. "
+            "Verified against the pre-merge base commit."
+        )
+        result = assess_premerge_baseline_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("missing required proof field" in r for r in result["reasons"])
+        )
+
+    def test_premerge_base_proof_accepted(self):
+        report = (
+            "Full suite: 1 failed. This is a pre-existing baseline failure, proven "
+            "on the PR pre-merge base commit.\n" + _FIELDS
+        )
+        result = assess_premerge_baseline_proof(report)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["proven"])
+        self.assertEqual(result["label"], PREMERGE_BASELINE_PROVEN_FAILURE)
+
+    def test_documented_known_failure_record_accepted(self):
+        report = (
+            "Full suite: 1 failed. Baseline failure: cited known-failure record #529 "
+            "which predates the PR.\n" + _FIELDS
+        )
+        result = assess_premerge_baseline_proof(report)
+        self.assertFalse(result["block"])
+        self.assertEqual(result["label"], PREMERGE_BASELINE_PROVEN_FAILURE)
+
+
+class TestPremergeBaselineProofValidatorIntegration(unittest.TestCase):
+    def _has_rule(self, report):
+        result = assess_final_report_validator(report, "review_pr")
+        return any(
+            f["rule_id"] == "reviewer.premerge_baseline_proof"
+            for f in result["findings"]
+        )
+
+    def test_review_report_current_master_only_flagged(self):
+        report = (
+            "## Reviewer final report\n"
+            "Full suite: 1 failed. Pre-existing baseline failure — reproduced on "
+            "post-merge master.\n" + _CURRENT_ONLY_FIELDS
+        )
+        self.assertTrue(self._has_rule(report))
+
+    def test_review_report_premerge_proof_clean(self):
+        report = (
+            "## Reviewer final report\n"
+            "Full suite: 1 failed. Pre-existing baseline failure proven on the "
+            "pre-merge base commit.\n" + _FIELDS
+        )
+        self.assertFalse(self._has_rule(report))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_state_handoff_ledger.py

@@ -0,0 +1,243 @@
+"""Tests for canonical state handoff ledger (#494)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from final_report_validator import assess_final_report_validator  # noqa: E402
+from state_handoff_ledger import (  # noqa: E402
+    ISSUE_STATE_FIELDS,
+    assess_contradictory_state_handoff,
+    assess_final_report_next_action_handoff,
+    assess_state_comment,
+    assess_state_handoff_ledger_report,
+    parse_state_comment,
+    render_issue_state_comment,
+    render_pr_state_comment,
+    render_queue_controller_report,
+)
+
+
+def _work_issue_handoff(**overrides):
+    lines = [
+        "## Controller Handoff",
+        "",
+        "- Task: work-issue",
+        "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+        "- Role: author",
+        "- Identity: jcwalker3 / prgs-author",
+        "- Selected issue: #494",
+        "- Current status: implementation complete; PR opened awaiting review",
+        "- Next actor: reviewer",
+        "- Next action: review PR at pinned head",
+        "- Next prompt: Review PR #500 for issue #494 at current head in reviewer worktree",
+        "- Safe next action: switch to reviewer session",
+        "- External-state mutations: none",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        needle = f"- {key}:"
+        if needle in text:
+            prefix = text.split(needle)[0]
+            rest = text.split(needle, 1)[1]
+            suffix = rest.split("\n", 1)[1] if "\n" in rest else ""
+            text = f"{prefix}{needle} {value}\n{suffix}".rstrip()
+        else:
+            text += f"\n- {key}: {value}"
+    return text
+
+
+class TestStateCommentTemplates(unittest.TestCase):
+    def test_render_issue_state_includes_all_fields(self):
+        body = render_issue_state_comment(
+            STATE="issue_ready_for_author",
+            NEXT_ACTOR="author",
+            NEXT_ACTION="lock and implement",
+            NEXT_PROMPT="Find issue #494 and open a PR",
+            LAST_UPDATED_BY="jcwalker3",
+        )
+        parsed = parse_state_comment(body)
+        for field in ISSUE_STATE_FIELDS:
+            self.assertIn(field, parsed)
+
+    def test_render_pr_state_parses(self):
+        body = render_pr_state_comment(
+            STATE="needs_review",
+            NEXT_ACTOR="reviewer",
+            NEXT_ACTION="review at head",
+            NEXT_PROMPT="Review PR #500",
+            ISSUE="#494",
+            HEAD_SHA="a" * 40,
+            LAST_UPDATED_BY="sysadmin",
+        )
+        parsed = parse_state_comment(body)
+        self.assertEqual(parsed["STATE"], "needs_review")
+        self.assertEqual(parsed["NEXT_ACTOR"], "reviewer")
+
+    def test_queue_controller_template(self):
+        body = render_queue_controller_report(
+            QUEUE_STATE="open_prs_need_review",
+            SELECTED_OBJECT="PR #500",
+            SELECTED_OBJECT_TYPE="pr",
+            NEXT_ACTOR="reviewer",
+            NEXT_ACTION="start review session",
+            NEXT_PROMPT="Review PR #500",
+            LAST_UPDATED_BY="controller",
+        )
+        result = assess_state_comment(body, kind="queue_controller")
+        self.assertTrue(result["complete"])
+
+
+class TestStateCommentValidation(unittest.TestCase):
+    def test_complete_issue_state_comment(self):
+        body = render_issue_state_comment(
+            STATE="in_progress",
+            NEXT_ACTOR="author",
+            NEXT_ACTION="push branch",
+            NEXT_PROMPT="Continue issue #494",
+            STATUS="open",
+            RELATED_PRS="none",
+            BRANCH="feat/issue-494-state-handoff-ledger",
+            HEAD_SHA="b" * 40,
+            VALIDATION="pytest passed",
+            BLOCKERS="none",
+            DUPLICATES_OR_SUPERSEDES="none",
+            LAST_UPDATED_BY="jcwalker3",
+        )
+        result = assess_state_comment(body, kind="issue_state")
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["block"])
+
+    def test_missing_fields_blocked(self):
+        body = "STATE: open\nNEXT_ACTOR: author"
+        result = assess_state_comment(body, kind="issue_state")
+        self.assertFalse(result["complete"])
+        self.assertIn("NEXT_ACTION", result["missing_fields"])
+
+    def test_discussion_requires_five_comments_by_default(self):
+        body = render_issue_state_comment(
+            STATE="collecting_feedback",
+            NEXT_ACTOR="controller",
+            NEXT_ACTION="facilitate discussion",
+            NEXT_PROMPT="Add acceptance criteria",
+            STATUS="open",
+            RELATED_PRS="none",
+            BRANCH="none",
+            HEAD_SHA="none",
+            VALIDATION="none",
+            BLOCKERS="none",
+            DUPLICATES_OR_SUPERSEDES="none",
+            LAST_UPDATED_BY="controller",
+        )
+        # Re-map to discussion fields manually for comment-count test
+        from state_handoff_ledger import render_discussion_state_comment
+
+        disc = render_discussion_state_comment(
+            STATE="collecting_feedback",
+            NEXT_ACTOR="controller",
+            NEXT_ACTION="wait for comments",
+            NEXT_PROMPT="Continue discussion",
+            COMMENT_COUNT="2",
+            SUBSTANTIVE_COMMENTS="yes",
+            URGENCY="normal",
+            BLOCKERS="none",
+            LAST_UPDATED_BY="controller",
+        )
+        result = assess_state_comment(disc, kind="discussion_state")
+        self.assertFalse(result["complete"])
+        self.assertTrue(any("5" in reason for reason in result["reasons"]))
+
+    def test_discussion_urgent_exception(self):
+        from state_handoff_ledger import render_discussion_state_comment
+
+        disc = render_discussion_state_comment(
+            STATE="ready_for_summary",
+            NEXT_ACTOR="controller",
+            NEXT_ACTION="summarize",
+            NEXT_PROMPT="Post summary",
+            COMMENT_COUNT="2",
+            SUBSTANTIVE_COMMENTS="yes",
+            URGENCY="urgent",
+            BLOCKERS="none",
+            LAST_UPDATED_BY="controller",
+        )
+        result = assess_state_comment(disc, kind="discussion_state")
+        self.assertTrue(result["complete"])
+
+
+class TestFinalReportNextAction(unittest.TestCase):
+    def test_complete_next_action_handoff(self):
+        report = _work_issue_handoff()
+        result = assess_final_report_next_action_handoff(report)
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["block"])
+
+    def test_missing_next_prompt_blocked(self):
+        report = _work_issue_handoff(**{"Next prompt": "none"})
+        result = assess_final_report_next_action_handoff(report)
+        self.assertFalse(result["complete"])
+        self.assertIn("Next prompt", result["missing_fields"])
+
+    def test_missing_handoff_section_blocked(self):
+        result = assess_final_report_next_action_handoff("no handoff here")
+        self.assertTrue(result["block"])
+
+
+class TestContradictoryState(unittest.TestCase):
+    def test_ready_to_merge_without_approval_blocked(self):
+        report = _work_issue_handoff(
+            **{
+                "Current status": "PR ready to merge",
+                "Review decision": "not attempted",
+            }
+        )
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+    def test_ready_to_merge_with_approval_allowed(self):
+        report = _work_issue_handoff(
+            **{
+                "Current status": "PR ready to merge",
+                "Review decision": "approve",
+            }
+        )
+        result = assess_contradictory_state_handoff(report)
+        self.assertFalse(result["block"])
+
+    def test_issue_done_without_pr_proof_blocked(self):
+        report = _work_issue_handoff(**{"Current status": "issue complete"})
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+    def test_external_none_with_lock_activity_blocked(self):
+        report = _work_issue_handoff(
+            **{
+                "External-state mutations": "none",
+            }
+        )
+        report += "\n- Issue mutations: gitea_lock_issue adopted branch"
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+    def test_discussion_complete_without_summary_blocked(self):
+        report = _work_issue_handoff(**{"Current status": "discussion complete"})
+        result = assess_contradictory_state_handoff(report)
+        self.assertTrue(result["block"])
+
+
+class TestFinalReportValidatorIntegration(unittest.TestCase):
+    def test_work_issue_validator_flags_missing_next_prompt(self):
+        report = _work_issue_handoff(**{"Next prompt": ""})
+        result = assess_final_report_validator(report, "work_issue")
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("shared.state_handoff_next_action", rule_ids)
+
+    def test_work_issue_validator_accepts_complete_handoff(self):
+        report = _work_issue_handoff()
+        result = assess_state_handoff_ledger_report(report)
+        self.assertTrue(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/lease_loader.py

@@ -11,7 +11,8 @@ from urllib.parse import urlparse
 
 from gitea_auth import api_fetch_page, get_auth_header, repo_api_url
 from issue_claim_heartbeat import build_claim_inventory
-from merged_cleanup_reconcile import ISSUE_LOCK_FILE, read_issue_lock
+from merged_cleanup_reconcile import read_issue_lock
+from issue_lock_provenance import ISSUE_LOCK_FILE
 
 from webui.project_registry import ProjectRecord, load_registry
 from webui.queue_loader import _extract_linked_issue, _fetch_issues, _fetch_prs