fix: resolve conflicts for PR #422

Merge prgs/master into feat/issue-403-workflow-load-boundaries.
Preserve workflow-load boundary rules (#403) alongside master's
audit reconciliation boundary and workspace verification gates.

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, []

final_report_validator.py

@@ -118,6 +118,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",
@@ -1024,6 +1040,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,
     *,
@@ -1084,6 +1164,7 @@ _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,
@@ -1100,6 +1181,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,

gitea_mcp_server.py

@@ -714,6 +714,8 @@ import role_session_router  # noqa: E402
 import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
+import review_workflow_boundary  # noqa: E402
+import review_workflow_load  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
@@ -729,6 +731,7 @@ import reviewer_pr_lease  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
+import audit_reconciliation_mode  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import pr_work_lease  # noqa: E402
 import native_mcp_preference  # noqa: E402
@@ -1446,7 +1449,6 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
-@mcp.tool()
 def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]:
     """Fetch all OPEN pull requests for a repo (used for stacked-base proof, #484)."""
     try:
@@ -1457,6 +1459,7 @@ def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]:
         )
 
 
+@mcp.tool()
 def gitea_lock_issue(
     issue_number: int,
     branch_name: str,
@@ -2384,6 +2387,7 @@ def init_review_decision_lock(remote: str | None, task: str | None):
     """Seed read-only-until-ready state for reviewer PR review tasks."""
     if task != "review_pr":
         return
+    review_workflow_load.clear_review_workflow_load()
     profile = get_profile()
     profile_name = (profile.get("profile_name") or "").strip()
     session_lock = (
@@ -2410,6 +2414,11 @@ def init_review_decision_lock(remote: str | None, task: str | None):
     })
 
 
+def _review_workflow_load_gate_reasons() -> list[str]:
+    """Fail closed when canonical review workflow was not loaded (#389)."""
+    return review_workflow_load.review_workflow_load_blockers(PROJECT_ROOT)
+
+
 def check_review_decision_gate(
     pr_number: int,
     action: str,
@@ -2420,7 +2429,10 @@ def check_review_decision_gate(
     repo: str | None = None,
 ) -> list[str]:
     """Fail closed unless validation completed and the final decision is ready."""
-    reasons = []
+    reasons = list(_review_workflow_load_gate_reasons())
+    if reasons:
+        reasons.extend(review_workflow_load.recovery_handoff_without_replay())
+        return reasons
     lock = _load_review_decision_lock()
     if lock is None:
         reasons.append(
@@ -2835,6 +2847,7 @@ def _evaluate_pr_review_submission(
         remote, worktree_path=worktree_path, task="review_pr"
     )
     action = (action or "").strip().lower()
+    workflow_blockers = _review_workflow_load_gate_reasons() if live else []
     result = {
         "requested_action": action,
         "performed": False,
@@ -2850,6 +2863,10 @@ def _evaluate_pr_review_submission(
         "reasons": [],
     }
     reasons = result["reasons"]
+    if workflow_blockers:
+        reasons.extend(workflow_blockers)
+        reasons.extend(review_workflow_load.recovery_handoff_without_replay())
+        return result
 
     if action not in _REVIEW_ACTIONS:
         reasons.append(
@@ -3068,6 +3085,13 @@ def gitea_mark_final_review_decision(
         }
     org = resolved_org
     repo = resolved_repo
+    workflow_blockers = _review_workflow_load_gate_reasons()
+    if workflow_blockers:
+        return {
+            "marked_ready": False,
+            "reasons": workflow_blockers + (
+                review_workflow_load.recovery_handoff_without_replay()),
+        }
     hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready")
     if hard_stop:
         return {"marked_ready": False, "reasons": hard_stop}
@@ -3703,6 +3727,7 @@ def gitea_merge_pr(
     _verify_role_mutation_workspace(
         remote, worktree_path=worktree_path, task="merge_pr"
     )
+    workflow_blockers = _review_workflow_load_gate_reasons()
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -3720,6 +3745,10 @@ def gitea_merge_pr(
         "reasons": [],
     }
     reasons = result["reasons"]
+    if workflow_blockers:
+        reasons.extend(workflow_blockers)
+        reasons.extend(review_workflow_load.recovery_handoff_without_replay())
+        return result
 
     # Gate 1 — valid merge method (no API call on a bad method).
     if do not in _MERGE_METHODS:
@@ -4231,6 +4260,18 @@ def gitea_delete_branch(
             "permission_report": _permission_block_report("gitea.branch.delete"),
         }
 
+    audit_allowed, audit_reasons = (
+        audit_reconciliation_mode.check_audit_mutation_allowed("delete_branch")
+    )
+    if not audit_allowed:
+        return {
+            "success": False,
+            "performed": False,
+            "required_permission": "gitea.branch.delete",
+            "reasons": audit_reasons,
+            "audit_phase": audit_reconciliation_mode.current_phase(),
+        }
+
     verify_preflight_purity(remote, task="delete_branch")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -4300,6 +4341,18 @@ def gitea_reconcile_merged_cleanups(
             "execute_confirmed must be True when dry_run=False (fail closed)"
         )
 
+    if not dry_run:
+        exec_allowed, exec_reasons = (
+            audit_reconciliation_mode.check_cleanup_execution_allowed()
+        )
+        if not exec_allowed:
+            return {
+                "success": False,
+                "performed": False,
+                "reasons": exec_reasons,
+                "audit_phase": audit_reconciliation_mode.current_phase(),
+            }
+
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -4382,6 +4435,53 @@ def gitea_reconcile_merged_cleanups(
     return {"success": True, "performed": True, **report}
 
 
+@mcp.tool()
+def gitea_authorize_reconciliation_cleanup_phase(
+    operator_approved: bool = False,
+    workflow_authorized: bool = False,
+    delete_capability_proven: bool = False,
+    safe_to_delete_remote: bool = False,
+    safe_to_remove_worktree: bool = False,
+    before_state: str = "",
+    after_state: str = "",
+) -> dict:
+    """Authorize cleanup phase after audit-only reconciliation (#419).
+
+    Requires operator or workflow approval, exact delete_branch capability proof,
+    branch/worktree safety proof, and before/after state snapshots. Audit phase
+    forbids branch deletion, worktree removal, pushes, and issue/PR mutations.
+    """
+    delete_gate = _profile_operation_gate("gitea.branch.delete")
+    capability_ok = not bool(delete_gate)
+    if delete_capability_proven and delete_gate:
+        return {
+            "authorized": False,
+            "performed": False,
+            "delete_capability_verified": False,
+            "reasons": [
+                "delete_capability_proven=true but active profile lacks "
+                "gitea.branch.delete",
+            ] + delete_gate,
+            "audit_phase": audit_reconciliation_mode.current_phase(),
+        }
+    result = audit_reconciliation_mode.authorize_cleanup_phase(
+        operator_approved=operator_approved,
+        workflow_authorized=workflow_authorized,
+        delete_capability_proven=delete_capability_proven and capability_ok,
+        safety_proof={
+            "safe_to_delete_remote": safe_to_delete_remote,
+            "safe_to_remove_worktree": safe_to_remove_worktree,
+        },
+        before_after_snapshot={
+            "before": (before_state or "").strip(),
+            "after": (after_state or "").strip(),
+        },
+    )
+    result["performed"] = bool(result.get("authorized"))
+    result["delete_capability_verified"] = capability_ok
+    return result
+
+
 @mcp.tool()
 def gitea_assess_already_landed_reconciliation(
     pr_number: int,
@@ -5195,6 +5295,13 @@ def gitea_acquire_reviewer_pr_lease(
 
     comments = _fetch_pr_comments(
         pr_number, remote=remote, host=host, org=org, repo=repo)
+    # Refuse merge-oriented lease acquisition/adoption on an already-merged or
+    # closed PR: the lease is moot and adopting it for merge work is unsafe (#515).
+    pr_live = api_request(
+        "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
+    pr_merged_or_closed = bool(
+        pr_live.get("merged") or pr_live.get("merged_at")
+    ) or (str(pr_live.get("state") or "").strip().lower() == "closed")
     assessment = reviewer_pr_lease.assess_acquire_lease(
         comments,
         pr_number=pr_number,
@@ -5207,6 +5314,7 @@ def gitea_acquire_reviewer_pr_lease(
         candidate_head=candidate_head,
         target_branch=target_branch,
         target_branch_sha=target_branch_sha,
+        pr_merged_or_closed=pr_merged_or_closed,
     )
     if not assessment.get("acquire_allowed"):
         return {
@@ -5214,6 +5322,7 @@ def gitea_acquire_reviewer_pr_lease(
             "acquired": False,
             "reasons": assessment.get("reasons") or [],
             "existing_lease": assessment.get("existing_lease"),
+            "post_merge_moot": assessment.get("post_merge_moot", False),
         }
 
     body = assessment["lease_body"]
@@ -5362,6 +5471,126 @@ def gitea_assess_reviewer_pr_lease(
     }
 
 
+@mcp.tool()
+def gitea_cleanup_post_merge_moot_lease(
+    pr_number: int,
+    apply: bool = False,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Safely resolve a moot reviewer lease left on an ALREADY-MERGED/closed PR (#515).
+
+    Read-first and fail-safe. This tool never merges and never adopts a lease.
+    It only acts when the live PR state is merged/closed, and it never steals or
+    force-cleans an active *foreign* lease while the PR is still open. When
+    ``apply`` is true and a lease is still active on a merged/closed PR, it posts
+    a terminal ``phase: released`` lease marker (``blocker: post-merge-moot``) —
+    an append-only comment that neutralises the moot lease without deleting any
+    other session's comment.
+
+    Args:
+        pr_number: The PR whose lingering lease to assess/clean.
+        apply: When false (default) report only (read-only). When true, post the
+            terminal released marker if — and only if — cleanup is allowed.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict reporting PR merged/closed state, merge_commit_sha, linked-issue
+        closure state, whether the lease is moot, whether cleanup was performed
+        or skipped (and why), and ``no_merge_or_adoption`` True — this path never
+        merges or adopts.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "cleanup_performed": False,
+            "no_merge_or_adoption": True,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    pr_live = api_request(
+        "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
+    comments = _fetch_pr_comments(
+        pr_number, remote=remote, host=host, org=org, repo=repo)
+    assessment = reviewer_pr_lease.assess_post_merge_moot_lease(
+        comments,
+        pr_number=pr_number,
+        pr_merged=bool(pr_live.get("merged") or pr_live.get("merged_at")),
+        pr_state=pr_live.get("state"),
+        merge_commit_sha=pr_live.get("merge_commit_sha"),
+    )
+
+    # Best-effort linked-issue closure state for the report.
+    active = assessment.get("active_lease") or {}
+    issue_no = active.get("issue_number")
+    linked_issue_state = None
+    if issue_no:
+        try:
+            issue = api_request(
+                "GET", f"{repo_api_url(h, o, r)}/issues/{issue_no}", auth) or {}
+            linked_issue_state = issue.get("state")
+        except Exception:
+            linked_issue_state = None
+
+    report = {
+        "success": True,
+        "pr_number": pr_number,
+        "pr_state": assessment.get("pr_state"),
+        "pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
+        "merge_commit_sha": assessment.get("merge_commit_sha"),
+        "linked_issue_number": issue_no,
+        "linked_issue_state": linked_issue_state,
+        "lease_moot": assessment.get("is_moot"),
+        "active_lease": assessment.get("active_lease"),
+        "cleanup_allowed": assessment.get("cleanup_allowed"),
+        "cleanup_performed": False,
+        "no_merge_or_adoption": True,
+        "mode": "apply" if apply else "read_only",
+        "reasons": assessment.get("reasons") or [],
+    }
+
+    if not apply:
+        return report
+    if not assessment.get("cleanup_allowed"):
+        report["cleanup_skipped_reason"] = (
+            assessment.get("reasons") or ["cleanup not allowed"]
+        )
+        return report
+
+    comment_block = _profile_operation_gate("gitea.pr.comment")
+    if comment_block:
+        report["success"] = False
+        report["reasons"] = comment_block
+        report["permission_report"] = _permission_block_report("gitea.pr.comment")
+        return report
+
+    verify_preflight_purity(remote)
+    body = assessment["release_body"]
+    comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+    with _audited(
+        "comment_pr",
+        host=h,
+        remote=remote,
+        org=o,
+        repo=r,
+        pr_number=pr_number,
+        request_metadata={"source": "cleanup_post_merge_moot_lease"},
+    ):
+        posted = api_request("POST", comment_url, auth, {"body": body})
+
+    report["cleanup_performed"] = True
+    report["released_comment_id"] = posted.get("id")
+    return report
+
+
 @mcp.tool()
 def gitea_list_issue_comments(
     issue_number: int,
@@ -5761,6 +5990,8 @@ _PROJECT_SKILLS = {
         "steps": [
             "Resolve task first: gitea_resolve_task_capability(task='review_pr') "
             "to confirm reviewer namespace and avoid author-profile blocks.",
+            "Load canonical workflow proof with gitea_load_review_workflow "
+            "before any review/merge mutation (#389).",
             "Verify reviewer identity with gitea_whoami; the PR author "
             "must be a different user.",
             "Reconcile live queue state FIRST (do not trust prior handoffs): "
@@ -6684,6 +6915,8 @@ def gitea_get_runtime_context(
         ),
         "role_kind": _role_kind(allowed, forbidden),
         "shell_health": native_mcp_preference.shell_health_status(),
+        "workflow_load_proof": review_workflow_load.workflow_load_status(
+            PROJECT_ROOT),
     }
 
     if reveal and h:
@@ -6692,6 +6925,80 @@ def gitea_get_runtime_context(
     return result
 
 
+@mcp.tool()
+def gitea_record_pre_review_command(
+    command: str,
+    cwd: str | None = None,
+    classification: str | None = None,
+) -> dict:
+    """Classify and record a command executed before workflow load (#403).
+
+    Read-only with respect to Gitea API. Pre-review inventory/diagnostic commands
+    may be recorded as allowed; boundary violations block reviewer mutations.
+    """
+    recorded = review_workflow_boundary.record_pre_review_command(
+        command,
+        cwd=cwd,
+        project_root=PROJECT_ROOT,
+        classification=classification,
+    )
+    boundary_state = review_workflow_boundary.assess_boundary_status(PROJECT_ROOT)
+    return {
+        "success": True,
+        "recorded": recorded,
+        "boundary_status": boundary_state.get("boundary_status"),
+        "boundary_clean": boundary_state.get("boundary_clean"),
+        "reasons": list(boundary_state.get("reasons") or []),
+    }
+
+
+@mcp.tool()
+def gitea_load_review_workflow(
+    prompt_text: str | None = None,
+) -> dict:
+    """Load and record canonical review-merge workflow proof for this session (#389, #403).
+
+    Read-only with respect to Gitea API; records in-process workflow source/hash
+    proof and session boundary state required before reviewer review or merge
+    mutations.
+    """
+    try:
+        recorded = review_workflow_load.record_review_workflow_load(
+            PROJECT_ROOT, prompt_text=prompt_text)
+    except OSError as exc:
+        return {
+            "success": False,
+            "loaded": False,
+            "reasons": [str(exc)],
+            "recovery_handoff": review_workflow_load.recovery_handoff_without_replay(),
+        }
+    boundary_reasons = review_workflow_boundary.boundary_blockers(PROJECT_ROOT)
+    helper = review_workflow_boundary.workflow_load_helper_result(
+        recorded, PROJECT_ROOT)
+    return {
+        "success": not boundary_reasons,
+        "loaded": True,
+        "workflow_source": recorded["workflow_source"],
+        "task_mode": recorded["task_mode"],
+        "workflow_hash": recorded["workflow_hash"],
+        "workflow_version": recorded["workflow_version"],
+        "final_report_schema_path": recorded["final_report_schema_path"],
+        "final_report_schema_hash": recorded["final_report_schema_hash"],
+        "prompt_conflicts_with_workflow": recorded[
+            "prompt_conflicts_with_workflow"],
+        "prompt_conflict_reasons": recorded.get("prompt_conflict_reasons") or [],
+        "workflow_load_proof_present": True,
+        "boundary_status": recorded.get("boundary_status"),
+        "boundary_clean": recorded.get("boundary_clean"),
+        "workflow_load_helper_result": helper,
+        "reasons": boundary_reasons,
+        "recovery_handoff": (
+            review_workflow_load.recovery_handoff_without_replay()
+            if boundary_reasons else []
+        ),
+    }
+
+
 @mcp.tool()
 def gitea_list_profiles() -> dict:
     """Read-only: list all Gitea MCP profiles with redacted metadata.
@@ -7860,7 +8167,20 @@ def gitea_resolve_task_capability(
     }
     if reason_msg:
         result["reason"] = reason_msg
+    if task in ("review_pr", "merge_pr"):
+        result["workflow_load_proof"] = review_workflow_load.workflow_load_status(
+            PROJECT_ROOT)
+        if not result["workflow_load_proof"].get("workflow_load_valid"):
+            guidance = (
+                "Call gitea_load_review_workflow before any reviewer review "
+                "or merge mutation."
+            )
+            if guidance not in task_role_guidance:
+                task_role_guidance.append(guidance)
     role_session_router.sync_route_from_capability(result)
+    if audit_reconciliation_mode.check_audit_task_enters_phase(task):
+        phase_record = audit_reconciliation_mode.enter_audit_phase(task)
+        result["reconciliation_phase"] = phase_record.get("phase")
     was_terminal = capability_stop_terminal.is_active()
     terminal = capability_stop_terminal.sync_from_capability_result(result)
     if terminal:

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

@@ -217,12 +217,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,6 +282,101 @@ def assess_acquire_lease(
         "existing_lease": existing,
         "lease_body": body,
         "session_id": session_id,
+        "post_merge_moot": post_merge_moot,
+    }
+
+
+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,
     }
 
 

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

@@ -36,6 +36,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:

task_capability_map.py

@@ -104,6 +104,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,13 @@ 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_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._lease_patch = _install_owned_reviewer_lease(8)
         self._lease_patch.start()
         self._auth_identity_patch = patch(
@@ -337,6 +339,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 +359,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_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_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'."""
@@ -179,6 +185,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 +678,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 +1057,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 +1885,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 +2005,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 +2425,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 +2509,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()

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