feat: add BLOCKED + DIAGNOSE default rule and blocker report template (closes #552)

- Updated llm-project-workflow/SKILL.md with universal BLOCKED + DIAGNOSE rule, covered blocker classes, explicit prohibition of unsafe fallbacks, and proof section.
- Added templates/blocked-diagnose-report.md (standard template with all required fields).
- Updated workflows/work-issue.md, review-merge-pr.md, create-issue.md and templates/start-issue.md, review-pr.md, merge-pr.md, recover-bad-state.md to require and reference the rule at load.
- Updated docs/llm-workflow-runbooks.md to reinforce the rule for controller prompts and all workflows.
- All changes in isolated worktree branches/issue-552-blocked-diagnose. No root mutation. No unsafe fallbacks used.
- Tests (workflow + cleanup proofs): 26+ passed in run.
- git diff --check prgs/master...HEAD: clean.

.env.example

@@ -46,3 +46,12 @@ GITEA_TOKEN_SOURCE=GITEA_TOKEN
 # profile's values. Leave unset for pure env-based configuration.
 GITEA_MCP_CONFIG=/Users/jasonwalker/.config/gitea-tools/profiles.json
 GITEA_MCP_PROFILE=prgs
+
+# Namespace-scoped active task workspaces (#510). Each MCP namespace uses only
+# its own role env var; foreign bindings (e.g. GITEA_AUTHOR_WORKTREE in a
+# merger process) are ignored.
+# GITEA_AUTHOR_WORKTREE=/path/to/repo/branches/issue-123-work
+# GITEA_REVIEWER_WORKTREE=/path/to/repo/branches/review-pr456
+# GITEA_MERGER_WORKTREE=/path/to/repo/branches/merge-pr456
+# GITEA_RECONCILER_WORKTREE=/path/to/repo/branches/reconcile-pr456
+# GITEA_ACTIVE_WORKTREE=/path/to/repo/branches/session-override

audit_reconciliation_mode.py

@@ -0,0 +1,344 @@
+"""Audit vs cleanup phase gates for reconciliation workflows (#419).
+
+Audit/reconciliation tasks are read-only unless a separate cleanup phase is
+explicitly authorized with exact capability proof, safety proof, and
+before/after snapshots. Cleanup mutations must be classified in final reports;
+audit reports must not claim ``no mutations`` when cleanup occurred.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+RECONCILE_WORKFLOW_PATH = "workflows/reconcile-landed-pr.md"
+
+PHASE_AUDIT = "audit"
+PHASE_CLEANUP = "cleanup"
+
+# Tasks that enter audit phase on capability resolution (read-only default).
+AUDIT_PHASE_TASKS = frozenset({
+    "reconcile-landed-pr",
+    "reconcile_landed_pr",
+    "reconcile_issue_claims",
+    "reconcile_merged_cleanups",
+})
+
+# Mutation tasks forbidden during audit phase (fail closed).
+AUDIT_FORBIDDEN_TASKS = frozenset({
+    "delete_branch",
+    "create_branch",
+    "push_branch",
+    "create_pr",
+    "commit_files",
+    "gitea_commit_files",
+    "mark_issue",
+    "lock_issue",
+    "claim_issue",
+    "close_pr",
+    "close_issue",
+    "create_issue",
+    "merge_pr",
+    "review_pr",
+    "submit_pr_review",
+    "comment_pr",
+    "comment_issue",
+    "set_issue_labels",
+})
+
+# Shell/git commands audit phase must not run.
+AUDIT_FORBIDDEN_COMMAND_RE = re.compile(
+    r"(?:^|\s)(?:git\s+(?:push|branch\s+-D|worktree\s+remove)|"
+    r"gitea_delete_branch|delete_remote_branch)",
+    re.IGNORECASE,
+)
+
+_NO_MUTATIONS_RE = re.compile(
+    r"(?:no\s+mutations|mutations\s*:\s*none|no\s+unsafe\s+mutation)",
+    re.IGNORECASE,
+)
+_CLEANUP_OCCURRED_RE = re.compile(
+    r"(?:delete_remote_branch|remove_local_worktree|git\s+branch\s+-D|"
+    r"git\s+worktree\s+remove|remote branch.*deleted|worktree.*removed|"
+    r"cleanup\s+phase\s*:\s*(?!none\b)\S)",
+    re.IGNORECASE,
+)
+_EXTERNAL_STATE_RE = re.compile(
+    r"^\s*[-*]?\s*external[- ]state mutations\s*:",
+    re.IGNORECASE | re.MULTILINE,
+)
+_GIT_REF_RE = re.compile(
+    r"^\s*[-*]?\s*git ref mutations\s*:",
+    re.IGNORECASE | re.MULTILINE,
+)
+_CLEANUP_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*cleanup mutations\s*:",
+    re.IGNORECASE | re.MULTILINE,
+)
+_CLEANUP_PHASE_AUTH_RE = re.compile(
+    r"^\s*[-*]?\s*cleanup phase (?:authorized|authorization)\s*:\s*true",
+    re.IGNORECASE | re.MULTILINE,
+)
+_DELETE_CAPABILITY_RE = re.compile(
+    r"^\s*[-*]?\s*delete.?branch capability(?: proven)?\s*:\s*true",
+    re.IGNORECASE | re.MULTILINE,
+)
+_BEFORE_AFTER_RE = re.compile(
+    r"^\s*[-*]?\s*before/after (?:state )?snapshot\s*:",
+    re.IGNORECASE | re.MULTILINE,
+)
+_SAFETY_PROOF_RE = re.compile(
+    r"^\s*[-*]?\s*(?:branch|worktree) safe to remove\s*:\s*true",
+    re.IGNORECASE | re.MULTILINE,
+)
+
+_session: dict[str, Any] | None = None
+
+
+def _blank_session() -> dict[str, Any]:
+    return {
+        "phase": PHASE_AUDIT,
+        "entered_from_task": None,
+        "cleanup_authorized": False,
+        "cleanup_authorization": {},
+    }
+
+
+def current_phase() -> str | None:
+    """Return active reconciliation phase or None when unset."""
+    if not _session:
+        return None
+    return _session.get("phase")
+
+
+def active_record() -> dict[str, Any] | None:
+    """Return a copy of the session record, if any."""
+    return dict(_session) if _session else None
+
+
+def clear_phase() -> None:
+    """Clear reconciliation phase state."""
+    global _session
+    _session = None
+
+
+def enter_audit_phase(task: str) -> dict[str, Any]:
+    """Enter read-only audit phase for a reconciliation task."""
+    global _session
+    normalized = (task or "").strip().lower()
+    _session = _blank_session()
+    _session["entered_from_task"] = normalized
+    return dict(_session)
+
+
+def authorize_cleanup_phase(
+    *,
+    operator_approved: bool = False,
+    workflow_authorized: bool = False,
+    delete_capability_proven: bool = False,
+    safety_proof: dict[str, Any] | None = None,
+    before_after_snapshot: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Authorize cleanup phase after explicit approval and safety proofs."""
+    reasons: list[str] = []
+    if not (operator_approved or workflow_authorized):
+        reasons.append(
+            "cleanup phase requires operator approval or explicit workflow "
+            "authorization"
+        )
+    if not delete_capability_proven:
+        reasons.append(
+            "cleanup phase requires exact delete_branch capability proof "
+            "(gitea.branch.delete)"
+        )
+    safety = dict(safety_proof or {})
+    if not safety.get("safe_to_delete_remote") and not safety.get(
+        "safe_to_remove_worktree"
+    ):
+        reasons.append(
+            "cleanup phase requires proof that branch/worktree is safe to remove"
+        )
+    snapshot = dict(before_after_snapshot or {})
+    if not snapshot.get("before") or not snapshot.get("after"):
+        reasons.append(
+            "cleanup phase requires before/after state snapshot"
+        )
+
+    if reasons:
+        return {
+            "authorized": False,
+            "phase": current_phase() or PHASE_AUDIT,
+            "reasons": reasons,
+            "safe_next_action": (
+                "remain in audit-only mode or supply operator approval, "
+                "delete_branch capability proof, safety proof, and "
+                "before/after snapshot before cleanup"
+            ),
+        }
+
+    global _session
+    if _session is None:
+        _session = _blank_session()
+    _session["phase"] = PHASE_CLEANUP
+    _session["cleanup_authorized"] = True
+    _session["cleanup_authorization"] = {
+        "operator_approved": operator_approved,
+        "workflow_authorized": workflow_authorized,
+        "delete_capability_proven": delete_capability_proven,
+        "safety_proof": safety,
+        "before_after_snapshot": snapshot,
+    }
+    return {
+        "authorized": True,
+        "phase": PHASE_CLEANUP,
+        "reasons": [],
+        "cleanup_authorization": dict(_session["cleanup_authorization"]),
+        "safe_next_action": "proceed with authorized cleanup mutations only",
+    }
+
+
+def check_audit_task_enters_phase(task: str) -> bool:
+    """Return whether resolving *task* should enter audit phase."""
+    return (task or "").strip().lower() in AUDIT_PHASE_TASKS
+
+
+def check_audit_mutation_allowed(task: str) -> tuple[bool, list[str]]:
+    """Fail closed when a mutation task runs during audit phase."""
+    normalized = (task or "").strip().lower()
+    phase = current_phase()
+    if phase != PHASE_AUDIT:
+        return True, []
+    if normalized in AUDIT_FORBIDDEN_TASKS:
+        return False, [
+            f"task '{normalized}' is forbidden in audit-only reconciliation "
+            "mode: switch to an explicit cleanup phase with operator approval "
+            "and exact delete_branch capability proof before cleanup mutations"
+        ]
+    return True, []
+
+
+def check_cleanup_execution_allowed() -> tuple[bool, list[str]]:
+    """Fail closed when cleanup execution is attempted without authorization."""
+    phase = current_phase()
+    if phase == PHASE_CLEANUP and (_session or {}).get("cleanup_authorized"):
+        return True, []
+    if phase is None:
+        return False, [
+            "cleanup execution requires an active reconciliation session; "
+            "resolve a reconciliation audit task first"
+        ]
+    return False, [
+        "cleanup execution forbidden in audit-only reconciliation mode; "
+        "call gitea_authorize_reconciliation_cleanup_phase with operator "
+        "approval, delete_branch capability proof, safety proof, and "
+        "before/after snapshot"
+    ]
+
+
+def classify_cleanup_mutation(action: str) -> str:
+    """Map a cleanup action to the required mutation ledger category (#419)."""
+    normalized = (action or "").strip().lower()
+    if "delete_remote" in normalized or normalized in {
+        "delete_branch",
+        "gitea_delete_branch",
+    }:
+        return "external-state"
+    if "branch" in normalized and "delete" in normalized:
+        return "git-ref"
+    if "worktree" in normalized or "remove_local" in normalized:
+        return "cleanup"
+    return "cleanup"
+
+
+def assess_audit_reconciliation_report(report_text: str) -> dict[str, Any]:
+    """Validate audit/cleanup reconciliation reports (fail closed)."""
+    text = report_text or ""
+    reasons: list[str] = []
+
+    cleanup_occurred = bool(_CLEANUP_OCCURRED_RE.search(text))
+    claims_no_mutations = bool(_NO_MUTATIONS_RE.search(text))
+
+    if cleanup_occurred and claims_no_mutations:
+        reasons.append(
+            "report claims no mutations but documents cleanup mutations; "
+            "audit-only reports must not perform cleanup and cleanup reports "
+            "must not claim no mutations"
+        )
+
+    if cleanup_occurred:
+        if not _CLEANUP_PHASE_AUTH_RE.search(text):
+            reasons.append(
+                "cleanup mutations reported without "
+                "'Cleanup phase authorized: true'"
+            )
+        if not _DELETE_CAPABILITY_RE.search(text):
+            reasons.append(
+                "cleanup mutations reported without delete_branch capability "
+                "proof"
+            )
+        if not _BEFORE_AFTER_RE.search(text):
+            reasons.append(
+                "cleanup mutations reported without before/after state snapshot"
+            )
+        if not _SAFETY_PROOF_RE.search(text):
+            reasons.append(
+                "cleanup mutations reported without branch/worktree safety proof"
+            )
+
+        if re.search(r"delete_remote|remote branch.*delet", text, re.I):
+            if not _EXTERNAL_STATE_RE.search(text):
+                reasons.append(
+                    "remote branch deletion must be classified under "
+                    "External-state mutations"
+                )
+        if re.search(r"git\s+branch\s+-D|local branch.*delet", text, re.I):
+            if not _GIT_REF_RE.search(text):
+                reasons.append(
+                    "local branch deletion must be classified under "
+                    "Git ref mutations"
+                )
+        if re.search(r"worktree.*remov|remove_local_worktree", text, re.I):
+            if not _CLEANUP_MUTATIONS_RE.search(text):
+                reasons.append(
+                    "worktree removal must be classified under Cleanup mutations"
+                )
+
+    if (
+        RECONCILE_WORKFLOW_PATH.replace("workflows/", "") in text
+        or "reconcile-landed-pr" in text.lower()
+    ):
+        if cleanup_occurred and "audit phase" in text.lower():
+            if "cleanup phase" not in text.lower():
+                reasons.append(
+                    "report mixes audit phase with cleanup mutations without "
+                    "documenting cleanup phase transition"
+                )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "cleanup_occurred": cleanup_occurred,
+        "claims_no_mutations": claims_no_mutations,
+        "safe_next_action": (
+            "proceed"
+            if proven
+            else "fix audit/cleanup report: separate audit from cleanup phase, "
+            "classify mutations, and do not claim no mutations after cleanup"
+        ),
+    }
+
+
+def assess_audit_command_allowed(command: str) -> tuple[bool, list[str]]:
+    """Block shell commands that perform cleanup during audit phase."""
+    phase = current_phase()
+    if phase != PHASE_AUDIT:
+        return True, []
+    cmd = (command or "").strip()
+    if AUDIT_FORBIDDEN_COMMAND_RE.search(cmd):
+        return False, [
+            f"command forbidden in audit-only reconciliation mode: {cmd!r}; "
+            "authorize cleanup phase before branch/worktree deletion or push"
+        ]
+    return True, []

author_mutation_worktree.py

@@ -12,6 +12,8 @@ import subprocess
 BASE_BRANCHES = frozenset({"master", "main", "dev"})
 ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
 AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
+# Author-only: reviewer/merger/reconciler namespaces use role-specific env vars
+# via namespace_workspace_binding (#510).
 
 
 def _normalize_path(path: str) -> str:

docs/llm-workflow-runbooks.md

@@ -28,6 +28,8 @@ audit logging). See [Related documents](#related-documents).
 > to discover the available project workflows and `mcp_get_skill_guide(<name>)`
 > for step-by-step instructions. This replaces long pasted operator prompts for
 > the standard rules; operator prompts still control task-specific scope.
+>
+> **BLOCKED + DIAGNOSE (default for any missing required step):** If a required workflow skill, guide, tool, capability, preflight, terminal, worktree binding, profile, or instruction is unavailable or fails, STOP. State BLOCKED. Use the canonical blocker report template (see skills/llm-project-workflow/templates/blocked-diagnose-report.md and the llm-project-workflow/SKILL.md universal rules). Only non-mutating recovery. Report fully. No unsafe fallbacks (temp scripts, direct API, MCP internals, direct imports, in-memory restoration, manual bypasses) unless controller authorizes in the handoff for this case. Missing required steps must fail closed *before* any git or Gitea mutation. Controller prompts and all workflows must reinforce: BLOCKED + DIAGNOSE, then stop.
 > See issue #129 for the skill registry design.
 
 Jenkins and GlitchTip workflows use separate MCP servers, not this Gitea MCP
@@ -377,6 +379,37 @@ explicit control-checkout repair.
 
 Portable wording: [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
 
+### Root checkout guard (#475)
+
+The MCP server enforces a fail-closed **root checkout guard** before author,
+reviewer, and merger mutations when the active workspace is not an isolated
+`branches/...` worktree (reconciler close paths remain exempt per #468).
+
+The guard blocks when the **control checkout** (repository root) is:
+
+- on a non-stable branch (`master` / `main` / `dev` expected),
+- detached HEAD,
+- dirty (tracked edits),
+- or its `HEAD` does not match `prgs/master` when that ref is available.
+
+**Remediation (never auto-reset or stash):**
+
+> Root checkout is not on master. Preserve state, switch root back to master,
+> and use `scripts/worktree-review` or the sanctioned issue worktree flow.
+
+**Recovery after root hijack:**
+
+1. Preserve any in-progress edits (copy paths, note branch name, or commit on a
+   rescue branch from a `branches/...` worktree).
+2. From the repository root: `git checkout master` (or `main` / `dev` per repo
+   policy) and `git fetch prgs && git merge --ff-only prgs/master` when safe.
+3. Confirm `git status` is clean and `git branch --show-current` is `master`.
+4. Resume work only inside `branches/issue-<n>-<slug>` via `gitea_lock_issue` /
+   `git worktree add`.
+
+`branches/...` directories are disposable role worktrees; the root checkout is
+the stable orchestration surface only.
+
 ## Shell Spawn Hard-Stop Rule
 
 Symptom: a shell tool call returns `exit_code: -1` with empty stdout/stderr.
@@ -492,25 +525,6 @@ Root-level matches are listed in `.gitignore` so they never get committed.
 `gitea_get_runtime_context` and `gitea_lock_issue` surface **warnings** (not
 hard blocks) when these artifacts are still present.
 
-## Capability preflight lifetime (#470)
-
-After `gitea_resolve_task_capability(task=…)` proves the mutation is allowed,
-interleaved **read-only** calls preserve that proof until a gated mutation
-consumes it:
-
-- Safe reads: `gitea_whoami`, `gitea_view_pr`, `gitea_view_issue`, `gitea_list_*`,
-  `gitea_get_runtime_context`, `gitea_check_pr_eligibility`, and related
-  read-only inventory/eligibility tools (see `preflight_contract.py`).
-- Each mutation consumes the proof once; call `gitea_resolve_task_capability`
-  again immediately before the next mutation on the same task.
-- Resolving capability for a **different** task replaces the prior task binding.
-- Workspace edits before resolve, profile switches, or a dirty whoami baseline
-  invalidate proof (fail closed).
-
-If a mutation fails with “capability has not been resolved” or “task mismatch”,
-re-run `gitea_resolve_task_capability(task="<mutation>")` immediately before
-retrying — do not guess or skip the resolve step.
-
 Implementation work and review work must use separate branch folders. For
 example, an implementation branch might live under
 `branches/fix-issue-123-example`, while a review branch for the resulting PR
@@ -660,6 +674,33 @@ loop and do **not** substitute WebFetch/Playwright/manual base64.
 - **Prompt:** `Use any eligible merger profile to merge PR #N if checks pass and
   it is mergeable. Confirm with "MERGE PR N". Do not force-merge.`
 
+#### Merger lease adoption (#536)
+
+When review and merge run in **separate sessions**, the merger must **not**
+manually seed `reviewer_pr_lease._SESSION_LEASE` or equivalent in-process state.
+That ad hoc pattern was used incidentally for PR #493 and PR #421; it is not
+canonical proof and is rejected by mutation gates.
+
+**Canonical merger handoff:**
+
+1. Confirm a reviewer session holds an active PR lease and posted **APPROVED**
+   at the current live head (`gitea_get_pr_review_feedback`).
+2. In a clean merger worktree under `branches/`, call
+   `gitea_adopt_merger_pr_lease` with `worktree`, `expected_head_sha`, and
+   optional `issue_number`.
+3. The tool posts durable adoption proof on the PR thread (`<!-- mcp-review-lease-adoption:v1 -->`)
+   recording actor, profiles, adopted-from session/comment, adoption reason, and
+   timestamp, then records sanctioned in-session provenance.
+4. Call `gitea_merge_pr` with the same pinned `expected_head_sha`.
+
+**Forbidden:** Python one-liners or scripts that call `record_session_lease()`
+without provenance from `gitea_acquire_reviewer_pr_lease`,
+`gitea_adopt_merger_pr_lease`, or `gitea_heartbeat_reviewer_pr_lease`.
+
+**Same-session review+merge:** the reviewer session may use
+`gitea_acquire_reviewer_pr_lease` directly; adoption is only for cross-session
+merger handoff.
+
 ### Close the issue after merge / Reconciliation
 
 - **Profile:** issue-manager or merger.
@@ -842,6 +883,45 @@ scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md
 scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md --push
 ```
 
+## Namespace workspace binding (#510)
+
+Each MCP namespace resolves its **own** active task workspace. Foreign role
+worktree environment variables must not poison another namespace's purity
+checks.
+
+| Namespace | Workspace env vars (in priority under `GITEA_ACTIVE_WORKTREE`) | Allowed roots |
+|-----------|------------------------------------------------------------------|---------------|
+| author | `GITEA_AUTHOR_WORKTREE` | `branches/<task>` worktree only (#274) |
+| reviewer | `GITEA_REVIEWER_WORKTREE` | clean `branches/<review>` worktree |
+| merger | `GITEA_MERGER_WORKTREE` | clean `branches/<merge>` worktree **or** clean control checkout |
+| reconciler | `GITEA_RECONCILER_WORKTREE` | clean `branches/<reconcile>` worktree **or** clean control checkout |
+
+`GITEA_AUTHOR_WORKTREE` is **author-only**. Reviewer, merger, and reconciler
+MCP processes ignore it even when it points at a dirty author WIP tree.
+
+### Safe reconnect / rebind procedure
+
+When a mutation blocks on workspace binding:
+
+1. Read the error — it names the **resolved workspace path**, **role
+   namespace**, and **binding source** (tool arg, env var, or process root).
+2. Reconnect or relaunch the correct namespace MCP server from the intended
+   workspace (or set the role-specific env var before launch).
+3. Pass `worktree_path` on reviewer/merger mutation tools when the active
+   branches/ worktree differs from the MCP process root.
+4. **Do not** clean, reset, or discard foreign role worktrees to unblock your
+   own namespace — that destroys another agent's WIP.
+
+### CTH guidance for workspace binding blockers
+
+When posting a Canonical Thread Handoff after a binding blocker:
+
+- State which namespace was active (author / reviewer / merger / reconciler).
+- Quote the resolved workspace path and binding source from the error.
+- Name the safe reconnect action (relaunch MCP from `branches/...`, set
+  `GITEA_*_WORKTREE`, or pass `worktree_path`).
+- Explicitly note that foreign worktrees must not be cleaned to unblock.
+
 ## Safety notes
 
 - Never place raw tokens or passwords in any LLM MCP config; reference secrets

docs/wiki/MCP-Tools.md

@@ -24,6 +24,8 @@
 - `gitea_dry_run_pr_review` — validation-phase review mechanics.
 - `gitea_mark_final_review_decision` — mark validation complete.
 - `gitea_submit_pr_review` / `gitea_review_pr` — gated live review.
+- `gitea_acquire_reviewer_pr_lease` / `gitea_heartbeat_reviewer_pr_lease` — per-PR reviewer lease (#407).
+- `gitea_adopt_merger_pr_lease` — guarded cross-session merger lease adoption (#536).
 - `gitea_merge_pr` — gated merge (only merge path).
 
 ## Read tools

final_report_validator.py

@@ -12,6 +12,8 @@ import re
 from typing import Any, Callable
 
 import issue_lock_provenance
+from mcp_native_cleanup_proof import assess_mcp_native_cleanup_proof
+from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
 from review_proofs import (
     HANDOFF_HEADING,
     assess_controller_handoff,
@@ -117,6 +119,22 @@ _TARGET_BRANCH_SHA_RE = re.compile(
     r"target branch sha\s*:\s*[0-9a-f]{40}",
     re.IGNORECASE,
 )
+_WORKFLOW_LOAD_HELPER_RE = re.compile(
+    r"workflow[- ]load helper result\s*:",
+    re.IGNORECASE,
+)
+_WORKFLOW_LOAD_HASH_RE = re.compile(
+    r"workflow[- ]load helper result[\s\S]{0,400}?workflow[_ ]hash\s*:\s*[0-9a-f]{12}",
+    re.IGNORECASE,
+)
+_WORKFLOW_LOAD_BOUNDARY_RE = re.compile(
+    r"workflow[- ]load helper result[\s\S]{0,400}?boundary[_ ]status\s*:\s*(?:clean|violation)",
+    re.IGNORECASE,
+)
+_WORKFLOW_FILE_VIEW_NARRATIVE_RE = re.compile(
+    r"(?:read|viewed|loaded)\s+(?:the\s+)?(?:canonical\s+)?(?:workflow|review-merge-pr\.md)",
+    re.IGNORECASE,
+)
 _FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
 _RECONCILE_STALE_FIELDS = (
     "pr number opened",
@@ -567,6 +585,27 @@ def _rule_conflict_fix_push_proof(report_text: str) -> list[dict[str, str]]:
     )
 
 
+def _rule_worktree_cleanup_audit_proof(report_text: str) -> list[dict[str, str]]:
+    from worktree_cleanup_audit import assess_cleanup_audit_final_report
+
+    text = report_text or ""
+    if "cleanup audit" not in text.lower() and "reconciliation table" not in text.lower():
+        return []
+    result = assess_cleanup_audit_final_report(text)
+    if result.get("proven"):
+        return []
+    return _findings_from_reasons(
+        "author.worktree_cleanup_audit_proof",
+        result.get("reasons") or [],
+        field="Worktree cleanup audit",
+        severity="block",
+        safe_next_action=(
+            "include reconciliation table counts, disposition rows, and "
+            "final git worktree list proof"
+        ),
+    )
+
+
 def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
     text = report_text or ""
     if not _BARE_PYTEST_RE.search(text):
@@ -1023,6 +1062,70 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st
     )
 
 
+def _rule_reviewer_workflow_load_boundary(report_text: str) -> list[dict[str, str]]:
+    """#403: require structured workflow-load helper result, not file-view narrative."""
+    if not report_text.strip():
+        return []
+    findings: list[dict[str, str]] = []
+    has_helper = bool(_WORKFLOW_LOAD_HELPER_RE.search(report_text))
+    has_hash = bool(_WORKFLOW_LOAD_HASH_RE.search(report_text))
+    has_boundary = bool(_WORKFLOW_LOAD_BOUNDARY_RE.search(report_text))
+    has_narrative_only = bool(_WORKFLOW_FILE_VIEW_NARRATIVE_RE.search(report_text))
+
+    if has_narrative_only and not has_helper:
+        findings.append(validator_finding(
+            "reviewer.workflow_load_boundary",
+            "block",
+            "Workflow-load helper result",
+            (
+                "canonical workflow file-view narrative without structured "
+                "gitea_load_review_workflow helper result"
+            ),
+            (
+                "include Workflow-load helper result with workflow_hash and "
+                "boundary_status from gitea_load_review_workflow"
+            ),
+        ))
+        return findings
+
+    if has_helper and (not has_hash or not has_boundary):
+        missing = []
+        if not has_hash:
+            missing.append("workflow_hash")
+        if not has_boundary:
+            missing.append("boundary_status")
+        findings.append(validator_finding(
+            "reviewer.workflow_load_boundary",
+            "block",
+            "Workflow-load helper result",
+            (
+                "workflow-load helper result incomplete; missing "
+                + ", ".join(missing)
+            ),
+            (
+                "copy workflow_load_helper_result fields from "
+                "gitea_load_review_workflow into the final report"
+            ),
+        ))
+    return findings
+
+
+def _rule_audit_reconciliation_boundary(report_text: str) -> list[dict[str, str]]:
+    from audit_reconciliation_mode import assess_audit_reconciliation_report
+
+    result = assess_audit_reconciliation_report(report_text)
+    if result.get("proven"):
+        return []
+    return _findings_from_reasons(
+        "reconcile.audit_cleanup_boundary",
+        result.get("reasons") or [],
+        field="Audit/cleanup phase",
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or "separate audit from authorized cleanup and classify mutations",
+    )
+
+
 def _rule_reviewer_review_mutation(
     report_text: str,
     *,
@@ -1042,12 +1145,44 @@ def _rule_reviewer_review_mutation(
     )
 
 
+def _rule_reviewer_post_merge_cleanup_proof(report_text: str) -> list[dict[str, str]]:
+    result = assess_post_merge_cleanup_proof(report_text)
+    if not result.get("block"):
+        return []
+    return _findings_from_reasons(
+        "reviewer.post_merge_cleanup_proof",
+        result.get("reasons") or [],
+        field="Cleanup status",
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or "report CLEANUP_SKIPPED with blocker or full cleanup checklist",
+    )
+
+
+def _rule_shared_mcp_native_cleanup_proof(report_text: str) -> list[dict[str, str]]:
+    result = assess_mcp_native_cleanup_proof(report_text)
+    if not result.get("block"):
+        return []
+    return _findings_from_reasons(
+        "shared.mcp_native_cleanup_proof",
+        result.get("reasons") or [],
+        field="Cleanup mutations",
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or "use authorized reconciler MCP cleanup tools; never raw scripts",
+    )
+
+
 _SHARED_ISSUE_LOCK_RULES = (
     _rule_shared_issue_lock_external_state,
     _rule_shared_manual_lock_pr_override,
     _rule_shared_author_reviewer_same_run,
 )
 
+_SHARED_CLEANUP_PROOF_RULES = (
+    _rule_shared_mcp_native_cleanup_proof,
+)
+
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "review_pr": [
         _rule_shared_controller_handoff,
@@ -1069,14 +1204,18 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_already_landed_eligible,
         _rule_reviewer_already_landed_state,
         _rule_reviewer_target_branch_freshness,
+        _rule_reviewer_workflow_load_boundary,
         _rule_reviewer_mutation_ledger,
         _rule_reviewer_review_mutation,
+        _rule_reviewer_post_merge_cleanup_proof,
+        *_SHARED_CLEANUP_PROOF_RULES,
         _rule_reviewer_stale_head_proof,
     ],
     "reconcile_already_landed": [
         _rule_reconcile_controller_handoff,
         _rule_shared_email_disclosure,
         *_SHARED_ISSUE_LOCK_RULES,
+        *_SHARED_CLEANUP_PROOF_RULES,
         _rule_reconcile_stale_author_fields,
         _rule_reconcile_eligible_reviewed,
         _rule_reconcile_linked_issue_live,
@@ -1084,6 +1223,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_git_fetch_readonly,
         _rule_reviewer_legacy_workspace_mutations,
         _rule_reviewer_vague_mutations_none,
+        _rule_audit_reconciliation_boundary,
     ],
     "author_issue": [
         _rule_shared_controller_handoff,
@@ -1097,6 +1237,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
         _rule_conflict_fix_push_proof,
+        _rule_worktree_cleanup_audit_proof,
     ],
     "issue_filing": [
         _rule_shared_controller_handoff,

mcp_native_cleanup_proof.py

@@ -0,0 +1,359 @@
+"""MCP-native post-merge cleanup proof verifier (#517).
+
+Post-merge cleanup of leases, comments, branches, and worktrees must be
+proven through explicit MCP tools (or approved helpers), never raw scripts or
+ad hoc git/API commands. Merger/reviewer sessions must hand cleanup to a
+reconciler profile with the right capability proof.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+AUTHORIZED_CLEANUP_TOOLS = frozenset({
+    "gitea_cleanup_post_merge_moot_lease",
+    "gitea_reconcile_merged_cleanups",
+    "gitea_delete_branch",
+    "gitea_cleanup_merged_pr_branch",
+    "gitea_audit_worktree_cleanup",
+    "gitea_capture_branches_worktree_snapshot",
+    "gitea_assess_worktree_cleanup_integrity",
+    "gitea_authorize_reconciliation_cleanup_phase",
+})
+
+_RAW_BRANCH_DELETE_PATTERNS = (
+    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+branch\s+-[dD]\b[^\n\r]*", re.I),
+    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s--delete\b[^\n\r]*", re.I),
+    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s:[^\s`]+", re.I),
+)
+
+_RAW_COMMENT_DELETE_PATTERNS = (
+    re.compile(
+        r"(?:curl|wget|httpie)\b[^\n\r]*\b(?:DELETE|delete)\b[^\n\r]*"
+        r"(?:/comments/|issues/\d+/comments)",
+        re.I,
+    ),
+    re.compile(
+        r"\bDELETE\b[^\n\r]*/repos/[^\n\r]*/issues/\d+/comments/\d+",
+        re.I,
+    ),
+    re.compile(
+        r"\b(?:delete_issue_comment|remove_issue_comment|purge_comments?)\b",
+        re.I,
+    ),
+    re.compile(
+        r"\b(?:raw|ad hoc|adhoc)\b[^\n\r]{0,40}\bcomment\b[^\n\r]{0,40}\bdelete",
+        re.I,
+    ),
+)
+
+_CLEANUP_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*cleanup mutations\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_GIT_REF_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*git ref mutations\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_WORKTREE_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*worktree mutations\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_MERGE_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*merge mutations\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_NONE_LEDGER_VALUES = frozenset({"", "none", "n/a"})
+_MUTATION_LEDGER_PATTERNS = (
+    _CLEANUP_MUTATIONS_RE,
+    _GIT_REF_MUTATIONS_RE,
+    _WORKTREE_MUTATIONS_RE,
+)
+_RAW_WORKTREE_REMOVE_PATTERNS = (
+    re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+worktree\s+remove\b", re.I),
+)
+_AUTHORIZED_TOOL_RE = re.compile(
+    r"\b(" + "|".join(re.escape(t) for t in sorted(AUTHORIZED_CLEANUP_TOOLS)) + r")\b",
+    re.IGNORECASE,
+)
+_RECONCILER_CAPABILITY_RE = re.compile(
+    r"(?:reconciler|gitea\.branch\.delete|delete_branch|reconcile_merged_cleanups)",
+    re.IGNORECASE,
+)
+_MERGER_CLEANUP_ROLE_RE = re.compile(
+    r"(?:merger|reviewer).{0,80}(?:deleted|removed|cleaned).{0,80}"
+    r"(?:branch|worktree|comment|lease)",
+    re.IGNORECASE,
+)
+_HANDOFF_TO_RECONCILER_RE = re.compile(
+    r"(?:hand(?:ed)? off|defer(?:red)?|next actor).{0,60}reconciler",
+    re.IGNORECASE,
+)
+
+
+def _ledger_field_body(text: str, pattern: re.Pattern[str]) -> str | None:
+    match = pattern.search(text)
+    if not match:
+        return None
+    return (match.group(1) or "").strip()
+
+
+def _is_none_ledger_value(value: str) -> bool:
+    return value.strip().lower() in _NONE_LEDGER_VALUES
+
+
+def scoped_cleanup_mutation_ledger_text(text: str | None) -> str:
+    """Return mutation-ledger bodies scoped to cleanup enforcement (#517)."""
+    text = text or ""
+    parts: list[str] = []
+    for pattern in _MUTATION_LEDGER_PATTERNS:
+        body = _ledger_field_body(text, pattern)
+        if body is not None and not _is_none_ledger_value(body):
+            parts.append(body)
+    return "\n".join(parts)
+
+
+def _git_ref_cleanup_claimed(body: str) -> bool:
+    return bool(raw_branch_delete_commands(body))
+
+
+def _worktree_cleanup_claimed(body: str) -> bool:
+    for pattern in _RAW_WORKTREE_REMOVE_PATTERNS:
+        if pattern.search(body):
+            return True
+    return bool(
+        re.search(
+            r"\b(?:removed|deleted)\b[^\n]{0,40}\bworktree\b",
+            body,
+            re.IGNORECASE,
+        )
+    )
+
+
+def _cleanup_claiming_ledger_fields(text: str) -> list[tuple[str, str]]:
+    claiming: list[tuple[str, str]] = []
+    cleanup_body = _ledger_field_body(text, _CLEANUP_MUTATIONS_RE)
+    if cleanup_body is not None and not _is_none_ledger_value(cleanup_body):
+        claiming.append(("Cleanup mutations", cleanup_body))
+
+    git_ref_body = _ledger_field_body(text, _GIT_REF_MUTATIONS_RE)
+    if git_ref_body is not None and not _is_none_ledger_value(git_ref_body):
+        if _git_ref_cleanup_claimed(git_ref_body):
+            claiming.append(("Git ref mutations", git_ref_body))
+
+    worktree_body = _ledger_field_body(text, _WORKTREE_MUTATIONS_RE)
+    if worktree_body is not None and not _is_none_ledger_value(worktree_body):
+        if _worktree_cleanup_claimed(worktree_body):
+            claiming.append(("Worktree mutations", worktree_body))
+    return claiming
+
+
+def raw_branch_delete_commands(text: str | None) -> list[str]:
+    """Return raw git branch-delete commands cited in *text*."""
+    if not text:
+        return []
+    commands: list[str] = []
+    for pattern in _RAW_BRANCH_DELETE_PATTERNS:
+        commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text))
+    return list(dict.fromkeys(commands))
+
+
+def raw_comment_delete_commands(text: str | None) -> list[str]:
+    """Return raw comment-deletion commands/scripts cited in *text*."""
+    if not text:
+        return []
+    commands: list[str] = []
+    for pattern in _RAW_COMMENT_DELETE_PATTERNS:
+        commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text))
+    return list(dict.fromkeys(commands))
+
+
+def assess_raw_branch_delete_report(text: str | None) -> dict[str, Any]:
+    """Fail closed when mutation ledgers cite raw git branch deletion."""
+    commands = raw_branch_delete_commands(scoped_cleanup_mutation_ledger_text(text))
+    reasons = [
+        (
+            "raw git branch deletion bypasses MCP branch.delete cleanup gates: "
+            f"{command}"
+        )
+        for command in commands
+    ]
+    return {
+        "proven": not reasons,
+        "block": bool(reasons),
+        "commands": commands,
+        "reasons": reasons,
+        "safe_next_action": (
+            "use gitea_delete_branch, gitea_reconcile_merged_cleanups, or another "
+            "approved MCP cleanup helper with explicit branch.delete capability"
+            if reasons
+            else "proceed"
+        ),
+    }
+
+
+def assess_raw_comment_delete_report(text: str | None) -> dict[str, Any]:
+    """Fail closed when mutation ledgers cite raw comment deletion."""
+    commands = raw_comment_delete_commands(scoped_cleanup_mutation_ledger_text(text))
+    reasons = [
+        (
+            "raw comment deletion bypasses MCP lease cleanup gates: "
+            f"{command}"
+        )
+        for command in commands
+    ]
+    return {
+        "proven": not reasons,
+        "block": bool(reasons),
+        "commands": commands,
+        "reasons": reasons,
+        "safe_next_action": (
+            "use gitea_cleanup_post_merge_moot_lease (append-only release comment) "
+            "or hand cleanup to a reconciler session; never delete lease comments"
+            if reasons
+            else "proceed"
+        ),
+    }
+
+
+def assess_merge_cleanup_mutation_separation(text: str | None) -> dict[str, Any]:
+    """Require merge mutations and cleanup mutations in separate ledger fields."""
+    text = text or ""
+    merge_match = _MERGE_MUTATIONS_RE.search(text)
+    cleanup_match = _CLEANUP_MUTATIONS_RE.search(text)
+    reasons: list[str] = []
+
+    if cleanup_match and not merge_match:
+        combined = re.search(
+            r"merge.{0,40}cleanup mutations|cleanup.{0,40}merge mutations",
+            text,
+            re.IGNORECASE,
+        )
+        if combined:
+            reasons.append(
+                "merge and cleanup mutations must use separate 'Merge mutations' "
+                "and 'Cleanup mutations' ledger fields"
+            )
+
+    if merge_match and cleanup_match:
+        merge_val = (merge_match.group(1) or "").strip().lower()
+        cleanup_val = (cleanup_match.group(1) or "").strip().lower()
+        if merge_val == cleanup_val and merge_val not in {"", "none"}:
+            reasons.append(
+                "merge mutations and cleanup mutations must not duplicate the "
+                "same ledger entry"
+            )
+
+    return {
+        "proven": not reasons,
+        "block": bool(reasons),
+        "reasons": reasons,
+        "safe_next_action": (
+            "split merge mutations (gitea_merge_pr) from cleanup mutations "
+            "(reconciler MCP tools) in the controller handoff"
+            if reasons
+            else "proceed"
+        ),
+    }
+
+
+def assess_authorized_reconciler_cleanup_path(text: str | None) -> dict[str, Any]:
+    """Validate cleanup claims cite authorized MCP tools and reconciler capability."""
+    text = text or ""
+    claiming = _cleanup_claiming_ledger_fields(text)
+    if not claiming:
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+
+    reasons: list[str] = []
+    for field_name, body in claiming:
+        if not _AUTHORIZED_TOOL_RE.search(body):
+            reasons.append(
+                f"{field_name} cleanup must name an authorized MCP cleanup tool "
+                f"({', '.join(sorted(AUTHORIZED_CLEANUP_TOOLS))})"
+            )
+    if not _RECONCILER_CAPABILITY_RE.search(text):
+        reasons.append(
+            "post-merge cleanup requires reconciler capability proof "
+            "(reconciler profile, gitea.branch.delete, or reconcile_merged_cleanups)"
+        )
+
+    return {
+        "proven": not reasons,
+        "block": bool(reasons),
+        "reasons": reasons,
+        "safe_next_action": (
+            "hand cleanup to a prgs-reconciler session and cite the exact MCP tool "
+            "plus delete_branch/reconcile_merged_cleanups capability proof"
+            if reasons
+            else "proceed"
+        ),
+    }
+
+
+def assess_merger_cleanup_handoff_guidance(text: str | None) -> dict[str, Any]:
+    """Merger sessions that performed cleanup must hand off to reconciler."""
+    text = text or ""
+    if not _MERGER_CLEANUP_ROLE_RE.search(text):
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+    if _HANDOFF_TO_RECONCILER_RE.search(text):
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+    return {
+        "proven": False,
+        "block": True,
+        "reasons": [
+            "merger/reviewer session performed cleanup but did not hand off to "
+            "reconciler for MCP-native cleanup"
+        ],
+        "safe_next_action": (
+            "merger sessions must end with cleanup handed to prgs-reconciler; "
+            "do not perform ad hoc branch/comment/worktree cleanup as merger"
+        ),
+    }
+
+
+def assess_mcp_native_cleanup_proof(report_text: str | None) -> dict[str, Any]:
+    """Composite #517 verifier for MCP-native post-merge cleanup proof."""
+    text = report_text or ""
+    checks = (
+        assess_raw_branch_delete_report(text),
+        assess_raw_comment_delete_report(text),
+        assess_merge_cleanup_mutation_separation(text),
+        assess_authorized_reconciler_cleanup_path(text),
+        assess_merger_cleanup_handoff_guidance(text),
+    )
+    reasons: list[str] = []
+    safe_next = "proceed"
+    for result in checks:
+        reasons.extend(result.get("reasons") or [])
+        if result.get("block") and result.get("safe_next_action"):
+            safe_next = result["safe_next_action"]
+
+    block = bool(reasons)
+    return {
+        "proven": not block,
+        "block": block,
+        "reasons": reasons,
+        "safe_next_action": safe_next,
+        "raw_branch_commands": raw_branch_delete_commands(
+            scoped_cleanup_mutation_ledger_text(text)
+        ),
+        "raw_comment_commands": raw_comment_delete_commands(
+            scoped_cleanup_mutation_ledger_text(text)
+        ),
+    }

merger_lease_adoption.py

@@ -0,0 +1,253 @@
+"""Guarded merger adoption of an existing reviewer PR lease (#536).
+
+Replaces manual in-process ``_SESSION_LEASE`` seeding with an auditable,
+comment-backed adoption path for merger sessions that inherit a reviewer lease
+after formal approval at the current PR head.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+import reviewer_pr_lease as leases
+
+ADOPTION_MARKER = "<!-- mcp-review-lease-adoption:v1 -->"
+DEFAULT_ADOPTION_REASON = "merger-handoff-approved-head"
+
+SOURCE_ADOPT = "gitea_adopt_merger_pr_lease"
+SOURCE_ACQUIRE = "gitea_acquire_reviewer_pr_lease"
+SOURCE_HEARTBEAT = "gitea_heartbeat_reviewer_pr_lease"
+
+SANCTIONED_PROVENANCE_SOURCES = frozenset({
+    SOURCE_ADOPT,
+    SOURCE_ACQUIRE,
+    SOURCE_HEARTBEAT,
+})
+
+_MERGER_ADOPTABLE_FRESHNESS = frozenset({"active", "stale_warning"})
+
+
+def format_adoption_body(
+    *,
+    repo: str,
+    pr_number: int,
+    issue_number: int | None,
+    adopter_identity: str,
+    adopter_profile: str,
+    adopter_session_id: str,
+    worktree: str,
+    candidate_head: str | None,
+    target_branch: str,
+    target_branch_sha: str | None,
+    adopted_from_session_id: str,
+    adopted_from_profile: str,
+    adopted_from_reviewer_identity: str,
+    adopted_from_comment_id: int | None,
+    adoption_reason: str = DEFAULT_ADOPTION_REASON,
+    adopted_at: datetime | None = None,
+) -> str:
+    """Render a durable merger adoption proof comment."""
+    adopted_at = adopted_at or datetime.now(timezone.utc)
+    adopted_text = adopted_at.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
+        "+00:00", "Z"
+    )
+    lease_body = leases.format_lease_body(
+        repo=repo,
+        pr_number=pr_number,
+        issue_number=issue_number,
+        reviewer_identity=adopter_identity,
+        profile=adopter_profile,
+        session_id=adopter_session_id,
+        worktree=worktree,
+        phase="adopted",
+        candidate_head=candidate_head,
+        target_branch=target_branch,
+        target_branch_sha=target_branch_sha,
+        last_activity=adopted_at,
+        blocker="none",
+    )
+    lines = [
+        ADOPTION_MARKER,
+        f"adopted_at: {adopted_text}",
+        f"adopted_by_identity: {adopter_identity}",
+        f"adopted_by_profile: {adopter_profile}",
+        f"adopted_from_session_id: {adopted_from_session_id}",
+        f"adopted_from_profile: {adopted_from_profile}",
+        f"adopted_from_reviewer_identity: {adopted_from_reviewer_identity}",
+        f"adopted_from_comment_id: {adopted_from_comment_id or 'none'}",
+        f"adoption_reason: {adoption_reason}",
+        lease_body,
+    ]
+    return "\n".join(lines)
+
+
+def is_adoption_comment(body: str) -> bool:
+    return ADOPTION_MARKER in (body or "")
+
+
+def build_lease_provenance(
+    *,
+    source: str,
+    comment_id: int | None = None,
+    adopted_from_session_id: str | None = None,
+    adopted_from_profile: str | None = None,
+    adopted_from_reviewer_identity: str | None = None,
+    adoption_reason: str | None = None,
+    recorded_at: datetime | None = None,
+) -> dict[str, Any]:
+    recorded_at = recorded_at or datetime.now(timezone.utc)
+    recorded_text = recorded_at.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
+        "+00:00", "Z"
+    )
+    proof = {
+        "source": source,
+        "recorded_at": recorded_text,
+    }
+    if comment_id is not None:
+        proof["comment_id"] = comment_id
+    if adopted_from_session_id:
+        proof["adopted_from_session_id"] = adopted_from_session_id
+    if adopted_from_profile:
+        proof["adopted_from_profile"] = adopted_from_profile
+    if adopted_from_reviewer_identity:
+        proof["adopted_from_reviewer_identity"] = adopted_from_reviewer_identity
+    if adoption_reason:
+        proof["adoption_reason"] = adoption_reason
+    return proof
+
+
+def is_sanctioned_session_lease(session: dict[str, Any] | None) -> bool:
+    if not session:
+        return False
+    provenance = session.get("lease_provenance") or {}
+    source = (provenance.get("source") or "").strip()
+    if source not in SANCTIONED_PROVENANCE_SOURCES:
+        return False
+    if source == SOURCE_ADOPT:
+        return bool(provenance.get("comment_id")) and bool(
+            provenance.get("adopted_from_session_id")
+        )
+    if source in {SOURCE_ACQUIRE, SOURCE_HEARTBEAT}:
+        return bool(session.get("comment_id") or provenance.get("comment_id"))
+    return False
+
+
+def assess_adopt_merger_lease(
+    comments: list[dict],
+    *,
+    pr_number: int,
+    adopter_identity: str,
+    adopter_profile: str,
+    adopter_session_id: str,
+    repo: str,
+    issue_number: int | None,
+    worktree: str,
+    expected_head_sha: str | None,
+    live_head_sha: str | None,
+    approval_at_head: bool,
+    pr_open: bool = True,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Decide whether a merger may adopt the active reviewer lease (#536)."""
+    now = now or datetime.now(timezone.utc)
+    reasons: list[str] = []
+    pinned = leases._normalize_sha(expected_head_sha)
+    live = leases._normalize_sha(live_head_sha)
+
+    if not pr_open:
+        reasons.append(
+            f"PR #{pr_number} is not open; merger lease adoption is only for open PRs"
+        )
+    if not approval_at_head:
+        reasons.append(
+            "formal APPROVED review at the current PR head is required before "
+            "merger lease adoption (fail closed)"
+        )
+    if not pinned:
+        reasons.append("expected_head_sha is required for merger lease adoption")
+    if not live:
+        reasons.append("live PR head SHA unavailable (fail closed)")
+    elif pinned and live and pinned != live:
+        reasons.append(
+            "expected_head_sha does not match live PR head; refresh approval before adoption"
+        )
+
+    active = leases.find_active_reviewer_lease(
+        comments, pr_number=pr_number, now=now
+    )
+    if not active:
+        reasons.append(
+            f"no active reviewer lease on PR #{pr_number} to adopt; reviewer "
+            "must acquire first"
+        )
+    else:
+        owner_session = (active.get("session_id") or "").strip()
+        freshness = active.get("freshness") or leases.classify_lease_freshness(
+            active, now=now
+        )
+        if freshness not in _MERGER_ADOPTABLE_FRESHNESS:
+            reasons.append(
+                f"active reviewer lease freshness is '{freshness}'; explicit "
+                "reclaim is not implemented (fail closed)"
+            )
+        lease_head = active.get("candidate_head")
+        if lease_head and live and lease_head != live:
+            reasons.append(
+                "active reviewer lease candidate_head differs from live PR head"
+            )
+        if owner_session and owner_session == adopter_session_id:
+            # Same session already holds the thread lease — allow idempotent adopt
+            # only when the newest entry is not yet an adoption by this session.
+            entries = leases._lease_entries(comments, pr_number=pr_number)
+            newest = entries[-1] if entries else None
+            if newest and (newest.get("phase") or "") == "adopted":
+                reasons.append(
+                    "merger session already recorded an adoption lease on this PR"
+                )
+        elif owner_session and owner_session != adopter_session_id:
+            # Cross-session merger handoff: adopt the foreign reviewer lease.
+            pass
+
+    if not (adopter_identity or "").strip():
+        reasons.append("adopter identity required")
+    if not (adopter_session_id or "").strip():
+        reasons.append("adopter session_id required")
+    if not (worktree or "").strip():
+        reasons.append("merger worktree path required")
+    if "merger" not in (adopter_profile or "").lower():
+        reasons.append(
+            f"profile '{adopter_profile}' is not a merger profile; adoption is "
+            "merger-only (fail closed)"
+        )
+
+    adopt_allowed = not reasons
+    adoption_body = None
+    if adopt_allowed and active:
+        adoption_body = format_adoption_body(
+            repo=repo,
+            pr_number=pr_number,
+            issue_number=issue_number or active.get("issue_number"),
+            adopter_identity=adopter_identity,
+            adopter_profile=adopter_profile,
+            adopter_session_id=adopter_session_id,
+            worktree=worktree,
+            candidate_head=live,
+            target_branch=active.get("target_branch") or "master",
+            target_branch_sha=active.get("target_branch_sha"),
+            adopted_from_session_id=active.get("session_id") or "",
+            adopted_from_profile=active.get("profile") or "unknown",
+            adopted_from_reviewer_identity=active.get("reviewer_identity") or "",
+            adopted_from_comment_id=active.get("comment_id"),
+            adopted_at=now,
+        )
+
+    return {
+        "adopt_allowed": adopt_allowed,
+        "reasons": reasons,
+        "active_lease": active,
+        "adoption_body": adoption_body,
+        "adopter_session_id": adopter_session_id,
+        "expected_head_sha": pinned,
+        "live_head_sha": live,
+    }

namespace_workspace_binding.py

@@ -0,0 +1,307 @@
+"""Namespace-scoped MCP workspace binding (#510).
+
+Each role namespace (author, reviewer, merger, reconciler) resolves its own
+active task workspace. Foreign role worktree environment variables must not
+poison workspace purity checks in another namespace.
+"""
+
+from __future__ import annotations
+
+import os
+
+import author_mutation_worktree as amw
+
+ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV
+AUTHOR_WORKTREE_ENV = amw.AUTHOR_WORKTREE_ENV
+REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE"
+MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
+RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
+
+ROLE_WORKTREE_ENVS: dict[str, str] = {
+    "author": AUTHOR_WORKTREE_ENV,
+    "reviewer": REVIEWER_WORKTREE_ENV,
+    "merger": MERGER_WORKTREE_ENV,
+    "reconciler": RECONCILER_WORKTREE_ENV,
+}
+
+NON_AUTHOR_ROLES = frozenset({"reviewer", "merger", "reconciler"})
+
+
+def normalize_role_kind(
+    role_kind: str | None,
+    *,
+    profile_name: str | None = None,
+) -> str:
+    """Map profile/task role to a workspace namespace key."""
+    role = (role_kind or "author").strip().lower()
+    profile = (profile_name or "").strip().lower()
+    if role == "reviewer" and "merger" in profile:
+        return "merger"
+    if role in ROLE_WORKTREE_ENVS:
+        return role
+    return "author"
+
+
+def _env_value(env: dict[str, str] | os._Environ, key: str) -> str | None:
+    text = (env.get(key) or "").strip()
+    return text or None
+
+
+def resolve_namespace_workspace(
+    *,
+    role_kind: str,
+    worktree_path: str | None = None,
+    worktree: str | None = None,
+    process_project_root: str,
+    env: dict[str, str] | os._Environ | None = None,
+    session_lease_worktree: str | None = None,
+    profile_name: str | None = None,
+) -> tuple[str, str]:
+    """Return ``(resolved_path, binding_source)`` for *role_kind*."""
+    env_map = env if env is not None else os.environ
+    role = normalize_role_kind(role_kind, profile_name=profile_name)
+    role_env_key = ROLE_WORKTREE_ENVS[role]
+
+    for candidate, source in (
+        (worktree_path, "worktree_path argument"),
+        (worktree, "worktree argument"),
+        (_env_value(env_map, ACTIVE_WORKTREE_ENV), f"{ACTIVE_WORKTREE_ENV} environment variable"),
+        (_env_value(env_map, role_env_key), f"{role_env_key} environment variable"),
+        (session_lease_worktree if role in {"reviewer", "merger"} else None,
+         "reviewer PR lease worktree"),
+    ):
+        text = (candidate or "").strip()
+        if text:
+            return os.path.realpath(os.path.abspath(text)), source
+
+    return os.path.realpath(process_project_root), "MCP server process root (default)"
+
+
+def resolve_namespace_mutation_context(
+    *,
+    role_kind: str,
+    worktree_path: str | None,
+    process_project_root: str,
+    env: dict[str, str] | os._Environ | None = None,
+    session_lease_worktree: str | None = None,
+    worktree: str | None = None,
+    profile_name: str | None = None,
+) -> dict:
+    """Shared workspace resolution for runtime_context and mutation guards."""
+    workspace, binding_source = resolve_namespace_workspace(
+        role_kind=role_kind,
+        worktree_path=worktree_path,
+        worktree=worktree,
+        process_project_root=process_project_root,
+        env=env,
+        session_lease_worktree=session_lease_worktree,
+        profile_name=profile_name,
+    )
+    process_root = os.path.realpath(process_project_root)
+    role = normalize_role_kind(role_kind, profile_name=profile_name)
+    pollution = assess_foreign_role_worktree_pollution(
+        role_kind=role,
+        resolved_workspace=workspace,
+        binding_source=binding_source,
+        env=env,
+        profile_name=profile_name,
+    )
+    canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
+    return {
+        "workspace_path": workspace,
+        "workspace_binding_source": binding_source,
+        "workspace_role_kind": role,
+        "ignored_bindings": pollution.get("ignored_bindings") or [],
+        "process_project_root": process_root,
+        "canonical_repo_root": canonical_root,
+        "roots_aligned": canonical_root == process_root,
+    }
+
+
+def assess_foreign_role_worktree_pollution(
+    *,
+    role_kind: str,
+    resolved_workspace: str,
+    binding_source: str,
+    env: dict[str, str] | os._Environ | None = None,
+    profile_name: str | None = None,
+) -> dict:
+    """Detect when a foreign role env would have hijacked workspace binding."""
+    env_map = env if env is not None else os.environ
+    role = normalize_role_kind(role_kind, profile_name=profile_name)
+    if role == "author":
+        return {"would_pollute": False, "ignored_bindings": []}
+
+    ignored: list[str] = []
+    author_path = _env_value(env_map, AUTHOR_WORKTREE_ENV)
+    if author_path:
+        author_real = os.path.realpath(os.path.abspath(author_path))
+        resolved_real = os.path.realpath(resolved_workspace)
+        if author_real != resolved_real and binding_source != f"{AUTHOR_WORKTREE_ENV} environment variable":
+            ignored.append(
+                f"{AUTHOR_WORKTREE_ENV}={author_real} (ignored for {role} namespace)"
+            )
+    return {
+        "would_pollute": bool(ignored),
+        "ignored_bindings": ignored,
+    }
+
+
+def assess_metadata_only_worktree_binding(
+    *,
+    role_kind: str,
+    declared_worktree_path: str | None,
+    mutation_workspace: str,
+    process_project_root: str,
+    profile_name: str | None = None,
+) -> dict:
+    """Fail closed when declared worktree_path would not redirect mutations."""
+    declared = (declared_worktree_path or "").strip()
+    process_root = os.path.realpath(process_project_root)
+    mutation_root = os.path.realpath(mutation_workspace)
+    role = normalize_role_kind(role_kind, profile_name=profile_name)
+    if not declared:
+        return {"block": False, "reasons": [], "metadata_only": False}
+
+    declared_root = os.path.realpath(os.path.abspath(declared))
+    if declared_root == mutation_root:
+        return {"block": False, "reasons": [], "metadata_only": False}
+
+    if declared_root != process_root and mutation_root == process_root:
+        return {
+            "block": True,
+            "metadata_only": True,
+            "reasons": [
+                f"worktree_path is metadata-only for {role} mutations: preflight "
+                f"inspected '{declared_root}' but mutation tools would still "
+                f"validate MCP server process root '{process_root}'"
+            ],
+            "declared_worktree_path": declared_root,
+            "mutation_workspace": mutation_root,
+            "process_project_root": process_root,
+        }
+
+    return {"block": False, "reasons": [], "metadata_only": False}
+
+
+def format_namespace_workspace_binding_error(
+    *,
+    role_kind: str,
+    workspace_path: str,
+    binding_source: str,
+    reasons: list[str] | None = None,
+    ignored_bindings: list[str] | None = None,
+    dirty_files: list[str] | None = None,
+) -> str:
+    """Canonical error when namespace workspace binding blocks mutations."""
+    role = normalize_role_kind(role_kind)
+    workspace = os.path.realpath(workspace_path)
+    parts = [
+        f"Namespace workspace binding blocked ({role} namespace, #510): "
+        f"resolved workspace '{workspace}' via {binding_source}."
+    ]
+    if ignored_bindings:
+        parts.append(
+            "Foreign role bindings ignored: " + "; ".join(ignored_bindings) + "."
+        )
+    if dirty_files:
+        parts.append(
+            "Dirty tracked files in active task workspace: "
+            + ", ".join(dirty_files)
+            + "."
+        )
+    if reasons:
+        parts.append("Details: " + "; ".join(reasons) + ".")
+    parts.append(
+        "Remediation: reconnect or relaunch the MCP server from a clean dedicated "
+        f"branches/ {role} worktree, set "
+        f"{ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} or {ACTIVE_WORKTREE_ENV} "
+        "to that path, or pass worktree_path on mutation tools. Do not clean or "
+        "reset foreign role worktrees to unblock this namespace."
+    )
+    return " ".join(parts)
+
+
+def assess_namespace_mutation_workspace(
+    *,
+    role_kind: str,
+    worktree_path: str | None,
+    worktree: str | None,
+    process_project_root: str,
+    env: dict[str, str] | os._Environ | None = None,
+    session_lease_worktree: str | None = None,
+    profile_name: str | None = None,
+    current_branch: str | None = None,
+) -> dict:
+    """Evaluate namespace workspace binding before preflight/mutation."""
+    ctx = resolve_namespace_mutation_context(
+        role_kind=role_kind,
+        worktree_path=worktree_path,
+        worktree=worktree,
+        process_project_root=process_project_root,
+        env=env,
+        session_lease_worktree=session_lease_worktree,
+        profile_name=profile_name,
+    )
+    mutation_workspace = ctx["workspace_path"]
+    binding_source = ctx["workspace_binding_source"]
+    role = ctx["workspace_role_kind"]
+    process_root = ctx["process_project_root"]
+
+    metadata = assess_metadata_only_worktree_binding(
+        role_kind=role,
+        declared_worktree_path=worktree_path,
+        mutation_workspace=mutation_workspace,
+        process_project_root=process_root,
+        profile_name=profile_name,
+    )
+    pollution = assess_foreign_role_worktree_pollution(
+        role_kind=role,
+        resolved_workspace=mutation_workspace,
+        binding_source=binding_source,
+        env=env,
+        profile_name=profile_name,
+    )
+
+    reasons = list(metadata.get("reasons") or [])
+    if role == "author":
+        branches = amw.assess_author_mutation_worktree(
+            workspace_path=mutation_workspace,
+            project_root=ctx["canonical_repo_root"],
+            current_branch=current_branch,
+        )
+        if branches["block"]:
+            reasons.extend(branches["reasons"])
+    elif (
+        role == "reviewer"
+        and mutation_workspace == process_root
+        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
+    ):
+        reasons.append(
+            f"{role} mutation blocked: workspace is the stable control checkout; "
+            f"create or reconnect to a session-owned worktree under branches/ "
+            f"or set {ROLE_WORKTREE_ENVS[role]} / {ACTIVE_WORKTREE_ENV}"
+        )
+    elif (
+        role in {"reviewer", "merger"}
+        and mutation_workspace != process_root
+        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
+    ):
+        reasons.append(
+            f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
+            f"'{ctx['canonical_repo_root']}/branches/'"
+        )
+
+    block = bool(reasons)
+    return {
+        "block": block,
+        "reasons": reasons,
+        "mutation_workspace": mutation_workspace,
+        "workspace_binding_source": binding_source,
+        "workspace_role_kind": role,
+        "process_project_root": process_root,
+        "canonical_repo_root": ctx["canonical_repo_root"],
+        "metadata_only": metadata.get("metadata_only", False),
+        "declared_worktree_path": metadata.get("declared_worktree_path"),
+        "ignored_bindings": pollution.get("ignored_bindings") or [],
+    }

post_merge_cleanup_proof.py

@@ -0,0 +1,239 @@
+"""Post-merge cleanup proof verifier for reviewer final reports (#402)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+CLEANUP_SKIPPED = "CLEANUP_SKIPPED"
+CLEANUP_PERFORMED = "CLEANUP_PERFORMED"
+
+_CLEANUP_SECTION_HINT = re.compile(
+    r"(?:cleanup (?:status|result|mutations)|post-merge cleanup|"
+    r"gitea_delete_branch|remote branch.*deleted|worktree remove)",
+    re.IGNORECASE,
+)
+_CLEANUP_SKIPPED_RE = re.compile(r"\bCLEANUP_SKIPPED\b", re.IGNORECASE)
+_CLEANUP_BLOCKER_RE = re.compile(
+    r"(?:cleanup blocker|cleanup skip(?:ped)? reason)\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+_REMOTE_DELETE_CLAIM_RE = re.compile(
+    r"(?:gitea_delete_branch|remote (?:head )?branch (?:was )?deleted|"
+    r"deleted remote branch|delete_branch)",
+    re.IGNORECASE,
+)
+_WORKTREE_REMOVE_CLAIM_RE = re.compile(
+    r"(?:git worktree remove|worktree (?:was )?removed|removed (?:local )?worktree|"
+    r"worktree cleanup performed)",
+    re.IGNORECASE,
+)
+_DELETE_CAPABILITY_RE = re.compile(
+    r"(?:delete[- ]branch capability resolved|gitea\.branch\.delete)\s*:\s*"
+    r".*(?:gitea\.branch\.delete|delete_branch).*(?:resolved|allowed|proven)|"
+    r"gitea\.branch\.delete\s+(?:resolved|allowed|proven)",
+    re.IGNORECASE,
+)
+_DELETE_TASK_RE = re.compile(
+    r"(?:delete_branch|cleanup_branch|reconcile_merged_cleanups)",
+    re.IGNORECASE,
+)
+_MERGE_RESULT_RE = re.compile(
+    r"merge result\s*:\s*(?:merged|success|performed)",
+    re.IGNORECASE,
+)
+_MERGE_COMMIT_SHA_RE = re.compile(
+    r"(?:merge commit sha|merged commit sha|merge commit)\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE,
+)
+_PR_HEAD_BRANCH_RE = re.compile(
+    r"(?:merged pr head branch|pr head branch|deleted branch)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_BRANCH_NOT_PROTECTED_RE = re.compile(
+    r"branch (?:is )?not protected|branch protection\s*:\s*(?:none|false|no)",
+    re.IGNORECASE,
+)
+_OPEN_PR_INVENTORY_RE = re.compile(
+    r"(?:no other open pr(?:\s+references)?(?:\s+\S+)?|open pr inventory proof|"
+    r"open pr references).*(?:none|zero|0|clear|inventory complete)|"
+    r"no other open pr references branch",
+    re.IGNORECASE,
+)
+_ACTIVE_CLAIM_LEASE_RE = re.compile(
+    r"(?:no active (?:heartbeat|claim|lease)|"
+    r"(?:active )?(?:heartbeat|claim|lease)(?:/(?:claim|lease))*\s*:\s*none)",
+    re.IGNORECASE,
+)
+_SESSION_OWNED_WORKTREE_RE = re.compile(
+    r"(?:removed worktree path|cleanup worktree path|session-owned worktree)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_BRANCHES_PATH_RE = re.compile(r"\bbranches/", re.IGNORECASE)
+_CLEAN_TRACKED_RE = re.compile(
+    r"(?:pre-removal tracked state|tracked state before removal)\s*:\s*clean",
+    re.IGNORECASE,
+)
+_CLEAN_UNTRACKED_RE = re.compile(
+    r"(?:pre-removal untracked state|untracked state before removal)\s*:\s*clean",
+    re.IGNORECASE,
+)
+_WORKTREE_LIST_AFTER_RE = re.compile(
+    r"(?:git worktree list after|post-removal worktree list|worktree list after)",
+    re.IGNORECASE,
+)
+_WRONG_BRANCH_RE = re.compile(
+    r"deleted branch (?:does not match|!=|differs from) (?:merged )?pr head",
+    re.IGNORECASE,
+)
+
+
+def _claims_remote_delete(text: str) -> bool:
+    return bool(_REMOTE_DELETE_CLAIM_RE.search(text))
+
+
+def _claims_worktree_remove(text: str) -> bool:
+    return bool(_WORKTREE_REMOVE_CLAIM_RE.search(text))
+
+
+def _branch_safety_fields_present(text: str) -> list[str]:
+    missing: list[str] = []
+    if not _DELETE_CAPABILITY_RE.search(text):
+        missing.append("delete-branch capability resolved (gitea.branch.delete)")
+    if not _DELETE_TASK_RE.search(text):
+        missing.append("delete-branch task named (delete_branch or cleanup)")
+    if not _MERGE_RESULT_RE.search(text):
+        missing.append("merge result: merged")
+    if not _MERGE_COMMIT_SHA_RE.search(text):
+        missing.append("merge commit SHA")
+    if not _PR_HEAD_BRANCH_RE.search(text):
+        missing.append("merged PR head branch / deleted branch name")
+    if not _BRANCH_NOT_PROTECTED_RE.search(text):
+        missing.append("branch not protected proof")
+    if not _OPEN_PR_INVENTORY_RE.search(text):
+        missing.append("open PR inventory proof (no other PR references branch)")
+    if not _ACTIVE_CLAIM_LEASE_RE.search(text):
+        missing.append("no active heartbeat/claim/lease proof")
+    return missing
+
+
+def _worktree_cleanup_fields_present(text: str) -> list[str]:
+    missing: list[str] = []
+    match = _SESSION_OWNED_WORKTREE_RE.search(text)
+    path = match.group(1).strip() if match else ""
+    if not path:
+        missing.append("session-owned worktree path")
+    elif not _BRANCHES_PATH_RE.search(path.replace("\\", "/")):
+        missing.append("worktree path under branches/")
+    if not _CLEAN_TRACKED_RE.search(text):
+        missing.append("pre-removal tracked state: clean")
+    if not _CLEAN_UNTRACKED_RE.search(text):
+        missing.append("pre-removal untracked state: clean")
+    if not _WORKTREE_LIST_AFTER_RE.search(text):
+        missing.append("git worktree list after removal")
+    return missing
+
+
+def assess_post_merge_cleanup_proof(
+    report_text: str,
+    *,
+    cleanup_session: dict | None = None,
+) -> dict[str, Any]:
+    """Validate post-merge cleanup claims carry safety-gate proof (#402)."""
+    text = report_text or ""
+    session = dict(cleanup_session or {})
+    reasons: list[str] = []
+
+    if _CLEANUP_SKIPPED_RE.search(text) or session.get("outcome") == CLEANUP_SKIPPED:
+        blocker = (session.get("blocker") or "").strip()
+        if not blocker:
+            match = _CLEANUP_BLOCKER_RE.search(text)
+            blocker = match.group(1).strip() if match else ""
+        if blocker.upper() == CLEANUP_SKIPPED:
+            blocker = ""
+        if not blocker:
+            reasons.append(
+                "CLEANUP_SKIPPED requires exact cleanup blocker reason (#402)"
+            )
+        return {
+            "block": bool(reasons),
+            "proven": not reasons,
+            "outcome": CLEANUP_SKIPPED,
+            "remote_delete_claimed": False,
+            "worktree_remove_claimed": False,
+            "reasons": reasons,
+            "safe_next_action": (
+                "report CLEANUP_SKIPPED with exact blocker; do not claim performed cleanup"
+                if reasons
+                else "proceed"
+            ),
+        }
+
+    if not _CLEANUP_SECTION_HINT.search(text) and not session.get("cleanup_claimed"):
+        return {
+            "block": False,
+            "proven": True,
+            "outcome": None,
+            "remote_delete_claimed": False,
+            "worktree_remove_claimed": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+
+    remote_delete = bool(
+        session.get("remote_delete_claimed") or _claims_remote_delete(text)
+    )
+    worktree_remove = bool(
+        session.get("worktree_remove_claimed") or _claims_worktree_remove(text)
+    )
+
+    if _WRONG_BRANCH_RE.search(text):
+        reasons.append(
+            "cleanup report claims deleted branch that is not the merged PR head branch"
+        )
+
+    if remote_delete:
+        reasons.extend(
+            f"remote branch deletion missing {field}"
+            for field in _branch_safety_fields_present(text)
+        )
+
+    if worktree_remove:
+        reasons.extend(
+            f"worktree removal missing {field}"
+            for field in _worktree_cleanup_fields_present(text)
+        )
+
+    if (remote_delete or worktree_remove) and not (remote_delete or worktree_remove):
+        pass
+
+    if not remote_delete and not worktree_remove:
+        cleanup_mutations = re.search(
+            r"cleanup mutations\s*:\s*(?!none\b)\S",
+            text,
+            re.IGNORECASE,
+        )
+        if cleanup_mutations:
+            reasons.append(
+                "cleanup mutations reported without post-merge cleanup proof checklist"
+            )
+
+    outcome = CLEANUP_PERFORMED if (remote_delete or worktree_remove) and not reasons else None
+    if remote_delete or worktree_remove:
+        outcome = CLEANUP_PERFORMED if not reasons else "CLEANUP_CLAIMED_UNPROVEN"
+
+    block = bool(reasons)
+    return {
+        "block": block,
+        "proven": not block,
+        "outcome": outcome,
+        "remote_delete_claimed": remote_delete,
+        "worktree_remove_claimed": worktree_remove,
+        "reasons": reasons,
+        "safe_next_action": (
+            "report CLEANUP_SKIPPED with exact blocker or include the full cleanup "
+            "checklist before claiming remote delete or worktree removal"
+            if reasons
+            else "proceed"
+        ),
+    }

preflight_contract.py

@@ -1,66 +0,0 @@
-"""Capability preflight lifetime contract (#470).
-
-Defines which read-only MCP tools preserve an existing capability proof and the
-canonical sequencing operators must follow before mutations.
-"""
-
-from __future__ import annotations
-
-# Read-only tools that must not invalidate capability proof (#470).
-# gitea_whoami is read-only but re-pins identity; capability is preserved when
-# identity was already verified clean (#469).
-READ_ONLY_PREFLIGHT_TOOLS = frozenset({
-    "gitea_whoami",
-    "gitea_get_authenticated_user",
-    "gitea_get_current_user",
-    "gitea_view_pr",
-    "gitea_view_issue",
-    "gitea_list_prs",
-    "gitea_list_issues",
-    "gitea_list_issue_comments",
-    "gitea_get_runtime_context",
-    "gitea_resolve_task_capability",
-    "gitea_check_pr_eligibility",
-    "gitea_get_pr_review_feedback",
-    "gitea_assess_work_issue_duplicate",
-    "gitea_route_task_session",
-    "gitea_audit_config",
-    "gitea_list_labels",
-    "gitea_get_profile",
-    "gitea_list_profiles",
-})
-
-PREFLIGHT_CONTRACT_SUMMARY = (
-    "Capability preflight is session-scoped per MCP process, profile, and task. "
-    "After gitea_resolve_task_capability(task=...), interleaved read-only calls "
-    "(whoami, view_*, list_*, get_runtime_context, eligibility checks) preserve "
-    "the proof until a gated mutation consumes it. Each mutation consumes the proof "
-    "once; re-resolve immediately before the next mutation. A new resolve for a "
-    "different task replaces the prior task binding. Profile/session changes or "
-    "workspace edits before resolve invalidate proof."
-)
-
-
-def format_missing_capability_error(task: str | None = None) -> str:
-    base = (
-        "Pre-flight order violation: Task capability "
-        "(gitea_resolve_task_capability) has not been resolved (fail closed)"
-    )
-    if task:
-        return (
-            f"{base}. Re-run gitea_resolve_task_capability(task=\"{task}\") "
-            "immediately before this mutation."
-        )
-    return (
-        f"{base}. Re-run gitea_resolve_task_capability for the mutation task "
-        "immediately before acting."
-    )
-
-
-def format_task_mismatch_error(resolved: str, required: str) -> str:
-    return (
-        "Pre-flight task mismatch: "
-        f"resolved '{resolved}' but mutation requires '{required}' (fail closed). "
-        f"Re-run gitea_resolve_task_capability(task=\"{required}\") "
-        "immediately before this mutation."
-    )

review_proofs.py

@@ -5115,6 +5115,15 @@ def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
     return _assess(report_text or "")
 
 
+def assess_audit_reconciliation_report(report_text: str | None) -> dict:
+    """#419: validate audit vs cleanup reconciliation report boundaries."""
+    from audit_reconciliation_mode import (
+        assess_audit_reconciliation_report as _assess,
+    )
+
+    return _assess(report_text or "")
+
+
 _GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
 
 _NOT_APPLICABLE_VALUE = re.compile(

review_workflow_boundary.py

@@ -0,0 +1,259 @@
+"""Reviewer session boundary tracking for workflow-load gate (#403).
+
+Pre-review commands executed before ``gitea_load_review_workflow`` must be
+classified. Boundary violations block downstream reviewer mutations even when
+workflow hash proof is present.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from typing import Any
+
+CLASSIFICATION_READ_ONLY_INVENTORY = "read_only_inventory"
+CLASSIFICATION_DIAGNOSTIC = "diagnostic"
+CLASSIFICATION_BOUNDARY_VIOLATION = "boundary_violation"
+CLASSIFICATION_UNCLASSIFIED = "unclassified"
+
+ALLOWED_CLASSIFICATIONS = frozenset({
+    CLASSIFICATION_READ_ONLY_INVENTORY,
+    CLASSIFICATION_DIAGNOSTIC,
+    CLASSIFICATION_BOUNDARY_VIOLATION,
+    CLASSIFICATION_UNCLASSIFIED,
+})
+
+_PRE_REVIEW_COMMANDS: list[dict[str, Any]] = []
+
+_READ_ONLY_INVENTORY_PATTERNS = (
+    re.compile(
+        r"\bgitea[_-](?:list|view|whoami|get[-_]|resolve[-_]task|check[-_]pr|route[-_]task)",
+        re.I,
+    ),
+    re.compile(r"\bgit\s+(?:fetch|remote\s+update|branch\s+-a|log|show|rev-parse)\b", re.I),
+    re.compile(r"\bgit\s+status\b", re.I),
+    re.compile(r"\bgit\s+worktree\s+list\b", re.I),
+)
+
+_DIAGNOSTIC_PATTERNS = (
+    re.compile(r"\bgit\s+diff(?:\s+--stat)?\b", re.I),
+    re.compile(r"\bwhich\s+pytest\b", re.I),
+    re.compile(r"\bpytest\s+--version\b", re.I),
+)
+
+_BOUNDARY_VIOLATION_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
+    (re.compile(r"\b(?:pytest|python\s+-m\s+pytest|python\s+-m\s+unittest)\b", re.I),
+     "validation command before workflow load"),
+    (re.compile(r"\bprofiles\.json\b", re.I), "local profile config inspection"),
+    (re.compile(r"\bgitea-mcp(?:\.v2-contexts)?\.json\b", re.I),
+     "local Gitea MCP config inspection"),
+    (re.compile(r"\b\.env(?:\.|$|\b)", re.I), "credential file inspection"),
+    (re.compile(r"\bkeychain\b", re.I), "credential store inspection"),
+    (re.compile(r"\bpkill\b", re.I), "MCP repair activity"),
+    (re.compile(r"\b(?:edit|write|modify).{0,40}\bmcp\b", re.I),
+     "MCP config exploration"),
+    (re.compile(r"\bgit\s+(?:add|commit|reset|clean|checkout|merge|rebase|push)\b", re.I),
+     "git mutation before workflow load"),
+)
+
+
+def clear_pre_review_commands() -> None:
+    """Test helper and session reset."""
+    global _PRE_REVIEW_COMMANDS
+    _PRE_REVIEW_COMMANDS = []
+
+
+def pre_review_commands() -> list[dict[str, Any]]:
+    """Return a shallow copy of recorded pre-review commands."""
+    return [dict(entry) for entry in _PRE_REVIEW_COMMANDS]
+
+
+def _normalize_path(path: str | None) -> str:
+    return os.path.realpath(os.path.abspath((path or "").strip() or os.getcwd()))
+
+
+def is_main_checkout_path(cwd: str | None, project_root: str | None) -> bool:
+    """True when *cwd* is the stable control checkout (not under branches/)."""
+    if not project_root:
+        return False
+    root = _normalize_path(project_root)
+    path = _normalize_path(cwd)
+    if path != root:
+        return False
+    marker = f"{os.sep}branches{os.sep}"
+    return marker not in path
+
+
+def classify_pre_review_command(
+    command: str,
+    *,
+    cwd: str | None = None,
+    project_root: str | None = None,
+) -> dict[str, Any]:
+    """Classify a command executed before workflow load."""
+    text = (command or "").strip()
+    path = _normalize_path(cwd)
+    root = _normalize_path(project_root) if project_root else None
+    reasons: list[str] = []
+
+    for pattern, label in _BOUNDARY_VIOLATION_PATTERNS:
+        if pattern.search(text):
+            if label.startswith("validation") and root and not is_main_checkout_path(path, root):
+                continue
+            if label.startswith("git mutation") and root and not is_main_checkout_path(path, root):
+                continue
+            reasons.append(label)
+            return {
+                "command": text,
+                "cwd": path,
+                "classification": CLASSIFICATION_BOUNDARY_VIOLATION,
+                "reasons": reasons,
+            }
+
+    for pattern in _READ_ONLY_INVENTORY_PATTERNS:
+        if pattern.search(text):
+            return {
+                "command": text,
+                "cwd": path,
+                "classification": CLASSIFICATION_READ_ONLY_INVENTORY,
+                "reasons": [],
+            }
+
+    for pattern in _DIAGNOSTIC_PATTERNS:
+        if pattern.search(text):
+            return {
+                "command": text,
+                "cwd": path,
+                "classification": CLASSIFICATION_DIAGNOSTIC,
+                "reasons": [],
+            }
+
+    if root and is_main_checkout_path(path, root):
+        if re.search(r"\b(?:cat|head|less|read)\b", text, re.I):
+            if re.search(r"workflow|skill|runbook", text, re.I):
+                return {
+                    "command": text,
+                    "cwd": path,
+                    "classification": CLASSIFICATION_BOUNDARY_VIOLATION,
+                    "reasons": [
+                        "canonical workflow viewed as local file without "
+                        "gitea_load_review_workflow (narrative load is not proof)"
+                    ],
+                }
+
+    return {
+        "command": text,
+        "cwd": path,
+        "classification": CLASSIFICATION_UNCLASSIFIED,
+        "reasons": [
+            "pre-review command not classified; record via "
+            "gitea_record_pre_review_command before workflow load"
+        ],
+    }
+
+
+def record_pre_review_command(
+    command: str,
+    *,
+    cwd: str | None = None,
+    project_root: str | None = None,
+    classification: str | None = None,
+) -> dict[str, Any]:
+    """Record and classify a pre-review command for the current session."""
+    assessed = classify_pre_review_command(
+        command, cwd=cwd, project_root=project_root)
+    if classification:
+        if classification not in ALLOWED_CLASSIFICATIONS:
+            assessed["classification"] = CLASSIFICATION_UNCLASSIFIED
+            assessed["reasons"] = [
+                f"unknown classification '{classification}'; fail closed"
+            ]
+        else:
+            assessed["classification"] = classification
+            assessed["reasons"] = []
+    entry = {
+        **assessed,
+        "session_pid": os.getpid(),
+    }
+    _PRE_REVIEW_COMMANDS.append(entry)
+    return dict(entry)
+
+
+def assess_boundary_status(project_root: str | None = None) -> dict[str, Any]:
+    """Summarize pre-review boundary state for session proof and reports."""
+    violations = [
+        entry for entry in _PRE_REVIEW_COMMANDS
+        if entry.get("classification") == CLASSIFICATION_BOUNDARY_VIOLATION
+    ]
+    unclassified = [
+        entry for entry in _PRE_REVIEW_COMMANDS
+        if entry.get("classification") == CLASSIFICATION_UNCLASSIFIED
+    ]
+    reasons: list[str] = []
+    for entry in violations:
+        reasons.extend(entry.get("reasons") or [
+            f"boundary violation: {entry.get('command', '')[:80]}"
+        ])
+    for entry in unclassified:
+        reasons.extend(entry.get("reasons") or [
+            "unclassified pre-review command blocks reviewer mutations"
+        ])
+
+    clean = not reasons
+    return {
+        "boundary_status": "clean" if clean else "violation",
+        "boundary_clean": clean,
+        "pre_review_command_count": len(_PRE_REVIEW_COMMANDS),
+        "boundary_violation_count": len(violations),
+        "unclassified_command_count": len(unclassified),
+        "violations": [
+            {
+                "command": v.get("command"),
+                "cwd": v.get("cwd"),
+                "reasons": list(v.get("reasons") or []),
+            }
+            for v in violations
+        ],
+        "reasons": reasons,
+    }
+
+
+def boundary_blockers(project_root: str | None = None) -> list[str]:
+    """Reasons reviewer mutations must fail closed due to boundary state."""
+    status = assess_boundary_status(project_root)
+    if status.get("boundary_clean"):
+        return []
+    return list(status.get("reasons") or [
+        "reviewer session boundary violation before workflow load"
+    ])
+
+
+def workflow_load_helper_result(
+    load: dict | None,
+    project_root: str | None = None,
+) -> dict[str, Any]:
+    """Structured helper result for final reports (#403)."""
+    boundary = assess_boundary_status(project_root)
+    if load is None:
+        return {
+            "workflow_load_proof_present": False,
+            "workflow_source": None,
+            "workflow_hash": None,
+            "final_report_schema_hash": None,
+            "boundary_status": boundary.get("boundary_status"),
+            "boundary_clean": False,
+            "reasons": [
+                "gitea_load_review_workflow helper result missing from report"
+            ],
+        }
+    return {
+        "workflow_load_proof_present": True,
+        "workflow_source": load.get("workflow_source"),
+        "workflow_hash": load.get("workflow_hash"),
+        "final_report_schema_path": load.get("final_report_schema_path"),
+        "final_report_schema_hash": load.get("final_report_schema_hash"),
+        "boundary_status": load.get("boundary_status", boundary.get("boundary_status")),
+        "boundary_clean": bool(load.get("boundary_clean", boundary.get("boundary_clean"))),
+        "pre_review_command_count": boundary.get("pre_review_command_count"),
+        "reasons": [],
+    }

review_workflow_load.py

@@ -0,0 +1,217 @@
+"""Canonical review-merge workflow load proof for reviewer mutations (#389, #403)."""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+from pathlib import Path
+
+import review_workflow_boundary as boundary
+
+WORKFLOW_REL_PATH = (
+    "skills/llm-project-workflow/workflows/review-merge-pr.md"
+)
+SCHEMA_REL_PATH = (
+    "skills/llm-project-workflow/schemas/review-merge-final-report.md"
+)
+TASK_MODE = "review-merge-pr"
+LOAD_TOOL_NAME = "gitea_load_review_workflow"
+
+_REVIEW_WORKFLOW_LOAD: dict | None = None
+
+
+def compute_content_hash(text: str) -> str:
+    """Short deterministic hash for workflow/schema version proof."""
+    return hashlib.sha256((text or "").encode("utf-8")).hexdigest()[:12]
+
+
+def _read_text(path: Path) -> str:
+    return path.read_text(encoding="utf-8")
+
+
+def _canonical_paths(project_root: str) -> tuple[Path, Path]:
+    root = Path(project_root)
+    workflow = root / WORKFLOW_REL_PATH
+    schema = root / SCHEMA_REL_PATH
+    if not workflow.is_file():
+        raise FileNotFoundError(f"canonical workflow missing: {workflow}")
+    if not schema.is_file():
+        raise FileNotFoundError(f"final report schema missing: {schema}")
+    return workflow, schema
+
+
+def build_canonical_workflow_metadata(
+    project_root: str,
+    *,
+    prompt_text: str | None = None,
+) -> dict:
+    """Load workflow + schema from disk and compute proof metadata."""
+    workflow_path, schema_path = _canonical_paths(project_root)
+    workflow_text = _read_text(workflow_path)
+    schema_text = _read_text(schema_path)
+    workflow_hash = compute_content_hash(workflow_text)
+    schema_hash = compute_content_hash(schema_text)
+    conflict, conflict_reasons = assess_prompt_conflict(prompt_text)
+    return {
+        "workflow_source": WORKFLOW_REL_PATH,
+        "workflow_path": str(workflow_path),
+        "task_mode": TASK_MODE,
+        "workflow_hash": workflow_hash,
+        "workflow_version": workflow_hash,
+        "final_report_schema_path": SCHEMA_REL_PATH,
+        "final_report_schema_hash": schema_hash,
+        "prompt_conflicts_with_workflow": conflict,
+        "prompt_conflict_reasons": conflict_reasons,
+        "load_tool": LOAD_TOOL_NAME,
+    }
+
+
+def assess_prompt_conflict(prompt_text: str | None) -> tuple[bool, list[str]]:
+    """Detect obvious task-mode conflicts between prompt and review workflow."""
+    if not (prompt_text or "").strip():
+        return False, []
+    text = prompt_text.lower()
+    reasons: list[str] = []
+    conflicting = (
+        (r"\bwork[- ]issue\b", "work-issue author mode"),
+        (r"\bcreate[- ]issue\b", "create-issue mode"),
+        (r"\bauthor/coder\b", "author/coder mode"),
+        (r"\breconcile[- ]landed\b", "reconcile-landed mode"),
+    )
+    for pattern, label in conflicting:
+        if re.search(pattern, text):
+            reasons.append(
+                f"active prompt appears to request {label} while loading "
+                f"{TASK_MODE} workflow"
+            )
+    return bool(reasons), reasons
+
+
+def record_review_workflow_load(
+    project_root: str,
+    *,
+    prompt_text: str | None = None,
+) -> dict:
+    """Record in-process workflow load proof for the current MCP session."""
+    global _REVIEW_WORKFLOW_LOAD
+    meta = build_canonical_workflow_metadata(
+        project_root, prompt_text=prompt_text)
+    boundary_state = boundary.assess_boundary_status(project_root)
+    _REVIEW_WORKFLOW_LOAD = {
+        **meta,
+        "session_pid": os.getpid(),
+        "loaded": True,
+        "boundary_status": boundary_state.get("boundary_status"),
+        "boundary_clean": boundary_state.get("boundary_clean"),
+        "pre_review_command_count": boundary_state.get("pre_review_command_count"),
+        "boundary_violation_count": boundary_state.get("boundary_violation_count"),
+        "boundary_reasons": list(boundary_state.get("reasons") or []),
+    }
+    return dict(_REVIEW_WORKFLOW_LOAD)
+
+
+def clear_review_workflow_load() -> None:
+    """Test helper and review_pr session reset."""
+    global _REVIEW_WORKFLOW_LOAD
+    _REVIEW_WORKFLOW_LOAD = None
+    boundary.clear_pre_review_commands()
+
+
+def workflow_load_status(project_root: str | None = None) -> dict:
+    """Non-throwing status for capability/runtime reports."""
+    load = _REVIEW_WORKFLOW_LOAD
+    if load is None:
+        return {
+            "workflow_load_proof_present": False,
+            "workflow_load_valid": False,
+            "workflow_source": None,
+            "workflow_hash": None,
+            "final_report_schema_path": SCHEMA_REL_PATH,
+            "reasons": [
+                f"{LOAD_TOOL_NAME} has not been called in this session "
+                "(fail closed for reviewer mutations)"
+            ],
+        }
+    reasons = _session_validation_reasons(load, project_root)
+    boundary_reasons = boundary.boundary_blockers(project_root)
+    if boundary_reasons:
+        reasons = list(reasons) + boundary_reasons
+    return {
+        "workflow_load_proof_present": True,
+        "workflow_load_valid": not reasons,
+        "workflow_source": load.get("workflow_source"),
+        "workflow_hash": load.get("workflow_hash"),
+        "task_mode": load.get("task_mode"),
+        "final_report_schema_path": load.get("final_report_schema_path"),
+        "final_report_schema_hash": load.get("final_report_schema_hash"),
+        "prompt_conflicts_with_workflow": load.get(
+            "prompt_conflicts_with_workflow"),
+        "session_pid": load.get("session_pid"),
+        "boundary_status": load.get("boundary_status"),
+        "boundary_clean": load.get("boundary_clean"),
+        "workflow_load_helper_result": boundary.workflow_load_helper_result(
+            load, project_root),
+        "reasons": reasons,
+    }
+
+
+def _session_validation_reasons(
+    load: dict,
+    project_root: str | None,
+) -> list[str]:
+    reasons: list[str] = []
+    if load.get("session_pid") != os.getpid():
+        reasons.append(
+            "workflow load proof was recorded in a different process "
+            "(fail closed)"
+        )
+        return reasons
+    if load.get("prompt_conflicts_with_workflow"):
+        reasons.extend(load.get("prompt_conflict_reasons") or [
+            "active prompt conflicts with loaded review-merge workflow"
+        ])
+    if project_root:
+        try:
+            current = build_canonical_workflow_metadata(project_root)
+        except OSError as exc:
+            reasons.append(f"cannot re-verify workflow hash: {exc}")
+            return reasons
+        if current["workflow_hash"] != load.get("workflow_hash"):
+            reasons.append(
+                "stored workflow hash is stale; reload via "
+                f"{LOAD_TOOL_NAME} (fail closed)"
+            )
+        if current["final_report_schema_hash"] != load.get(
+                "final_report_schema_hash"):
+            reasons.append(
+                "stored final-report schema hash is stale; reload via "
+                f"{LOAD_TOOL_NAME} (fail closed)"
+            )
+    return reasons
+
+
+def review_workflow_load_blockers(
+    project_root: str | None = None,
+) -> list[str]:
+    """Reasons reviewer mutations must fail closed."""
+    boundary_reasons = boundary.boundary_blockers(project_root)
+    if boundary_reasons and _REVIEW_WORKFLOW_LOAD is None:
+        return boundary_reasons
+    status = workflow_load_status(project_root)
+    if not status.get("workflow_load_proof_present"):
+        return list(status.get("reasons") or []) + boundary_reasons
+    if not status.get("workflow_load_valid"):
+        return list(status.get("reasons") or [])
+    return []
+
+
+def recovery_handoff_without_replay() -> list[str]:
+    """Safe next-step lines that must not include approve/merge replay."""
+    return [
+        "Reload the canonical workflow via gitea_load_review_workflow, then "
+        "rerun the full review-merge workflow from inventory.",
+        "Do not call gitea_submit_pr_review, gitea_mark_final_review_decision, "
+        "or gitea_merge_pr until workflow-load proof is present.",
+        "Do not include approve/merge replay commands in the recovery handoff.",
+    ]

reviewer_pr_lease.py

@@ -23,6 +23,7 @@ _ACTIVE_PHASES = frozenset({
     "approved",
     "request-changes",
     "merging",
+    "adopted",
 })
 
 DEFAULT_LEASE_TTL_MINUTES = 120
@@ -217,12 +218,24 @@ def assess_acquire_lease(
     candidate_head: str | None,
     target_branch: str,
     target_branch_sha: str | None,
+    pr_merged_or_closed: bool = False,
     now: datetime | None = None,
 ) -> dict[str, Any]:
-    """Fail closed when another session holds an active lease."""
+    """Fail closed when another session holds an active lease.
+
+    When *pr_merged_or_closed* is true the PR has already merged/closed, so any
+    reviewer-lease acquisition or adoption for merge work is moot: fail closed
+    with a ``post_merge_moot`` reason and never mint a lease body (#515).
+    """
     now = now or datetime.now(timezone.utc)
     reasons: list[str] = []
     existing = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
+    post_merge_moot = bool(pr_merged_or_closed)
+    if post_merge_moot:
+        reasons.append(
+            f"post_merge_moot: PR #{pr_number} is already merged/closed; reviewer "
+            "lease adoption for merge is moot (fail closed)"
+        )
     if existing:
         owner_session = (existing.get("session_id") or "").strip()
         freshness = existing.get("freshness") or classify_lease_freshness(existing, now=now)
@@ -270,12 +283,120 @@ def assess_acquire_lease(
         "existing_lease": existing,
         "lease_body": body,
         "session_id": session_id,
+        "post_merge_moot": post_merge_moot,
     }
 
 
-def record_session_lease(lease: dict[str, Any]) -> dict[str, Any]:
+def assess_post_merge_moot_lease(
+    comments: list[dict],
+    *,
+    pr_number: int,
+    pr_merged: bool = False,
+    pr_state: str | None = None,
+    merge_commit_sha: str | None = None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Assess a reviewer lease left lingering on an already-merged/closed PR (#515).
+
+    Read-first and fail-safe:
+
+    - Only treats a lease as moot when the live PR state is merged/closed.
+    - Never proposes touching an *active* lease while the PR is still open
+      (``cleanup_allowed`` stays false and a refusal reason is returned).
+    - When the PR is merged/closed and a lease is still active, ``cleanup_allowed``
+      is true and a terminal ``phase: released`` lease body (``blocker:
+      post-merge-moot``) is provided so the moot lease can be neutralised by an
+      append-only comment — never by deleting a foreign session's comment, and
+      never by adopting or merging.
+
+    Posting the released body makes that lease terminal, so a subsequent call
+    finds no active lease and reports nothing left to clean (idempotent).
+    """
+    now = now or datetime.now(timezone.utc)
+    merged_or_closed = bool(pr_merged) or (
+        str(pr_state or "").strip().lower() == "closed"
+    )
+    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
+    # The newest lease comment is authoritative: once a terminal marker
+    # (released/done/blocked) is the latest entry, the lease is resolved even if
+    # an earlier non-terminal comment from the same session still lingers. This
+    # keeps post-merge cleanup idempotent.
+    entries = _lease_entries(comments, pr_number=pr_number)
+    newest = entries[-1] if entries else None
+    newest_terminal = bool(newest) and (
+        (newest.get("phase") or "").strip().lower() in _TERMINAL_PHASES
+    )
+    reasons: list[str] = []
+    cleanup_allowed = False
+    release_body: str | None = None
+    is_moot = bool(active) and merged_or_closed and not newest_terminal
+
+    if not merged_or_closed:
+        if active:
+            reasons.append(
+                f"PR #{pr_number} is still open; refusing to touch active reviewer "
+                "lease (fail closed)"
+            )
+        else:
+            reasons.append(
+                f"PR #{pr_number} is still open; no post-merge lease cleanup applicable"
+            )
+    elif newest_terminal:
+        reasons.append(
+            f"PR #{pr_number} reviewer lease already released/terminal; nothing to clean"
+        )
+    elif active:
+        cleanup_allowed = True
+        release_body = format_lease_body(
+            repo=active.get("repo") or "",
+            pr_number=pr_number,
+            issue_number=active.get("issue_number"),
+            reviewer_identity=active.get("reviewer_identity") or "",
+            profile=active.get("profile") or "unknown",
+            session_id=active.get("session_id") or "",
+            worktree=active.get("worktree") or "",
+            phase="released",
+            candidate_head=active.get("candidate_head"),
+            target_branch=active.get("target_branch") or "master",
+            target_branch_sha=active.get("target_branch_sha"),
+            last_activity=now,
+            blocker="post-merge-moot",
+        )
+    else:
+        reasons.append(
+            f"PR #{pr_number} is merged/closed but no active reviewer lease remains; "
+            "nothing to clean"
+        )
+
+    return {
+        "pr_number": pr_number,
+        "pr_state": pr_state,
+        "pr_merged_or_closed": merged_or_closed,
+        "merge_commit_sha": merge_commit_sha,
+        "active_lease": active,
+        "is_moot": is_moot,
+        "cleanup_allowed": cleanup_allowed,
+        "release_body": release_body,
+        "reasons": reasons,
+    }
+
+
+def record_session_lease(
+    lease: dict[str, Any],
+    *,
+    lease_provenance: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Record the in-session lease mirror for mutation gates.
+
+    *lease_provenance* must be supplied by sanctioned MCP tools (#536). Bare
+    manual seeding without provenance cannot satisfy merger/reviewer mutation
+    gates.
+    """
     global _SESSION_LEASE
-    _SESSION_LEASE = dict(lease)
+    stored = dict(lease)
+    if lease_provenance:
+        stored["lease_provenance"] = dict(lease_provenance)
+    _SESSION_LEASE = stored
     return dict(_SESSION_LEASE)
 
 
@@ -308,13 +429,25 @@ def assess_mutation_lease_gate(
     if not session:
         reasons.append(
             f"no in-session reviewer lease recorded; acquire via "
-            f"gitea_acquire_reviewer_pr_lease before {mutation}"
+            f"gitea_acquire_reviewer_pr_lease or adopt via "
+            f"gitea_adopt_merger_pr_lease before {mutation}"
         )
-    elif session.get("pr_number") != pr_number:
+    else:
+        import merger_lease_adoption as mla
+
+        if not mla.is_sanctioned_session_lease(session):
+            reasons.append(
+                "in-session lease lacks sanctioned provenance; manual "
+                "_SESSION_LEASE seeding is not canonical proof — use "
+                "gitea_acquire_reviewer_pr_lease or gitea_adopt_merger_pr_lease"
+            )
+    if session and session.get("pr_number") != pr_number:
         reasons.append(
             f"session lease is for PR #{session.get('pr_number')}, not #{pr_number}"
         )
-    elif (session.get("session_id") or "") != (session_id or session.get("session_id")):
+    elif session and (session.get("session_id") or "") != (
+        session_id or session.get("session_id")
+    ):
         reasons.append("session lease session_id mismatch (fail closed)")
 
     if active:

root_checkout_guard.py

@@ -0,0 +1,133 @@
+"""Root checkout guard (#475).
+
+The project root checkout is the stable control checkout on master/prgs/master.
+Author/reviewer/merge flows must fail closed when the control checkout is
+contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of
+prgs/master). Isolated ``branches/...`` worktrees remain allowed.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+from author_mutation_worktree import is_path_under_branches
+from reviewer_worktree import parse_dirty_tracked_files
+
+REMEDIATION = (
+    "Root checkout is not on master. Preserve state, switch root back to master, "
+    "and use scripts/worktree-review or the sanctioned issue worktree flow."
+)
+
+BASE_BRANCHES = frozenset({"master", "main", "dev"})
+REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master")
+
+
+def resolve_remote_master_sha(
+    canonical_repo_root: str,
+    *,
+    remote_refs: tuple[str, ...] | None = None,
+) -> str | None:
+    """Return the commit SHA for the tracking master ref when available."""
+    root = (canonical_repo_root or "").strip()
+    if not root:
+        return None
+    for ref in remote_refs or REMOTE_MASTER_REFS:
+        res = subprocess.run(
+            ["git", "-C", root, "rev-parse", "--verify", ref],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        if res.returncode == 0:
+            sha = (res.stdout or "").strip()
+            if sha:
+                return sha
+    return None
+
+
+resolve_tracking_master_sha = resolve_remote_master_sha
+
+
+def assess_root_checkout_guard(
+    *,
+    workspace_path: str,
+    canonical_repo_root: str,
+    current_branch: str | None,
+    head_sha: str | None,
+    porcelain_status: str,
+    remote_master_sha: str | None,
+    resolved_role: str | None = None,
+) -> dict:
+    """Fail closed when the control checkout is not clean master/prgs/master."""
+    reasons: list[str] = []
+    root = os.path.realpath(canonical_repo_root)
+    workspace = os.path.realpath(workspace_path)
+    branch = (current_branch or "").strip()
+    dirty_files = parse_dirty_tracked_files(porcelain_status)
+
+    if resolved_role == "reconciler":
+        return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
+
+    if resolved_role != "merger" and is_path_under_branches(workspace, root):
+        return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
+
+    if dirty_files:
+        reasons.append(
+            "control checkout has tracked local edits before role work "
+            f"(dirty files: {', '.join(dirty_files)})"
+        )
+
+    if not branch:
+        reasons.append("control checkout is detached HEAD; expected branch 'master'")
+    elif branch not in BASE_BRANCHES:
+        reasons.append(
+            f"control checkout branch '{branch}' is not a stable base branch "
+            f"({'/'.join(sorted(BASE_BRANCHES))})"
+        )
+
+    if remote_master_sha and head_sha and head_sha != remote_master_sha:
+        reasons.append(
+            "control checkout HEAD does not match prgs/master "
+            f"(HEAD {head_sha[:12]}, prgs/master {remote_master_sha[:12]})"
+        )
+
+    proven = not reasons
+    return _assessment(proven, reasons, root, workspace, branch or None, head_sha, dirty_files)
+
+
+def format_root_checkout_guard_error(assessment: dict) -> str:
+    """Single RuntimeError message for MCP preflight gates."""
+    root = assessment.get("canonical_repo_root") or "(unknown)"
+    workspace = assessment.get("workspace_path") or "(unknown)"
+    reasons = "; ".join(assessment.get("reasons") or ["unknown root checkout violation"])
+    return (
+        f"Root checkout guard (#475): {reasons}. "
+        f"canonical repository root: {root}; workspace: {workspace}. "
+        f"{REMEDIATION}"
+    )
+
+
+def _assessment(
+    proven: bool,
+    reasons: list[str],
+    canonical_repo_root: str,
+    workspace_path: str,
+    current_branch: str | None,
+    head_sha: str | None,
+    dirty_files: list[str],
+) -> dict:
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "canonical_repo_root": canonical_repo_root,
+        "workspace_path": workspace_path,
+        "current_branch": current_branch,
+        "head_sha": head_sha,
+        "dirty_files": dirty_files,
+        "remediation": REMEDIATION,
+    }
+
+
+assess_root_checkout = assess_root_checkout_guard

skills/llm-project-workflow/SKILL.md

@@ -31,12 +31,29 @@ workflow file.
 - A nearby capability does not count.
 - Do not self-review or self-merge.
 - Do not mix modes in one run.
+- **BLOCKED + DIAGNOSE default rule (required):** If any required workflow step, skill, tool, capability, preflight, instruction, profile, worktree binding, or terminal/MCP operation cannot be performed or loaded (including the canonical ones listed in this skill and its loaded workflow), immediately enter `BLOCKED + DIAGNOSE`. Stop before any git or Gitea mutation. Diagnose using the standard template in [`templates/blocked-diagnose-report.md`](templates/blocked-diagnose-report.md). Attempt *only* safe non-mutating recovery. Report using the template. Do not continue, use fallbacks, or treat the missing requirement as harmless.
 - If the required workflow cannot be loaded, stop and produce a recovery handoff
-  only.
+  only (see BLOCKED + DIAGNOSE rule above).
 - Final report must use the schema for the loaded workflow.
 - If a task requires a different mode, stop and produce a handoff for the
   correct workflow.
 
+## Covered blocker classes (BLOCKED + DIAGNOSE must trigger for these)
+
+- missing required skill or workflow guide (e.g. gitea-workflow, llm-project-workflow)
+- broken terminal/tool runner or shell spawn failure
+- MCP capability failure, reset, or deadlock (e.g. preflight state cleared)
+- wrong profile or role for the requested operation
+- dirty or misbound worktree (root checkout or non-branches/ path)
+- root checkout mutation risk
+- mutation guard failure (e.g. branches-only guard)
+- missing required MCP tool/schema or operation
+- stale or inconsistent runtime state (e.g. lease vs actual, dirty state disagreement)
+- unavailable project instructions or checked-in guides
+- any other failure of a step the current workflow or controller prompt declares "required"
+
+**Prohibited unless controller authorizes in writing for this instance:** temp scripts, direct API fallback, MCP internals, direct imports, in-memory state restoration, manual bypasses, or any continuation that hides the blocker. All such cases must be reported as process/tooling defects.
+
 ## Mode isolation
 
 A run that starts in `review-merge-pr` mode may not create process issues,
@@ -182,4 +199,16 @@ 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`.
+
+## Proof: missing required workflow steps stop before mutation
+
+- The llm-project-workflow router (this file) and every loaded workflow (work-issue.md, review-merge-pr.md, create-issue.md, etc.) now declare at the top: if required step/skill/tool/capability/instruction/profile/worktree binding/preflight fails, STOP, state BLOCKED, use blocked-diagnose-report.md template, only non-mutating recovery.
+- Controller prompts (start-issue.md, review-pr.md, merge-pr.md, recover-bad-state.md, etc.) and the runbooks (docs/llm-workflow-runbooks.md) explicitly require the same and prohibit unsafe fallbacks.
+- MCP guards (branches-only mutation guard #274, worktree binding #510, preflight purity, role checks, lease gates, gitea_lock_issue, etc.) plus the "prove before mutation" rules ensure that a BLOCKED state prevents git/Gitea mutations.
+- When a skill/guide/tool is missing (e.g. gitea-workflow not mounted for a runtime), the load step in the router/prompt fails the "required" check → BLOCKED + report before any gitea_* call or git command that mutates.
+- Terminal/shell failures, capability deadlocks, wrong profile, dirty/misbound worktree, root risk, guard failures, missing schema, stale state, unavailable instructions all map to the covered blocker classes and trigger the same stop + report.
+- No code path in the canonical workflows allows continuation past a declared required step without the BLOCKED report.
+- See also: Global LLM Worktree Rule, Shell Spawn Hard-Stop Rule, Identity and profile safety, Subagent Tool-Budget Guardrails, and the explicit prohibition list in Universal rules.
+
+Tests / proof docs updated in this change + runbooks. Full relevant test runs (see PR handoff) pass; `git diff --check` clean. Missing-step cases are now documented to fail closed before mutation.

skills/llm-project-workflow/schemas/review-merge-final-report.md

@@ -63,8 +63,14 @@ Do not use legacy fields: `Pinned reviewed head`, `Scratch worktree used`,
 - Current status:
 - Safe next action:
 - Safety statement:
+- Workflow-load helper result:
 ```
 
+The **Workflow-load helper result** field must carry structured output from
+`gitea_load_review_workflow` (workflow_hash, final_report_schema_hash,
+boundary_status). Narrative claims that workflow files were viewed locally are
+not sufficient (#403).
+
 ### Already-landed handoff overrides
 
 When eligibility class is `ALREADY_LANDED_RECONCILE_REQUIRED`:

skills/llm-project-workflow/templates/blocked-diagnose-report.md

@@ -0,0 +1,52 @@
+# Blocker Report Template (BLOCKED + DIAGNOSE)
+
+Use this exact structure whenever a required workflow step cannot be performed. Emit this report and stop. Do not proceed to mutation or fallback unless a controller explicitly authorizes an exception in writing.
+
+## Required step
+<Describe the exact step, skill, tool, capability, instruction, profile, or preflight that was required. Include the canonical name and where it is defined (e.g. gitea-workflow skill, specific workflow file, gitea_xxx tool).>
+
+## Observed failure
+<Exact symptom, error message, missing output, guard error, 404, schema error, dirty state, wrong profile, etc. Quote relevant output or tool response.>
+
+## Expected behavior
+<What the workflow/docs/prompts say should happen. Reference the specific rule, template, or preflight that requires this step.>
+
+## Checks performed
+- List every verification attempted (e.g. gitea_whoami, resolve_task_capability, ls skills/, git status, mcp_list_*, worktree list, etc.)
+- Note any discrepancies found (e.g. skill not mounted for this runtime, capability not in profile, cwd not under branches/, etc.)
+
+## Safe recovery attempted
+- Only non-mutating actions (reads, lists, views, status, whoami, resolve, fetch --dry, etc.)
+- List what was tried and the result.
+- If no safe recovery possible, state that explicitly.
+
+## Likely classification
+Choose one or more:
+- missing required skill or workflow guide
+- broken terminal/tool runner
+- MCP capability failure or deadlock
+- wrong profile or role
+- dirty or misbound worktree
+- root checkout mutation risk
+- mutation guard failure
+- missing required MCP tool/schema
+- stale or inconsistent runtime state
+- unavailable project instructions
+- other: <describe>
+
+## Durable fix recommendation
+<Specific, actionable recommendation that fixes the root process/tooling issue (e.g. "Mount gitea-workflow skill for Codex under canonical name in ~/.codex/skills/", "Add preflight in llm-project-workflow/SKILL.md that hard-stops before any gitea_ call if X is unavailable", "Update controller prompt to require BLOCKED + this report before any fallback", "Grant capability in profile config", etc.). Do not suggest temp workarounds.>
+
+## Mutation occurred?
+- No (preferred and required unless explicitly authorized)
+- Yes — describe exactly what was mutated and why it was unavoidable after diagnosis. (This should be rare and will trigger additional review.)
+
+## Single next action
+<One concrete next step for the current actor (e.g. "Controller to approve or reject recovery", "File follow-up issue #XXX for skill mounting", "Re-launch session from clean branches/ worktree after skill installed", "Stop and wait for profile update").>
+
+---
+
+**Rule reminder (do not bypass):**
+If the required step is unavailable, you are BLOCKED. Diagnose using this template. Report. Stop. Unsafe fallbacks (temp scripts, direct API, MCP internals, direct imports, in-memory restoration, manual bypasses) are prohibited unless a controller has authorized them for this specific instance in a prior handoff.
+
+This report must appear in the final output / handoff before any further action.

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

@@ -10,6 +10,7 @@ Load the canonical workflow first:
 Final report schema: `schemas/review-merge-final-report.md`.
 
 Rules (llm-project-workflow):
+- **BLOCKED + DIAGNOSE default (required):** If any required step (load workflow, lease, profile/role, capability, worktree under branches/, preflight, tool, instruction, etc.) cannot be performed, STOP. State BLOCKED. Use [`blocked-diagnose-report.md`](../templates/blocked-diagnose-report.md) template exactly. Only safe non-mutating recovery. Report. Do not continue or fallback.
 - Only an eligible, NON-author reviewer merges. If authenticated user == PR
   author → STOP.
 - Do not merge unless the PR is open, mergeable, and its checks/review pass.
@@ -41,7 +42,15 @@ Steps:
 8. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
    *Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
 
-Then run the cleanup template (worktree-cleanup.md):
+Post-merge cleanup (#517): merger sessions must NOT perform ad hoc cleanup.
+- Record merge mutations separately from cleanup mutations in the controller handoff.
+- Hand cleanup to a `prgs-reconciler` session — never raw `git branch -d`,
+  `git push --delete`, curl/API comment deletion, or local scripts.
+- Reconciler cleanup must cite authorized MCP tools
+  (`gitea_reconcile_merged_cleanups`, `gitea_cleanup_post_merge_moot_lease`,
+  `gitea_delete_branch`, etc.) plus `gitea.branch.delete` capability proof.
+
+Then run the cleanup template (worktree-cleanup.md) in a reconciler session:
 - Verify expected file/commit presence on master (post-merge file-presence verification):
   - Run: git fetch <remote> --prune; git checkout master; git pull <remote> master --ff-only
   - Verify that the expected files added/modified in the PR are present on master (or absent if deleted).

skills/llm-project-workflow/templates/post-merge-cleanup-handoff.md

@@ -0,0 +1,27 @@
+# Template: post-merge cleanup (reconciler only)
+
+Copy, fill the `<...>` fields, and paste as the task prompt. Run only after merge
+is confirmed on remote master. Merger sessions must hand off here — never perform
+this cleanup inline (#517).
+
+```text
+Task: MCP-native post-merge cleanup for PR #<pr> / issue #<n>.
+
+Rules:
+- Active profile must be reconciler (`prgs-reconciler`) with `gitea.branch.delete`.
+- Use MCP tools only — no raw git branch delete, no API comment deletion scripts.
+- Record cleanup mutations separately from merge mutations in the handoff.
+
+Steps:
+1. `gitea_resolve_task_capability(task="reconcile_merged_cleanups", remote=prgs)`
+2. Confirm PR #<pr> merged on <remote>/master.
+3. `gitea_cleanup_post_merge_moot_lease` if a reviewer lease remains (append-only).
+4. `gitea_reconcile_merged_cleanups` for branch/worktree cleanup with dry-run first.
+5. Report authorized cleanup tools used and reconciler capability proof.
+
+Handoff ledger (required fields):
+- Merge mutations: (none — merger already recorded gitea_merge_pr)
+- Cleanup mutations: list exact MCP tools invoked
+- Reconciler capability: profile + gitea.branch.delete proof
+- Next actor: controller acceptance or none
+```

skills/llm-project-workflow/templates/recover-bad-state.md

@@ -3,12 +3,15 @@
 Copy, fill the `<...>` fields, paste as the task prompt. Recovery is read-then-
 act: gather facts first, never discard unmerged work.
 
+**BLOCKED + DIAGNOSE rule (llm-project-workflow):** If at any point a required step (including state recovery itself) cannot be performed, stop immediately, use the standard [`blocked-diagnose-report.md`](blocked-diagnose-report.md) template, attempt only safe non-mutating recovery, and report. Do not continue or fallback.
+
 ```text
 Task: recover repo state for <situation>. Do not lose unmerged work.
 
 Rules (llm-project-workflow):
-- Fail closed: if state is unclear or a step would delete unmerged work, STOP.
+- BLOCKED + DIAGNOSE default: if state is unclear, a required check fails, or a step would delete unmerged work or bypass a guard, STOP and emit a full blocked-diagnose-report.md using the template. Clearly state BLOCKED. Diagnose. Only non-mutating recovery.
 - Never push master. Never discard commits not safely pushed to <remote>.
+- Prove you are in a branches/ worktree before any recovery mutation.
 
 Diagnose first:
 1. git fetch <remote> --prune
@@ -16,8 +19,19 @@ Diagnose first:
 3. git rev-list --left-right --count <remote>/master...master   # ahead/behind
 4. For any PR involved: confirm state (open/closed/merged) AND whether
    <remote>/master actually contains its commits ("closed" != "merged").
+5. Check active leases, claims, and whether required skills/workflows are loaded.
 
-Act per case:
+If a required diagnostic or recovery step itself is unavailable (e.g. terminal broken, skill missing, guard blocks, wrong profile), emit:
+
+## Required step
+<the step>
+
+## Observed failure
+<...>
+
+(complete the full blocked-diagnose-report.md template)
+
+Act per case (only after clean diagnosis; if blocked, use the template and stop):
 - Dirty worktree from another issue: leave it; start yours in a new worktree.
 - Local master ahead of remote: confirm the extra commits live on a branch
   pushed to <remote>, THEN git reset --hard <remote>/master. Verify with
@@ -26,6 +40,7 @@ Act per case:
 - Branch deleted before merge: recover commits from a local branch/reflog (or
   git fsck --lost-found), re-push, reopen the PR.
 - Unauthorized untracked file: do not commit it; leave pre-existing artifacts.
+- Any blocker: use blocked-diagnose-report.md template and stop.
 
-Handoff: what was wrong, evidence, action taken, current state, what remains.
+Handoff: what was wrong, evidence, action taken, current state, what remains. If BLOCKED, include the full blocker report.
 ```

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

@@ -35,6 +35,7 @@ Load the canonical workflow first:
 Final report schema: `schemas/review-merge-final-report.md`.
 
 Rules (llm-project-workflow):
+- **BLOCKED + DIAGNOSE default (required):** If any required step (load workflow, lease, profile/role, capability, worktree under branches/, preflight, tool, instruction, etc.) cannot be performed, STOP. State BLOCKED. Use [`blocked-diagnose-report.md`](../templates/blocked-diagnose-report.md) template exactly. Only safe non-mutating recovery. Report. Do not continue or fallback.
 - Review in a SEPARATE detached review worktree, never the author's folder.
 - Worktree safety (#233): before checkout, diff, validation, review, or merge,
   report the starting worktree path and whether it was dirty. If unrelated

skills/llm-project-workflow/templates/start-issue.md

@@ -10,6 +10,7 @@ Final report schema: skills/llm-project-workflow/schemas/work-issue-final-report
 Router: skills/llm-project-workflow/SKILL.md (task mode: work-issue)
 
 Rules (llm-project-workflow):
+- **BLOCKED + DIAGNOSE default (required):** If any required step (load workflow, acquire lease, prove worktree under branches/, capability, profile, tool, instruction, preflight, etc.) cannot be performed, STOP. State BLOCKED. Use [`blocked-diagnose-report.md`](../templates/blocked-diagnose-report.md) template exactly. Only safe non-mutating recovery. Report. Do not continue or fallback.
 - No repo changes without a tracking issue. If none exists, create one first;
   if it can't be created, stop.
 - Work only in an isolated branch worktree under branches/. The main checkout

skills/llm-project-workflow/templates/worktree-cleanup.md

@@ -26,3 +26,43 @@ Steps:
 
 Handoff: merge confirmed, issue closed, branch+worktree removed, checkout clean.
 ```
+
+## Branches cleanup audit integrity (#404)
+
+Any bulk or multi-path cleanup under `branches/` must capture auditable before/after
+identity for every initial directory and registered worktree. Use
+`worktree_cleanup_audit.capture_cleanup_snapshot` before and after cleanup, record
+every intentional removal in a removal log (path, method, order, timestamp,
+pre-removal proof), then run `reconcile_cleanup_audit` and
+`assess_cleanup_audit_integrity`.
+
+The cleanup report must include a reconciliation table:
+
+* initial count
+* removed count
+* preserved count
+* missing-unexplained count
+* final count
+
+Fail closed when:
+
+* a preserved (active PR, dirty, claim/lease, or unsafe) worktree disappears without
+  a removal log entry or explicit explanation
+* the removal log omits a removed clean-stale path
+* final counts do not reconcile with initial minus removed
+
+If another session removes or mutates a worktree during cleanup, record the path
+under explained missing entries — never treat silent disappearance as success.
+
+## Bulk `branches/` cleanup audit (#404)
+
+Before removing multiple session-owned worktrees:
+
+1. Call `gitea_capture_branches_worktree_snapshot` and record the before snapshot.
+2. Remove only paths classified as `clean_stale_removable` with explicit per-path proof.
+3. Log every removal with path, method, and timestamp/order.
+4. Capture an after snapshot with the same tool.
+5. Call `gitea_assess_worktree_cleanup_integrity` with before, after, and the removal log.
+6. Fail closed when any protected path (active PR, dirty, claim/lease) disappears
+   without an explained state transition.
+7. Final report must include the reconciliation table and `git worktree list` proof.

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

@@ -12,6 +12,8 @@ This file is the canonical issue-creation workflow for Gitea-Tools. Load it
 before any issue mutation. Final report schema:
 [`schemas/create-issue-final-report.md`](../schemas/create-issue-final-report.md).
 
+**BLOCKED + DIAGNOSE (universal default):** If at any point you cannot perform a required step (skill not available, terminal broken, capability missing, wrong profile, guard blocks, worktree misbound, instructions unavailable, preflight fails, etc.), STOP. Clearly state BLOCKED. Use the standard [`../templates/blocked-diagnose-report.md`](../templates/blocked-diagnose-report.md) template. Only safe non-mutating recovery. Report using the template. Do not continue or use any fallback (temp scripts, direct API, etc.) unless controller authorizes. All blocker classes listed in llm-project-workflow/SKILL.md must trigger this.
+
 **Default task prompt:**
 
 > Create or update Gitea issues in this project only if every identity,

skills/llm-project-workflow/workflows/reconcile-landed-pr.md

@@ -304,6 +304,40 @@ If any required mutation capability is missing:
 * include safe next action (profile switch, human close, or dedicated reconciler
   profile)
 
+## 15A. Audit vs cleanup phase (#419)
+
+Reconciliation audits are **read-only** unless a separate cleanup phase is
+explicitly authorized.
+
+**Audit phase forbids** (``audit_reconciliation_mode.check_audit_mutation_allowed``
+fails closed):
+
+* ``gitea_delete_branch``
+* ``git branch -D``
+* ``git worktree remove``
+* pushes
+* issue/PR mutations
+* file edits
+
+Dry-run merged-cleanup reconciliation (``gitea_reconcile_merged_cleanups`` with
+``dry_run=True``) stays in audit phase. Execution requires:
+
+1. Operator approval or workflow authorization
+2. Exact ``delete_branch`` capability proof (``gitea.branch.delete``)
+3. Proof branch/worktree is safe to remove
+4. Before/after state snapshot
+
+Call ``gitea_authorize_reconciliation_cleanup_phase`` before any cleanup
+mutation. Final reports must not claim ``no mutations`` if cleanup occurred.
+Classify cleanup mutations as:
+
+* remote branch deletion → **External-state mutations**
+* local branch deletion → **Git ref mutations**
+* worktree removal → **Cleanup mutations**
+
+``audit_reconciliation_mode.assess_audit_reconciliation_report`` validates
+these boundaries in final reports.
+
 ## 16. Mutation classification
 
 Use precise mutation categories in the final report:

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

@@ -12,6 +12,8 @@ This file is the canonical PR review/merge workflow for Gitea-Tools. Load it
 before any PR mutation. Final report schema:
 [`schemas/review-merge-final-report.md`](../schemas/review-merge-final-report.md).
 
+**BLOCKED + DIAGNOSE (universal default):** If at any point you cannot perform a required step (skill not available, terminal broken, capability missing, wrong profile, guard blocks, worktree misbound, instructions unavailable, preflight fails, etc.), STOP. Clearly state BLOCKED. Use the standard [`../templates/blocked-diagnose-report.md`](../templates/blocked-diagnose-report.md) template. Only safe non-mutating recovery. Report using the template. Do not continue or use any fallback (temp scripts, direct API, etc.) unless controller authorizes. All blocker classes listed in llm-project-workflow/SKILL.md must trigger this.
+
 **Default task prompt:**
 
 > Review the next eligible open PR in this project. Merge it only if every
@@ -36,6 +38,44 @@ If available, load it first and report:
 
 If the canonical workflow cannot be loaded and the project requires it, stop and produce a recovery handoff only.
 
+## 0A. Workflow-load and session boundary anchor (#403)
+
+The MCP gate is the authority — not local file viewing.
+
+Before any reviewer mutation:
+
+1. Record pre-review commands with `gitea_record_pre_review_command` when they
+   are not automatically classified (inventory/diagnostic commands may be
+   recorded explicitly for proof).
+2. Call `gitea_load_review_workflow` to establish workflow hash proof **and**
+   session boundary state in the same in-process session proof.
+3. Do not claim the workflow was loaded by reading
+   `skills/llm-project-workflow/workflows/review-merge-pr.md` as a local file;
+   that narrative does not satisfy the validator.
+
+Allowed before workflow load (classify as `read_only_inventory` or
+`diagnostic`):
+
+* `gitea_whoami`, `gitea_resolve_task_capability`, `gitea_list_prs`,
+  `gitea_view_pr`, `gitea_get_runtime_context`
+* `git fetch` / `git remote update` for inventory
+* `git status`, `git worktree list` (read-only)
+
+Boundary violations (block downstream reviewer mutations even after load):
+
+* validation commands (`pytest`, `python -m unittest`) in the main checkout
+* local profile/credential/config inspection (`profiles.json`, `gitea-mcp.json`,
+  `.env`, keychain dumps)
+* MCP repair (`pkill`, MCP config edits)
+* git mutations before workflow load
+
+Final reports must include a structured **Workflow-load helper result** block
+copied from `gitea_load_review_workflow`, including at minimum:
+
+* `workflow_hash`
+* `final_report_schema_hash`
+* `boundary_status` (`clean` or `violation`)
+
 ## 1. Start with live identity, profile, runtime, and capability checks
 
 Prove:
@@ -796,6 +836,21 @@ inventory before continuing.
 Final reports must include lease session id, acquisition proof, heartbeat
 status, and release/blocked status.
 
+## 26B-1. Merger lease adoption (#536)
+
+When review and merge are **different sessions**, the merger must adopt the
+reviewer's lease — never manually seed `reviewer_pr_lease._SESSION_LEASE`.
+
+Before merge in a merger-only session:
+
+1. Confirm `approval_at_current_head` via `gitea_get_pr_review_feedback`.
+2. Call `gitea_adopt_merger_pr_lease` from a clean merger worktree under
+   `branches/`, passing `expected_head_sha` pinned to the approved head.
+3. Quote the adoption comment id and `adopted_from_session_id` in the handoff.
+4. Proceed to `gitea_merge_pr` only after adoption succeeds.
+
+Manual in-process lease seeding is rejected by mutation gates (fail closed).
+
 ## 26C. Conflict-fix lease and stale-head protection (#399)
 
 Before validating, approving, or merging a PR:
@@ -871,6 +926,16 @@ Confirm:
 
 Clean only the session-owned `branches/` review worktree if the project workflow explicitly allows cleanup.
 
+Review, baseline, and merge-simulation worktrees created during this run are
+transient and are removed automatically at successful completion once they are
+clean, carry no open PR, and hold no active lease (#401). Use
+`gitea_audit_worktree_cleanup` (read-only) to classify `branches/` entries; only
+`clean_stale_removable` and `detached_review_leftover` may be removed, one-by-one,
+after `git worktree list` proof plus per-worktree proof of path, branch/HEAD,
+clean/dirty status, no active PR/lease, and the removal result. Dirty,
+active-PR, active-issue, and leased worktrees are never deleted automatically; a
+failed removal must be reported with the leftover path and reason.
+
 Do not delete or mutate unrelated branches/worktrees.
 
 Do not touch the main checkout except to update the stable branch after merge if explicitly allowed by the workflow.
@@ -881,6 +946,61 @@ Do not update the main checkout if merge failed, was blocked, or produced reconc
 
 If any local artifact is created after final cleanup, run and report a new final status check.
 
+## 28A. Post-merge cleanup proof checklist (#402)
+
+Successful tool execution is not proof that cleanup was authorized. Before claiming remote branch deletion or local worktree removal, the final report must carry the full safety checklist below. If any gate is missing, report `CLEANUP_SKIPPED` with the exact blocker — never perform cleanup and never claim it was performed.
+
+### Remote branch deletion checklist
+
+When `gitea_delete_branch` (or equivalent) deletes the merged PR head branch, report:
+
+* Delete-branch capability resolved: name the task (`delete_branch` / `cleanup_branch` / `reconcile_merged_cleanups`) and permission (`gitea.branch.delete`) with resolver proof before the delete call
+* Merge result: merged
+* Merge commit SHA: full 40-character SHA
+* Merged PR head branch / deleted branch: exact branch name (must match)
+* Branch protection: none / branch is not protected
+* Open PR inventory proof: no other open PR references the branch
+* Active heartbeat/claim/lease: none
+
+### Local worktree removal checklist
+
+When removing session-owned review/simulation worktrees under `branches/`, report:
+
+* Session-owned worktree path: exact path under `branches/`
+* Pre-removal tracked state: clean
+* Pre-removal untracked state: clean
+* Git worktree list after removal: command output or equivalent proof
+
+### Skipped cleanup
+
+If any gate fails, report:
+
+* Cleanup outcome: `CLEANUP_SKIPPED`
+* Cleanup blocker: exact missing gate (for example `gitea.branch.delete capability not resolved`)
+
+Skipped cleanup with an exact blocker passes validation. Performed-cleanup claims without the checklist fail validation.
+
+## 28B. MCP-native cleanup only (#517)
+
+Post-merge cleanup of leases, comments, branches, and worktrees must go through
+explicit MCP tools — never raw git, curl/API scripts, or ad hoc helper scripts.
+
+Merger and reviewer sessions must **not** perform cleanup inline. Record:
+
+* **Merge mutations** — only `gitea_merge_pr` (or review mutations for review-only runs)
+* **Cleanup mutations** — only authorized reconciler MCP tools, cited by exact tool name
+
+Authorized cleanup tools include `gitea_reconcile_merged_cleanups`,
+`gitea_cleanup_post_merge_moot_lease`, `gitea_delete_branch`, and
+`gitea_cleanup_merged_pr_branch` (when available). Lease cleanup uses
+append-only release comments — never delete another session's lease comment.
+
+Hand cleanup to a `prgs-reconciler` session with `gitea.branch.delete` capability
+proof. Raw `git branch -d`, `git push --delete`, and comment-deletion API calls
+are blocked in final-report validation.
+
+Template: `templates/post-merge-cleanup-handoff.md`.
+
 ## 29. Recovery handoff rules
 
 If blocked, produce a recovery handoff with:

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

@@ -12,6 +12,8 @@ This file is the canonical author/coder workflow for Gitea-Tools. Load it
 before any issue implementation mutation. Final report schema:
 [`schemas/work-issue-final-report.md`](../schemas/work-issue-final-report.md).
 
+**BLOCKED + DIAGNOSE (universal default):** If at any point you cannot perform a required step (skill not available, terminal broken, capability missing, wrong profile, guard blocks, worktree misbound, instructions unavailable, preflight fails, etc.), STOP. Clearly state BLOCKED. Use the standard [`../templates/blocked-diagnose-report.md`](../templates/blocked-diagnose-report.md) template. Only safe non-mutating recovery. Report using the template. Do not continue or use any fallback (temp scripts, direct API, etc.) unless controller authorizes. All blocker classes listed in llm-project-workflow/SKILL.md must trigger this.
+
 **Default task prompt:**
 
 > Find the next eligible issue in this project, work on it only if all gates
@@ -699,6 +701,39 @@ Do not update the main checkout unless the canonical workflow explicitly allows
 
 Any cleanup is a mutation and must be reported.
 
+### 22A. Session-owned worktree cleanup and TTL (#401)
+
+Every session-owned worktree created under `branches/` has ownership metadata:
+path, workflow type, issue number, PR number, branch/head SHA, creator
+identity/profile, created timestamp, last-used timestamp, and cleanup
+eligibility.
+
+Cleanup is classification-driven. `gitea_audit_worktree_cleanup` (read-only)
+classifies every `branches/` entry as exactly one of:
+
+* `active_open_pr` — branch has an open PR; never auto-removed.
+* `active_issue_work` — active claim/lease or fresh issue worktree; never
+  auto-removed.
+* `dirty_local_worktree` — uncommitted changes; never auto-removed.
+* `clean_stale_removable` — clean, no PR, no lease; removable.
+* `detached_review_leftover` — clean detached review/baseline/merge-simulation
+  worktree; removable.
+* `unsafe_unknown` — protected base checkout or unknown workflow type; never
+  auto-removed.
+
+Only `clean_stale_removable` and `detached_review_leftover` may be removed, and
+only one-by-one after `git worktree list` proof plus per-worktree proof of:
+worktree path, branch/HEAD, clean/dirty status, no active PR/lease, and the
+removal result. Review, baseline, and merge-simulation worktrees are removed
+automatically at successful workflow completion; issue/conflict-fix worktrees
+are removed only after their TTL (`GITEA_WORKTREE_TTL_HOURS`, default 24h)
+expires and no lock/lease is held.
+
+Dirty, active-PR, active-issue, and leased worktrees are never deleted
+automatically. If a removal fails, the final report must list the leftover
+worktree path and the reason. Include the `git worktree list` output as final
+cleanup verification.
+
 ## 23. Recovery handoff rules
 
 If blocked, produce a recovery handoff with:

task_capability_map.py

@@ -68,6 +68,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.merge",
         "role": "reviewer",
     },
+    "adopt_merger_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "reviewer",
+    },
     "blind_pr_queue_review": {
         "permission": "gitea.pr.review",
         "role": "reviewer",
@@ -104,6 +108,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
+    "reconciliation_cleanup": {
+        "permission": "gitea.branch.delete",
+        "role": "author",
+    },
     "work_issue": {
         "permission": "gitea.pr.create",
         "role": "author",

tests/test_audit.py

@@ -157,6 +157,7 @@ class _AuditWiringBase(unittest.TestCase):
 
     def tearDown(self):
         mcp_server._IDENTITY_CACHE.clear()
+        mcp_server.review_workflow_load.clear_review_workflow_load()
         self._dir.cleanup()
 
     def _env(self, **extra):
@@ -291,12 +292,15 @@ class TestGatedToolAudit(_AuditWiringBase):
 
     def setUp(self):
         super().setUp()
-        from mcp_server import init_review_decision_lock
+        from tests.test_mcp_server import _init_reviewer_session, _install_owned_reviewer_lease
-        from tests.test_mcp_server import _install_owned_reviewer_lease
         import reviewer_pr_lease
+        import review_workflow_boundary
+        import review_workflow_load
 
-        # init_review_decision_lock clears any prior session lease (#407).
+        # Session init clears any prior session lease (#407) and loads workflow (#389).
-        init_review_decision_lock("prgs", "review_pr")
+        _init_reviewer_session("prgs")
+        self.addCleanup(review_workflow_load.clear_review_workflow_load)
+        self.addCleanup(review_workflow_boundary.clear_pre_review_commands)
         self._lease_patch = _install_owned_reviewer_lease(8)
         self._lease_patch.start()
         self._auth_identity_patch = patch(
@@ -337,6 +341,7 @@ class TestGatedToolAudit(_AuditWiringBase):
         env = self._env(GITEA_PROFILE_NAME="gitea-merger",
                         GITEA_ALLOWED_OPERATIONS="read,merge")
         with patch.dict(os.environ, env, clear=True):
+            mcp_server.gitea_load_review_workflow()
             r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8",
                                expected_head_sha="abc123", remote="prgs")
         self.assertTrue(r["performed"])
@@ -356,6 +361,7 @@ class TestGatedToolAudit(_AuditWiringBase):
         env = self._env(GITEA_PROFILE_NAME="gitea-merger",
                         GITEA_ALLOWED_OPERATIONS="read,merge")
         with patch.dict(os.environ, env, clear=True):
+            mcp_server.gitea_load_review_workflow()
             r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8", remote="prgs")
         self.assertFalse(r["performed"])
         recs = self._records()

tests/test_audit_reconciliation_mode.py

@@ -0,0 +1,291 @@
+"""Tests for audit vs cleanup reconciliation mode (#419)."""
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import audit_reconciliation_mode as arm
+import mcp_server
+from audit_reconciliation_mode import (
+    AUDIT_FORBIDDEN_TASKS,
+    PHASE_AUDIT,
+    PHASE_CLEANUP,
+    assess_audit_command_allowed,
+    assess_audit_reconciliation_report,
+    authorize_cleanup_phase,
+    check_audit_mutation_allowed,
+    check_cleanup_execution_allowed,
+    classify_cleanup_mutation,
+    clear_phase,
+    enter_audit_phase,
+)
+from final_report_validator import assess_final_report_validator
+from review_proofs import assess_audit_reconciliation_report as proofs_assess
+from task_capability_map import required_permission, required_role
+
+DELETE_PROFILE = {
+    "profile_name": "prgs-author-delete",
+    "allowed_operations": ["gitea.read", "gitea.branch.delete"],
+    "forbidden_operations": [],
+    "audit_label": "prgs-author-delete",
+}
+
+READ_PROFILE = {
+    "profile_name": "prgs-author",
+    "allowed_operations": ["gitea.read", "gitea.issue.comment"],
+    "forbidden_operations": ["gitea.branch.delete"],
+    "audit_label": "prgs-author",
+}
+
+READ_ENV = {
+    "GITEA_MCP_CONFIG": os.path.join(
+        os.path.dirname(__file__), "..", "profiles.json"
+    ),
+    "GITEA_MCP_PROFILE": "prgs-author",
+}
+
+
+def _authorize_cleanup(**kwargs):
+    defaults = {
+        "operator_approved": True,
+        "delete_capability_proven": True,
+        "safety_proof": {"safe_to_delete_remote": True},
+        "before_after_snapshot": {
+            "before": "remote branch exists",
+            "after": "remote branch absent",
+        },
+    }
+    defaults.update(kwargs)
+    return authorize_cleanup_phase(**defaults)
+
+
+def _cleanup_report(**overrides):
+    base = (
+        "Task mode: reconcile-landed-pr\n"
+        "Workflow source: workflows/reconcile-landed-pr.md\n"
+        "Audit phase: read-only assessment\n"
+        "Cleanup phase authorized: true\n"
+        "Delete-branch capability proven: true\n"
+        "Branch safe to remove: true\n"
+        "Before/after state snapshot: remote branch feat/x present → absent\n"
+        "External-state mutations: deleted remote branch feat/x\n"
+        "Git ref mutations: none\n"
+        "Cleanup mutations: removed worktree branches/feat-x\n"
+    )
+    for key, value in overrides.items():
+        base = base.replace(key, value)
+    return base
+
+
+class TestAuditPhaseGates(unittest.TestCase):
+    def setUp(self):
+        clear_phase()
+        enter_audit_phase("reconcile-landed-pr")
+
+    def tearDown(self):
+        clear_phase()
+
+    def test_audit_blocks_delete_branch_task(self):
+        allowed, reasons = check_audit_mutation_allowed("delete_branch")
+        self.assertFalse(allowed)
+        self.assertTrue(reasons)
+
+    def test_audit_blocks_worktree_shell_commands(self):
+        allowed, reasons = assess_audit_command_allowed(
+            "git worktree remove branches/feat-x"
+        )
+        self.assertFalse(allowed)
+        self.assertTrue(reasons)
+
+    def test_audit_blocks_local_branch_delete_command(self):
+        allowed, reasons = assess_audit_command_allowed("git branch -D feat/x")
+        self.assertFalse(allowed)
+
+    def test_audit_allows_read_tasks(self):
+        allowed, _ = check_audit_mutation_allowed("reconcile_landed_pr")
+        self.assertTrue(allowed)
+
+    def test_forbidden_set_covers_issue_and_pr_mutations(self):
+        for task in ("close_pr", "comment_issue", "commit_files", "push_branch"):
+            self.assertIn(task, AUDIT_FORBIDDEN_TASKS)
+
+
+class TestCleanupAuthorization(unittest.TestCase):
+    def setUp(self):
+        clear_phase()
+        enter_audit_phase("reconcile_merged_cleanups")
+
+    def tearDown(self):
+        clear_phase()
+
+    def test_cleanup_without_approval_blocked(self):
+        result = authorize_cleanup_phase(
+            delete_capability_proven=True,
+            safety_proof={"safe_to_delete_remote": True},
+            before_after_snapshot={"before": "a", "after": "b"},
+        )
+        self.assertFalse(result["authorized"])
+
+    def test_cleanup_without_capability_proof_blocked(self):
+        result = _authorize_cleanup(delete_capability_proven=False)
+        self.assertFalse(result["authorized"])
+
+    def test_cleanup_without_snapshot_blocked(self):
+        result = _authorize_cleanup(before_after_snapshot={"before": "", "after": ""})
+        self.assertFalse(result["authorized"])
+
+    def test_authorized_cleanup_switches_phase(self):
+        result = _authorize_cleanup()
+        self.assertTrue(result["authorized"])
+        self.assertEqual(result["phase"], PHASE_CLEANUP)
+
+    def test_cleanup_execution_allowed_only_after_authorization(self):
+        self.assertFalse(check_cleanup_execution_allowed()[0])
+        _authorize_cleanup()
+        self.assertTrue(check_cleanup_execution_allowed()[0])
+
+
+class TestReportVerifier(unittest.TestCase):
+    def test_false_no_mutations_after_cleanup_blocked(self):
+        report = (
+            "Task mode: reconcile-landed-pr\n"
+            "No mutations performed.\n"
+            "delete_remote_branch feat/dup\n"
+        )
+        result = assess_audit_reconciliation_report(report)
+        self.assertFalse(result["proven"])
+        self.assertIn("no mutations", result["reasons"][0].lower())
+
+    def test_cleanup_without_authorization_fields_blocked(self):
+        report = (
+            "Task mode: reconcile-landed-pr\n"
+            "remove_local_worktree branches/feat-x\n"
+        )
+        result = assess_audit_reconciliation_report(report)
+        self.assertFalse(result["proven"])
+
+    def test_authorized_cleanup_report_passes(self):
+        result = assess_audit_reconciliation_report(_cleanup_report())
+        self.assertTrue(result["proven"])
+
+    def test_mutation_classification_enforced(self):
+        report = (
+            "Task mode: reconcile-landed-pr\n"
+            "Cleanup phase authorized: true\n"
+            "Delete-branch capability proven: true\n"
+            "Branch safe to remove: true\n"
+            "Before/after state snapshot: present\n"
+            "remove_local_worktree branches/feat-x\n"
+        )
+        result = assess_audit_reconciliation_report(report)
+        self.assertFalse(result["proven"])
+
+    def test_proofs_export_matches_module(self):
+        report = "No mutations performed.\ndelete_remote_branch feat/dup"
+        self.assertEqual(
+            proofs_assess(report)["proven"],
+            assess_audit_reconciliation_report(report)["proven"],
+        )
+
+    def test_final_report_validator_includes_boundary_rule(self):
+        report = "No mutations performed.\ndelete_remote_branch feat/dup"
+        result = assess_final_report_validator(
+            report_text=report,
+            task_kind="reconcile_already_landed",
+        )
+        self.assertTrue(result["blocked"])
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reconcile.audit_cleanup_boundary", rule_ids)
+
+
+class TestMutationClassification(unittest.TestCase):
+    def test_remote_delete_is_external_state(self):
+        self.assertEqual(
+            classify_cleanup_mutation("delete_remote_branch"),
+            "external-state",
+        )
+
+    def test_worktree_remove_is_cleanup(self):
+        self.assertEqual(
+            classify_cleanup_mutation("remove_local_worktree"),
+            "cleanup",
+        )
+
+
+class TestMcpGates(unittest.TestCase):
+    def setUp(self):
+        clear_phase()
+        enter_audit_phase("reconcile_merged_cleanups")
+        self.mock_api = patch("mcp_server.api_request").start()
+        self.mock_auth = patch(
+            "mcp_server.get_auth_header", return_value="token test"
+        ).start()
+
+    def tearDown(self):
+        patch.stopall()
+        clear_phase()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    @patch.dict(os.environ, READ_ENV, clear=True)
+    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
+    def test_delete_branch_blocked_in_audit_phase(self, _profile):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", resolved_role="author")
+        result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
+        self.assertFalse(result["success"])
+        self.assertEqual(result["audit_phase"], PHASE_AUDIT)
+        self.mock_api.assert_not_called()
+
+    @patch.dict(os.environ, READ_ENV, clear=True)
+    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
+    def test_reconcile_execute_blocked_without_cleanup_auth(self, _profile):
+        with self.assertRaises(ValueError):
+            mcp_server.gitea_reconcile_merged_cleanups(
+                dry_run=False,
+                execute_confirmed=False,
+                remote="prgs",
+            )
+
+    @patch.dict(os.environ, READ_ENV, clear=True)
+    @patch("mcp_server.get_profile", return_value=READ_PROFILE)
+    def test_cleanup_auth_fails_without_delete_capability(self, _profile):
+        result = mcp_server.gitea_authorize_reconciliation_cleanup_phase(
+            operator_approved=True,
+            delete_capability_proven=True,
+            safe_to_delete_remote=True,
+            before_state="exists",
+            after_state="gone",
+        )
+        self.assertFalse(result["authorized"])
+        self.assertFalse(result["delete_capability_verified"])
+
+    @patch.dict(os.environ, READ_ENV, clear=True)
+    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
+    def test_delete_branch_allowed_after_cleanup_authorization(self, _profile):
+        mcp_server.gitea_authorize_reconciliation_cleanup_phase(
+            operator_approved=True,
+            delete_capability_proven=True,
+            safe_to_delete_remote=True,
+            before_state="exists",
+            after_state="gone",
+        )
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", resolved_role="author")
+        self.mock_api.return_value = {}
+        result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
+        self.assertTrue(result["success"])
+
+
+class TestTaskCapabilityMap(unittest.TestCase):
+    def test_reconciliation_cleanup_maps_delete_permission(self):
+        self.assertEqual(
+            required_permission("reconciliation_cleanup"),
+            "gitea.branch.delete",
+        )
+        self.assertEqual(required_role("reconciliation_cleanup"), "author")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_create_issue_workspace_guard.py

@@ -38,8 +38,18 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
     @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
     @patch("gitea_mcp_server.api_request")
     @patch("gitea_mcp_server.api_get_all", return_value=[])
-    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", return_value={"current_branch": "feat/issue-1"})
+    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
-    def test_create_issue_stable_checkout_rejected(self, _git, _get_all, mock_api, _role, _ns, _prof, _auth):
+    @patch(
+        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={
+            "current_branch": "master",
+            "head_sha": "a" * 40,
+            "porcelain_status": "",
+        },
+    )
+    def test_create_issue_stable_checkout_rejected(
+        self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth,
+    ):
         # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
         # path is the stable control checkout (not under branches/), mutation must fail.
         with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):

tests/test_llm_agent_sha.py

@@ -22,6 +22,7 @@ from unittest.mock import patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
+import mcp_server  # noqa: E402
 from mcp_server import (  # noqa: E402
     gitea_check_pr_eligibility,
     gitea_merge_pr,

tests/test_llm_workflow_split.py

@@ -87,6 +87,14 @@ def test_reconcile_landed_workflow_contract():
     assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text
     assert "RECOVERY_HANDOFF_ONLY" in text
     assert "resolve_partial_reconciliation_plan" in text
+    assert "check_audit_mutation_allowed" in text
+    assert "gitea_authorize_reconciliation_cleanup_phase" in text
+
+
+def test_audit_reconciliation_verifier_exported():
+    from review_proofs import assess_audit_reconciliation_report
+
+    assert callable(assess_audit_reconciliation_report)
 
 
 def test_create_issue_workflow_contract():

tests/test_lock_issue_mcp_registration.py

@@ -0,0 +1,127 @@
+"""Regression tests for gitea_lock_issue MCP tool registration (#521)."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import issue_lock_store
+import mcp_server
+from mcp_server import gitea_create_pr, gitea_lock_issue
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+
+ISSUE_WRITE_ENV = {
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
+    ),
+}
+
+CREATE_PR_ENV = {
+    "GITEA_PROFILE_NAME": "author-test",
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.read,gitea.pr.create,gitea.branch.push,"
+        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
+    ),
+    "GITEA_FORBIDDEN_OPERATIONS": (
+        "gitea.pr.approve,gitea.pr.merge,gitea.pr.review"
+    ),
+}
+
+
+def _clean_master_git_state_for_lock():
+    return {
+        "current_branch": "master",
+        "porcelain_status": "",
+        "base_equivalent": True,
+        "inspected_git_root": "/scratch/wt",
+        "base_branch": "origin/master",
+    }
+
+
+def _registered_tool_names() -> set[str]:
+    manager = mcp_server.mcp._tool_manager
+    tools = getattr(manager, "_tools", None) or {}
+    return set(tools.keys())
+
+
+class TestLockIssueMcpRegistration(unittest.TestCase):
+    def test_gitea_lock_issue_registered_as_public_mcp_tool(self):
+        names = _registered_tool_names()
+        self.assertIn("gitea_lock_issue", names)
+
+    def test_internal_list_open_pulls_not_exposed_as_mcp_tool(self):
+        names = _registered_tool_names()
+        self.assertNotIn("_list_open_pulls", names)
+
+
+class TestCreatePrLockRegistrationFlow(unittest.TestCase):
+    def setUp(self):
+        self._lock_dir = tempfile.TemporaryDirectory()
+        self._env_patcher = patch.dict(
+            os.environ,
+            {
+                **CREATE_PR_ENV,
+                "GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
+            },
+            clear=True,
+        )
+        self._env_patcher.start()
+        self._dup_fetcher_patcher = patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            return_value=([], [], {"status": "not_claimed"}),
+        )
+        self._dup_fetcher_patcher.start()
+
+    def tearDown(self):
+        self._dup_fetcher_patcher.stop()
+        self._env_patcher.stop()
+        self._lock_dir.cleanup()
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_still_fails_closed_without_issue_lock(self, _auth, _role):
+        with self.assertRaises(RuntimeError) as ctx:
+            gitea_create_pr(
+                title="feat: X Closes #521",
+                head="feat/issue-521-lock-issue-registration",
+                remote="prgs",
+            )
+        self.assertIn("Issue lock is missing", str(ctx.exception))
+
+    @patch("mcp_server.api_request")
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_proceeds_after_valid_issue_lock(
+        self, _auth, _role, _api, _git_state, mock_api_request
+    ):
+        worktree = os.path.realpath(os.getcwd())
+        mock_api_request.return_value = {"number": 521, "html_url": "https://example/pr/521"}
+        lock_res = gitea_lock_issue(
+            issue_number=521,
+            branch_name="feat/issue-521-lock-issue-registration",
+            remote="prgs",
+            worktree_path=worktree,
+        )
+        self.assertTrue(lock_res["success"])
+        lock_path = lock_res["lock_file_path"]
+        self.assertTrue(os.path.exists(lock_path))
+        lock = issue_lock_store.read_lock_file(lock_path)
+        self.assertEqual(lock["issue_number"], 521)
+
+        res = gitea_create_pr(
+            title="fix: restore lock tool registration Closes #521",
+            head="feat/issue-521-lock-issue-registration",
+            remote="prgs",
+            worktree_path=worktree,
+        )
+        self.assertEqual(res["number"], 521)

tests/test_mcp_native_cleanup_proof.py

@@ -0,0 +1,251 @@
+"""Tests for MCP-native post-merge cleanup proof (#517)."""
+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 mcp_native_cleanup_proof import (  # noqa: E402
+    assess_authorized_reconciler_cleanup_path,
+    assess_mcp_native_cleanup_proof,
+    assess_merger_cleanup_handoff_guidance,
+    assess_raw_branch_delete_report,
+    assess_raw_comment_delete_report,
+)
+
+
+def _authorized_cleanup_report(**overrides):
+    fields = {
+        "Merge mutations": "gitea_merge_pr on PR #100",
+        "Cleanup mutations": (
+            "gitea_reconcile_merged_cleanups deleted remote branch; "
+            "gitea_cleanup_post_merge_moot_lease released lease"
+        ),
+        "Reconciler capability": "prgs-reconciler / gitea.branch.delete resolved",
+        "Next actor": "reconciler session complete",
+    }
+    fields.update(overrides)
+    lines = ["## Controller Handoff", ""]
+    lines.extend(f"- {key}: {value}" for key, value in fields.items())
+    return "\n".join(lines)
+
+
+class TestRawBranchDeleteBlocked(unittest.TestCase):
+    def test_git_branch_d_blocked(self):
+        report = "Cleanup mutations: git branch -d feat/issue-1-x"
+        result = assess_raw_branch_delete_report(report)
+        self.assertTrue(result["block"])
+        self.assertIn("git branch -d", result["commands"][0])
+
+    def test_git_push_delete_in_ledger_blocked(self):
+        report = "- Git ref mutations: git push origin --delete feat/x"
+        result = assess_raw_branch_delete_report(report)
+        self.assertTrue(result["block"])
+
+    def test_narrative_git_push_delete_not_blocked(self):
+        report = "\n".join([
+            "## Validation",
+            "Workflow §28B forbids raw git push --delete for cleanup.",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: none",
+            "- Worktree mutations: none",
+        ])
+        result = assess_raw_branch_delete_report(report)
+        self.assertFalse(result["block"])
+
+    def test_authorized_mcp_path_passes(self):
+        report = _authorized_cleanup_report()
+        result = assess_raw_branch_delete_report(report)
+        self.assertFalse(result["block"])
+
+
+class TestRawCommentDeleteBlocked(unittest.TestCase):
+    def test_curl_delete_comment_in_ledger_blocked(self):
+        report = (
+            "- Cleanup mutations: curl -X DELETE https://gitea.example/api/v1/repos/o/r/"
+            "issues/9/comments/42"
+        )
+        result = assess_raw_comment_delete_report(report)
+        self.assertTrue(result["block"])
+
+    def test_narrative_comment_delete_not_blocked(self):
+        report = "\n".join([
+            "Files reviewed: post_merge_cleanup_proof.py (raw comment delete patterns)",
+            "Validation note: never use delete_issue_comment for lease cleanup.",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: none",
+            "- Worktree mutations: none",
+        ])
+        result = assess_raw_comment_delete_report(report)
+        self.assertFalse(result["block"])
+
+    def test_delete_issue_comment_helper_in_ledger_blocked(self):
+        report = "- Cleanup mutations: delete_issue_comment purged lease comments"
+        result = assess_raw_comment_delete_report(report)
+        self.assertTrue(result["block"])
+
+    def test_moot_lease_append_only_passes(self):
+        report = _authorized_cleanup_report()
+        result = assess_raw_comment_delete_report(report)
+        self.assertFalse(result["block"])
+
+
+class TestAuthorizedReconcilerPath(unittest.TestCase):
+    def test_cleanup_without_mcp_tool_blocked(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: removed branch manually after merge",
+            "- Reconciler capability: prgs-reconciler",
+        ])
+        result = assess_authorized_reconciler_cleanup_path(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("authorized MCP" in r for r in result["reasons"]))
+
+    def test_authorized_cleanup_succeeds(self):
+        result = assess_authorized_reconciler_cleanup_path(_authorized_cleanup_report())
+        self.assertFalse(result["block"])
+
+    def test_git_ref_cleanup_without_mcp_tool_blocked(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: git branch -D feat/issue-517-test",
+            "- Reconciler capability: prgs-reconciler",
+        ])
+        result = assess_authorized_reconciler_cleanup_path(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("Git ref mutations cleanup must name" in r for r in result["reasons"])
+        )
+
+    def test_worktree_cleanup_without_mcp_tool_blocked(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Worktree mutations: git worktree remove branches/review-pr542",
+            "- Reconciler capability: prgs-reconciler",
+        ])
+        result = assess_authorized_reconciler_cleanup_path(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("Worktree mutations cleanup must name" in r for r in result["reasons"])
+        )
+
+    def test_git_fetch_in_git_ref_mutations_not_cleanup_claim(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: git fetch prgs master",
+        ])
+        result = assess_authorized_reconciler_cleanup_path(report)
+        self.assertFalse(result["block"])
+
+
+class TestMergerHandoffGuidance(unittest.TestCase):
+    def test_merger_cleanup_without_handoff_blocked(self):
+        report = "Merger deleted remote branch and removed worktree after merge"
+        result = assess_merger_cleanup_handoff_guidance(report)
+        self.assertTrue(result["block"])
+
+    def test_merger_cleanup_with_reconciler_handoff_passes(self):
+        report = (
+            "Merger deleted remote branch; next actor: reconciler for worktree audit"
+        )
+        result = assess_merger_cleanup_handoff_guidance(report)
+        self.assertFalse(result["block"])
+
+
+class TestCompositeVerifier(unittest.TestCase):
+    def test_full_authorized_report_passes(self):
+        result = assess_mcp_native_cleanup_proof(_authorized_cleanup_report())
+        self.assertFalse(result["block"], result["reasons"])
+
+    def test_raw_bypasses_fail_composite(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Merge mutations: gitea_merge_pr",
+            "- Cleanup mutations: git branch -D feat/x; curl -X DELETE .../comments/1",
+        ])
+        result = assess_mcp_native_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertGreaterEqual(len(result["reasons"]), 2)
+
+    def test_narrative_cleanup_mentions_pass_when_ledgers_none(self):
+        report = "\n".join([
+            "## Review summary",
+            "Validated workflow §28B MCP-native cleanup guidance.",
+            "Files reviewed describe raw git branch -d and delete_issue_comment blocks.",
+            "",
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: none",
+            "- Worktree mutations: none",
+        ])
+        result = assess_mcp_native_cleanup_proof(report)
+        self.assertFalse(result["block"], result["reasons"])
+
+    def test_git_ref_cleanup_bypass_blocked_in_composite(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: git branch -D feat/issue-517-test",
+            "- Reconciler capability: prgs-reconciler",
+        ])
+        result = assess_mcp_native_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("Git ref mutations cleanup must name" in r for r in result["reasons"])
+        )
+
+
+class TestFinalReportValidatorIntegration(unittest.TestCase):
+    def test_validator_blocks_raw_branch_delete_in_review_report(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: git branch -d feat/issue-517-test",
+        ])
+        result = assess_final_report_validator(report, "review_pr")
+        rules = {f["rule_id"] for f in result["findings"]}
+        self.assertIn("shared.mcp_native_cleanup_proof", rules)
+
+    def test_validator_passes_authorized_cleanup_report(self):
+        result = assess_final_report_validator(_authorized_cleanup_report(), "review_pr")
+        blocked = [f for f in result["findings"] if f.get("severity") == "block"]
+        mcp_blocks = [
+            f for f in blocked if f.get("rule_id") == "shared.mcp_native_cleanup_proof"
+        ]
+        self.assertEqual(mcp_blocks, [])
+
+    def test_validator_passes_narrative_cleanup_mentions_with_none_ledgers(self):
+        report = "\n".join([
+            "## Review summary",
+            "Workflow §28B documents raw git push --delete and delete_issue_comment blocks.",
+            "",
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: none",
+            "- Worktree mutations: none",
+        ])
+        result = assess_final_report_validator(report, "review_pr")
+        mcp_blocks = [
+            f for f in result["findings"]
+            if f.get("rule_id") == "shared.mcp_native_cleanup_proof"
+            and f.get("severity") == "block"
+        ]
+        self.assertEqual(mcp_blocks, [])
+
+    def test_validator_blocks_git_ref_cleanup_bypass(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup mutations: none",
+            "- Git ref mutations: git branch -D feat/issue-517-test",
+            "- Reconciler capability: prgs-reconciler",
+        ])
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        rules = {f["rule_id"] for f in result["findings"] if f.get("severity") == "block"}
+        self.assertIn("shared.mcp_native_cleanup_proof", rules)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mcp_server.py

@@ -58,6 +58,12 @@ _NO_BLOCKER_FEEDBACK = {
 }
 
 
+def _init_reviewer_session(remote="prgs"):
+    """Seed review decision lock and required workflow-load proof (#389)."""
+    init_review_decision_lock(remote, "review_pr")
+    mcp_server.gitea_load_review_workflow()
+
+
 def _mark_request_changes_ready(pr_number=8, **kwargs):
     """Mark a request_changes decision ready with the #332 duplicate-
     suppression feedback fetch stubbed to 'no existing blocker'."""
@@ -121,6 +127,7 @@ def _install_owned_reviewer_lease(
     session_id=_DEFAULT_LEASE_SESSION,
     head_sha="abc123",
 ):
+    import merger_lease_adoption as mla
     import reviewer_pr_lease
 
     reviewer_pr_lease.clear_session_lease()
@@ -129,7 +136,11 @@ def _install_owned_reviewer_lease(
         "session_id": session_id,
         "candidate_head": head_sha,
         "target_branch": "master",
-    })
+        "comment_id": 9001,
+    }, lease_provenance=mla.build_lease_provenance(
+        source=mla.SOURCE_ACQUIRE,
+        comment_id=9001,
+    ))
     return patch(
         "mcp_server._fetch_pr_comments",
         return_value=[
@@ -179,6 +190,7 @@ def _seed_ready_review_decision(
         "correction_authorized": False,
         "correction_reason": None,
     })
+    _m.gitea_load_review_workflow()
 
 
 # Issue-write tools are profile-gated (#69).
@@ -671,6 +683,7 @@ class TestMergePR(unittest.TestCase):
     def setUp(self):
         import reviewer_pr_lease
 
+        mcp_server.gitea_load_review_workflow()
         self._lease_patch = _install_owned_reviewer_lease(8)
         self._lease_patch.start()
         self._auth_identity_patch = patch(
@@ -1049,7 +1062,8 @@ class TestMergePR(unittest.TestCase):
                "GITEA_ALLOWED_OPERATIONS": "read,merge"}
         with patch.dict(os.environ, env, clear=True):
             r = gitea_merge_pr(
-                pr_number=8, confirmation=self._confirm(8), remote="prgs")
+                pr_number=8, confirmation=self._confirm(8), remote="prgs",
+                expected_head_sha=new_sha)
         self.assertFalse(r["performed"])
         self.assertTrue(r.get("approval_visible"))
         self.assertFalse(r.get("approval_at_current_head"))
@@ -1876,7 +1890,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
     def setUp(self):
         import reviewer_pr_lease
 
-        init_review_decision_lock("prgs", "review_pr")
+        _init_reviewer_session("prgs")
         self._lease_patch = _install_owned_reviewer_lease(
             self.PR, head_sha=self.SHA,
         )
@@ -1996,7 +2010,7 @@ class TestSubmitPrReview(unittest.TestCase):
     def setUp(self):
         import reviewer_pr_lease
 
-        init_review_decision_lock("prgs", "review_pr")
+        _init_reviewer_session("prgs")
         self._lease_patch = _install_owned_reviewer_lease(8)
         self._lease_patch.start()
         self._auth_identity_patch = patch(
@@ -2416,7 +2430,7 @@ class TestSubmitPrReview(unittest.TestCase):
                 os.remove(spoof_path)
 
     def test_mark_final_decision_rejects_remote_mismatch(self):
-        init_review_decision_lock("prgs", "review_pr")
+        _init_reviewer_session("prgs")
         r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools", expected_head_sha="abc123")
         self.assertFalse(r["marked_ready"])
         self.assertTrue(any("does not match locked remote" in x for x in r["reasons"]))
@@ -2500,6 +2514,7 @@ if __name__ == "__main__":
 class TestTrackerHygieneCleanup(unittest.TestCase):
 
     def setUp(self):
+        mcp_server.gitea_load_review_workflow()
         self.mock_api = patch("mcp_server.api_request").start()
         self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
         patch("gitea_audit.audit_enabled", return_value=True).start()
@@ -3879,7 +3894,7 @@ class TestPreflightVerification(unittest.TestCase):
         os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n"
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity()
-        self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
+        self.assertIn("forbidden from modifying tracked workspace files", str(ctx.exception))
         self.assertIn("reviewer_edit.py", str(ctx.exception))
 
         # Foreign pre-existing dirty state does not block when unchanged.
@@ -3945,7 +3960,8 @@ class TestPreflightVerification(unittest.TestCase):
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity(worktree_path=worktree)
         msg = str(ctx.exception)
-        self.assertIn("active task workspace root", msg)
+        self.assertIn("resolved workspace", msg)
-        self.assertIn("inspected git root", msg)
+        self.assertIn(worktree, msg)
-        self.assertIn("dirty files: task_file.py", msg)
+        self.assertIn("worktree_path argument", msg)
-        self.assertIn("dirty scope:", msg)
+        self.assertIn("task_file.py", msg)
+        self.assertIn("author namespace", msg)

tests/test_merger_lease_adoption.py

@@ -0,0 +1,368 @@
+"""Merger lease adoption / recovery (#536)."""
+
+import os
+import sys
+import unittest
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import merger_lease_adoption as mla
+import reviewer_pr_lease as leases
+
+
+def _lease_comment(
+    pr_number: int,
+    session_id: str,
+    *,
+    phase: str = "claimed",
+    profile: str = "prgs-reviewer",
+    identity: str = "sysadmin",
+    candidate_head: str = "a" * 40,
+    comment_id: int = 100,
+) -> dict:
+    body = leases.format_lease_body(
+        repo="Scaled-Tech-Consulting/Gitea-Tools",
+        pr_number=pr_number,
+        issue_number=536,
+        reviewer_identity=identity,
+        profile=profile,
+        session_id=session_id,
+        worktree="branches/review-pr536",
+        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": identity}}
+
+
+HEAD = "f" * 40
+
+
+class TestSanctionedProvenance(unittest.TestCase):
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def test_manual_session_seed_rejected_by_mutation_gate(self):
+        comments = [_lease_comment(536, "reviewer-session", candidate_head=HEAD)]
+        leases.record_session_lease({
+            "pr_number": 536,
+            "session_id": "merger-session",
+            "candidate_head": HEAD,
+            "comment_id": 999,
+        })
+        result = leases.assess_mutation_lease_gate(
+            pr_number=536,
+            comments=comments,
+            reviewer_identity="sysadmin",
+            session_id="merger-session",
+            mutation="merge",
+            live_head_sha=HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("sanctioned provenance" in r for r in result["reasons"])
+        )
+
+    def test_sanctioned_acquire_provenance_allows_mutation(self):
+        comments = [_lease_comment(536, "my-session", candidate_head=HEAD)]
+        leases.record_session_lease(
+            {
+                "pr_number": 536,
+                "session_id": "my-session",
+                "candidate_head": HEAD,
+                "comment_id": 101,
+            },
+            lease_provenance=mla.build_lease_provenance(
+                source=mla.SOURCE_ACQUIRE,
+                comment_id=101,
+            ),
+        )
+        result = leases.assess_mutation_lease_gate(
+            pr_number=536,
+            comments=comments,
+            reviewer_identity="sysadmin",
+            session_id="my-session",
+            mutation="merge",
+            live_head_sha=HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertFalse(result["block"])
+
+
+class TestAdoptAssessment(unittest.TestCase):
+    def test_adopt_allowed_for_merger_with_reviewer_lease_and_approval(self):
+        comments = [_lease_comment(536, "reviewer-sess", candidate_head=HEAD)]
+        result = mla.assess_adopt_merger_lease(
+            comments,
+            pr_number=536,
+            adopter_identity="sysadmin",
+            adopter_profile="prgs-merger",
+            adopter_session_id="merger-sess",
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            issue_number=536,
+            worktree="branches/merge-pr536",
+            expected_head_sha=HEAD,
+            live_head_sha=HEAD,
+            approval_at_head=True,
+            pr_open=True,
+        )
+        self.assertTrue(result["adopt_allowed"])
+        self.assertIn(mla.ADOPTION_MARKER, result["adoption_body"])
+        self.assertIn("adopted_from_session_id: reviewer-sess", result["adoption_body"])
+
+    def test_adopt_blocked_without_approval(self):
+        comments = [_lease_comment(536, "reviewer-sess", candidate_head=HEAD)]
+        result = mla.assess_adopt_merger_lease(
+            comments,
+            pr_number=536,
+            adopter_identity="sysadmin",
+            adopter_profile="prgs-merger",
+            adopter_session_id="merger-sess",
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            issue_number=536,
+            worktree="branches/merge-pr536",
+            expected_head_sha=HEAD,
+            live_head_sha=HEAD,
+            approval_at_head=False,
+            pr_open=True,
+        )
+        self.assertFalse(result["adopt_allowed"])
+        self.assertTrue(any("APPROVED" in r for r in result["reasons"]))
+
+    def test_adopt_blocked_for_non_merger_profile(self):
+        comments = [_lease_comment(536, "reviewer-sess", candidate_head=HEAD)]
+        result = mla.assess_adopt_merger_lease(
+            comments,
+            pr_number=536,
+            adopter_identity="sysadmin",
+            adopter_profile="prgs-reviewer",
+            adopter_session_id="merger-sess",
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            issue_number=536,
+            worktree="branches/review-pr536",
+            expected_head_sha=HEAD,
+            live_head_sha=HEAD,
+            approval_at_head=True,
+            pr_open=True,
+        )
+        self.assertFalse(result["adopt_allowed"])
+        self.assertTrue(any("merger profile" in r for r in result["reasons"]))
+
+    def test_adopt_blocked_when_pr_not_open(self):
+        comments = [_lease_comment(536, "reviewer-sess", candidate_head=HEAD)]
+        result = mla.assess_adopt_merger_lease(
+            comments,
+            pr_number=536,
+            adopter_identity="sysadmin",
+            adopter_profile="prgs-merger",
+            adopter_session_id="merger-sess",
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            issue_number=536,
+            worktree="branches/merge-pr536",
+            expected_head_sha=HEAD,
+            live_head_sha=HEAD,
+            approval_at_head=True,
+            pr_open=False,
+        )
+        self.assertFalse(result["adopt_allowed"])
+
+
+class TestMergerHandoffMutationGate(unittest.TestCase):
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def test_cross_session_merge_blocked_without_adoption_comment(self):
+        reviewer_comment = _lease_comment(536, "reviewer-sess", candidate_head=HEAD)
+        provenance = mla.build_lease_provenance(
+            source=mla.SOURCE_ADOPT,
+            comment_id=7001,
+            adopted_from_session_id="reviewer-sess",
+        )
+        leases.record_session_lease(
+            {
+                "pr_number": 536,
+                "session_id": "merger-sess",
+                "candidate_head": HEAD,
+                "comment_id": 7001,
+                "phase": "adopted",
+            },
+            lease_provenance=provenance,
+        )
+        result = leases.assess_mutation_lease_gate(
+            pr_number=536,
+            comments=[reviewer_comment],
+            reviewer_identity="sysadmin",
+            session_id="merger-sess",
+            mutation="merge",
+            live_head_sha=HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(any("owned by session_id=reviewer-sess" in r for r in result["reasons"]))
+
+    def test_cross_session_merge_allowed_after_adoption_comment(self):
+        reviewer_comment = _lease_comment(536, "reviewer-sess", candidate_head=HEAD)
+        adoption_body = mla.format_adoption_body(
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            pr_number=536,
+            issue_number=536,
+            adopter_identity="sysadmin",
+            adopter_profile="prgs-merger",
+            adopter_session_id="merger-sess",
+            worktree="branches/merge-pr536",
+            candidate_head=HEAD,
+            target_branch="master",
+            target_branch_sha="b" * 40,
+            adopted_from_session_id="reviewer-sess",
+            adopted_from_profile="prgs-reviewer",
+            adopted_from_reviewer_identity="sysadmin",
+            adopted_from_comment_id=100,
+        )
+        adoption_comment = {
+            "id": 7001,
+            "body": adoption_body,
+            "user": {"login": "sysadmin"},
+        }
+        provenance = mla.build_lease_provenance(
+            source=mla.SOURCE_ADOPT,
+            comment_id=7001,
+            adopted_from_session_id="reviewer-sess",
+        )
+        leases.record_session_lease(
+            {
+                "pr_number": 536,
+                "session_id": "merger-sess",
+                "candidate_head": HEAD,
+                "comment_id": 7001,
+                "phase": "adopted",
+            },
+            lease_provenance=provenance,
+        )
+        result = leases.assess_mutation_lease_gate(
+            pr_number=536,
+            comments=[reviewer_comment, adoption_comment],
+            reviewer_identity="sysadmin",
+            session_id="merger-sess",
+            mutation="merge",
+            live_head_sha=HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertFalse(result["block"])
+
+
+class TestAdoptTool(unittest.TestCase):
+    def setUp(self):
+        leases.clear_session_lease()
+        patch("mcp_server.verify_preflight_purity", return_value=None).start()
+        patch("mcp_server._verify_role_mutation_workspace", return_value=None).start()
+        patch("gitea_audit.audit_enabled", return_value=False).start()
+        mcp_server = __import__("mcp_server")
+        mcp_server._IDENTITY_CACHE.clear()
+        mcp_server.init_review_decision_lock("prgs", "adopt_merger_pr_lease")
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", "reviewer", resolved_task="adopt_merger_pr_lease"
+        )
+
+    def tearDown(self):
+        patch.stopall()
+        leases.clear_session_lease()
+
+    @patch("mcp_server._authenticated_username", return_value="sysadmin")
+    @patch("mcp_server.get_auth_header", return_value="Basic dGVzdDp0ZXN0")
+    @patch("mcp_server.get_profile")
+    @patch("mcp_server.api_request")
+    def test_adopt_tool_posts_proof_and_records_sanctioned_session(
+        self, mock_api, mock_profile, _mock_auth, _mock_user
+    ):
+        mock_profile.return_value = {
+            "profile_name": "prgs-merger",
+            "allowed_operations": [
+                "gitea.read",
+                "gitea.pr.comment",
+                "gitea.pr.merge",
+            ],
+            "forbidden_operations": [],
+        }
+        reviewer_comment = _lease_comment(536, "reviewer-sess", candidate_head=HEAD)
+        posted_bodies: list[str] = []
+
+        def _side(method, url, auth=None, payload=None, *a, **k):
+            m = (method or "").upper()
+            if m == "POST":
+                if payload and payload.get("body"):
+                    posted_bodies.append(payload["body"])
+                return {"id": 7001}
+            if "/pulls/" in url and "/files" not in url:
+                return {
+                    "state": "open",
+                    "number": 536,
+                    "head": {"sha": HEAD},
+                    "merged": False,
+                }
+            if "/reviews" in url:
+                return []
+            if "/comments" in url:
+                return [reviewer_comment]
+            return {}
+
+        mock_api.side_effect = _side
+        from mcp_server import gitea_adopt_merger_pr_lease
+
+        with patch.object(
+            __import__("mcp_server"),
+            "gitea_get_pr_review_feedback",
+            return_value={
+                "approval_at_current_head": True,
+                "latest_approved_head_sha": HEAD,
+            },
+        ):
+            with patch.dict(
+                os.environ,
+                {
+                    "GITEA_PROFILE_NAME": "prgs-merger",
+                    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment,gitea.pr.merge",
+                },
+                clear=True,
+            ):
+                result = gitea_adopt_merger_pr_lease(
+                    pr_number=536,
+                    worktree="branches/merge-pr536",
+                    expected_head_sha=HEAD,
+                    issue_number=536,
+                    remote="prgs",
+                )
+        self.assertTrue(result["success"])
+        self.assertTrue(result["adopted"])
+        self.assertEqual(result["adopted_from_session_id"], "reviewer-sess")
+        self.assertEqual(result["adoption_comment_id"], 7001)
+        self.assertEqual(len(posted_bodies), 1)
+        self.assertIn(mla.ADOPTION_MARKER, posted_bodies[0])
+
+        session = leases.get_session_lease()
+        self.assertTrue(mla.is_sanctioned_session_lease(session))
+
+        adoption_comment = {
+            "id": 7001,
+            "body": posted_bodies[0],
+            "user": {"login": "sysadmin"},
+        }
+        gate = leases.assess_mutation_lease_gate(
+            pr_number=536,
+            comments=[reviewer_comment, adoption_comment],
+            reviewer_identity="sysadmin",
+            session_id=result["session_id"],
+            mutation="merge",
+            live_head_sha=HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertFalse(gate["block"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_namespace_workspace_binding.py

@@ -0,0 +1,240 @@
+"""Tests for namespace-scoped MCP workspace binding (#510)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+from unittest.mock import MagicMock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as srv  # noqa: E402
+import namespace_workspace_binding as nwb  # noqa: E402
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+if REPO_ROOT.parent.name == "branches":
+    CONTROL_ROOT = str(REPO_ROOT.parent.parent)
+else:
+    CONTROL_ROOT = str(REPO_ROOT)
+
+AUTHOR_DIRTY = f"{CONTROL_ROOT}/branches/mcp-author-worktree"
+MERGER_CLEAN = f"{CONTROL_ROOT}/branches/merge-pr487-submit"
+REVIEWER_CLEAN = f"{CONTROL_ROOT}/branches/review-pr487-submit"
+RECONCILER_CLEAN = f"{CONTROL_ROOT}/branches/reconcile-pr487"
+MCP_PROCESS_ROOT = CONTROL_ROOT
+
+
+class TestNamespaceWorkspaceModule(unittest.TestCase):
+    def test_author_env_ignored_for_merger_namespace(self):
+        workspace, source = nwb.resolve_namespace_workspace(
+            role_kind="merger",
+            worktree_path=None,
+            process_project_root=MCP_PROCESS_ROOT,
+            env={
+                nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
+                nwb.MERGER_WORKTREE_ENV: MERGER_CLEAN,
+            },
+            profile_name="gitea-merger",
+        )
+        self.assertEqual(workspace, os.path.realpath(MERGER_CLEAN))
+        self.assertEqual(source, f"{nwb.MERGER_WORKTREE_ENV} environment variable")
+
+    def test_author_env_ignored_for_reviewer_namespace(self):
+        workspace, source = nwb.resolve_namespace_workspace(
+            role_kind="reviewer",
+            worktree_path=None,
+            process_project_root=MCP_PROCESS_ROOT,
+            env={
+                nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
+                nwb.REVIEWER_WORKTREE_ENV: REVIEWER_CLEAN,
+            },
+        )
+        self.assertEqual(workspace, os.path.realpath(REVIEWER_CLEAN))
+        self.assertEqual(source, f"{nwb.REVIEWER_WORKTREE_ENV} environment variable")
+
+    def test_author_env_ignored_for_reconciler_namespace(self):
+        workspace, source = nwb.resolve_namespace_workspace(
+            role_kind="reconciler",
+            worktree_path=None,
+            process_project_root=MCP_PROCESS_ROOT,
+            env={
+                nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
+                nwb.RECONCILER_WORKTREE_ENV: RECONCILER_CLEAN,
+            },
+        )
+        self.assertEqual(workspace, os.path.realpath(RECONCILER_CLEAN))
+        self.assertEqual(source, f"{nwb.RECONCILER_WORKTREE_ENV} environment variable")
+
+    def test_merger_profile_maps_to_merger_namespace(self):
+        role = nwb.normalize_role_kind("reviewer", profile_name="gitea-merger")
+        self.assertEqual(role, "merger")
+
+    def test_error_message_includes_path_and_binding_source(self):
+        msg = nwb.format_namespace_workspace_binding_error(
+            role_kind="merger",
+            workspace_path=AUTHOR_DIRTY,
+            binding_source=f"{nwb.AUTHOR_WORKTREE_ENV} environment variable",
+            dirty_files=["gitea_mcp_server.py"],
+            ignored_bindings=[f"{nwb.AUTHOR_WORKTREE_ENV}={AUTHOR_DIRTY} (ignored for merger namespace)"],
+        )
+        self.assertIn(AUTHOR_DIRTY, msg)
+        self.assertIn("via", msg.lower())
+        self.assertIn(nwb.AUTHOR_WORKTREE_ENV, msg)
+        self.assertIn("Do not clean or reset foreign role worktrees", msg)
+
+
+class TestNamespaceWorkspaceIntegration(unittest.TestCase):
+    def setUp(self):
+        self._saved = {
+            "whoami_called": srv._preflight_whoami_called,
+            "capability_called": srv._preflight_capability_called,
+            "resolved_role": srv._preflight_resolved_role,
+            "whoami_violation": srv._preflight_whoami_violation,
+            "capability_violation": srv._preflight_capability_violation,
+            "in_test": srv._preflight_in_test_mode,
+        }
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_whoami_violation = False
+        srv._preflight_capability_violation = False
+        srv._preflight_in_test_mode = lambda: False
+        self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False)
+        self._env_patch.start()
+
+    def tearDown(self):
+        srv._preflight_whoami_called = self._saved["whoami_called"]
+        srv._preflight_capability_called = self._saved["capability_called"]
+        srv._preflight_resolved_role = self._saved["resolved_role"]
+        srv._preflight_whoami_violation = self._saved["whoami_violation"]
+        srv._preflight_capability_violation = self._saved["capability_violation"]
+        srv._preflight_in_test_mode = self._saved["in_test"]
+        self._env_patch.stop()
+        for key in (
+            nwb.AUTHOR_WORKTREE_ENV,
+            nwb.MERGER_WORKTREE_ENV,
+            nwb.REVIEWER_WORKTREE_ENV,
+            nwb.RECONCILER_WORKTREE_ENV,
+            nwb.ACTIVE_WORKTREE_ENV,
+        ):
+            os.environ.pop(key, None)
+
+    def _merger_profile(self):
+        return {
+            "profile_name": "gitea-merger",
+            "allowed_operations": ["gitea.pr.merge", "gitea.read"],
+            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
+        }
+
+    def _reviewer_profile(self):
+        return {
+            "profile_name": "prgs-reviewer",
+            "allowed_operations": ["gitea.pr.approve", "gitea.pr.review", "gitea.read"],
+            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
+        }
+
+    def _reconciler_profile(self):
+        return {
+            "profile_name": "prgs-reconciler",
+            "allowed_operations": ["gitea.pr.close", "gitea.read"],
+            "forbidden_operations": [
+                "gitea.pr.approve",
+                "gitea.pr.merge",
+                "gitea.pr.create",
+            ],
+        }
+
+    def _author_profile(self):
+        return {
+            "profile_name": "prgs-author",
+            "allowed_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.read"],
+            "forbidden_operations": ["gitea.pr.merge", "gitea.pr.approve"],
+        }
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_dirty_author_worktree_does_not_block_merger_with_clean_workspace(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
+        os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN
+        srv._preflight_resolved_role = "reviewer"
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
+                srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN)
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_dirty_author_worktree_does_not_block_reviewer_with_clean_workspace(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
+        os.environ[nwb.REVIEWER_WORKTREE_ENV] = REVIEWER_CLEAN
+        srv._preflight_resolved_role = "reviewer"
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._reviewer_profile()):
+                srv.verify_preflight_purity("prgs", worktree_path=REVIEWER_CLEAN)
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_dirty_author_worktree_does_not_block_reconciler_with_clean_workspace(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
+        os.environ[nwb.RECONCILER_WORKTREE_ENV] = RECONCILER_CLEAN
+        srv._preflight_resolved_role = "reconciler"
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._reconciler_profile()):
+                srv.verify_preflight_purity("prgs", worktree_path=RECONCILER_CLEAN)
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_dirty_active_task_workspace_still_blocks_mutations(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN
+        srv._preflight_resolved_role = "reviewer"
+        dirty = " M namespace_workspace_binding.py\n"
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
+                with mock.patch("gitea_mcp_server._get_workspace_porcelain", return_value=dirty):
+                    with self.assertRaises(RuntimeError) as ctx:
+                        srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN)
+        self.assertIn(MERGER_CLEAN, str(ctx.exception))
+        self.assertIn("binding", str(ctx.exception).lower())
+
+    def test_root_workspace_mutation_still_blocked_for_author(self):
+        srv._preflight_resolved_role = "author"
+        with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._author_profile()):
+                with mock.patch(
+                    "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+                    return_value={"current_branch": "master"},
+                ):
+                    with self.assertRaises(RuntimeError) as ctx:
+                        srv.verify_preflight_purity("prgs")
+        self.assertIn("stable control checkout", str(ctx.exception))
+
+    @mock.patch("subprocess.run")
+    @mock.patch("os.path.isdir", return_value=True)
+    @mock.patch("os.path.exists", return_value=True)
+    def test_pr487_style_merge_binds_clean_merger_workspace(
+        self, _exists, _isdir, mock_run
+    ):
+        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
+        srv._preflight_resolved_role = "reviewer"
+        with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
+            with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
+                resolved = srv._verify_role_mutation_workspace("prgs")
+        self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT))

tests/test_permission_reports.py

@@ -95,10 +95,15 @@ class PermissionReportBase(unittest.TestCase):
         self._dir = tempfile.TemporaryDirectory()
         self.config_path = os.path.join(self._dir.name, "profiles.json")
         self._write_config(CONFIG)
+        import review_workflow_load
+        review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
 
     def tearDown(self):
+        import review_workflow_load
+        review_workflow_load.clear_review_workflow_load()
         self._remotes.stop()
         mcp_server._IDENTITY_CACHE.clear()
+        mcp_server.review_workflow_load.clear_review_workflow_load()
         gitea_config._active_profile_override = None
         self._dir.cleanup()
 
@@ -250,6 +255,8 @@ class TestEligibilityDenialReport(PermissionReportBase):
                 return {"login": "author-user"}
             return PR_PAYLOAD
         mock_api.side_effect = fake_api
+        mcp_server.init_review_decision_lock("prgs", "review_pr")
+        mcp_server.gitea_load_review_workflow()
         with patch.dict(os.environ, self._env("author-profile")):
             res = mcp_server.gitea_merge_pr(
                 pr_number=42, confirmation="MERGE PR 42",

tests/test_post_merge_cleanup_proof.py

@@ -0,0 +1,169 @@
+"""Tests for post-merge cleanup proof enforcement (#402)."""
+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 post_merge_cleanup_proof import (  # noqa: E402
+    CLEANUP_SKIPPED,
+    assess_post_merge_cleanup_proof,
+)
+
+MERGE_SHA = "a" * 40
+HEAD_BRANCH = "feat/issue-274-branches-only-worktrees"
+WORKTREE = "branches/review-pr374-conflicts"
+
+
+def _full_remote_cleanup_report(**overrides):
+    fields = {
+        "Task": "review PR #374",
+        "Merge result": "merged",
+        "Merge commit SHA": MERGE_SHA,
+        "Cleanup status": "remote branch deleted",
+        "Delete-branch capability resolved": "gitea.branch.delete allowed via delete_branch task",
+        "Merged PR head branch": HEAD_BRANCH,
+        "Deleted branch": HEAD_BRANCH,
+        "Branch protection": "none",
+        "Open PR inventory proof": "no other open PR references branch (inventory complete)",
+        "Active heartbeat/claim/lease": "none",
+        "Cleanup mutations": "gitea_delete_branch on remote head branch",
+    }
+    fields.update(overrides)
+    lines = ["## Controller Handoff", ""]
+    lines.extend(f"- {key}: {value}" for key, value in fields.items())
+    return "\n".join(lines)
+
+
+def _full_worktree_cleanup_report(**overrides):
+    fields = {
+        "Task": "review PR #374",
+        "Merge result": "merged",
+        "Cleanup status": "local worktree removed",
+        "Removed worktree path": WORKTREE,
+        "Pre-removal tracked state": "clean",
+        "Pre-removal untracked state": "clean",
+        "Git worktree list after removal": "only main checkout listed",
+        "Cleanup mutations": "git worktree remove on session-owned review worktree",
+    }
+    fields.update(overrides)
+    lines = ["## Controller Handoff", ""]
+    lines.extend(f"- {key}: {value}" for key, value in fields.items())
+    return "\n".join(lines)
+
+
+class TestPostMergeCleanupProof(unittest.TestCase):
+    def test_no_cleanup_claim_passes(self):
+        report = "## Controller Handoff\n- Task: review PR #1\n- Merge result: merged"
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertFalse(result["block"])
+
+    def test_missing_capability_proof_blocked(self):
+        report = _full_remote_cleanup_report(
+            **{"Delete-branch capability resolved": "delete succeeded"}
+        )
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("capability" in r.lower() for r in result["reasons"]))
+
+    def test_unmerged_pr_blocked(self):
+        report = _full_remote_cleanup_report(**{"Merge result": "not merged"})
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("merge result" in r.lower() for r in result["reasons"]))
+
+    def test_wrong_branch_blocked(self):
+        report = _full_remote_cleanup_report(
+            **{"Deleted branch": "feat/other-branch"}
+        )
+        report += "\nDeleted branch does not match merged PR head branch"
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_protected_branch_blocked(self):
+        report = _full_remote_cleanup_report(**{"Branch protection": "enabled"})
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_open_pr_reference_blocked(self):
+        report = _full_remote_cleanup_report(
+            **{"Open PR inventory proof": "PR #999 still references branch"}
+        )
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_active_claim_blocked(self):
+        report = _full_remote_cleanup_report(
+            **{"Active heartbeat/claim/lease": "status:in-progress on linked issue"}
+        )
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_dirty_worktree_blocked(self):
+        report = _full_worktree_cleanup_report(
+            **{"Pre-removal tracked state": "dirty"}
+        )
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("tracked" in r.lower() for r in result["reasons"]))
+
+    def test_foreign_worktree_blocked(self):
+        report = _full_worktree_cleanup_report(
+            **{"Removed worktree path": "/tmp/foreign-worktree"}
+        )
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(any("branches/" in r.lower() for r in result["reasons"]))
+
+    def test_cleanup_skipped_with_blocker_passes(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup outcome: CLEANUP_SKIPPED",
+            "- Cleanup blocker: gitea.branch.delete capability not resolved in active profile",
+        ])
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertFalse(result["block"])
+        self.assertEqual(result["outcome"], CLEANUP_SKIPPED)
+
+    def test_cleanup_skipped_without_blocker_blocked(self):
+        report = "## Controller Handoff\n- Cleanup outcome: CLEANUP_SKIPPED"
+        result = assess_post_merge_cleanup_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_full_remote_cleanup_passes(self):
+        result = assess_post_merge_cleanup_proof(_full_remote_cleanup_report())
+        self.assertFalse(result["block"])
+        self.assertTrue(result["remote_delete_claimed"])
+
+    def test_full_worktree_cleanup_passes(self):
+        result = assess_post_merge_cleanup_proof(_full_worktree_cleanup_report())
+        self.assertFalse(result["block"])
+        self.assertTrue(result["worktree_remove_claimed"])
+
+    def test_validator_integration_blocks_unproven_delete(self):
+        report = (
+            "## Controller Handoff\n"
+            "- Cleanup mutations: gitea_delete_branch deleted remote branch\n"
+        )
+        result = assess_final_report_validator(report, task_kind="review_pr")
+        self.assertTrue(result["blocked"])
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.post_merge_cleanup_proof", rule_ids)
+
+    def test_validator_integration_allows_skipped(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Cleanup outcome: CLEANUP_SKIPPED",
+            "- Cleanup blocker: branch still referenced by open PR #414",
+        ])
+        result = assess_final_report_validator(report, task_kind="review_pr")
+        cleanup_findings = [
+            f for f in result["findings"]
+            if f["rule_id"] == "reviewer.post_merge_cleanup_proof"
+        ]
+        self.assertEqual(cleanup_findings, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_post_merge_moot_lease.py

@@ -0,0 +1,257 @@
+"""Post-merge moot reviewer-lease handling (#515).
+
+Covers:
+  a. Reviewer-lease acquisition/adoption is refused on an already-merged/closed
+     PR (fail closed, no mutation).
+  b. The post-merge moot cleanup path is safe and idempotent.
+  c. An active foreign lease on an *open* PR is never force-cleaned.
+"""
+
+import os
+import sys
+import unittest
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import reviewer_pr_lease as leases  # noqa: E402
+from mcp_server import (  # noqa: E402
+    gitea_acquire_reviewer_pr_lease,
+    gitea_cleanup_post_merge_moot_lease,
+)
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+MERGER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-merger",
+    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment",
+}
+PR = 487
+ISSUE = 485
+SESSION = "97274-676d20a825c4"
+
+
+def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed",
+                   candidate_head="a" * 40):
+    body = leases.format_lease_body(
+        repo="Scaled-Tech-Consulting/Gitea-Tools",
+        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": 6603, "body": body, "user": {"login": "sysadmin"}}
+
+
+# --------------------------------------------------------------------------- #
+# Pure logic
+# --------------------------------------------------------------------------- #
+class TestAcquireRefusedOnMergedPR(unittest.TestCase):
+    def test_acquire_refused_when_pr_merged_or_closed(self):
+        result = leases.assess_acquire_lease(
+            [_lease_comment()],
+            pr_number=PR,
+            reviewer_identity="sysadmin",
+            profile="prgs-merger",
+            session_id="new-session",
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            issue_number=ISSUE,
+            worktree="branches/merge-pr487",
+            candidate_head="a" * 40,
+            target_branch="master",
+            target_branch_sha="b" * 40,
+            pr_merged_or_closed=True,
+        )
+        self.assertFalse(result["acquire_allowed"])
+        self.assertTrue(result["post_merge_moot"])
+        self.assertIsNone(result["lease_body"])
+        self.assertTrue(any("post_merge_moot" in r for r in result["reasons"]))
+
+    def test_acquire_still_allowed_on_open_pr_without_flag(self):
+        result = leases.assess_acquire_lease(
+            [],
+            pr_number=PR,
+            reviewer_identity="sysadmin",
+            profile="prgs-reviewer",
+            session_id="s1",
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            issue_number=ISSUE,
+            worktree="branches/review-pr487",
+            candidate_head="a" * 40,
+            target_branch="master",
+            target_branch_sha="b" * 40,
+        )
+        self.assertTrue(result["acquire_allowed"])
+        self.assertFalse(result["post_merge_moot"])
+
+
+class TestPostMergeMootAssessment(unittest.TestCase):
+    def test_merged_pr_with_active_lease_is_moot_and_cleanable(self):
+        a = leases.assess_post_merge_moot_lease(
+            [_lease_comment()],
+            pr_number=PR,
+            pr_merged=True,
+            pr_state="closed",
+            merge_commit_sha="c" * 40,
+        )
+        self.assertTrue(a["pr_merged_or_closed"])
+        self.assertTrue(a["is_moot"])
+        self.assertTrue(a["cleanup_allowed"])
+        self.assertIsNotNone(a["release_body"])
+        self.assertIn("phase: released", a["release_body"])
+        self.assertIn("blocker: post-merge-moot", a["release_body"])
+
+    def test_open_pr_active_lease_never_cleaned(self):
+        a = leases.assess_post_merge_moot_lease(
+            [_lease_comment()],
+            pr_number=PR,
+            pr_merged=False,
+            pr_state="open",
+        )
+        self.assertFalse(a["pr_merged_or_closed"])
+        self.assertFalse(a["is_moot"])
+        self.assertFalse(a["cleanup_allowed"])
+        self.assertIsNone(a["release_body"])
+        self.assertTrue(any("still open" in r for r in a["reasons"]))
+
+    def test_merged_pr_without_lease_nothing_to_clean(self):
+        a = leases.assess_post_merge_moot_lease(
+            [], pr_number=PR, pr_merged=True, pr_state="closed")
+        self.assertTrue(a["pr_merged_or_closed"])
+        self.assertFalse(a["is_moot"])
+        self.assertFalse(a["cleanup_allowed"])
+        self.assertTrue(any("nothing to clean" in r for r in a["reasons"]))
+
+    def test_cleanup_is_idempotent(self):
+        """After the released marker is posted, a re-assess finds nothing to clean."""
+        first = leases.assess_post_merge_moot_lease(
+            [_lease_comment()], pr_number=PR, pr_merged=True, pr_state="closed")
+        self.assertTrue(first["cleanup_allowed"])
+        released_comment = {
+            "id": 7000, "body": first["release_body"], "user": {"login": "sysadmin"}}
+        second = leases.assess_post_merge_moot_lease(
+            [_lease_comment(), released_comment],
+            pr_number=PR, pr_merged=True, pr_state="closed")
+        self.assertFalse(second["is_moot"])
+        self.assertFalse(second["cleanup_allowed"])
+
+
+# --------------------------------------------------------------------------- #
+# Server tools
+# --------------------------------------------------------------------------- #
+def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999):
+    """Build an api_request side effect keyed on method + url."""
+    calls = {"post": []}
+
+    def _side(method, url, auth=None, payload=None, *a, **k):
+        m = (method or "").upper()
+        if m == "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
+
+
+class TestAcquireToolRefusesMergedPR(unittest.TestCase):
+    def setUp(self):
+        leases.clear_session_lease()
+
+    @patch("mcp_server.verify_preflight_purity", return_value=None)
+    @patch("mcp_server._authenticated_username", return_value="sysadmin")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    @patch("mcp_server.api_request")
+    def test_acquire_tool_fails_closed_on_merged_pr_without_posting(
+            self, mock_api, _auth, _user, _purity):
+        side, calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=[])
+        mock_api.side_effect = side
+        with patch.dict(os.environ, MERGER_ENV, clear=True):
+            result = gitea_acquire_reviewer_pr_lease(
+                pr_number=PR,
+                worktree="branches/merge-pr487",
+                candidate_head="a" * 40,
+                issue_number=ISSUE,
+                remote="prgs",
+            )
+        self.assertFalse(result["success"])
+        self.assertFalse(result["acquired"])
+        self.assertTrue(result.get("post_merge_moot"))
+        self.assertEqual(calls["post"], [], "must not post a lease comment")
+
+
+class TestCleanupTool(unittest.TestCase):
+    def setUp(self):
+        leases.clear_session_lease()
+
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    @patch("mcp_server.api_request")
+    def test_read_only_reports_moot_without_mutating(self, mock_api, _auth):
+        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):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
+        self.assertTrue(result["success"])
+        self.assertTrue(result["pr_merged_or_closed"])
+        self.assertTrue(result["lease_moot"])
+        self.assertFalse(result["cleanup_performed"])
+        self.assertTrue(result["no_merge_or_adoption"])
+        self.assertEqual(result["mode"], "read_only")
+        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")
+    def test_apply_posts_released_marker_on_merged_pr(
+            self, mock_api, _auth, _purity):
+        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):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs")
+        self.assertTrue(result["success"])
+        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.get_auth_header", return_value=FAKE_AUTH)
+    @patch("mcp_server.api_request")
+    def test_apply_refuses_to_clean_open_pr(self, mock_api, _auth, _purity):
+        side, calls = _api_side_effect(
+            pr_state="open", pr_merged=False, comments=[_lease_comment()])
+        mock_api.side_effect = side
+        with patch.dict(os.environ, MERGER_ENV, clear=True):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs")
+        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", [])))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pr_queue_inventory.py

@@ -11,6 +11,7 @@ from mcp_server import (
     gitea_view_pr,
     gitea_review_pr,
     gitea_check_pr_eligibility,
+    gitea_load_review_workflow,
 )
 import gitea_config
 

tests/test_preflight_read_survival.py

@@ -1,4 +1,4 @@
-"""#469/#470: capability preflight lifetime across safe read-only calls."""
+"""#469: capability preflight survives interleaved read-only whoami calls."""
 
 import os
 import unittest
@@ -101,59 +101,6 @@ class TestPreflightReadSurvival(unittest.TestCase):
         )
         mcp_server.verify_preflight_purity(task="review_pr")
 
-    def test_close_pr_sequence_with_interleaved_reads(self):
-        """resolve(close_pr) → whoami/view reads → close_pr gate (#470)."""
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        # Simulate live-state revalidation between resolve and mutation.
-        mcp_server.record_preflight_check("whoami")
-        self.assertTrue(mcp_server._preflight_capability_called)
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-        mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertFalse(mcp_server._preflight_capability_called)
-
-    def test_fresh_capability_resolve_replaces_prior_task(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
-        mcp_server.verify_preflight_purity(task="close_pr")
-
-    def test_missing_capability_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        msg = str(ctx.exception)
-        self.assertIn("gitea_resolve_task_capability", msg)
-        self.assertIn('task="close_pr"', msg)
-
-    def test_task_mismatch_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="author", resolved_task="create_issue"
-        )
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        msg = str(ctx.exception)
-        self.assertIn("task mismatch", msg)
-        self.assertIn('task="close_pr"', msg)
-
-    def test_consumed_capability_error_names_re_resolve(self):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check(
-            "capability", resolved_role="reconciler", resolved_task="close_pr"
-        )
-        mcp_server.verify_preflight_purity(task="close_pr")
-        with self.assertRaises(RuntimeError) as ctx:
-            mcp_server.verify_preflight_purity(task="close_pr")
-        self.assertIn('task="close_pr"', str(ctx.exception))
-
 
 if __name__ == "__main__":
     unittest.main()

tests/test_review_workflow_boundary.py

@@ -0,0 +1,138 @@
+"""Tests for workflow-load session boundary tracking (#403)."""
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import final_report_validator
+import review_workflow_boundary
+import review_workflow_load
+import mcp_server
+
+
+class TestPreReviewClassification(unittest.TestCase):
+    def setUp(self):
+        review_workflow_boundary.clear_pre_review_commands()
+        review_workflow_load.clear_review_workflow_load()
+
+    def test_inventory_command_allowed(self):
+        result = review_workflow_boundary.classify_pre_review_command(
+            "gitea_list_prs remote=prgs",
+            cwd="/tmp",
+            project_root="/repo/Gitea-Tools",
+        )
+        self.assertEqual(
+            result["classification"],
+            review_workflow_boundary.CLASSIFICATION_READ_ONLY_INVENTORY,
+        )
+
+    def test_main_checkout_pytest_is_boundary_violation(self):
+        root = "/repo/Gitea-Tools"
+        result = review_workflow_boundary.classify_pre_review_command(
+            "python -m pytest tests/",
+            cwd=root,
+            project_root=root,
+        )
+        self.assertEqual(
+            result["classification"],
+            review_workflow_boundary.CLASSIFICATION_BOUNDARY_VIOLATION,
+        )
+
+    def test_profiles_json_inspection_is_boundary_violation(self):
+        result = review_workflow_boundary.classify_pre_review_command(
+            "cat profiles.json",
+            cwd="/repo/Gitea-Tools",
+            project_root="/repo/Gitea-Tools",
+        )
+        self.assertEqual(
+            result["classification"],
+            review_workflow_boundary.CLASSIFICATION_BOUNDARY_VIOLATION,
+        )
+
+
+class TestWorkflowLoadBoundaryGate(unittest.TestCase):
+    def setUp(self):
+        review_workflow_boundary.clear_pre_review_commands()
+        review_workflow_load.clear_review_workflow_load()
+        mcp_server.init_review_decision_lock("prgs", "review_pr")
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", "reviewer")
+
+    def _root(self) -> str:
+        return str(__import__("pathlib").Path(__file__).resolve().parent.parent)
+
+    def test_boundary_violation_blocks_mutation_after_load(self):
+        root = self._root()
+        review_workflow_boundary.record_pre_review_command(
+            "python -m pytest tests/",
+            cwd=root,
+            project_root=root,
+        )
+        res = mcp_server.gitea_load_review_workflow()
+        self.assertFalse(res["success"])
+        self.assertEqual(res["boundary_status"], "violation")
+        blocked = mcp_server.gitea_mark_final_review_decision(
+            42, "approve", remote="prgs")
+        self.assertFalse(blocked["marked_ready"])
+        joined = " ".join(blocked["reasons"]).lower()
+        self.assertTrue(
+            "validation" in joined or "boundary" in joined or "workflow" in joined
+        )
+
+    def test_clean_inventory_then_load_passes(self):
+        root = self._root()
+        review_workflow_boundary.record_pre_review_command(
+            "gitea_list_prs remote=prgs",
+            cwd=root,
+            project_root=root,
+        )
+        res = mcp_server.gitea_load_review_workflow()
+        self.assertTrue(res["success"])
+        self.assertEqual(res["boundary_status"], "clean")
+        blockers = review_workflow_load.review_workflow_load_blockers(root)
+        self.assertEqual(blockers, [])
+
+    def test_file_view_narrative_fails_validator_without_helper(self):
+        report = (
+            "## Controller Handoff\n"
+            "- Task: review-merge-pr\n"
+            "- I read the canonical workflow review-merge-pr.md before review.\n"
+        )
+        findings = final_report_validator.assess_final_report_validator(
+            report,
+            task_kind="review_pr",
+        )
+        self.assertTrue(any(
+            f["rule_id"] == "reviewer.workflow_load_boundary"
+            for f in findings.get("findings") or []
+        ))
+
+    def test_helper_result_passes_validator(self):
+        root = self._root()
+        review_workflow_load.record_review_workflow_load(root)
+        helper = review_workflow_boundary.workflow_load_helper_result(
+            review_workflow_load._REVIEW_WORKFLOW_LOAD,
+            root,
+        )
+        report = (
+            "## Controller Handoff\n"
+            "- Task: review-merge-pr\n"
+            f"- Workflow-load helper result: workflow_hash: {helper['workflow_hash']}; "
+            f"boundary_status: {helper['boundary_status']}\n"
+        )
+        findings = final_report_validator.assess_final_report_validator(
+            report,
+            task_kind="review_pr",
+        )
+        boundary_findings = [
+            f for f in (findings.get("findings") or [])
+            if f.get("rule_id") == "reviewer.workflow_load_boundary"
+        ]
+        self.assertEqual(boundary_findings, [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_review_workflow_load.py

@@ -0,0 +1,153 @@
+"""Tests for canonical review workflow load proof (#389)."""
+
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import review_workflow_load
+import mcp_server
+
+
+class TestReviewWorkflowLoadModule(unittest.TestCase):
+    def setUp(self):
+        review_workflow_load.clear_review_workflow_load()
+        mcp_server._save_review_decision_lock(None)
+
+    def test_load_records_hash_and_schema(self):
+        root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
+        recorded = review_workflow_load.record_review_workflow_load(root)
+        self.assertEqual(
+            recorded["workflow_source"],
+            review_workflow_load.WORKFLOW_REL_PATH,
+        )
+        self.assertEqual(recorded["task_mode"], "review-merge-pr")
+        self.assertRegex(recorded["workflow_hash"], r"^[0-9a-f]{12}$")
+        self.assertEqual(
+            recorded["final_report_schema_path"],
+            review_workflow_load.SCHEMA_REL_PATH,
+        )
+        status = review_workflow_load.workflow_load_status(root)
+        self.assertTrue(status["workflow_load_proof_present"])
+        self.assertTrue(status["workflow_load_valid"])
+
+    def test_stale_session_pid_blocks(self):
+        root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
+        review_workflow_load.record_review_workflow_load(root)
+        review_workflow_load._REVIEW_WORKFLOW_LOAD["session_pid"] = 0
+        blockers = review_workflow_load.review_workflow_load_blockers(root)
+        self.assertTrue(any("different process" in b for b in blockers))
+
+    def test_prompt_conflict_detected(self):
+        conflict, reasons = review_workflow_load.assess_prompt_conflict(
+            "Run work-issue author implementation only")
+        self.assertTrue(conflict)
+        self.assertTrue(reasons)
+
+
+class TestReviewWorkflowLoadGates(unittest.TestCase):
+    def setUp(self):
+        review_workflow_load.clear_review_workflow_load()
+        mcp_server.init_review_decision_lock("prgs", "review_pr")
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", "reviewer")
+
+    def _load_workflow(self):
+        return mcp_server.gitea_load_review_workflow()
+
+    def test_mcp_helper_returns_required_fields(self):
+        res = self._load_workflow()
+        self.assertTrue(res["success"])
+        self.assertTrue(res["loaded"])
+        self.assertIn("workflow_source", res)
+        self.assertIn("workflow_hash", res)
+        self.assertIn("final_report_schema_path", res)
+        self.assertIn("final_report_schema_hash", res)
+
+    def test_mark_final_blocked_without_load(self):
+        res = mcp_server.gitea_mark_final_review_decision(
+            42, "approve", remote="prgs")
+        self.assertFalse(res["marked_ready"])
+        self.assertTrue(any(
+            "gitea_load_review_workflow" in r for r in res["reasons"]))
+        self.assertTrue(any(
+            "approve/merge replay" in r.lower() or "Do not call" in r
+            for r in res["reasons"]))
+
+    def test_submit_review_blocked_without_load(self):
+        with patch("mcp_server.gitea_check_pr_eligibility") as elig:
+            elig.return_value = {
+                "eligible": True,
+                "authenticated_user": "rev",
+                "profile_name": "prgs-reviewer",
+                "pr_author": "author",
+                "head_sha": "abc123",
+                "reasons": [],
+            }
+            res = mcp_server.gitea_submit_pr_review(
+                42,
+                "approve",
+                remote="prgs",
+                final_review_decision_ready=True,
+            )
+        self.assertFalse(res["performed"])
+        self.assertTrue(any(
+            "gitea_load_review_workflow" in r for r in res["reasons"]))
+
+    def test_merge_blocked_without_load(self):
+        res = mcp_server.gitea_merge_pr(
+            42,
+            confirmation="MERGE PR 42",
+            remote="prgs",
+        )
+        self.assertFalse(res["performed"])
+        self.assertTrue(any(
+            "gitea_load_review_workflow" in r for r in res["reasons"]))
+
+    def test_resolve_capability_reports_missing_load(self):
+        with patch.object(mcp_server, "_ensure_matching_profile"):
+            with patch.object(
+                    mcp_server.gitea_config, "is_runtime_switching_enabled",
+                    return_value=False):
+                with patch.object(
+                        mcp_server, "_authenticated_username",
+                        return_value="rev"):
+                    res = mcp_server.gitea_resolve_task_capability(
+                        "review_pr", remote="prgs")
+        proof = res.get("workflow_load_proof") or {}
+        self.assertFalse(proof.get("workflow_load_valid"))
+        self.assertTrue(any(
+            "gitea_load_review_workflow" in g
+            for g in res.get("task_role_guidance") or []))
+
+    def test_init_review_lock_clears_prior_load(self):
+        self._load_workflow()
+        mcp_server.init_review_decision_lock("prgs", "review_pr")
+        blockers = review_workflow_load.review_workflow_load_blockers(
+            str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+        self.assertTrue(blockers)
+
+    def test_dry_run_allowed_without_load(self):
+        with patch("mcp_server.get_auth_header", return_value="Basic dGVzdA=="), \
+                patch("mcp_server._list_pr_lease_comments", return_value=[]), \
+                patch("mcp_server.gitea_check_pr_eligibility") as elig:
+            elig.return_value = {
+                "eligible": True,
+                "authenticated_user": "rev",
+                "profile_name": "prgs-reviewer",
+                "pr_author": "author",
+                "head_sha": "abc123",
+                "reasons": [],
+            }
+            res = mcp_server.gitea_dry_run_pr_review(
+                42, "approve", remote="prgs")
+        self.assertNotIn(
+            "gitea_load_review_workflow",
+            " ".join(res.get("reasons") or []),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_pr_lease.py

@@ -7,6 +7,7 @@ from unittest.mock import patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
+import merger_lease_adoption as mla
 import reviewer_pr_lease as leases
 
 
@@ -123,7 +124,11 @@ class TestReviewerLeaseMutationGate(unittest.TestCase):
             "session_id": "my-session",
             "candidate_head": head,
             "target_branch": "master",
-        })
+            "comment_id": 101,
+        }, lease_provenance=mla.build_lease_provenance(
+            source=mla.SOURCE_ACQUIRE,
+            comment_id=101,
+        ))
         result = leases.assess_mutation_lease_gate(
             pr_number=382,
             comments=comments,
@@ -143,7 +148,11 @@ class TestReviewerLeaseMutationGate(unittest.TestCase):
             "pr_number": 382,
             "session_id": "my-session",
             "candidate_head": reviewed,
-        })
+            "comment_id": 102,
+        }, lease_provenance=mla.build_lease_provenance(
+            source=mla.SOURCE_ACQUIRE,
+            comment_id=102,
+        ))
         result = leases.assess_mutation_lease_gate(
             pr_number=382,
             comments=comments,

tests/test_root_checkout_guard.py

@@ -0,0 +1,161 @@
+"""Tests for root checkout guard (#475)."""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as srv  # noqa: E402
+import root_checkout_guard as rcg  # noqa: E402
+
+CONTROL_ROOT = str(Path(__file__).resolve().parents[3])
+BRANCHES_WORKTREE = str(Path(__file__).resolve().parents[1])
+MASTER_SHA = "a" * 40
+OTHER_SHA = "b" * 40
+
+
+class TestAssessRootCheckoutGuard(unittest.TestCase):
+    def _assess(self, **kwargs):
+        defaults = {
+            "workspace_path": CONTROL_ROOT,
+            "canonical_repo_root": CONTROL_ROOT,
+            "current_branch": "master",
+            "head_sha": MASTER_SHA,
+            "porcelain_status": "",
+            "remote_master_sha": MASTER_SHA,
+            "resolved_role": "author",
+        }
+        defaults.update(kwargs)
+        return rcg.assess_root_checkout_guard(**defaults)
+
+    def test_clean_master_control_checkout_allowed(self):
+        result = self._assess()
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_branches_worktree_allowed_for_author(self):
+        result = self._assess(
+            workspace_path=BRANCHES_WORKTREE,
+            current_branch="feat/issue-475-root-checkout-guard",
+            head_sha=OTHER_SHA,
+            resolved_role="author",
+        )
+        self.assertTrue(result["proven"])
+
+    def test_branches_worktree_allowed_for_reviewer(self):
+        result = self._assess(
+            workspace_path=f"{CONTROL_ROOT}/branches/review-pr-1",
+            current_branch="review-pr-1",
+            resolved_role="reviewer",
+        )
+        self.assertTrue(result["proven"])
+
+    def test_reconciler_always_allowed(self):
+        result = self._assess(
+            current_branch="feat/some-branch",
+            head_sha=OTHER_SHA,
+            porcelain_status=" M gitea_mcp_server.py\n",
+            resolved_role="reconciler",
+        )
+        self.assertTrue(result["proven"])
+
+    def test_feature_branch_on_control_checkout_blocked(self):
+        result = self._assess(
+            current_branch="feat/issue-99-example",
+            head_sha=OTHER_SHA,
+        )
+        self.assertTrue(result["block"])
+        self.assertIn("not a stable base branch", result["reasons"][0])
+
+    def test_detached_head_blocked(self):
+        result = self._assess(current_branch=None)
+        self.assertTrue(result["block"])
+        self.assertIn("detached HEAD", result["reasons"][0])
+
+    def test_dirty_control_checkout_blocked(self):
+        result = self._assess(porcelain_status=" M gitea_mcp_server.py\n")
+        self.assertTrue(result["block"])
+        self.assertIn("tracked local edits", result["reasons"][0])
+
+    def test_head_behind_prgs_master_blocked(self):
+        result = self._assess(
+            head_sha=OTHER_SHA,
+            remote_master_sha=MASTER_SHA,
+        )
+        self.assertTrue(result["block"])
+        self.assertIn("does not match prgs/master", result["reasons"][0])
+
+    def test_merger_requires_clean_control_checkout(self):
+        result = self._assess(
+            workspace_path=BRANCHES_WORKTREE,
+            current_branch="feat/issue-475-root-checkout-guard",
+            head_sha=OTHER_SHA,
+            resolved_role="merger",
+        )
+        self.assertTrue(result["block"])
+
+
+class TestVerifyPreflightRootGuardIntegration(unittest.TestCase):
+    def setUp(self):
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_resolved_role = "reviewer"
+        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
+
+    @patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
+    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value=MASTER_SHA)
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state")
+    @patch("gitea_mcp_server._resolve_author_mutation_context")
+    def test_reviewer_from_contaminated_root_blocked(
+        self, mock_ctx, mock_git, _remote_sha, _porcelain,
+    ):
+        srv._preflight_capability_baseline_porcelain = ""
+        mock_ctx.return_value = {
+            "workspace_path": CONTROL_ROOT,
+            "canonical_repo_root": CONTROL_ROOT,
+            "process_project_root": CONTROL_ROOT,
+        }
+        mock_git.return_value = {
+            "current_branch": "feat/hijacked-root",
+            "head_sha": OTHER_SHA,
+            "porcelain_status": "",
+        }
+        with self.assertRaises(RuntimeError) as ctx:
+            srv.verify_preflight_purity("prgs", worktree_path=CONTROL_ROOT)
+        self.assertIn("Root checkout guard (#475)", str(ctx.exception))
+        self.assertIn(rcg.REMEDIATION, str(ctx.exception))
+
+    @patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
+    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value=MASTER_SHA)
+    @patch("gitea_mcp_server.issue_lock_worktree.read_worktree_git_state")
+    @patch("gitea_mcp_server._resolve_author_mutation_context")
+    def test_reviewer_from_branches_worktree_allowed(
+        self, mock_ctx, mock_git, _remote_sha, _porcelain,
+    ):
+        srv._preflight_capability_baseline_porcelain = ""
+        mock_ctx.return_value = {
+            "workspace_path": BRANCHES_WORKTREE,
+            "canonical_repo_root": CONTROL_ROOT,
+            "process_project_root": BRANCHES_WORKTREE,
+        }
+        mock_git.return_value = {
+            "current_branch": "feat/issue-475-root-checkout-guard",
+            "head_sha": OTHER_SHA,
+            "porcelain_status": "",
+        }
+        srv.verify_preflight_purity("prgs", worktree_path=BRANCHES_WORKTREE)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_terminal_review_hard_stop.py

@@ -36,7 +36,11 @@ def _lock(mutations=None, correction=False):
 
 
 def _seed(mutations=None, correction=False):
+    import review_workflow_load
+    review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
     mcp_server._save_review_decision_lock(_lock(mutations, correction))
+    mcp_server.gitea_load_review_workflow()
+
 
 
 APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
@@ -48,6 +52,7 @@ RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
 class TestTerminalHardStopReasons(unittest.TestCase):
     def tearDown(self):
         mcp_server._save_review_decision_lock(None)
+        mcp_server.review_workflow_load.clear_review_workflow_load()
 
     def test_no_lock_no_reasons(self):
         mcp_server._save_review_decision_lock(None)
@@ -92,7 +97,10 @@ class TestTerminalHardStopReasons(unittest.TestCase):
 
 class TestMergeHardStopWiring(unittest.TestCase):
     def tearDown(self):
+        import review_workflow_load
+        review_workflow_load.clear_review_workflow_load()
         mcp_server._save_review_decision_lock(None)
+        mcp_server.review_workflow_load.clear_review_workflow_load()
 
     def test_merge_blocked_after_request_changes(self):
         _seed([RC_A])
@@ -113,7 +121,10 @@ class TestMergeHardStopWiring(unittest.TestCase):
 
 class TestMarkFinalHardStopWiring(unittest.TestCase):
     def tearDown(self):
+        import review_workflow_load
+        review_workflow_load.clear_review_workflow_load()
         mcp_server._save_review_decision_lock(None)
+        mcp_server.review_workflow_load.clear_review_workflow_load()
 
     def test_mark_ready_blocked_after_terminal_mutation(self):
         _seed([RC_A])
@@ -145,7 +156,10 @@ def _mark(action, pr_number=6, **kwargs):
 
 class TestDuplicateRequestChangesSuppression(unittest.TestCase):
     def tearDown(self):
+        import review_workflow_load
+        review_workflow_load.clear_review_workflow_load()
         mcp_server._save_review_decision_lock(None)
+        mcp_server.review_workflow_load.clear_review_workflow_load()
 
     def test_duplicate_request_changes_blocked_at_same_head(self):
         _seed()

tests/test_workspace_guard_alignment.py

@@ -126,17 +126,37 @@ class TestRuntimeContextGuardAlignment(unittest.TestCase):
             with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
                 srv.verify_preflight_purity(worktree_path=BRANCHES_WORKTREE)
 
-    def test_stable_checkout_still_rejected(self):
+    @mock.patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
+    @mock.patch(
+        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={
+            "current_branch": "master",
+            "head_sha": "a" * 40,
+            "porcelain_status": "",
+        },
+    )
+    def test_stable_checkout_still_rejected(self, _git, _remote_sha):
         with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
             with mock.patch.dict("os.environ", {"GITEA_TEST_PORCELAIN": ""}, clear=False):
                 with self.assertRaises(RuntimeError) as ctx:
                     srv.verify_preflight_purity()
         self.assertIn("stable control checkout", str(ctx.exception))
 
+    @mock.patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
+    @mock.patch(
+        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={
+            "current_branch": "master",
+            "head_sha": "a" * 40,
+            "porcelain_status": "",
+        },
+    )
     @mock.patch("os.path.isdir", return_value=True)
     @mock.patch("os.path.exists", return_value=True)
     @mock.patch("subprocess.run")
-    def test_non_branches_worktree_rejected(self, mock_run, *_exists):
+    def test_non_branches_worktree_rejected(
+        self, mock_run, mock_exists, mock_isdir, _git, _remote_sha,
+    ):
         outside = "/tmp/outside-repo-checkout"
         mock_run.return_value = MagicMock(
             returncode=0,

tests/test_worktree_cleanup_audit.py

@@ -0,0 +1,532 @@
+"""Tests for session-owned worktree cleanup audit, TTL, and integrity (#401, #404)."""
+
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import worktree_cleanup_audit as wca  # noqa: E402
+
+
+NOW = datetime(2026, 7, 7, 12, 0, 0, tzinfo=timezone.utc)
+
+
+def _iso(dt):
+    return dt.isoformat().replace("+00:00", "Z")
+
+
+class TestWorkflowTypeInference(unittest.TestCase):
+    def test_review_pr_path(self):
+        self.assertEqual(
+            wca.infer_workflow_type("branches/review-pr376", "review-pr376"),
+            wca.WORKFLOW_REVIEW,
+        )
+
+    def test_baseline_path(self):
+        self.assertEqual(
+            wca.infer_workflow_type("branches/baseline-master-issue-401"),
+            wca.WORKFLOW_BASELINE,
+        )
+
+    def test_merge_simulation_path(self):
+        self.assertEqual(
+            wca.infer_workflow_type("branches/merge-sim-pr380"),
+            wca.WORKFLOW_MERGE_SIMULATION,
+        )
+
+    def test_issue_work_branch(self):
+        self.assertEqual(
+            wca.infer_workflow_type(
+                "branches/issue-401-worktree", "feat/issue-401-worktree"
+            ),
+            wca.WORKFLOW_ISSUE_WORK,
+        )
+
+    def test_conflict_fix_path(self):
+        self.assertEqual(
+            wca.infer_workflow_type("branches/conflict-fix-pr376"),
+            wca.WORKFLOW_CONFLICT_FIX,
+        )
+
+    def test_unknown_path(self):
+        self.assertEqual(
+            wca.infer_workflow_type("branches/scratchpad"), wca.WORKFLOW_UNKNOWN
+        )
+
+
+class TestMetadata(unittest.TestCase):
+    def test_metadata_fields_present(self):
+        meta = wca.build_worktree_metadata(
+            path="branches/issue-401-worktree",
+            branch="feat/issue-401-worktree",
+            head_sha="abc123",
+            creator="jcwalker3",
+            profile="prgs-author",
+            created_at=_iso(NOW),
+            last_used_at=_iso(NOW),
+        )
+        for field in (
+            "path",
+            "workflow_type",
+            "issue_number",
+            "pr_number",
+            "branch",
+            "head_sha",
+            "creator",
+            "profile",
+            "created_at",
+            "last_used_at",
+            "cleanup_eligibility",
+        ):
+            self.assertIn(field, meta)
+        self.assertEqual(meta["issue_number"], 401)
+        self.assertEqual(meta["workflow_type"], wca.WORKFLOW_ISSUE_WORK)
+
+    def test_review_metadata_auto_removable_flag(self):
+        meta = wca.build_worktree_metadata(path="branches/review-pr42")
+        self.assertTrue(meta["auto_remove_on_success"])
+
+
+class TestTTL(unittest.TestCase):
+    def test_expired(self):
+        old = _iso(NOW - timedelta(hours=48))
+        self.assertTrue(wca.is_ttl_expired(last_used_at=old, now=NOW, ttl_hours=24))
+
+    def test_not_expired(self):
+        recent = _iso(NOW - timedelta(hours=1))
+        self.assertFalse(
+            wca.is_ttl_expired(last_used_at=recent, now=NOW, ttl_hours=24)
+        )
+
+    def test_unknown_timestamp_fails_safe(self):
+        self.assertFalse(wca.is_ttl_expired(last_used_at=None, now=NOW))
+        self.assertFalse(
+            wca.is_ttl_expired(last_used_at="not-a-date", now=NOW)
+        )
+
+
+class TestClassification(unittest.TestCase):
+    def test_successful_review_cleanup(self):
+        # Scenario 1: clean review worktree -> removable.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_REVIEW, is_dirty=False
+        )
+        self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
+        self.assertTrue(wca.is_removable(cls))
+
+    def test_dirty_worktree_preserved(self):
+        # Scenario 3: dirty worktree is never removable.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_REVIEW, is_dirty=True
+        )
+        self.assertEqual(cls, wca.CLASS_DIRTY_LOCAL)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_active_pr_worktree_preserved(self):
+        # Scenario 4: open PR wins over an otherwise-removable review worktree.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_REVIEW,
+            is_dirty=False,
+            has_open_pr=True,
+        )
+        self.assertEqual(cls, wca.CLASS_ACTIVE_OPEN_PR)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_stale_clean_issue_worktree_removable(self):
+        # Scenario 5: clean issue worktree, TTL expired, no lock -> removable.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_ISSUE_WORK,
+            is_dirty=False,
+            ttl_expired=True,
+        )
+        self.assertEqual(cls, wca.CLASS_CLEAN_STALE_REMOVABLE)
+        self.assertTrue(wca.is_removable(cls))
+
+    def test_fresh_clean_issue_worktree_preserved(self):
+        # Clean issue worktree not yet TTL-expired stays active.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_ISSUE_WORK,
+            is_dirty=False,
+            ttl_expired=False,
+        )
+        self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_detached_review_worktree_classified(self):
+        # Scenario 6: detached review worktree -> detached_review_leftover.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_REVIEW,
+            is_dirty=False,
+            is_detached=True,
+        )
+        self.assertEqual(cls, wca.CLASS_DETACHED_REVIEW_LEFTOVER)
+        self.assertTrue(wca.is_removable(cls))
+
+    def test_ttl_expired_but_dirty_preserved(self):
+        # Scenario 7: dirty wins over TTL expiry.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_ISSUE_WORK,
+            is_dirty=True,
+            ttl_expired=True,
+        )
+        self.assertEqual(cls, wca.CLASS_DIRTY_LOCAL)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_lease_protected_worktree_preserved(self):
+        # Scenario 8: active lease is never removable.
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_REVIEW,
+            is_dirty=False,
+            has_active_lease=True,
+        )
+        self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_active_issue_lock_preserved(self):
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_ISSUE_WORK,
+            is_dirty=False,
+            has_active_issue_lock=True,
+            ttl_expired=True,
+        )
+        self.assertEqual(cls, wca.CLASS_ACTIVE_ISSUE_WORK)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_protected_base_worktree_never_removable(self):
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_ISSUE_WORK,
+            is_dirty=False,
+            is_protected=True,
+            ttl_expired=True,
+        )
+        self.assertEqual(cls, wca.CLASS_UNSAFE_UNKNOWN)
+        self.assertFalse(wca.is_removable(cls))
+
+    def test_unknown_workflow_type_unsafe(self):
+        cls = wca.classify_worktree(
+            workflow_type=wca.WORKFLOW_UNKNOWN, is_dirty=False, ttl_expired=True
+        )
+        self.assertEqual(cls, wca.CLASS_UNSAFE_UNKNOWN)
+        self.assertFalse(wca.is_removable(cls))
+
+
+class TestRemovalDecision(unittest.TestCase):
+    def test_safe_removal_proof(self):
+        decision = wca.assess_worktree_removal(
+            path="branches/review-pr42",
+            branch="review-pr42",
+            head_sha="abc123",
+            is_dirty=False,
+            has_open_pr=False,
+            has_active_lease=False,
+            classification=wca.CLASS_CLEAN_STALE_REMOVABLE,
+        )
+        self.assertTrue(decision["safe_to_remove"])
+        self.assertEqual(decision["block_reasons"], [])
+        self.assertTrue(decision["clean"])
+        self.assertTrue(decision["no_active_pr"])
+        self.assertTrue(decision["no_active_lease"])
+
+    def test_dirty_blocks_removal(self):
+        decision = wca.assess_worktree_removal(
+            path="branches/review-pr42",
+            branch="review-pr42",
+            head_sha="abc123",
+            is_dirty=True,
+            has_open_pr=False,
+            has_active_lease=False,
+            classification=wca.CLASS_DIRTY_LOCAL,
+        )
+        self.assertFalse(decision["safe_to_remove"])
+        self.assertIn("worktree has uncommitted changes", decision["block_reasons"])
+
+    def test_open_pr_and_lease_block_removal(self):
+        decision = wca.assess_worktree_removal(
+            path="branches/review-pr42",
+            branch="review-pr42",
+            head_sha="abc123",
+            is_dirty=False,
+            has_open_pr=True,
+            has_active_lease=True,
+            classification=wca.CLASS_ACTIVE_OPEN_PR,
+        )
+        self.assertFalse(decision["safe_to_remove"])
+        self.assertIn("worktree branch has an open PR", decision["block_reasons"])
+        self.assertIn("worktree has an active lease", decision["block_reasons"])
+
+
+class TestSuccessCleanupPlan(unittest.TestCase):
+    def test_review_worktree_removed_at_success(self):
+        # Scenario 1 end-to-end: clean review worktree removed at success.
+        meta = wca.build_worktree_metadata(path="branches/review-pr42")
+        plan = wca.plan_success_cleanup(
+            metadata=meta, is_dirty=False, has_open_pr=False, has_active_lease=False
+        )
+        self.assertTrue(plan["remove"])
+
+    def test_failed_review_leaves_worktree_reported(self):
+        # Scenario 2: dirty review worktree preserved and reported at failure.
+        meta = wca.build_worktree_metadata(path="branches/review-pr42")
+        plan = wca.plan_success_cleanup(
+            metadata=meta, is_dirty=True, has_open_pr=False, has_active_lease=False
+        )
+        self.assertFalse(plan["remove"])
+        self.assertIn("uncommitted", plan["reason"])
+        report = wca.cleanup_failure_report(meta["path"], plan["reason"])
+        self.assertFalse(report["removed"])
+        self.assertEqual(report["path"], "branches/review-pr42")
+
+    def test_issue_worktree_preserved_by_policy(self):
+        meta = wca.build_worktree_metadata(
+            path="branches/issue-401-worktree", branch="feat/issue-401-worktree"
+        )
+        plan = wca.plan_success_cleanup(
+            metadata=meta, is_dirty=False, has_open_pr=False, has_active_lease=False
+        )
+        self.assertFalse(plan["remove"])
+        self.assertIn("preserved by policy", plan["reason"])
+
+
+class TestPorcelainParser(unittest.TestCase):
+    def test_parse_branch_and_detached(self):
+        text = (
+            "worktree /repo\n"
+            "HEAD 1111111111111111111111111111111111111111\n"
+            "branch refs/heads/master\n"
+            "\n"
+            "worktree /repo/branches/review-pr42\n"
+            "HEAD 2222222222222222222222222222222222222222\n"
+            "detached\n"
+        )
+        entries = wca.parse_worktree_porcelain(text)
+        self.assertEqual(len(entries), 2)
+        self.assertEqual(entries[0]["branch"], "master")
+        self.assertFalse(entries[0]["detached"])
+        self.assertIsNone(entries[1]["branch"])
+        self.assertTrue(entries[1]["detached"])
+
+
+class TestAuditReportAccuracy(unittest.TestCase):
+    """Scenario 9: cleanup report accuracy over a mixed branches/ directory."""
+
+    PORCELAIN = (
+        "worktree /repo\n"
+        "HEAD 1111111111111111111111111111111111111111\n"
+        "branch refs/heads/master\n"
+        "\n"
+        "worktree /repo/branches/review-pr42\n"
+        "HEAD 2222222222222222222222222222222222222222\n"
+        "branch refs/heads/review-pr42\n"
+        "\n"
+        "worktree /repo/branches/issue-400-open-pr\n"
+        "HEAD 3333333333333333333333333333333333333333\n"
+        "branch refs/heads/feat/issue-400-open-pr\n"
+        "\n"
+        "worktree /repo/branches/issue-401-dirty\n"
+        "HEAD 4444444444444444444444444444444444444444\n"
+        "branch refs/heads/feat/issue-401-dirty\n"
+    )
+
+    def _fake_dirty(self, path):
+        if path.endswith("issue-401-dirty"):
+            return {"exists": True, "dirty": True, "dirty_files": [" M x.py"]}
+        return {"exists": True, "dirty": False, "dirty_files": []}
+
+    def test_mixed_directory_classified(self):
+        with patch.object(
+            wca,
+            "list_worktrees",
+            return_value=wca.parse_worktree_porcelain(self.PORCELAIN),
+        ), patch.object(
+            wca, "read_worktree_dirty", side_effect=self._fake_dirty
+        ), patch.object(
+            wca, "git_worktree_list", return_value="(mocked)"
+        ):
+            report = wca.audit_branches_directory(
+                "/repo",
+                open_pr_branches={"feat/issue-400-open-pr"},
+            )
+
+        by_path = {wt["path"]: wt for wt in report["worktrees"]}
+        # main checkout on master -> protected -> unsafe/unknown, not removable
+        self.assertEqual(
+            by_path["/repo"]["classification"], wca.CLASS_UNSAFE_UNKNOWN
+        )
+        # clean review worktree -> removable
+        self.assertEqual(
+            by_path["/repo/branches/review-pr42"]["classification"],
+            wca.CLASS_CLEAN_STALE_REMOVABLE,
+        )
+        # open PR branch -> preserved
+        self.assertEqual(
+            by_path["/repo/branches/issue-400-open-pr"]["classification"],
+            wca.CLASS_ACTIVE_OPEN_PR,
+        )
+        # dirty worktree -> preserved
+        self.assertEqual(
+            by_path["/repo/branches/issue-401-dirty"]["classification"],
+            wca.CLASS_DIRTY_LOCAL,
+        )
+        # exactly one removable candidate (the clean review worktree)
+        self.assertEqual(report["removable_count"], 1)
+        self.assertEqual(
+            report["removable_candidates"][0]["path"],
+            "/repo/branches/review-pr42",
+        )
+        self.assertEqual(report["total"], 4)
+        self.assertEqual(report["git_worktree_list"], "(mocked)")
+
+
+def _integrity_entry(
+    path: str,
+    *,
+    classification: str,
+    registered: bool = True,
+    preserve: bool | None = None,
+) -> dict:
+    preserve_flag = preserve if preserve is not None else classification in {
+        "active_open_pr",
+        "active_issue_work",
+        "dirty_local_worktree",
+        "unsafe_unknown",
+    }
+    return {
+        "path": path,
+        "classification": classification,
+        "preserve": preserve_flag,
+        "registered_worktree": registered,
+        "worktree_state": {"exists": True, "clean": classification == "clean_stale_removable"},
+        "worktree_record": {"branch": "feat/x"} if registered else None,
+    }
+
+
+def _integrity_snapshot(entries: list[dict]) -> dict:
+    return {"entries": entries}
+
+
+class TestParseWorktreePorcelain(unittest.TestCase):
+    def test_parses_multiple_worktrees(self):
+        text = "\n".join([
+            "worktree /proj/branches/foo",
+            "HEAD abcdef0123456789abcdef0123456789abcdef0",
+            "branch refs/heads/feat/foo",
+            "",
+            "worktree /proj",
+            "HEAD 1111111111111111111111111111111111111111",
+            "branch refs/heads/master",
+        ])
+        parsed = wca.parse_worktree_list_porcelain(text)
+        self.assertEqual(len(parsed), 2)
+        self.assertEqual(parsed[0]["branch"], "feat/foo")
+
+
+class TestClassifyEntry(unittest.TestCase):
+    def test_active_pr_classification(self):
+        result = wca.classify_branches_entry(
+            rel_path="branches/feat-issue-1-x",
+            worktree_record={"branch": "feat/issue-1-x"},
+            worktree_state={"exists": True, "clean": True, "dirty_files": []},
+            open_pr_branches={"feat/issue-1-x"},
+        )
+        self.assertEqual(result, "active_open_pr")
+
+    def test_dirty_classification(self):
+        result = wca.classify_branches_entry(
+            rel_path="branches/dirty-one",
+            worktree_record={"branch": "feat/dirty-one"},
+            worktree_state={"exists": True, "dirty_files": ["a.py"]},
+            open_pr_branches=set(),
+        )
+        self.assertEqual(result, "dirty_local_worktree")
+
+
+class TestCleanupIntegrity(unittest.TestCase):
+    def test_preserved_worktree_remains(self):
+        path = "branches/keep-me"
+        before = _integrity_snapshot([
+            _integrity_entry(path, classification="clean_stale_removable", preserve=False),
+        ])
+        after = _integrity_snapshot([
+            _integrity_entry(path, classification="clean_stale_removable", preserve=False),
+        ])
+        result = wca.assess_worktree_cleanup_integrity(before=before, after=after)
+        self.assertTrue(result["integrity_passed"])
+
+    def test_intentional_removal_passes(self):
+        path = "branches/remove-me"
+        before = _integrity_snapshot([
+            _integrity_entry(path, classification="clean_stale_removable", preserve=False),
+        ])
+        after = _integrity_snapshot([])
+        result = wca.assess_worktree_cleanup_integrity(
+            before=before,
+            after=after,
+            removals=[{
+                "path": path,
+                "method": "git worktree remove",
+                "pre_removal_proof": "clean status",
+            }],
+        )
+        self.assertTrue(result["integrity_passed"])
+
+    def test_dirty_worktree_disappears_fails(self):
+        path = "branches/dirty-wt"
+        before = _integrity_snapshot([_integrity_entry(path, classification="dirty_local_worktree")])
+        after = _integrity_snapshot([])
+        recon = wca.reconcile_cleanup_audit(before, after)
+        result = wca.assess_cleanup_audit_integrity(recon)
+        self.assertFalse(result["proven"])
+
+    def test_active_pr_worktree_disappears_fails(self):
+        path = "branches/review-pr99"
+        before = _integrity_snapshot([_integrity_entry(path, classification="active_open_pr")])
+        after = _integrity_snapshot([])
+        result = wca.assess_worktree_cleanup_integrity(before=before, after=after)
+        self.assertFalse(result["integrity_passed"])
+
+    def test_explained_missing_allowed(self):
+        path = "branches/review-pr382"
+        before = _integrity_snapshot([
+            _integrity_entry(path, classification="detached_review_leftover", preserve=False),
+        ])
+        after = _integrity_snapshot([])
+        result = wca.assess_worktree_cleanup_integrity(
+            before=before,
+            after=after,
+            explained_missing={path: "removed concurrently by sibling session"},
+        )
+        self.assertTrue(result["integrity_passed"])
+
+    def test_removal_log_omits_clean_stale_fails(self):
+        path = "branches/a"
+        before = _integrity_snapshot([
+            _integrity_entry(path, classification="clean_stale_removable", preserve=False),
+        ])
+        after = _integrity_snapshot([])
+        recon = wca.reconcile_cleanup_audit(before, after, removal_log=[])
+        self.assertFalse(recon["removal_log_complete"])
+
+
+class TestCleanupReportProof(unittest.TestCase):
+    def test_complete_report_passes(self):
+        report = "\n".join([
+            "Cleanup audit reconciliation table:",
+            "Initial count: 10",
+            "Removed count: 3",
+            "Preserved count: 7",
+            "Missing-unexplained count: 0",
+            "Final count: 7",
+            "Final verification: git worktree list proof attached",
+        ])
+        result = wca.assess_cleanup_audit_final_report(report)
+        self.assertTrue(result["proven"])
+
+    def test_incomplete_report_fails(self):
+        result = wca.assess_cleanup_audit_final_report("removed some worktrees")
+        self.assertFalse(result["proven"])
+
+
+if __name__ == "__main__":
+    unittest.main()