fix: resolve conflicts for PR #511

Merge prgs/master into feat/issue-507-controller-thread-ledger and preserve PR intent
alongside compatible master guardrails.

.env.example

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

audit_reconciliation_mode.py

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

author_mutation_worktree.py

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

docs/llm-workflow-runbooks.md

@@ -863,6 +863,45 @@ scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md
 scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md --push
 ```
 
+## Namespace workspace binding (#510)
+
+Each MCP namespace resolves its **own** active task workspace. Foreign role
+worktree environment variables must not poison another namespace's purity
+checks.
+
+| Namespace | Workspace env vars (in priority under `GITEA_ACTIVE_WORKTREE`) | Allowed roots |
+|-----------|------------------------------------------------------------------|---------------|
+| author | `GITEA_AUTHOR_WORKTREE` | `branches/<task>` worktree only (#274) |
+| reviewer | `GITEA_REVIEWER_WORKTREE` | clean `branches/<review>` worktree |
+| merger | `GITEA_MERGER_WORKTREE` | clean `branches/<merge>` worktree **or** clean control checkout |
+| reconciler | `GITEA_RECONCILER_WORKTREE` | clean `branches/<reconcile>` worktree **or** clean control checkout |
+
+`GITEA_AUTHOR_WORKTREE` is **author-only**. Reviewer, merger, and reconciler
+MCP processes ignore it even when it points at a dirty author WIP tree.
+
+### Safe reconnect / rebind procedure
+
+When a mutation blocks on workspace binding:
+
+1. Read the error — it names the **resolved workspace path**, **role
+   namespace**, and **binding source** (tool arg, env var, or process root).
+2. Reconnect or relaunch the correct namespace MCP server from the intended
+   workspace (or set the role-specific env var before launch).
+3. Pass `worktree_path` on reviewer/merger mutation tools when the active
+   branches/ worktree differs from the MCP process root.
+4. **Do not** clean, reset, or discard foreign role worktrees to unblock your
+   own namespace — that destroys another agent's WIP.
+
+### CTH guidance for workspace binding blockers
+
+When posting a Canonical Thread Handoff after a binding blocker:
+
+- State which namespace was active (author / reviewer / merger / reconciler).
+- Quote the resolved workspace path and binding source from the error.
+- Name the safe reconnect action (relaunch MCP from `branches/...`, set
+  `GITEA_*_WORKTREE`, or pass `worktree_path`).
+- Explicitly note that foreign worktrees must not be cleaned to unblock.
+
 ## Safety notes
 
 - Never place raw tokens or passwords in any LLM MCP config; reference secrets

final_report_validator.py

@@ -13,6 +13,7 @@ from typing import Any, Callable
 
 import issue_lock_provenance
 import thread_state_ledger_validator
+from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
 from review_proofs import (
     HANDOFF_HEADING,
     assess_controller_handoff,
@@ -1024,6 +1025,22 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st
     )
 
 
+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,
     *,
@@ -1048,6 +1065,20 @@ def _rule_shared_two_comment_workflow(report_text: str) -> list[dict[str, str]]:
     return thread_state_ledger_validator.findings_for_final_report(report_text)
 
 
+def _rule_reviewer_post_merge_cleanup_proof(report_text: str) -> list[dict[str, str]]:
+    result = assess_post_merge_cleanup_proof(report_text)
+    if not result.get("block"):
+        return []
+    return _findings_from_reasons(
+        "reviewer.post_merge_cleanup_proof",
+        result.get("reasons") or [],
+        field="Cleanup status",
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or "report CLEANUP_SKIPPED with blocker or full cleanup checklist",
+    )
+
+
 _SHARED_ISSUE_LOCK_RULES = (
     _rule_shared_issue_lock_external_state,
     _rule_shared_manual_lock_pr_override,
@@ -1082,6 +1113,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_target_branch_freshness,
         _rule_reviewer_mutation_ledger,
         _rule_reviewer_review_mutation,
+        _rule_reviewer_post_merge_cleanup_proof,
         _rule_reviewer_stale_head_proof,
     ],
     "reconcile_already_landed": [
@@ -1096,6 +1128,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,

namespace_workspace_binding.py

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

post_merge_cleanup_proof.py

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

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

@@ -0,0 +1,198 @@
+"""Canonical review-merge workflow load proof for reviewer mutations (#389)."""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+from pathlib import Path
+
+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)
+    _REVIEW_WORKFLOW_LOAD = {
+        **meta,
+        "session_pid": os.getpid(),
+        "loaded": True,
+    }
+    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
+
+
+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)
+    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"),
+        "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."""
+    status = workflow_load_status(project_root)
+    if not status.get("workflow_load_proof_present"):
+        return list(status.get("reasons") or [])
+    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/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

@@ -881,6 +881,40 @@ Do not update the main checkout if merge failed, was blocked, or produced reconc
 
 If any local artifact is created after final cleanup, run and report a new final status check.
 
+## 28A. Post-merge cleanup proof checklist (#402)
+
+Successful tool execution is not proof that cleanup was authorized. Before claiming remote branch deletion or local worktree removal, the final report must carry the full safety checklist below. If any gate is missing, report `CLEANUP_SKIPPED` with the exact blocker — never perform cleanup and never claim it was performed.
+
+### Remote branch deletion checklist
+
+When `gitea_delete_branch` (or equivalent) deletes the merged PR head branch, report:
+
+* Delete-branch capability resolved: name the task (`delete_branch` / `cleanup_branch` / `reconcile_merged_cleanups`) and permission (`gitea.branch.delete`) with resolver proof before the delete call
+* Merge result: merged
+* Merge commit SHA: full 40-character SHA
+* Merged PR head branch / deleted branch: exact branch name (must match)
+* Branch protection: none / branch is not protected
+* Open PR inventory proof: no other open PR references the branch
+* Active heartbeat/claim/lease: none
+
+### Local worktree removal checklist
+
+When removing session-owned review/simulation worktrees under `branches/`, report:
+
+* Session-owned worktree path: exact path under `branches/`
+* Pre-removal tracked state: clean
+* Pre-removal untracked state: clean
+* Git worktree list after removal: command output or equivalent proof
+
+### Skipped cleanup
+
+If any gate fails, report:
+
+* Cleanup outcome: `CLEANUP_SKIPPED`
+* Cleanup blocker: exact missing gate (for example `gitea.branch.delete capability not resolved`)
+
+Skipped cleanup with an exact blocker passes validation. Performed-cleanup claims without the checklist fail validation.
+
 ## 29. Recovery handoff rules
 
 If blocked, produce a recovery handoff with:

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,11 @@ class TestGatedToolAudit(_AuditWiringBase):
 
     def setUp(self):
         super().setUp()
-        from mcp_server import init_review_decision_lock
-        from tests.test_mcp_server import _install_owned_reviewer_lease
+        from tests.test_mcp_server import _init_reviewer_session, _install_owned_reviewer_lease
         import reviewer_pr_lease
 
-        # init_review_decision_lock clears any prior session lease (#407).
-        init_review_decision_lock("prgs", "review_pr")
+        # Session init clears any prior session lease (#407) and loads workflow (#389).
+        _init_reviewer_session("prgs")
         self._lease_patch = _install_owned_reviewer_lease(8)
         self._lease_patch.start()
         self._auth_identity_patch = patch(
@@ -337,6 +337,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 +357,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(
@@ -1876,7 +1884,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 +2004,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 +2424,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 +2508,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()
@@ -3893,7 +3902,7 @@ class TestPreflightVerification(unittest.TestCase):
         os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n"
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity()
-        self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
+        self.assertIn("forbidden from modifying tracked workspace files", str(ctx.exception))
         self.assertIn("reviewer_edit.py", str(ctx.exception))
 
         # Foreign pre-existing dirty state does not block when unchanged.
@@ -3959,7 +3968,8 @@ class TestPreflightVerification(unittest.TestCase):
         with self.assertRaises(RuntimeError) as ctx:
             mcp_server.verify_preflight_purity(worktree_path=worktree)
         msg = str(ctx.exception)
-        self.assertIn("active task workspace root", msg)
-        self.assertIn("inspected git root", msg)
-        self.assertIn("dirty files: task_file.py", msg)
-        self.assertIn("dirty scope:", msg)
+        self.assertIn("resolved workspace", msg)
+        self.assertIn(worktree, msg)
+        self.assertIn("worktree_path argument", msg)
+        self.assertIn("task_file.py", msg)
+        self.assertIn("author namespace", msg)

tests/test_namespace_workspace_binding.py

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

tests/test_permission_reports.py

@@ -99,6 +99,7 @@ class PermissionReportBase(unittest.TestCase):
     def tearDown(self):
         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 +251,8 @@ class TestEligibilityDenialReport(PermissionReportBase):
                 return {"login": "author-user"}
             return PR_PAYLOAD
         mock_api.side_effect = fake_api
+        mcp_server.init_review_decision_lock("prgs", "review_pr")
+        mcp_server.gitea_load_review_workflow()
         with patch.dict(os.environ, self._env("author-profile")):
             res = mcp_server.gitea_merge_pr(
                 pr_number=42, confirmation="MERGE PR 42",

tests/test_post_merge_cleanup_proof.py

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

tests/test_post_merge_moot_lease.py

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

tests/test_pr_lease_comments_non_list_guard.py

@@ -0,0 +1,76 @@
+"""Regression tests for non-list API payloads on PR/issue comment listing (#485)."""
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+from mcp_server import (  # noqa: E402
+    _list_pr_lease_comments,
+    gitea_list_issue_comments,
+)
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+AUTHOR_ENV = {
+    "GITEA_PROFILE_NAME": "gitea-author",
+    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
+}
+
+
+class TestPrLeaseCommentsNonListGuard(unittest.TestCase):
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_list_pr_lease_comments_non_list_payload_returns_empty(self, _auth, mock_api):
+        mock_api.return_value = {"message": "Unauthorized"}
+        result = _list_pr_lease_comments(
+            12, remote="prgs", host=None, org=None, repo=None)
+        self.assertEqual(result, [])
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_list_pr_lease_comments_none_returns_empty(self, _auth, mock_api):
+        mock_api.return_value = None
+        result = _list_pr_lease_comments(
+            12, remote="prgs", host=None, org=None, repo=None)
+        self.assertEqual(result, [])
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_list_pr_lease_comments_list_payload_unchanged(self, _auth, mock_api):
+        comment = {"id": 7, "body": "<!-- mcp-review-lease:v1 -->"}
+        mock_api.return_value = [comment]
+        result = _list_pr_lease_comments(
+            12, remote="prgs", host=None, org=None, repo=None, limit=5)
+        self.assertEqual(result, [comment])
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_list_issue_comments_non_list_payload_returns_empty(self, _auth, mock_api):
+        mock_api.return_value = {"message": "Unauthorized"}
+        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            result = gitea_list_issue_comments(issue_number=9, remote="prgs")
+        self.assertTrue(result["success"])
+        self.assertEqual(result["comments"], [])
+
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_list_issue_comments_list_payload_unchanged(self, _auth, mock_api):
+        mock_api.return_value = [
+            {
+                "id": 101,
+                "user": {"login": "alice"},
+                "body": "hello",
+                "created_at": "2026-07-03T00:00:00Z",
+                "updated_at": "2026-07-03T01:00:00Z",
+            }
+        ]
+        with patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            result = gitea_list_issue_comments(issue_number=9, remote="prgs")
+        self.assertTrue(result["success"])
+        self.assertEqual(len(result["comments"]), 1)
+        self.assertEqual(result["comments"][0]["author"], "alice")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pr_queue_inventory.py

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

tests/test_preflight_read_survival.py

@@ -0,0 +1,106 @@
+"""#469: capability preflight survives interleaved read-only whoami calls."""
+
+import os
+import unittest
+
+import gitea_mcp_server as mcp_server
+
+
+class TestPreflightReadSurvival(unittest.TestCase):
+    def setUp(self):
+        self.orig_whoami = mcp_server._preflight_whoami_called
+        self.orig_capability = mcp_server._preflight_capability_called
+        self.orig_whoami_violation = mcp_server._preflight_whoami_violation
+        self.orig_capability_violation = mcp_server._preflight_capability_violation
+        self.orig_resolved_role = mcp_server._preflight_resolved_role
+        self.orig_resolved_task = mcp_server._preflight_resolved_task
+        self.orig_process_start = mcp_server._process_start_porcelain
+        self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
+        self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
+        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+            if key in os.environ:
+                del os.environ[key]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server._preflight_whoami_called = False
+        mcp_server._preflight_capability_called = False
+        mcp_server._preflight_whoami_violation = False
+        mcp_server._preflight_capability_violation = False
+        mcp_server._preflight_resolved_role = None
+        mcp_server._preflight_resolved_task = None
+        mcp_server._process_start_porcelain = ""
+        mcp_server._preflight_whoami_baseline_porcelain = None
+        mcp_server._preflight_capability_baseline_porcelain = None
+
+    def tearDown(self):
+        mcp_server._preflight_whoami_called = self.orig_whoami
+        mcp_server._preflight_capability_called = self.orig_capability
+        mcp_server._preflight_whoami_violation = self.orig_whoami_violation
+        mcp_server._preflight_capability_violation = self.orig_capability_violation
+        mcp_server._preflight_resolved_role = self.orig_resolved_role
+        mcp_server._preflight_resolved_task = self.orig_resolved_task
+        mcp_server._process_start_porcelain = self.orig_process_start
+        mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
+        mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
+        for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
+            if key in os.environ:
+                del os.environ[key]
+
+    def test_interleaved_whoami_preserves_capability(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reconciler", resolved_task="close_pr"
+        )
+        self.assertTrue(mcp_server._preflight_capability_called)
+        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
+
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_capability_called)
+        self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
+
+        mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertFalse(mcp_server._preflight_capability_called)
+
+    def test_missing_capability_still_fails_closed(self):
+        mcp_server.record_preflight_check("whoami")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertIn("has not been resolved", str(ctx.exception))
+
+    def test_task_mismatch_fails_closed(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="author", resolved_task="create_issue"
+        )
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="close_pr")
+        self.assertIn("task mismatch", str(ctx.exception))
+
+    def test_capability_consumed_after_mutation_gate(self):
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="author", resolved_task="create_issue"
+        )
+        mcp_server.verify_preflight_purity(task="create_issue")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_preflight_purity(task="create_issue")
+        self.assertIn("has not been resolved", str(ctx.exception))
+
+    def test_whoami_recovery_after_violation_clears_capability(self):
+        os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
+        mcp_server.record_preflight_check("whoami")
+        self.assertTrue(mcp_server._preflight_whoami_violation)
+
+        del os.environ["GITEA_TEST_FORCE_DIRTY"]
+        os.environ["GITEA_TEST_PORCELAIN"] = ""
+        mcp_server.record_preflight_check("whoami")
+        self.assertFalse(mcp_server._preflight_whoami_violation)
+        self.assertFalse(mcp_server._preflight_capability_called)
+
+        mcp_server.record_preflight_check(
+            "capability", resolved_role="reviewer", resolved_task="review_pr"
+        )
+        mcp_server.verify_preflight_purity(task="review_pr")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reconciler_close_workspace_guard.py

@@ -0,0 +1,89 @@
+"""Reconciler close_pr must not require author branches/ worktree (#468)."""
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as srv
+
+FAKE_AUTH = "token test"
+CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
+
+RECONCILER_PROFILE = {
+    "profile_name": "prgs-reconciler",
+    "allowed_operations": ["gitea.read", "gitea.pr.close", "gitea.pr.comment"],
+    "forbidden_operations": [
+        "gitea.pr.approve",
+        "gitea.pr.merge",
+        "gitea.pr.review",
+        "gitea.pr.create",
+        "gitea.branch.push",
+        "gitea.repo.commit",
+    ],
+    "audit_label": "prgs-reconciler",
+}
+
+
+class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
+    def setUp(self):
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_whoami_violation = False
+        srv._preflight_capability_violation = False
+        self._orig_in_test = srv._preflight_in_test_mode
+        srv._preflight_in_test_mode = lambda: False
+
+    def tearDown(self):
+        srv._preflight_in_test_mode = self._orig_in_test
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE)
+    @patch("gitea_mcp_server.api_request")
+    def test_reconciler_close_pr_from_control_checkout_succeeds(
+        self, mock_api, _profile, _ns, _auth
+    ):
+        srv._preflight_resolved_role = "reconciler"
+        mock_api.return_value = {
+            "number": 414,
+            "title": "old",
+            "body": "",
+            "state": "closed",
+            "html_url": "https://gitea.example.com/pulls/414",
+        }
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with patch.dict(os.environ, {}, clear=False):
+                os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
+                os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
+                result = srv.gitea_edit_pr(414, state="closed", remote="prgs")
+        self.assertTrue(result["success"])
+        self.assertEqual(result["state"], "closed")
+        mock_api.assert_called_once()
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch(
+        "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+        return_value=(True, []),
+    )
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch(
+        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={"current_branch": "master"},
+    )
+    def test_author_create_issue_still_blocked_on_control_checkout(
+        self, _git, _get_all, _role, _ns, _prof, _auth
+    ):
+        srv._preflight_resolved_role = "author"
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with self.assertRaises(RuntimeError) as ctx:
+                srv.gitea_create_issue(title="Test", body="body")
+        self.assertIn("stable control checkout", str(ctx.exception))
+
+
+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

@@ -37,6 +37,7 @@ def _lock(mutations=None, correction=False):
 
 def _seed(mutations=None, correction=False):
     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 +49,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)
@@ -93,6 +95,7 @@ class TestTerminalHardStopReasons(unittest.TestCase):
 class TestMergeHardStopWiring(unittest.TestCase):
     def tearDown(self):
         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])
@@ -114,6 +117,7 @@ class TestMergeHardStopWiring(unittest.TestCase):
 class TestMarkFinalHardStopWiring(unittest.TestCase):
     def tearDown(self):
         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])
@@ -146,6 +150,7 @@ def _mark(action, pr_number=6, **kwargs):
 class TestDuplicateRequestChangesSuppression(unittest.TestCase):
     def tearDown(self):
         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()