Merge branch 'prgs/master' into feat/issue-503-reviewer-active-worktree

Merge prgs/master into feat/issue-503-reviewer-active-worktree.
Preserve reviewer mutation workspace binding (#503) and fold in
master's review_pr task preflight checks via _verify_reviewer_mutation_workspace.

.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

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

@@ -325,6 +325,17 @@ shared state and manual writes can clobber another session's live lease. Use
 3. Operator override only when explicitly authorized — record
    `External-state mutations` and `operator override proof` in the final report.
 
+**Adoption proof in the live lock response (#477):** when `gitea_lock_issue`
+adopts an existing own branch, the response carries an `adoption` block with
+citable fields — `adoption_decision` (`ADOPT`), `adopted` (`true`),
+`adopted_branch`, `adopted_branch_head`, `matcher_summary` (boundary-safe reason
+the branch qualified), `competing_branch_check`, and `safe_next_action`. A normal
+lock instead returns an `adoption_check` block with `adoption_decision`
+(`NO_MATCH`) and `adopted: false`, so a non-adoption response can never be misread
+as claiming adoption. Recovery reports should quote the live lock response
+`adoption`/`adoption_check` block directly instead of inferring adoption from
+separate offline checks.
+
 `gitea_create_pr` rejects lock files that lack sanctioned `lock_provenance`
 metadata. Final-report validation blocks handoffs that hide lock read/write/delete
 under `External-state mutations: none` or mix author PR creation with reviewer
@@ -812,6 +823,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

@@ -12,6 +12,7 @@ import re
 from typing import Any, Callable
 
 import issue_lock_provenance
+from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
 from review_proofs import (
     HANDOFF_HEADING,
     assess_controller_handoff,
@@ -492,6 +493,42 @@ def _rule_reviewer_validation_failure_history(
     ]
 
 
+def _rule_reviewer_validation_cwd_proof(
+    report_text: str,
+    *,
+    validation_session: dict | None = None,
+) -> list[dict[str, str]]:
+    from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report
+
+    session = validation_session or {}
+    claims = (
+        session.get("validation_ran")
+        or session.get("command")
+        or session.get("baseline_validation_ran")
+    )
+    if not claims and "validation command:" not in (report_text or "").lower():
+        return []
+
+    result = assess_validation_cwd_proof_report(
+        report_text,
+        validation_session=session,
+    )
+    if result.get("proven") or not result.get("claims_validation"):
+        return []
+    severity = "block" if result.get("violations") else "downgrade"
+    return [
+        validator_finding(
+            "reviewer.validation_cwd_proof",
+            severity,
+            "Validation cwd/HEAD proof",
+            reason,
+            result.get("safe_next_action")
+            or "document pwd, HEAD SHA, and explicit cwd before validation",
+        )
+        for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
+    ]
+
+
 def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]:
     from pr_work_lease import assess_reviewer_stale_head_final_report
 
@@ -1006,6 +1043,20 @@ def _rule_reviewer_review_mutation(
     )
 
 
+def _rule_reviewer_post_merge_cleanup_proof(report_text: str) -> list[dict[str, str]]:
+    result = assess_post_merge_cleanup_proof(report_text)
+    if not result.get("block"):
+        return []
+    return _findings_from_reasons(
+        "reviewer.post_merge_cleanup_proof",
+        result.get("reasons") or [],
+        field="Cleanup status",
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or "report CLEANUP_SKIPPED with blocker or full cleanup checklist",
+    )
+
+
 _SHARED_ISSUE_LOCK_RULES = (
     _rule_shared_issue_lock_external_state,
     _rule_shared_manual_lock_pr_override,
@@ -1023,6 +1074,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_git_fetch_readonly,
         _rule_reviewer_validation_command,
         _rule_reviewer_validation_failure_history,
+        _rule_reviewer_validation_cwd_proof,
         _rule_reviewer_validation_structured,
         _rule_reviewer_linked_issue,
         _rule_reviewer_baseline_on_failure,
@@ -1034,6 +1086,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": [

gitea_mcp_server.py

@@ -165,6 +165,7 @@ _preflight_capability_called = False
 _preflight_whoami_violation = False
 _preflight_capability_violation = False
 _preflight_resolved_role = None
+_preflight_resolved_task: str | None = None
 _process_start_porcelain: str | None = None
 _preflight_whoami_baseline_porcelain: str | None = None
 _preflight_capability_baseline_porcelain: str | None = None
@@ -174,12 +175,76 @@ _preflight_reviewer_violation_files: list[str] = []
 
 ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
 AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
+REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE"
+MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
+RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
+
+import namespace_workspace_binding as nwb  # noqa: E402
 
 
 def _preflight_in_test_mode() -> bool:
     return "pytest" in sys.modules or "unittest" in sys.modules
 
 
+def _reviewer_session_worktree() -> str | None:
+    import reviewer_pr_lease as _reviewer_pr_lease
+
+    session = _reviewer_pr_lease.get_session_lease()
+    if not session:
+        return None
+    worktree = (session.get("worktree") or "").strip()
+    return worktree or None
+
+
+def _effective_workspace_role() -> str:
+    """Resolve the namespace key used for workspace binding (#510)."""
+    profile = get_profile()
+    role = _preflight_resolved_role
+    if not role:
+        role = _role_kind(
+            profile.get("allowed_operations") or [],
+            profile.get("forbidden_operations") or [],
+        )
+    return nwb.normalize_role_kind(
+        role,
+        profile_name=profile.get("profile_name"),
+    )
+
+
+def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
+    """Resolve the namespace-scoped workspace root inspected by pre-flight guards."""
+    role = _effective_workspace_role()
+    workspace, _source = nwb.resolve_namespace_workspace(
+        role_kind=role,
+        worktree_path=worktree_path,
+        process_project_root=PROJECT_ROOT,
+        session_lease_worktree=(
+            _reviewer_session_worktree() if role in {"reviewer", "merger"} else None
+        ),
+        profile_name=get_profile().get("profile_name"),
+    )
+    return workspace
+
+
+def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict:
+    """Canonical namespace workspace + repository root for guards (#460/#510)."""
+    role = _effective_workspace_role()
+    return nwb.resolve_namespace_mutation_context(
+        role_kind=role,
+        worktree_path=worktree_path,
+        process_project_root=PROJECT_ROOT,
+        session_lease_worktree=(
+            _reviewer_session_worktree() if role in {"reviewer", "merger"} else None
+        ),
+        profile_name=get_profile().get("profile_name"),
+    )
+
+
+def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
+    """Backward-compatible alias for namespace workspace context."""
+    return _resolve_namespace_mutation_context(worktree_path)
+
+
 def _ensure_process_start_porcelain() -> str:
     """Capture the shared-worktree baseline once per MCP process (#252)."""
     global _process_start_porcelain
@@ -188,26 +253,6 @@ def _ensure_process_start_porcelain() -> str:
     return _process_start_porcelain
 
 
-def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
-    """Resolve the workspace root inspected by pre-flight guards."""
-    return author_mutation_worktree.resolve_mutation_workspace(
-        worktree_path,
-        PROJECT_ROOT,
-        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
-        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
-    )
-
-
-def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
-    """Canonical workspace + repository root for runtime_context and guards (#460)."""
-    return author_mutation_worktree.resolve_author_mutation_context(
-        worktree_path,
-        PROJECT_ROOT,
-        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
-        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
-    )
-
-
 def _get_git_root(path: str) -> str | None:
     try:
         res = subprocess.run(
@@ -275,7 +320,7 @@ def _format_preflight_files(files: list[str]) -> str:
 
 
 def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
-    ctx = _resolve_author_mutation_context(worktree_path)
+    ctx = _resolve_namespace_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
     inspected_root = _get_git_root(workspace)
     process_root = ctx["process_project_root"]
@@ -293,6 +338,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
         "dirty_files": list(dirty_files),
         "dirty_scope": dirty_scope,
         "workspace_roots_aligned": ctx["roots_aligned"],
+        "workspace_role_kind": ctx.get("workspace_role_kind"),
+        "workspace_binding_source": ctx.get("workspace_binding_source"),
+        "ignored_bindings": list(ctx.get("ignored_bindings") or []),
     }
     if not ctx["roots_aligned"]:
         details["workspace_root_mismatch"] = (
@@ -303,13 +351,19 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
 
 
 def _format_preflight_workspace_details(details: dict) -> str:
-    return (
-        f"MCP server process root: {details.get('mcp_server_process_root')}; "
-        f"active task workspace root: {details.get('active_task_workspace_root')}; "
-        f"inspected git root: {details.get('inspected_git_root')}; "
-        f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; "
-        f"dirty scope: {details.get('dirty_scope')}"
-    )
+    parts = [
+        f"MCP server process root: {details.get('mcp_server_process_root')}",
+        f"active task workspace root: {details.get('active_task_workspace_root')}",
+        f"workspace role: {details.get('workspace_role_kind')}",
+        f"binding source: {details.get('workspace_binding_source')}",
+        f"inspected git root: {details.get('inspected_git_root')}",
+        f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}",
+        f"dirty scope: {details.get('dirty_scope')}",
+    ]
+    ignored = details.get("ignored_bindings") or []
+    if ignored:
+        parts.append(f"ignored foreign bindings: {'; '.join(ignored)}")
+    return "; ".join(parts)
 
 
 def assess_preflight_status(worktree_path: str | None = None) -> dict:
@@ -332,6 +386,25 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
                 "Active task workspace has tracked file edits before mutation "
                 f"({_format_preflight_workspace_details(workspace_details)})"
             )
+        role = _effective_workspace_role()
+        if role in nwb.NON_AUTHOR_ROLES:
+            binding = nwb.assess_metadata_only_worktree_binding(
+                role_kind=role,
+                declared_worktree_path=worktree_path,
+                mutation_workspace=_resolve_preflight_workspace_path(worktree_path),
+                process_project_root=PROJECT_ROOT,
+                profile_name=get_profile().get("profile_name"),
+            )
+            if binding.get("block"):
+                reasons.append(binding["reasons"][0])
+                reasons.append(
+                    nwb.format_namespace_workspace_binding_error(
+                        role_kind=role,
+                        workspace_path=binding["mutation_workspace"],
+                        binding_source="MCP server process root (default)",
+                        reasons=binding.get("reasons"),
+                    )
+                )
         return {
             "preflight_ready": not reasons,
             "preflight_block_reasons": reasons,
@@ -379,11 +452,30 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
     }
 
 
-def record_preflight_check(type_name: str, resolved_role: str | None = None):
+def _clear_preflight_capability_state() -> None:
+    """Drop resolved capability proof (consumed by a mutation or fresh resolve)."""
+    global _preflight_capability_called, _preflight_capability_violation
+    global _preflight_resolved_task
+    global _preflight_capability_baseline_porcelain, _preflight_capability_violation_files
+    global _preflight_reviewer_violation_files
+
+    _preflight_capability_called = False
+    _preflight_capability_violation = False
+    _preflight_capability_violation_files = []
+    _preflight_capability_baseline_porcelain = None
+    _preflight_resolved_task = None
+    _preflight_reviewer_violation_files = []
+
+
+def record_preflight_check(
+    type_name: str,
+    resolved_role: str | None = None,
+    resolved_task: str | None = None,
+):
     """Record a pre-flight check (whoami or capability) with session-scoped deltas."""
     global _preflight_whoami_called, _preflight_capability_called
     global _preflight_whoami_violation, _preflight_capability_violation
-    global _preflight_resolved_role
+    global _preflight_resolved_role, _preflight_resolved_task
     global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
     global _preflight_whoami_violation_files, _preflight_capability_violation_files
     global _preflight_reviewer_violation_files
@@ -391,13 +483,24 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
     current = _get_workspace_porcelain()
 
     if type_name == "whoami":
-        # Fresh whoami restarts the capability step and re-evaluates violations
-        # instead of replaying a sticky record (#252).
-        _preflight_capability_called = False
-        _preflight_capability_violation = False
-        _preflight_capability_violation_files = []
-        _preflight_capability_baseline_porcelain = None
-        _preflight_reviewer_violation_files = []
+        # Re-evaluate whoami violations instead of replaying sticky state (#252).
+        # Interleaved read-only whoami must not clear a valid capability proof (#469).
+        saved_capability = None
+        preserve_capability = (
+            _preflight_capability_called
+            and _preflight_whoami_called
+            and not _preflight_whoami_violation
+        )
+        if preserve_capability:
+            saved_capability = (
+                _preflight_capability_violation,
+                list(_preflight_capability_violation_files),
+                _preflight_capability_baseline_porcelain,
+                _preflight_resolved_role,
+                _preflight_resolved_task,
+            )
+        else:
+            _clear_preflight_capability_state()
 
         process_start = _ensure_process_start_porcelain()
         whoami_delta = _new_tracked_changes_since(process_start, current)
@@ -405,6 +508,20 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
         _preflight_whoami_violation_files = whoami_delta
         _preflight_whoami_baseline_porcelain = current
         _preflight_whoami_called = True
+
+        if (
+            preserve_capability
+            and saved_capability is not None
+            and not whoami_delta
+        ):
+            (
+                _preflight_capability_violation,
+                _preflight_capability_violation_files,
+                _preflight_capability_baseline_porcelain,
+                _preflight_resolved_role,
+                _preflight_resolved_task,
+            ) = saved_capability
+            _preflight_capability_called = True
     elif type_name == "capability":
         baseline = _preflight_whoami_baseline_porcelain or ""
         capability_delta = _new_tracked_changes_since(baseline, current)
@@ -414,13 +531,21 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
         _preflight_capability_called = True
         if resolved_role:
             _preflight_resolved_role = resolved_role
+        if resolved_task:
+            _preflight_resolved_task = resolved_task
 
 
 def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
-    """#274: author mutations must run from a branches/ session worktree."""
-    if _preflight_resolved_role == "reviewer":
+    """#274: author file/branch mutations must run from a branches/ worktree.
+
+    Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
+    is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
+    (#468). Non-author namespaces use dedicated workspace env vars (#510).
+    """
+    role = _effective_workspace_role()
+    if role in nwb.NON_AUTHOR_ROLES:
         return
-    ctx = _resolve_author_mutation_context(worktree_path)
+    ctx = _resolve_namespace_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
     git_state = issue_lock_worktree.read_worktree_git_state(workspace)
     assessment = author_mutation_worktree.assess_author_mutation_worktree(
@@ -434,7 +559,11 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
         )
 
 
-def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
+def verify_preflight_purity(
+    remote: str | None = None,
+    worktree_path: str | None = None,
+    task: str | None = None,
+):
     """Verify that identity and capability were verified prior to session edits."""
     global _preflight_reviewer_violation_files
 
@@ -453,12 +582,23 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
         raise RuntimeError(
             "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
         )
+    if (
+        task is not None
+        and _preflight_resolved_task is not None
+        and task != _preflight_resolved_task
+    ):
+        raise RuntimeError(
+            "Pre-flight task mismatch: "
+            f"resolved '{_preflight_resolved_task}' but mutation requires "
+            f"'{task}' (fail closed)"
+        )
 
-    ctx = _resolve_author_mutation_context(worktree_path)
+    ctx = _resolve_namespace_mutation_context(worktree_path)
     workspace = ctx["workspace_path"]
     canonical_root = ctx["canonical_repo_root"]
     process_root = ctx["process_project_root"]
     real_workspace = os.path.realpath(workspace)
+    role = ctx.get("workspace_role_kind") or _effective_workspace_role()
 
     if real_workspace != process_root:
         if not _preflight_in_test_mode():
@@ -475,11 +615,15 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
 
         dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace)))
         if dirty_files:
-            details = _preflight_workspace_details(workspace, dirty_files)
             raise RuntimeError(
-                "Pre-flight order violation: Active task workspace has tracked "
-                "file edits before mutation (fail closed). "
-                f"{_format_preflight_workspace_details(details)}"
+                nwb.format_namespace_workspace_binding_error(
+                    role_kind=role,
+                    workspace_path=workspace,
+                    binding_source=ctx.get("workspace_binding_source")
+                    or "unknown binding source",
+                    dirty_files=dirty_files,
+                    ignored_bindings=ctx.get("ignored_bindings"),
+                )
             )
     else:
         if _preflight_whoami_violation:
@@ -495,19 +639,59 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
                 f"{_format_preflight_files(_preflight_capability_violation_files)}"
             )
 
-        if _preflight_resolved_role == "reviewer":
+        if role in {"reviewer", "merger"}:
             current = _get_workspace_porcelain()
             baseline = _preflight_capability_baseline_porcelain or ""
             reviewer_delta = _new_tracked_changes_since(baseline, current)
             _preflight_reviewer_violation_files = reviewer_delta
             if reviewer_delta:
                 raise RuntimeError(
-                    "Reviewer role violation: Reviewer profile is forbidden from modifying "
+                    f"{role.title()} role violation: profile is forbidden from modifying "
                     "tracked workspace files (fail closed). Offending files: "
                     f"{_format_preflight_files(reviewer_delta)}"
                 )
 
     _enforce_branches_only_author_mutation(worktree_path)
+    _clear_preflight_capability_state()
+
+
+def _verify_role_mutation_workspace(
+    remote: str | None = None,
+    *,
+    worktree_path: str | None = None,
+    worktree: str | None = None,
+    task: str | None = None,
+) -> str:
+    """Bind reviewer/merger mutations to the active namespace workspace (#510)."""
+    role = _effective_workspace_role()
+    git_state = issue_lock_worktree.read_worktree_git_state(
+        _resolve_preflight_workspace_path(worktree_path)
+    )
+    assessment = nwb.assess_namespace_mutation_workspace(
+        role_kind=role,
+        worktree_path=worktree_path,
+        worktree=worktree,
+        process_project_root=PROJECT_ROOT,
+        session_lease_worktree=(
+            _reviewer_session_worktree() if role in {"reviewer", "merger"} else None
+        ),
+        profile_name=get_profile().get("profile_name"),
+        current_branch=git_state.get("current_branch"),
+    )
+    if assessment["block"]:
+        raise RuntimeError(
+            nwb.format_namespace_workspace_binding_error(
+                role_kind=role,
+                workspace_path=assessment["mutation_workspace"],
+                binding_source=assessment.get("workspace_binding_source")
+                or "unknown binding source",
+                reasons=assessment.get("reasons"),
+                ignored_bindings=assessment.get("ignored_bindings"),
+            )
+        )
+    resolved = assessment["mutation_workspace"]
+    verify_preflight_purity(remote, worktree_path=resolved, task=task)
+    return resolved
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
@@ -1223,7 +1407,7 @@ def gitea_create_issue(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
     base = repo_api_url(h, o, r)
     open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
     closed_issues = api_get_all(
@@ -1356,7 +1540,7 @@ def gitea_lock_issue(
         )
     else:
         git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
-    verify_preflight_purity(remote, worktree_path=resolved_worktree)
+    verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -1484,6 +1668,14 @@ def gitea_lock_issue(
             f"Adopted existing branch '{branch_name}' and locked issue "
             f"#{issue_number} for recovery (fail-closed check complete)."
         )
+    else:
+        # #477 AC2: normal (no-adoption) lock responses carry explicit,
+        # adoption-free proof metadata so they stay clear and cannot be
+        # misread as claiming a branch was adopted.
+        result["adoption_check"] = issue_lock_adoption.build_non_adoption_lock_proof(
+            issue_number=issue_number,
+            branch_name=branch_name,
+        )
     if agent_artifacts:
         result["warnings"] = [
             "Agent temp artifacts at repo root (delete before implementation): "
@@ -1572,7 +1764,7 @@ def gitea_create_pr(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
@@ -2588,7 +2780,12 @@ def _list_pr_lease_comments(
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
-    comments = api_request("GET", api, auth) or []
+    comments = api_request("GET", api, auth)
+    # Fail safe to no lease comments when the API returns a non-list payload
+    # (e.g. an error object such as an HTTP 401 body): lease state can only be
+    # proven from real comment entries, never inferred from an error shape (#485).
+    if not isinstance(comments, list):
+        return []
     return list(comments[:limit])
 
 
@@ -2631,9 +2828,12 @@ def _evaluate_pr_review_submission(
     *,
     live: bool,
     final_review_decision_ready: bool = False,
+    worktree_path: str | None = None,
 ) -> dict:
     """Shared gate chain for live submit and dry-run review tools."""
-    verify_preflight_purity(remote)
+    _verify_role_mutation_workspace(
+        remote, worktree_path=worktree_path, task="review_pr"
+    )
     action = (action or "").strip().lower()
     result = {
         "requested_action": action,
@@ -3033,6 +3233,7 @@ def gitea_dry_run_pr_review(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Validate review submission mechanics without a live PR mutation."""
     return _evaluate_pr_review_submission(
@@ -3045,6 +3246,7 @@ def gitea_dry_run_pr_review(
         org=org,
         repo=repo,
         live=False,
+        worktree_path=worktree_path,
     )
 
 
@@ -3060,6 +3262,7 @@ def gitea_submit_pr_review(
     org: str | None = None,
     repo: str | None = None,
     final_review_decision_ready: bool = False,
+    worktree_path: str | None = None,
 ) -> dict:
     """Gated PR review mutation: comment findings, request changes, or approve.
 
@@ -3078,6 +3281,7 @@ def gitea_submit_pr_review(
         repo=repo,
         live=True,
         final_review_decision_ready=final_review_decision_ready,
+        worktree_path=worktree_path,
     )
 
 
@@ -3140,12 +3344,12 @@ def gitea_edit_pr(
     if not payload:
         raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
 
-    verify_preflight_purity(remote)
+    closing = payload.get("state") == "closed"
+    verify_preflight_purity(remote, task="close_pr" if closing else None)
 
     # PR closure is a first-class capability, distinct from retitling or
     # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
     # never touches the network.
-    closing = payload.get("state") == "closed"
     if closing:
         gate_reasons = _profile_operation_gate("gitea.pr.close")
         if gate_reasons:
@@ -3390,7 +3594,7 @@ def gitea_commit_files(
             branch="",
         )
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="commit_files")
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
     h, o, r = _resolve(remote, host, org, repo)
@@ -3438,6 +3642,7 @@ def gitea_merge_pr(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    worktree_path: str | None = None,
 ) -> dict:
     """Gated merge of a Gitea pull request (#16).
 
@@ -3484,6 +3689,10 @@ def gitea_merge_pr(
         host: Override the Gitea host.
         org: Override the owner/organization.
         repo: Override the repository name.
+        worktree_path: Merger workspace path under ``branches/`` or clean
+            control checkout; defaults to ``GITEA_MERGER_WORKTREE``,
+            ``GITEA_ACTIVE_WORKTREE``, or the MCP server process root. Ignores
+            foreign ``GITEA_AUTHOR_WORKTREE`` bindings (#510).
 
     Returns:
         dict describing the attempt: performed, authenticated user, profile
@@ -3491,7 +3700,9 @@ def gitea_merge_pr(
         reasons/gates passed or blocked, and merge result / merge commit if
         available. Never secrets.
     """
-    verify_preflight_purity(remote)
+    _verify_role_mutation_workspace(
+        remote, worktree_path=worktree_path, task="merge_pr"
+    )
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -4020,7 +4231,7 @@ def gitea_delete_branch(
             "permission_report": _permission_block_report("gitea.branch.delete"),
         }
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="delete_branch")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     import urllib.parse
@@ -4129,7 +4340,7 @@ def gitea_reconcile_merged_cleanups(
         report["executed"] = False
         return {"success": True, "performed": False, **report}
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="reconcile_merged_cleanups")
     actions: list[dict] = []
     for entry in report.get("entries") or []:
         head_branch = entry.get("head_branch") or ""
@@ -4443,7 +4654,7 @@ def gitea_reconcile_already_landed_pr(
         )
         return result
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="reconcile_already_landed_pr")
 
     if post_comment and comment_body.strip():
         comment_block = _profile_operation_gate("gitea.pr.comment")
@@ -4546,7 +4757,7 @@ def gitea_close_issue(
         task_capability_map.required_permission("close_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="close_issue")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -4974,7 +5185,7 @@ def gitea_acquire_reviewer_pr_lease(
             "permission_report": _permission_block_report("gitea.pr.comment"),
         }
 
-    verify_preflight_purity(remote)
+    _verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     profile = get_profile()
@@ -5075,7 +5286,7 @@ def gitea_heartbeat_reviewer_pr_lease(
             ],
         }
 
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="review_pr")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     body = reviewer_pr_lease.format_lease_body(
@@ -5192,7 +5403,12 @@ def gitea_list_issue_comments(
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
-    comments = api_request("GET", api, auth) or []
+    comments = api_request("GET", api, auth)
+    # Fail safe to no comments when the API returns a non-list payload (e.g. an
+    # error object such as an HTTP 401 body) so listing never crashes on a
+    # malformed response (#485).
+    if not isinstance(comments, list):
+        comments = []
     reveal = _reveal_endpoints()
     out = []
     for c in comments[:limit]:
@@ -5246,7 +5462,7 @@ def gitea_create_issue_comment(
         (permission blocks also carry a structured 'permission_report',
         #142).
     """
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="comment_issue")
     gate_reasons = _profile_operation_gate("gitea.issue.comment")
     reasons = list(gate_reasons)
     if not (body or "").strip():
@@ -6780,7 +6996,7 @@ def gitea_mark_issue(
         task_capability_map.required_permission("mark_issue"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -6858,7 +7074,7 @@ def gitea_post_heartbeat(
         task_capability_map.required_permission("post_heartbeat"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="post_heartbeat")
     active_profile = profile or get_profile().get("profile_name")
     body = issue_claim_heartbeat.format_heartbeat_body(
         kind="progress",
@@ -7101,7 +7317,7 @@ def gitea_cleanup_stale_claims(
         task_capability_map.required_permission("cleanup_stale_claims"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="cleanup_stale_claims")
 
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
@@ -7255,7 +7471,7 @@ def gitea_set_issue_labels(
         task_capability_map.required_permission("set_issue_labels"))
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, task="set_issue_labels")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     base = repo_api_url(h, o, r)
@@ -7492,7 +7708,7 @@ def gitea_resolve_task_capability(
                 "exact_safe_next_action": next_safe_action,
             }
 
-    record_preflight_check("capability", required_role)
+    record_preflight_check("capability", required_role, resolved_task=task)
 
     # Try automatic dispatch switching
     _ensure_matching_profile(required_permission, required_role, remote, host)

issue_lock_adoption.py

@@ -20,6 +20,43 @@ ADOPT = "adopt_existing_branch"
 BLOCK_COMPETING = "block_competing_branch"
 NO_MATCH = "no_matching_branch"
 
+# Citable decision labels aligned with the ``assess_own_branch_adoption``
+# outcomes, surfaced verbatim in the live ``gitea_lock_issue`` response so
+# recovery reports (#473-style) can quote the lock tool output directly
+# instead of inferring adoption from separate offline checks (#477).
+DECISION_LABELS = {
+    ADOPT: "ADOPT",
+    BLOCK_COMPETING: "BLOCK_COMPETING",
+    NO_MATCH: "NO_MATCH",
+}
+
+_SAFE_NEXT_ACTIONS = {
+    ADOPT: (
+        "Own existing branch adopted for lock recovery; proceed to "
+        "gitea_create_pr for this issue and cite this adoption proof."
+    ),
+    BLOCK_COMPETING: (
+        "Competing same-issue branch(es) exist; resolve branch ownership "
+        "before locking. No adoption performed (fail closed)."
+    ),
+    NO_MATCH: (
+        "No existing branch carries this issue marker; normal lock path "
+        "applied. No adoption performed."
+    ),
+}
+
+
+def decision_label(outcome: str) -> str:
+    """Map an ``assess_own_branch_adoption`` outcome to its citable label."""
+    return DECISION_LABELS.get(outcome, "UNKNOWN")
+
+
+def safe_next_action(outcome: str) -> str:
+    """Return the safe next action string for an adoption *outcome*."""
+    return _SAFE_NEXT_ACTIONS.get(
+        outcome, "Unknown adoption outcome; treat as fail closed."
+    )
+
 
 def _branch_name(entry) -> str:
     if isinstance(entry, dict):
@@ -128,6 +165,42 @@ def assess_own_branch_adoption(
     }
 
 
+def _matcher_summary(issue_number: int, assessment: dict) -> str:
+    """Explain, citably, why the assessed branch did or did not qualify.
+
+    Names the numeric word-boundary rule so reports can show that
+    ``issue-42`` was not matched inside ``issue-420`` (#440 / #477 AC3).
+    """
+    outcome = assessment.get("outcome")
+    matched = assessment.get("matched_branch")
+    competing = assessment.get("competing_branches") or []
+    if outcome == ADOPT and matched:
+        return (
+            f"branch '{matched}' exactly matches the issue-{int(issue_number)} "
+            f"marker (numeric word-boundary; 'issue-{int(issue_number)}' is not "
+            f"matched inside 'issue-{int(issue_number)}0')"
+        )
+    if outcome == BLOCK_COMPETING:
+        return (
+            f"competing same-issue branch(es) {competing} carry the "
+            f"issue-{int(issue_number)} marker but are not the requested "
+            f"branch; ownership is ambiguous (fail closed)"
+        )
+    return (
+        f"no existing branch carries the issue-{int(issue_number)} marker "
+        f"under the numeric word-boundary rule"
+    )
+
+
+def _competing_branch_check(assessment: dict) -> dict:
+    """Structured competing-branch verdict for the proof block."""
+    competing = list(assessment.get("competing_branches") or [])
+    return {
+        "result": "blocked" if competing else "clear",
+        "competing_branches": competing,
+    }
+
+
 def build_adoption_proof(
     *,
     issue_number: int,
@@ -143,7 +216,18 @@ def build_adoption_proof(
     Requirement #4: adoption results must carry issue number, branch name,
     branch head commit, adoption reason, no-existing-PR proof, no-competing-
     live-lock proof, and lock file path/status.
+
+    #477: additionally surface explicit, citable adoption-proof fields tied to
+    the ``assess_own_branch_adoption`` outcome (``adoption_decision``,
+    ``adopted``, ``adopted_branch``, ``adopted_branch_head``,
+    ``matcher_summary``, ``competing_branch_check``, ``safe_next_action``) so a
+    recovery session can quote the live lock response directly. The explicit
+    fields are populated for any outcome; ``adopted_branch`` /
+    ``adopted_branch_head`` are set only when the outcome is ADOPT so a
+    non-adoption proof can never be misread as claiming adoption.
     """
+    outcome = assessment.get("outcome")
+    adopted = outcome == ADOPT
     return {
         "issue_number": issue_number,
         "branch_name": branch_name,
@@ -153,4 +237,36 @@ def build_adoption_proof(
         "no_competing_live_lock_proof": bool(competing_lock_checked),
         "lock_file_path": lock_file_path,
         "lock_file_status": lock_file_status,
+        # Explicit citable fields (#477).
+        "adoption_decision": decision_label(outcome),
+        "adopted": adopted,
+        "adopted_branch": branch_name if adopted else None,
+        "adopted_branch_head": assessment.get("matched_head_sha") if adopted else None,
+        "matcher_summary": _matcher_summary(issue_number, assessment),
+        "competing_branch_check": _competing_branch_check(assessment),
+        "safe_next_action": safe_next_action(outcome),
+    }
+
+
+def build_non_adoption_lock_proof(*, issue_number: int, branch_name: str) -> dict:
+    """Safe, adoption-free proof metadata for a normal (NO_MATCH) lock.
+
+    Requirement #477 AC2: non-adoption lock responses must stay clear and must
+    not imply adoption. This returns explicit ``adopted: False`` metadata with
+    the ``NO_MATCH`` decision so a normal lock response can carry citable proof
+    without ever asserting a branch was adopted.
+    """
+    return {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "adoption_decision": DECISION_LABELS[NO_MATCH],
+        "adopted": False,
+        "adopted_branch": None,
+        "adopted_branch_head": None,
+        "matcher_summary": (
+            f"no existing branch carries the issue-{int(issue_number)} marker; "
+            f"normal lock path (no adoption)"
+        ),
+        "competing_branch_check": {"result": "clear", "competing_branches": []},
+        "safe_next_action": safe_next_action(NO_MATCH),
     }

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

@@ -5515,6 +5515,13 @@ def assess_validation_failure_history_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
+def assess_validation_cwd_proof_report(report_text, **kwargs):
+    """#398: validation commands require explicit worktree cwd and HEAD proof."""
+    from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report as _assess
+
+    return _assess(report_text, **kwargs)
+
+
 def assess_already_landed_classification_report(report_text, **kwargs):
     """#295: already-landed PRs are reconciliation-only, not review eligible."""
     from reviewer_already_landed_classification import (

reviewer_validation_cwd_proof.py

@@ -0,0 +1,233 @@
+"""Explicit worktree and cwd proof for PR review validation (#398)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
+
+_PWD_RE = re.compile(
+    r"(?:^|\n)\s*(?:pwd|working\s+directory|cwd)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_HEAD_RE = re.compile(
+    r"(?:git\s+rev-parse\s+head|observed\s+head\s+sha)\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE | re.MULTILINE,
+)
+_EXPECTED_HEAD_RE = re.compile(
+    r"(?:expected\s+(?:pr\s+)?head\s+sha|candidate\s+head\s+sha|pinned\s+head)\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE,
+)
+_STATUS_RE = re.compile(
+    r"git\s+status\s+(?:--short\s+--branch|--short|-sb)\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+_VALIDATION_CMD_RE = re.compile(
+    r"validation\s+command\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+_GIT_C_CMD_RE = re.compile(r"git\s+-C\s+\S+", re.IGNORECASE)
+_CD_CMD_RE = re.compile(r"(?:^|&&\s*)cd\s+\S+", re.IGNORECASE)
+_BASELINE_CWD_RE = re.compile(
+    r"baseline\s+(?:worktree|working\s+directory|cwd)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_BASELINE_SHA_RE = re.compile(
+    r"baseline\s+(?:target\s+)?sha\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE,
+)
+_BASELINE_CMD_RE = re.compile(
+    r"baseline\s+validation\s+command\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+
+
+def _normalize_path(path: str) -> str:
+    return (path or "").replace("\\", "/").rstrip("/")
+
+
+def _path_under_branches(path: str, project_root: str | None = None) -> bool:
+    normalized = _normalize_path(path)
+    if not normalized:
+        return False
+    if "/branches/" in f"{normalized}/":
+        return True
+    if normalized.endswith("/branches"):
+        return True
+    if project_root:
+        root = _normalize_path(project_root)
+        if normalized.startswith(f"{root}/"):
+            rel = normalized[len(root) + 1 :]
+            return rel == "branches" or rel.startswith("branches/")
+    return False
+
+
+def _expand_sha(sha: str) -> str:
+    return (sha or "").strip().lower()
+
+
+def _sha_matches(expected: str, observed: str) -> bool:
+    exp = _expand_sha(expected)
+    obs = _expand_sha(observed)
+    if not exp or not obs:
+        return False
+    if len(exp) == 40 and len(obs) == 40:
+        return exp == obs
+    return obs.startswith(exp) or exp.startswith(obs)
+
+
+def _command_has_explicit_cwd(command: str, cwd: str) -> bool:
+    text = (command or "").strip()
+    if not text:
+        return False
+    if _GIT_C_CMD_RE.search(text):
+        return True
+    if _CD_CMD_RE.search(text):
+        return True
+    if cwd and cwd in text:
+        return True
+    return False
+
+
+def assess_validation_cwd_proof_report(
+    report_text: str,
+    *,
+    validation_session: dict | None = None,
+    project_root: str | None = None,
+) -> dict[str, Any]:
+    """Require cwd/HEAD proof before reviewer validation claims (#398)."""
+    text = report_text or ""
+    session = dict(validation_session or {})
+    reasons: list[str] = []
+    violations: list[str] = []
+
+    claims_validation = bool(
+        session.get("validation_ran")
+        or _VALIDATION_CMD_RE.search(text)
+        or session.get("command")
+    )
+    if not claims_validation:
+        return {
+            "proven": True,
+            "block": False,
+            "claims_validation": False,
+            "reasons": [],
+            "violations": [],
+            "safe_next_action": "proceed",
+        }
+
+    expected_head = (
+        session.get("expected_head_sha")
+        or session.get("candidate_head_sha")
+        or ""
+    ).strip()
+    if not expected_head:
+        match = _EXPECTED_HEAD_RE.search(text)
+        expected_head = (match.group(1) if match else "").strip()
+
+    observed_head = (session.get("observed_head_sha") or "").strip()
+    if not observed_head:
+        match = _HEAD_RE.search(text)
+        observed_head = (match.group(1) if match else "").strip()
+
+    cwd = (
+        session.get("working_directory")
+        or session.get("cwd")
+        or session.get("pwd")
+        or ""
+    ).strip()
+    if not cwd:
+        match = _PWD_RE.search(text)
+        cwd = (match.group(1) if match else "").strip().rstrip(",.;")
+
+    command = (session.get("command") or "").strip()
+    if not command:
+        match = _VALIDATION_CMD_RE.search(text)
+        command = (match.group(1) if match else "").strip().rstrip(".;")
+
+    if not cwd:
+        reasons.append(
+            "validation claimed without pwd/working-directory proof (#398)"
+        )
+    elif not _path_under_branches(cwd, project_root):
+        violations.append(
+            f"validation cwd {cwd!r} is not under branches/ (#398)"
+        )
+        reasons.append(
+            "reviewer validation must run from a branches/ worktree, "
+            "not the main checkout (#398)"
+        )
+
+    if not observed_head:
+        reasons.append(
+            "validation claimed without git rev-parse HEAD / observed HEAD SHA "
+            "proof (#398)"
+        )
+    elif expected_head and not _sha_matches(expected_head, observed_head):
+        violations.append(
+            f"observed HEAD {observed_head} does not match expected "
+            f"PR head {expected_head} (#398)"
+        )
+        reasons.append("validation HEAD SHA must match pinned PR head (#398)")
+
+    if not _STATUS_RE.search(text) and session.get("git_status") is None:
+        reasons.append(
+            "validation claimed without git status --short --branch proof (#398)"
+        )
+
+    if command and cwd and not _command_has_explicit_cwd(command, cwd):
+        if session.get("tool_working_directory") is not True:
+            reasons.append(
+                "validation command must use git -C <worktree>, "
+                "cd <worktree> && ..., or tool-provided cwd metadata (#398)"
+            )
+
+    baseline_ran = bool(
+        session.get("baseline_validation_ran")
+        or _BASELINE_CMD_RE.search(text)
+    )
+    if baseline_ran:
+        baseline_cwd = (session.get("baseline_worktree_path") or "").strip()
+        if not baseline_cwd:
+            match = _BASELINE_CWD_RE.search(text)
+            baseline_cwd = (match.group(1) if match else "").strip().rstrip(",.;")
+        if not baseline_cwd or not _path_under_branches(baseline_cwd, project_root):
+            reasons.append(
+                "baseline validation claimed without baseline worktree cwd "
+                "under branches/ (#398)"
+            )
+        baseline_sha = (session.get("baseline_target_sha") or "").strip()
+        if not baseline_sha:
+            match = _BASELINE_SHA_RE.search(text)
+            baseline_sha = (match.group(1) if match else "").strip()
+        if not baseline_sha:
+            reasons.append(
+                "baseline validation claimed without baseline target SHA (#398)"
+            )
+        baseline_cmd = (session.get("baseline_command") or "").strip()
+        if not baseline_cmd:
+            match = _BASELINE_CMD_RE.search(text)
+            baseline_cmd = (match.group(1) if match else "").strip()
+        if not baseline_cmd:
+            reasons.append(
+                "baseline validation claimed without exact baseline command (#398)"
+            )
+
+    proven = not reasons and not violations
+    return {
+        "proven": proven,
+        "block": bool(violations) or not proven,
+        "claims_validation": True,
+        "expected_head_sha": expected_head or None,
+        "observed_head_sha": observed_head or None,
+        "working_directory": cwd or None,
+        "reasons": reasons,
+        "violations": violations,
+        "safe_next_action": (
+            "before validation record pwd, git rev-parse HEAD, git status, "
+            "expected PR head SHA; run commands with git -C or cd in the same line"
+            if not proven
+            else "proceed"
+        ),
+    }

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

@@ -626,6 +626,28 @@ Do not claim “full-suite failures are pre-existing” unless baseline proof is
 
 ## 23. Validation command proof rule
 
+Before any diff, test, or compile validation, record in the same command
+transcript or final report:
+
+* `pwd` or explicit working directory
+* `git rev-parse HEAD`
+* `git status --short --branch`
+* expected PR head SHA (candidate head SHA)
+
+Validation commands must use one of:
+
+* `git -C <review_worktree> ...`
+* `cd <review_worktree> && ...` in the same command
+* tool-provided explicit working-directory metadata
+
+Do not rely on inferred shell cwd from a prior command in a different block.
+
+`gitea_validate_review_final_report` rejects validation claims without
+cwd/HEAD proof when `validation_session` is supplied.
+
+Baseline validation must document baseline worktree path, baseline target SHA,
+cwd proof, exact baseline command, and baseline result using the same rules.
+
 Report the exact validation command as executed.
 
 Report the working directory where validation ran.
@@ -859,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:
@@ -1189,6 +1245,7 @@ Controller Handoff:
 * Files reviewed:
 * Validation:
 * Validation failure history:
+* Validation cwd/HEAD proof:
 * Official validation integrity status:
 * Terminal review mutation:
 * Review decision:

tests/test_issue_lock_adoption.py

@@ -11,6 +11,7 @@ from issue_lock_adoption import (  # noqa: E402
     NO_MATCH,
     assess_own_branch_adoption,
     build_adoption_proof,
+    build_non_adoption_lock_proof,
 )
 
 REQ = "feat/issue-420-server-code-parity"
@@ -134,5 +135,93 @@ class TestBuildAdoptionProof(unittest.TestCase):
         self.assertTrue(proof["no_competing_live_lock_proof"])
 
 
+class TestExplicitAdoptionProofFields(unittest.TestCase):
+    """#477: explicit, citable adoption-proof fields for all outcomes."""
+
+    def _proof(self, assessment, branch):
+        return build_adoption_proof(
+            issue_number=420,
+            branch_name=branch,
+            assessment=assessment,
+            open_pr_checked=True,
+            competing_lock_checked=True,
+            lock_file_path="/tmp/example-lock.json",
+            lock_file_status="written",
+        )
+
+    def test_adopt_proof_exposes_explicit_fields(self):
+        assessment = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
+        )
+        proof = self._proof(assessment, REQ)
+        self.assertEqual(proof["adoption_decision"], "ADOPT")
+        self.assertTrue(proof["adopted"])
+        self.assertEqual(proof["adopted_branch"], REQ)
+        self.assertEqual(proof["adopted_branch_head"], "934688a")
+        self.assertEqual(proof["competing_branch_check"]["result"], "clear")
+        self.assertEqual(proof["competing_branch_check"]["competing_branches"], [])
+        self.assertIn("gitea_create_pr", proof["safe_next_action"])
+        self.assertIn("exactly matches", proof["matcher_summary"])
+
+    def test_block_proof_reports_competing_and_does_not_claim_adoption(self):
+        assessment = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": "feat/issue-420-rogue"}],
+        )
+        proof = self._proof(assessment, REQ)
+        self.assertEqual(proof["adoption_decision"], "BLOCK_COMPETING")
+        self.assertFalse(proof["adopted"])
+        self.assertIsNone(proof["adopted_branch"])
+        self.assertIsNone(proof["adopted_branch_head"])
+        self.assertEqual(proof["competing_branch_check"]["result"], "blocked")
+        self.assertIn(
+            "feat/issue-420-rogue",
+            proof["competing_branch_check"]["competing_branches"],
+        )
+        self.assertIn("fail closed", proof["safe_next_action"])
+
+    def test_no_match_proof_does_not_claim_adoption(self):
+        assessment = assess_own_branch_adoption(
+            issue_number=420,
+            requested_branch=REQ,
+            existing_branches=[{"name": "feat/issue-999-unrelated"}],
+        )
+        proof = self._proof(assessment, REQ)
+        self.assertEqual(proof["adoption_decision"], "NO_MATCH")
+        self.assertFalse(proof["adopted"])
+        self.assertIsNone(proof["adopted_branch"])
+        self.assertEqual(proof["competing_branch_check"]["result"], "clear")
+
+    def test_substring_collision_stays_boundary_safe(self):
+        # issue-42 must not adopt/claim against an issue-420 branch (#440/#477).
+        own = "feat/issue-42-widget"
+        assessment = assess_own_branch_adoption(
+            issue_number=42,
+            requested_branch=own,
+            existing_branches=[
+                {"name": own, "commit_sha": "abc1234"},
+                {"name": "feat/issue-420-server-code-parity"},
+            ],
+        )
+        proof = self._proof(assessment, own)
+        self.assertEqual(proof["adoption_decision"], "ADOPT")
+        self.assertEqual(proof["adopted_branch"], own)
+        self.assertEqual(proof["competing_branch_check"]["competing_branches"], [])
+
+    def test_non_adoption_lock_proof_is_adoption_free(self):
+        proof = build_non_adoption_lock_proof(
+            issue_number=196, branch_name="feat/issue-196-mutations"
+        )
+        self.assertEqual(proof["adoption_decision"], "NO_MATCH")
+        self.assertFalse(proof["adopted"])
+        self.assertIsNone(proof["adopted_branch"])
+        self.assertIsNone(proof["adopted_branch_head"])
+        self.assertEqual(proof["competing_branch_check"]["result"], "clear")
+        self.assertIn("no adoption", proof["safe_next_action"].lower())
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_llm_workflow_split.py

@@ -213,6 +213,12 @@ def test_validation_failure_history_verifier_exported():
     assert callable(assess_validation_failure_history_report)
 
 
+def test_validation_cwd_proof_verifier_exported():
+    from review_proofs import assess_validation_cwd_proof_report
+
+    assert callable(assess_validation_cwd_proof_report)
+
+
 def test_prior_blocker_skip_verifier_exported():
     from review_proofs import assess_prior_blocker_skip_proof
 

tests/test_mcp_server.py

@@ -3460,6 +3460,45 @@ class TestIssueLocking(unittest.TestCase):
         self.assertIn("adoption", res)
         self.assertEqual(res["adoption"]["branch_head_commit"], "abc123")
 
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_adoption_response_has_explicit_proof(self, _auth, mock_api, _git_state):
+        # #477 AC1: the live lock response must carry citable adoption proof.
+        branch = "feat/issue-196-mutations"
+        self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"})
+        mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
+        res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
+        proof = res["adoption"]
+        self.assertEqual(proof["adoption_decision"], "ADOPT")
+        self.assertTrue(proof["adopted"])
+        self.assertEqual(proof["adopted_branch"], branch)
+        self.assertEqual(proof["adopted_branch_head"], "abc123")
+        self.assertEqual(proof["competing_branch_check"]["result"], "clear")
+        self.assertIn("gitea_create_pr", proof["safe_next_action"])
+        self.assertIn("196", proof["matcher_summary"])
+
+    @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.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_no_match_response_does_not_claim_adoption(self, _auth, _api, _git_state):
+        # #477 AC2/AC3: a normal (NO_MATCH) lock must carry adoption-free proof
+        # and must NOT expose an ``adoption`` block.
+        res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertTrue(res["success"])
+        self.assertNotIn("adoption", res)
+        check = res["adoption_check"]
+        self.assertEqual(check["adoption_decision"], "NO_MATCH")
+        self.assertFalse(check["adopted"])
+        self.assertIsNone(check["adopted_branch"])
+        self.assertEqual(check["competing_branch_check"]["result"], "clear")
+
     @patch(
         "mcp_server.issue_lock_worktree.read_worktree_git_state",
         return_value=_clean_master_git_state_for_lock(),
@@ -3840,7 +3879,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.
@@ -3906,7 +3945,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_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_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_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_reviewer_validation_cwd_proof.py

@@ -0,0 +1,135 @@
+"""Tests for validation cwd/HEAD proof verifier (#398)."""
+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 reviewer_validation_cwd_proof import assess_validation_cwd_proof_report  # noqa: E402
+
+ROOT = "/Users/jasonwalker/Development/Gitea-Tools"
+WORKTREE = f"{ROOT}/branches/review-feat-issue-398"
+HEAD = "f5953549aad5e822f14f52d3ea3c6d7990109384"
+
+
+def _proof_backed_report() -> str:
+    return "\n".join([
+        f"Candidate head SHA: {HEAD}",
+        f"pwd: {WORKTREE}",
+        f"git rev-parse HEAD: {HEAD}",
+        "git status --short --branch: ## feat/issue-398...prgs/master",
+        f"Validation command: cd {WORKTREE} && venv/bin/python -m pytest tests/ -q",
+        "Result: 1497 passed, 6 skipped",
+    ])
+
+
+class TestValidationCwdProof(unittest.TestCase):
+    def test_no_validation_claim_passes(self):
+        result = assess_validation_cwd_proof_report("Review decision: approve")
+        self.assertTrue(result["proven"])
+
+    def test_missing_cwd_proof_fails(self):
+        result = assess_validation_cwd_proof_report(
+            f"Validation command: pytest tests/\nCandidate head SHA: {HEAD}",
+            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_main_checkout_cwd_blocks(self):
+        result = assess_validation_cwd_proof_report(
+            "\n".join([
+                f"Candidate head SHA: {HEAD}",
+                f"pwd: {ROOT}",
+                f"git rev-parse HEAD: {HEAD}",
+                "git status --short --branch: ## master",
+                "Validation command: pytest tests/ -q",
+            ]),
+            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
+            project_root=ROOT,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["violations"])
+
+    def test_wrong_head_blocks(self):
+        wrong = "a" * 40
+        result = assess_validation_cwd_proof_report(
+            "\n".join([
+                f"Candidate head SHA: {HEAD}",
+                f"pwd: {WORKTREE}",
+                f"git rev-parse HEAD: {wrong}",
+                "git status --short --branch: clean",
+                f"Validation command: cd {WORKTREE} && pytest -q",
+            ]),
+            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
+            project_root=ROOT,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["violations"])
+
+    def test_fully_proof_backed_passes(self):
+        result = assess_validation_cwd_proof_report(
+            _proof_backed_report(),
+            validation_session={"validation_ran": True, "expected_head_sha": HEAD},
+            project_root=ROOT,
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_baseline_without_cwd_fails(self):
+        result = assess_validation_cwd_proof_report(
+            "\n".join([
+                _proof_backed_report(),
+                "Baseline validation command: pytest tests/ -q",
+            ]),
+            validation_session={
+                "validation_ran": True,
+                "expected_head_sha": HEAD,
+                "baseline_validation_ran": True,
+            },
+            project_root=ROOT,
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(
+            any("baseline" in r.lower() for r in result["reasons"])
+        )
+
+    def test_baseline_with_full_proof_passes(self):
+        result = assess_validation_cwd_proof_report(
+            "\n".join([
+                _proof_backed_report(),
+                f"Baseline worktree: {ROOT}/branches/baseline-master-pr376",
+                f"Baseline target SHA: {HEAD}",
+                f"Baseline validation command: cd {ROOT}/branches/baseline-master-pr376 && pytest -q",
+            ]),
+            validation_session={
+                "validation_ran": True,
+                "expected_head_sha": HEAD,
+                "baseline_validation_ran": True,
+            },
+            project_root=ROOT,
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_final_report_validator_integration(self):
+        result = assess_final_report_validator(
+            "Validation command: pytest tests/ -q",
+            "review_pr",
+            validation_session={"validation_ran": True},
+        )
+        self.assertTrue(result["blocked"] or result["downgraded"])
+        self.assertTrue(
+            any(
+                f.get("rule_id") == "reviewer.validation_cwd_proof"
+                for f in result.get("findings") or []
+            )
+        )
+
+    def test_exported_from_review_proofs(self):
+        from review_proofs import assess_validation_cwd_proof_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()