fix: resolve conflicts for PR #422

Extend the review workflow load gate with pre-review command classification,
boundary violation tracking, and structured helper-result requirements in
final reports. Adds gitea_record_pre_review_command MCP tool.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

docs/llm-workflow-runbooks.md

@@ -281,6 +281,21 @@ same-issue/same-operation lease blocks duplicate work. An expired lease still
 blocks takeover until a recovery review records why the prior work is abandoned,
 completed, or unsafe to continue.
 
+**Issue-lock recovery (#447):** Do not manually seed, restore, or delete
+`/tmp/gitea_issue_lock.json` as a normal recovery path. That file is global
+shared state and manual writes can clobber another session's live lease. Use
+`sanctioned recovery` instead:
+
+1. `gitea_lock_issue` on a clean `branches/` worktree (normal path).
+2. Own-branch adoption via #442 when the issue's exact branch is already pushed.
+3. Operator override only when explicitly authorized — record
+   `External-state mutations` and `operator override proof` in the final report.
+
+`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
+approval in one run. See also #438 (global lock redesign).
+
 Remote branches matching the issue number are also treated as active work unless
 the recovery review proves the branch is abandoned or superseded. Never delete
 or clean up a branch when it has an active lease, dirty worktree, open PR, or is

final_report_validator.py

@@ -11,6 +11,7 @@ import inspect
 import re
 from typing import Any, Callable
 
+import issue_lock_provenance
 from review_proofs import (
     HANDOFF_HEADING,
     assess_controller_handoff,
@@ -115,6 +116,22 @@ _TARGET_BRANCH_SHA_RE = re.compile(
     r"target branch sha\s*:\s*[0-9a-f]{40}",
     re.IGNORECASE,
 )
+_WORKFLOW_LOAD_HELPER_RE = re.compile(
+    r"workflow[- ]load helper result\s*:",
+    re.IGNORECASE,
+)
+_WORKFLOW_LOAD_HASH_RE = re.compile(
+    r"workflow[- ]load helper result[\s\S]{0,400}?workflow[_ ]hash\s*:\s*[0-9a-f]{12}",
+    re.IGNORECASE,
+)
+_WORKFLOW_LOAD_BOUNDARY_RE = re.compile(
+    r"workflow[- ]load helper result[\s\S]{0,400}?boundary[_ ]status\s*:\s*(?:clean|violation)",
+    re.IGNORECASE,
+)
+_WORKFLOW_FILE_VIEW_NARRATIVE_RE = re.compile(
+    r"(?:read|viewed|loaded)\s+(?:the\s+)?(?:canonical\s+)?(?:workflow|review-merge-pr\.md)",
+    re.IGNORECASE,
+)
 _FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
 _RECONCILE_STALE_FIELDS = (
     "pr number opened",
@@ -870,6 +887,102 @@ def _rule_reviewer_mutation_ledger(
     )
 
 
+def _rule_shared_issue_lock_external_state(report_text: str) -> list[dict[str, str]]:
+    result = issue_lock_provenance.assess_issue_lock_external_state_report(report_text)
+    if result.get("proven"):
+        return []
+    return _findings_from_reasons(
+        "shared.issue_lock_external_state",
+        result.get("reasons") or [],
+        field="External-state mutations",
+        severity="block",
+        safe_next_action=(
+            "disclose gitea_issue_lock.json read/write/delete under "
+            "External-state mutations; never claim none after lock seeding"
+        ),
+    )
+
+
+def _rule_shared_manual_lock_pr_override(report_text: str) -> list[dict[str, str]]:
+    result = issue_lock_provenance.assess_manual_lock_pr_without_override(report_text)
+    if result.get("proven"):
+        return []
+    return _findings_from_reasons(
+        "shared.manual_lock_pr_override",
+        result.get("reasons") or [],
+        field="External-state mutations",
+        severity="block",
+        safe_next_action=(
+            "use gitea_lock_issue or #442 adoption instead of manual lock seeding; "
+            "if operator override was authorized, cite override proof"
+        ),
+    )
+
+
+def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, str]]:
+    result = issue_lock_provenance.assess_author_reviewer_same_run_report(report_text)
+    if result.get("proven"):
+        return []
+    return _findings_from_reasons(
+        "shared.author_reviewer_same_run",
+        result.get("reasons") or [],
+        field="Review mutations",
+        severity="block",
+        safe_next_action=(
+            "split author PR creation and reviewer approval across separate "
+            "sessions and handoffs"
+        ),
+    )
+
+
+def _rule_reviewer_workflow_load_boundary(report_text: str) -> list[dict[str, str]]:
+    """#403: require structured workflow-load helper result, not file-view narrative."""
+    if not report_text.strip():
+        return []
+    findings: list[dict[str, str]] = []
+    has_helper = bool(_WORKFLOW_LOAD_HELPER_RE.search(report_text))
+    has_hash = bool(_WORKFLOW_LOAD_HASH_RE.search(report_text))
+    has_boundary = bool(_WORKFLOW_LOAD_BOUNDARY_RE.search(report_text))
+    has_narrative_only = bool(_WORKFLOW_FILE_VIEW_NARRATIVE_RE.search(report_text))
+
+    if has_narrative_only and not has_helper:
+        findings.append(validator_finding(
+            "reviewer.workflow_load_boundary",
+            "block",
+            "Workflow-load helper result",
+            (
+                "canonical workflow file-view narrative without structured "
+                "gitea_load_review_workflow helper result"
+            ),
+            (
+                "include Workflow-load helper result with workflow_hash and "
+                "boundary_status from gitea_load_review_workflow"
+            ),
+        ))
+        return findings
+
+    if has_helper and (not has_hash or not has_boundary):
+        missing = []
+        if not has_hash:
+            missing.append("workflow_hash")
+        if not has_boundary:
+            missing.append("boundary_status")
+        findings.append(validator_finding(
+            "reviewer.workflow_load_boundary",
+            "block",
+            "Workflow-load helper result",
+            (
+                "workflow-load helper result incomplete; missing "
+                + ", ".join(missing)
+            ),
+            (
+                "copy workflow_load_helper_result fields from "
+                "gitea_load_review_workflow into the final report"
+            ),
+        ))
+    return findings
+
+
 def _rule_reviewer_review_mutation(
     report_text: str,
     *,
@@ -889,10 +1002,17 @@ def _rule_reviewer_review_mutation(
     )
 
 
+_SHARED_ISSUE_LOCK_RULES = (
+    _rule_shared_issue_lock_external_state,
+    _rule_shared_manual_lock_pr_override,
+    _rule_shared_author_reviewer_same_run,
+)
+
 _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "review_pr": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_legacy_workspace_mutations,
         _rule_reviewer_vague_mutations_none,
         _rule_reviewer_mutation_categories,
@@ -907,12 +1027,14 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_already_landed_eligible,
         _rule_reviewer_already_landed_state,
         _rule_reviewer_target_branch_freshness,
+        _rule_reviewer_workflow_load_boundary,
         _rule_reviewer_mutation_ledger,
         _rule_reviewer_review_mutation,
     ],
     "reconcile_already_landed": [
         _rule_reconcile_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_stale_author_fields,
         _rule_reconcile_eligible_reviewed,
         _rule_reconcile_linked_issue_live,
@@ -924,25 +1046,30 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
     "author_issue": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
     ],
     "work_issue": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_ISSUE_LOCK_RULES,
         _rule_reviewer_vague_mutations_none,
     ],
     "issue_filing": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_ISSUE_LOCK_RULES,
     ],
     "inventory": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_ISSUE_LOCK_RULES,
         _rule_reconcile_pagination_proof,
     ],
     "issue_selection": [
         _rule_shared_controller_handoff,
         _rule_shared_email_disclosure,
+        *_SHARED_ISSUE_LOCK_RULES,
     ],
 }
 

gitea_mcp_server.py

@@ -536,11 +536,15 @@ import role_session_router  # noqa: E402
 import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 import review_proofs  # noqa: E402
+import review_workflow_boundary  # noqa: E402
+import review_workflow_load  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
+import issue_lock_provenance  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
+import issue_work_duplicate_gate  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
@@ -585,7 +589,7 @@ def _load_existing_issue_lock() -> dict | None:
     if not os.path.exists(ISSUE_LOCK_FILE):
         return None
     try:
-        with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
+        with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
             data = json.load(f)
         return data if isinstance(data, dict) else None
     except Exception:
@@ -679,6 +683,119 @@ def _branch_entry_name(branch: dict | str) -> str:
     return str(branch.get("name") or branch.get("ref") or "")
 
 
+def _live_fetch_issue_duplicate_context(
+    h: str,
+    o: str,
+    r: str,
+    auth: str,
+    issue_number: int,
+) -> tuple[list[dict], list[str], dict]:
+    """Live open PRs, remote branch names, and claim state for one issue."""
+    base = repo_api_url(h, o, r)
+    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
+    branches = api_get_all(f"{base}/branches", auth)
+    branch_names = [_branch_entry_name(b) for b in branches]
+    issue = api_request("GET", f"{base}/issues/{issue_number}", auth) or {}
+    comments = api_request(
+        "GET", f"{base}/issues/{issue_number}/comments", auth
+    ) or []
+    claim_entry = issue_claim_heartbeat.classify_issue_claim(
+        issue=issue,
+        comments=comments,
+        open_prs=open_prs,
+        branch_names=branch_names,
+    )
+    return open_prs, branch_names, claim_entry
+
+
+# Injectable duplicate-work context fetcher (#400). Production uses the live
+# Gitea API path above; unit tests patch this symbol instead of hitting the
+# network.
+issue_duplicate_context_fetcher = _live_fetch_issue_duplicate_context
+
+
+def _collect_issue_duplicate_context(
+    h: str,
+    o: str,
+    r: str,
+    auth: str,
+    issue_number: int,
+) -> tuple[list[dict], list[str], dict]:
+    return issue_duplicate_context_fetcher(h, o, r, auth, issue_number)
+
+
+def _assess_issue_duplicate_gate(
+    issue_number: int,
+    *,
+    h: str,
+    o: str,
+    r: str,
+    auth: str,
+    locked_branch: str | None = None,
+    phase: str,
+) -> dict:
+    open_prs, branch_names, claim_entry = _collect_issue_duplicate_context(
+        h, o, r, auth, issue_number
+    )
+    return issue_work_duplicate_gate.assess_work_issue_duplicate_gate(
+        issue_number,
+        open_prs=open_prs,
+        branch_names=branch_names,
+        claim_entry=claim_entry,
+        locked_branch=locked_branch,
+        phase=phase,
+    )
+
+
+def _duplicate_gate_block_response(gate: dict, **extra) -> dict:
+    out = {
+        "success": False,
+        "performed": False,
+        "reasons": list(gate.get("reasons") or []),
+        "duplicate_gate": gate,
+        "safe_next_action": gate.get("safe_next_action"),
+    }
+    out.update(extra)
+    return out
+
+
+def _enforce_locked_issue_duplicate_recheck(
+    remote: str,
+    phase: str,
+    *,
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict | None:
+    """Re-check duplicate-work gates for the locked issue (#400)."""
+    lock_data = _load_existing_issue_lock()
+    if not lock_data:
+        return None
+    issue_number = int(lock_data.get("issue_number") or 0)
+    locked_branch = lock_data.get("branch_name")
+    if not issue_number:
+        return None
+    h, o, r = _resolve(
+        remote or lock_data.get("remote") or "dadeschools",
+        host or lock_data.get("host"),
+        org or lock_data.get("org"),
+        repo or lock_data.get("repo"),
+    )
+    auth = _auth(h)
+    gate = _assess_issue_duplicate_gate(
+        issue_number,
+        h=h,
+        o=o,
+        r=r,
+        auth=auth,
+        locked_branch=locked_branch,
+        phase=phase,
+    )
+    if gate.get("block"):
+        return gate
+    return None
+
+
 def _reveal_endpoints() -> bool:
     """Admin/debug opt-in (#120): include endpoint URLs and token source
     names in tool output. Off by default so normal LLM-facing responses
@@ -1151,48 +1268,21 @@ def gitea_lock_issue(
             issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
         )
 
-    # 2. Check if the issue already has an open PR (reuse protection)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
-    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
+    duplicate_gate = _assess_issue_duplicate_gate(
-
+        issue_number,
-    try:
+        h=h,
-        prs = api_get_all(url, auth)
+        o=o,
-    except Exception as e:
+        r=r,
-        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
+        auth=auth,
-
+        locked_branch=branch_name,
-    for pr in prs:
+        phase=issue_work_duplicate_gate.PHASE_LOCK,
-        pr_head = pr.get("head", {}).get("ref", "")
-        pr_title = pr.get("title", "")
-        pr_body = pr.get("body", "")
-
-        if expected_pattern in pr_head:
-            raise ValueError(
-                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
-            )
-
-        patterns = [
-            f"closes #{issue_number}",
-            f"fixes #{issue_number}",
-        ]
-        text_to_check = f"{pr_title} {pr_body}".lower()
-        if any(p in text_to_check for p in patterns):
-            raise ValueError(
-                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
-            )
-
-    branch_url = f"{repo_api_url(h, o, r)}/branches"
-    try:
-        branches = api_get_all(branch_url, auth)
-    except Exception as e:
-        raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
-    for branch in branches:
-        name = _branch_entry_name(branch)
-        if expected_pattern in name:
-            raise ValueError(
-                f"Issue #{issue_number} already has matching branch '{name}' "
-                "(fail closed)"
     )
+    if duplicate_gate.get("block"):
+        raise ValueError("; ".join(duplicate_gate.get("reasons") or [
+            f"duplicate work gate blocked issue #{issue_number} (fail closed)"
+        ]))
 
     work_lease = _build_author_issue_work_lease(
         issue_number=issue_number,
@@ -1208,6 +1298,10 @@ def gitea_lock_issue(
         "repo": r,
         "worktree_path": resolved_worktree,
         "work_lease": work_lease,
+        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+            tool="gitea_lock_issue",
+            claimant=work_lease.get("claimant"),
+        ),
     }
 
     try:
@@ -1238,6 +1332,39 @@ def gitea_lock_issue(
     return result
 
 
+@mcp.tool()
+def gitea_assess_work_issue_duplicate(
+    issue_number: int,
+    branch_name: str | None = None,
+    phase: str = issue_work_duplicate_gate.PHASE_LOCK,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only duplicate-work gate for author sessions before mutations (#400)."""
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    gate = _assess_issue_duplicate_gate(
+        issue_number,
+        h=h,
+        o=o,
+        r=r,
+        auth=auth,
+        locked_branch=branch_name,
+        phase=phase,
+    )
+    return {"success": not gate.get("block"), **gate}
+
+
 @mcp.tool()
 def gitea_create_pr(
     title: str,
@@ -1298,6 +1425,14 @@ def gitea_create_pr(
     except Exception as e:
         raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
 
+    lock_provenance_check = issue_lock_provenance.assess_lock_file_for_create_pr(
+        lock_data
+    )
+    if lock_provenance_check["block"]:
+        raise RuntimeError(
+            issue_lock_provenance.format_lock_provenance_error(lock_provenance_check)
+        )
+
     locked_issue = lock_data.get("issue_number")
     locked_branch = lock_data.get("branch_name")
     locked_worktree = lock_data.get("worktree_path")
@@ -1329,6 +1464,21 @@ def gitea_create_pr(
             f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
         )
 
+    duplicate_block = _enforce_locked_issue_duplicate_recheck(
+        remote,
+        issue_work_duplicate_gate.PHASE_CREATE_PR,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    if duplicate_block:
+        return _duplicate_gate_block_response(
+            duplicate_block,
+            number=None,
+            issue_number=locked_issue,
+            branch_name=locked_branch,
+        )
+
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls"
     payload = {"title": title, "body": body, "head": head, "base": base}
@@ -1862,6 +2012,7 @@ def init_review_decision_lock(remote: str | None, task: str | None):
     """Seed read-only-until-ready state for reviewer PR review tasks."""
     if task != "review_pr":
         return
+    review_workflow_load.clear_review_workflow_load()
     profile = get_profile()
     profile_name = (profile.get("profile_name") or "").strip()
     session_lock = (
@@ -1887,6 +2038,11 @@ def init_review_decision_lock(remote: str | None, task: str | None):
     })
 
 
+def _review_workflow_load_gate_reasons() -> list[str]:
+    """Fail closed when canonical review workflow was not loaded (#389)."""
+    return review_workflow_load.review_workflow_load_blockers(PROJECT_ROOT)
+
+
 def check_review_decision_gate(
     pr_number: int,
     action: str,
@@ -1897,7 +2053,10 @@ def check_review_decision_gate(
     repo: str | None = None,
 ) -> list[str]:
     """Fail closed unless validation completed and the final decision is ready."""
-    reasons = []
+    reasons = list(_review_workflow_load_gate_reasons())
+    if reasons:
+        reasons.extend(review_workflow_load.recovery_handoff_without_replay())
+        return reasons
     lock = _load_review_decision_lock()
     if lock is None:
         reasons.append(
@@ -2253,6 +2412,7 @@ def _evaluate_pr_review_submission(
     """Shared gate chain for live submit and dry-run review tools."""
     verify_preflight_purity(remote)
     action = (action or "").strip().lower()
+    workflow_blockers = _review_workflow_load_gate_reasons() if live else []
     result = {
         "requested_action": action,
         "performed": False,
@@ -2268,6 +2428,10 @@ def _evaluate_pr_review_submission(
         "reasons": [],
     }
     reasons = result["reasons"]
+    if workflow_blockers:
+        reasons.extend(workflow_blockers)
+        reasons.extend(review_workflow_load.recovery_handoff_without_replay())
+        return result
 
     if action not in _REVIEW_ACTIONS:
         reasons.append(
@@ -2452,6 +2616,13 @@ def gitea_mark_final_review_decision(
         }
     org = resolved_org
     repo = resolved_repo
+    workflow_blockers = _review_workflow_load_gate_reasons()
+    if workflow_blockers:
+        return {
+            "marked_ready": False,
+            "reasons": workflow_blockers + (
+                review_workflow_load.recovery_handoff_without_replay()),
+        }
     hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready")
     if hard_stop:
         return {"marked_ready": False, "reasons": hard_stop}
@@ -2933,6 +3104,20 @@ def gitea_commit_files(
     if blocked:
         return blocked
 
+    duplicate_block = _enforce_locked_issue_duplicate_recheck(
+        remote,
+        issue_work_duplicate_gate.PHASE_COMMIT,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    if duplicate_block:
+        return _duplicate_gate_block_response(
+            duplicate_block,
+            commit="",
+            branch="",
+        )
+
     verify_preflight_purity(remote)
     processed_files, source_proofs = _prepare_commit_payload_files(files)
 
@@ -3035,6 +3220,7 @@ def gitea_merge_pr(
         available. Never secrets.
     """
     verify_preflight_purity(remote)
+    workflow_blockers = _review_workflow_load_gate_reasons()
     do = (do or "").strip().lower()
     result = {
         "performed": False,
@@ -3052,6 +3238,10 @@ def gitea_merge_pr(
         "reasons": [],
     }
     reasons = result["reasons"]
+    if workflow_blockers:
+        reasons.extend(workflow_blockers)
+        reasons.extend(review_workflow_load.recovery_handoff_without_replay())
+        return result
 
     # Gate 1 — valid merge method (no API call on a bad method).
     if do not in _MERGE_METHODS:
@@ -4788,6 +4978,8 @@ _PROJECT_SKILLS = {
         "steps": [
             "Resolve task first: gitea_resolve_task_capability(task='review_pr') "
             "to confirm reviewer namespace and avoid author-profile blocks.",
+            "Load canonical workflow proof with gitea_load_review_workflow "
+            "before any review/merge mutation (#389).",
             "Verify reviewer identity with gitea_whoami; the PR author "
             "must be a different user.",
             "Reconcile live queue state FIRST (do not trust prior handoffs): "
@@ -5711,6 +5903,8 @@ def gitea_get_runtime_context(
         ),
         "role_kind": _role_kind(allowed, forbidden),
         "shell_health": native_mcp_preference.shell_health_status(),
+        "workflow_load_proof": review_workflow_load.workflow_load_status(
+            PROJECT_ROOT),
     }
 
     if reveal and h:
@@ -5719,6 +5913,80 @@ def gitea_get_runtime_context(
     return result
 
 
+@mcp.tool()
+def gitea_record_pre_review_command(
+    command: str,
+    cwd: str | None = None,
+    classification: str | None = None,
+) -> dict:
+    """Classify and record a command executed before workflow load (#403).
+
+    Read-only with respect to Gitea API. Pre-review inventory/diagnostic commands
+    may be recorded as allowed; boundary violations block reviewer mutations.
+    """
+    recorded = review_workflow_boundary.record_pre_review_command(
+        command,
+        cwd=cwd,
+        project_root=PROJECT_ROOT,
+        classification=classification,
+    )
+    boundary_state = review_workflow_boundary.assess_boundary_status(PROJECT_ROOT)
+    return {
+        "success": True,
+        "recorded": recorded,
+        "boundary_status": boundary_state.get("boundary_status"),
+        "boundary_clean": boundary_state.get("boundary_clean"),
+        "reasons": list(boundary_state.get("reasons") or []),
+    }
+
+
+@mcp.tool()
+def gitea_load_review_workflow(
+    prompt_text: str | None = None,
+) -> dict:
+    """Load and record canonical review-merge workflow proof for this session (#389, #403).
+
+    Read-only with respect to Gitea API; records in-process workflow source/hash
+    proof and session boundary state required before reviewer review or merge
+    mutations.
+    """
+    try:
+        recorded = review_workflow_load.record_review_workflow_load(
+            PROJECT_ROOT, prompt_text=prompt_text)
+    except OSError as exc:
+        return {
+            "success": False,
+            "loaded": False,
+            "reasons": [str(exc)],
+            "recovery_handoff": review_workflow_load.recovery_handoff_without_replay(),
+        }
+    boundary_reasons = review_workflow_boundary.boundary_blockers(PROJECT_ROOT)
+    helper = review_workflow_boundary.workflow_load_helper_result(
+        recorded, PROJECT_ROOT)
+    return {
+        "success": not boundary_reasons,
+        "loaded": True,
+        "workflow_source": recorded["workflow_source"],
+        "task_mode": recorded["task_mode"],
+        "workflow_hash": recorded["workflow_hash"],
+        "workflow_version": recorded["workflow_version"],
+        "final_report_schema_path": recorded["final_report_schema_path"],
+        "final_report_schema_hash": recorded["final_report_schema_hash"],
+        "prompt_conflicts_with_workflow": recorded[
+            "prompt_conflicts_with_workflow"],
+        "prompt_conflict_reasons": recorded.get("prompt_conflict_reasons") or [],
+        "workflow_load_proof_present": True,
+        "boundary_status": recorded.get("boundary_status"),
+        "boundary_clean": recorded.get("boundary_clean"),
+        "workflow_load_helper_result": helper,
+        "reasons": boundary_reasons,
+        "recovery_handoff": (
+            review_workflow_load.recovery_handoff_without_replay()
+            if boundary_reasons else []
+        ),
+    }
+
+
 @mcp.tool()
 def gitea_list_profiles() -> dict:
     """Read-only: list all Gitea MCP profiles with redacted metadata.
@@ -6774,6 +7042,16 @@ def gitea_resolve_task_capability(
     }
     if reason_msg:
         result["reason"] = reason_msg
+    if task in ("review_pr", "merge_pr"):
+        result["workflow_load_proof"] = review_workflow_load.workflow_load_status(
+            PROJECT_ROOT)
+        if not result["workflow_load_proof"].get("workflow_load_valid"):
+            guidance = (
+                "Call gitea_load_review_workflow before any reviewer review "
+                "or merge mutation."
+            )
+            if guidance not in task_role_guidance:
+                task_role_guidance.append(guidance)
     role_session_router.sync_route_from_capability(result)
     was_terminal = capability_stop_terminal.is_active()
     terminal = capability_stop_terminal.sync_from_capability_result(result)

issue_lock_provenance.py

@@ -0,0 +1,269 @@
+"""Issue-lock provenance and external-state disclosure (#447).
+
+Sanctioned locks are written only by ``gitea_lock_issue`` (or adoption recovery
+#442). Manual seeding of ``/tmp/gitea_issue_lock.json`` is unsafe and must be
+blocked at PR creation unless explicit operator override proof is recorded.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from datetime import datetime, timezone
+
+ISSUE_LOCK_FILE = os.environ.get("GITEA_ISSUE_LOCK_FILE", "/tmp/gitea_issue_lock.json")
+
+SOURCE_LOCK_ISSUE = "gitea_lock_issue"
+SOURCE_LOCK_ADOPTION = "gitea_lock_issue_adoption"
+SOURCE_OPERATOR_OVERRIDE = "operator_override"
+
+SANCTIONED_LOCK_SOURCES = frozenset({
+    SOURCE_LOCK_ISSUE,
+    SOURCE_LOCK_ADOPTION,
+    SOURCE_OPERATOR_OVERRIDE,
+})
+
+_OPERATOR_OVERRIDE_ENV = "GITEA_ISSUE_LOCK_OPERATOR_OVERRIDE"
+
+_ISSUE_LOCK_PATH_RE = re.compile(
+    r"(?:/tmp/)?gitea_issue_lock\.json",
+    re.IGNORECASE,
+)
+_LOCK_SEED_RE = re.compile(
+    r"(?:seed(?:ed|ing)?|restor(?:e|ed|ing)|wrote|written|write|programmatically|"
+    r"hand[- ]forg|manual(?:ly)?).{0,80}gitea_issue_lock",
+    re.IGNORECASE | re.DOTALL,
+)
+_LOCK_REMOVE_RE = re.compile(
+    r"(?:\brm\b|remove|deleted?|unlink).{0,80}gitea_issue_lock",
+    re.IGNORECASE | re.DOTALL,
+)
+_LOCK_READ_RE = re.compile(
+    r"(?:read|loaded?|parsed?).{0,80}gitea_issue_lock",
+    re.IGNORECASE | re.DOTALL,
+)
+_EXTERNAL_NONE_RE = re.compile(
+    r"external[- ]state mutations\s*:\s*none\b",
+    re.IGNORECASE,
+)
+_EXTERNAL_FIELD_RE = re.compile(
+    r"external[- ]state mutations\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_CLEANUP_ONLY_RE = re.compile(
+    r"cleanup mutations\s*:\s*(?:none|lock removed|removed issue lock)",
+    re.IGNORECASE,
+)
+_PR_CREATED_RE = re.compile(
+    r"(?:\bgitea_create_pr\b|PR\s*#\s*\d+\s+created|created\s+PR\s*#|opened\s+PR\s*#|"
+    r"PR\s+creation\s+(?:succeeded|complete))",
+    re.IGNORECASE,
+)
+_REVIEW_APPROVE_RE = re.compile(
+    r"(?:submitted\s+(?:['\"]approve['\"]|approve\s+review)|"
+    r"review decision\s*:\s*approve|approved\s+PR\s*#|gitea_review_pr.*approve)",
+    re.IGNORECASE,
+)
+_OVERRIDE_PROOF_RE = re.compile(
+    r"operator[- ]override\s+proof\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def build_sanctioned_lock_provenance(
+    *,
+    tool: str,
+    source: str = SOURCE_LOCK_ISSUE,
+    claimant: dict | None = None,
+    adoption: dict | None = None,
+) -> dict:
+    """Return provenance metadata stored with a sanctioned lock write."""
+    entry = {
+        "source": source,
+        "written_at": _utc_now_iso(),
+        "written_by_tool": tool,
+        "lock_file_path": ISSUE_LOCK_FILE,
+    }
+    if claimant:
+        entry["claimant"] = claimant
+    if adoption:
+        entry["adoption"] = adoption
+    return entry
+
+
+def operator_override_requested() -> bool:
+    return os.environ.get(_OPERATOR_OVERRIDE_ENV, "").strip().lower() in {
+        "1",
+        "true",
+        "yes",
+    }
+
+
+def build_operator_override_provenance(*, reason: str, claimant: dict | None = None) -> dict:
+    text = (reason or "").strip()
+    if not text:
+        raise ValueError(
+            "operator override requires a non-empty override reason (fail closed)"
+        )
+    entry = build_sanctioned_lock_provenance(
+        tool="operator_override",
+        source=SOURCE_OPERATOR_OVERRIDE,
+        claimant=claimant,
+    )
+    entry["override_reason"] = text
+    return entry
+
+
+def assess_lock_file_for_create_pr(lock_data: dict | None) -> dict:
+    """Fail closed when lock file lacks sanctioned provenance (#447)."""
+    data = lock_data if isinstance(lock_data, dict) else {}
+    reasons: list[str] = []
+    provenance = data.get("lock_provenance")
+    if not isinstance(provenance, dict):
+        reasons.append(
+            "issue lock file lacks sanctioned lock_provenance; manual seeding is "
+            "not a normal recovery path — call gitea_lock_issue or use #442 adoption"
+        )
+        return _provenance_result(False, reasons, provenance)
+
+    source = str(provenance.get("source") or "").strip()
+    if source not in SANCTIONED_LOCK_SOURCES:
+        reasons.append(
+            f"issue lock provenance source '{source or '(missing)'}' is not sanctioned"
+        )
+
+    if source == SOURCE_OPERATOR_OVERRIDE and not str(
+        provenance.get("override_reason") or ""
+    ).strip():
+        reasons.append(
+            "operator_override lock provenance requires override_reason proof"
+        )
+
+    if not data.get("work_lease"):
+        reasons.append("issue lock file missing work_lease metadata")
+
+    if not str(provenance.get("written_by_tool") or "").strip():
+        reasons.append("issue lock provenance missing written_by_tool")
+
+    proven = not reasons
+    return _provenance_result(proven, reasons, provenance)
+
+
+def _provenance_result(proven: bool, reasons: list[str], provenance: dict | None) -> dict:
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "lock_provenance": provenance,
+    }
+
+
+def format_lock_provenance_error(assessment: dict) -> str:
+    reasons = "; ".join(assessment.get("reasons") or ["unknown lock provenance violation"])
+    return f"Issue lock provenance guard (#447): {reasons} (fail closed)"
+
+
+def _lock_activity_detected(text: str) -> dict[str, bool]:
+    body = text or ""
+    return {
+        "seed_or_restore": bool(_LOCK_SEED_RE.search(body)),
+        "remove": bool(_LOCK_REMOVE_RE.search(body)),
+        "read": bool(_LOCK_READ_RE.search(body)),
+    }
+
+
+def _external_state_discloses_lock(text: str) -> bool:
+    match = _EXTERNAL_FIELD_RE.search(text or "")
+    if not match:
+        return False
+    value = (match.group(1) or "").strip().lower()
+    if value in {"", "none", "n/a"}:
+        return False
+    return "lock" in value or "gitea_issue_lock" in value or "issue-lock" in value
+
+
+def assess_issue_lock_external_state_report(report_text: str) -> dict:
+    """Require explicit external-state disclosure for issue-lock mutations (#447)."""
+    text = report_text or ""
+    activity = _lock_activity_detected(text)
+    if not any(activity.values()):
+        return {"proven": True, "block": False, "reasons": [], "activity": activity}
+
+    reasons: list[str] = []
+    disclosed = _external_state_discloses_lock(text)
+
+    if activity["seed_or_restore"] and _EXTERNAL_NONE_RE.search(text):
+        reasons.append(
+            "report mentions seeding/restoring gitea_issue_lock.json but claims "
+            "External-state mutations: none"
+        )
+    elif activity["seed_or_restore"] and not disclosed:
+        reasons.append(
+            "report mentions issue-lock file activity but External-state mutations "
+            "does not disclose read/write of gitea_issue_lock.json"
+        )
+
+    if activity["remove"]:
+        if _EXTERNAL_NONE_RE.search(text):
+            reasons.append(
+                "report mentions removing gitea_issue_lock.json but claims "
+                "External-state mutations: none"
+            )
+        elif not disclosed and _CLEANUP_ONLY_RE.search(text):
+            reasons.append(
+                "report removes issue lock but classifies it as cleanup only; "
+                "record under External-state mutations"
+            )
+        elif not disclosed:
+            reasons.append(
+                "report mentions deleting issue lock without External-state "
+                "mutation disclosure"
+            )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "activity": activity,
+    }
+
+
+def assess_manual_lock_pr_without_override(report_text: str) -> dict:
+    """Block reports that created a PR via manual lock seed without override proof."""
+    text = report_text or ""
+    seeded = bool(_LOCK_SEED_RE.search(text))
+    created = bool(_PR_CREATED_RE.search(text))
+    if not (seeded and created):
+        return {"proven": True, "block": False, "reasons": []}
+
+    if _OVERRIDE_PROOF_RE.search(text):
+        return {"proven": True, "block": False, "reasons": []}
+
+    return {
+        "proven": False,
+        "block": True,
+        "reasons": [
+            "report created/opened a PR after manual issue-lock seeding without "
+            "operator override proof"
+        ],
+    }
+
+
+def assess_author_reviewer_same_run_report(report_text: str) -> dict:
+    """Reviewer handoff must not create and approve the same PR in one run (#447)."""
+    text = report_text or ""
+    if not (_PR_CREATED_RE.search(text) and _REVIEW_APPROVE_RE.search(text)):
+        return {"proven": True, "block": False, "reasons": []}
+    return {
+        "proven": False,
+        "block": True,
+        "reasons": [
+            "report mixes author-side PR creation and reviewer approval in one "
+            "final handoff; split author and reviewer sessions"
+        ],
+    }

issue_work_duplicate_gate.py

@@ -0,0 +1,180 @@
+"""Early duplicate-work detection for author work-issue sessions (#400)."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import issue_claim_heartbeat as claim_hb
+
+PHASE_LOCK = "lock_issue"
+PHASE_COMMIT = "commit"
+PHASE_PUSH = "push"
+PHASE_CREATE_PR = "create_pr"
+
+OUTCOME_DUPLICATE_PR_PREVENTED = "duplicate_pr_prevented"
+OUTCOME_DUPLICATE_BRANCH_PREVENTED = "duplicate_branch_prevented"
+OUTCOME_DUPLICATE_COMMIT_PREVENTED = "duplicate_commit_prevented"
+OUTCOME_DUPLICATE_WORK_NOT_PREVENTED = "duplicate_work_not_prevented"
+
+_ACTIVE_CLAIM_STATUSES = frozenset({"active", "awaiting_review"})
+
+
+def _issue_pattern(issue_number: int) -> str:
+    return f"issue-{int(issue_number)}"
+
+
+def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None:
+    return claim_hb._linked_open_pr(issue_number, open_prs)
+
+
+def _matching_branches(
+    issue_number: int,
+    branch_names: list[str],
+    *,
+    locked_branch: str | None = None,
+) -> list[str]:
+    pattern = _issue_pattern(issue_number)
+    matches = [
+        name for name in (branch_names or [])
+        if pattern in (name or "").lower()
+    ]
+    if locked_branch:
+        locked = locked_branch.strip()
+        matches = [name for name in matches if name != locked]
+    return matches
+
+
+def assess_work_issue_duplicate_gate(
+    issue_number: int,
+    *,
+    open_prs: list[dict] | None = None,
+    branch_names: list[str] | None = None,
+    claim_entry: dict | None = None,
+    locked_branch: str | None = None,
+    phase: str = PHASE_LOCK,
+) -> dict[str, Any]:
+    """Fail closed when duplicate work is already in flight for an issue."""
+    reasons: list[str] = []
+    outcome = OUTCOME_DUPLICATE_WORK_NOT_PREVENTED
+    prs = list(open_prs or [])
+    branches = list(branch_names or [])
+    pattern = _issue_pattern(issue_number)
+
+    linked = _linked_open_pr(issue_number, prs)
+    if linked:
+        reasons.append(
+            f"open PR #{linked.get('number')} already covers issue "
+            f"#{issue_number} (fail closed)"
+        )
+        outcome = OUTCOME_DUPLICATE_PR_PREVENTED
+
+    conflicting_branches = _matching_branches(
+        issue_number, branches, locked_branch=locked_branch
+    )
+    if conflicting_branches:
+        names = ", ".join(conflicting_branches[:5])
+        reasons.append(
+            f"remote branch(es) already match issue pattern '{pattern}': "
+            f"{names} (fail closed)"
+        )
+        if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
+            outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED
+
+    entry = claim_entry or {}
+    if entry.get("linked_open_pr") and not linked:
+        reasons.append(
+            f"claim inventory reports open PR #{entry['linked_open_pr']} "
+            f"for issue #{issue_number} (fail closed)"
+        )
+        outcome = OUTCOME_DUPLICATE_PR_PREVENTED
+
+    status = (entry.get("status") or "").strip().lower()
+    if status in _ACTIVE_CLAIM_STATUSES and not linked:
+        heartbeat = entry.get("latest_heartbeat") or {}
+        claim_branch = (heartbeat.get("branch") or "").strip()
+        if locked_branch and claim_branch and claim_branch != locked_branch:
+            reasons.append(
+                f"active claim lease on branch '{claim_branch}' blocks "
+                f"work on '{locked_branch}' for issue #{issue_number} "
+                "(fail closed)"
+            )
+            if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
+                outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED
+        elif not locked_branch and status == "active":
+            reasons.append(
+                f"issue #{issue_number} has an active claim lease "
+                "(fail closed)"
+            )
+            if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
+                outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED
+
+    if phase in {PHASE_COMMIT, PHASE_PUSH} and reasons:
+        if outcome == OUTCOME_DUPLICATE_PR_PREVENTED:
+            outcome = OUTCOME_DUPLICATE_COMMIT_PREVENTED
+        elif outcome == OUTCOME_DUPLICATE_BRANCH_PREVENTED:
+            outcome = OUTCOME_DUPLICATE_COMMIT_PREVENTED
+
+    block = bool(reasons)
+    return {
+        "block": block,
+        "performed": not block,
+        "issue_number": issue_number,
+        "phase": phase,
+        "outcome": outcome,
+        "linked_open_pr": linked.get("number") if linked else entry.get("linked_open_pr"),
+        "conflicting_branches": conflicting_branches,
+        "claim_status": status or None,
+        "reasons": reasons,
+        "safe_next_action": (
+            "stop before mutating; preserve local work and produce a "
+            "reconciliation handoff if a concurrent PR appeared after push"
+            if block and phase == PHASE_CREATE_PR
+            else "stop before mutating; do not commit or push duplicate work"
+            if block
+            else "proceed"
+        ),
+    }
+
+
+def assess_work_issue_duplicate_report(report_text: str) -> dict[str, Any]:
+    """Require explicit duplicate-work outcome wording in work-issue reports."""
+    text = (report_text or "").lower()
+    markers = {
+        OUTCOME_DUPLICATE_PR_PREVENTED: (
+            "duplicate pr prevented",
+            "duplicate_pr_prevented",
+        ),
+        OUTCOME_DUPLICATE_BRANCH_PREVENTED: (
+            "duplicate branch prevented",
+            "duplicate_branch_prevented",
+        ),
+        OUTCOME_DUPLICATE_COMMIT_PREVENTED: (
+            "duplicate commit prevented",
+            "duplicate_commit_prevented",
+        ),
+        OUTCOME_DUPLICATE_WORK_NOT_PREVENTED: (
+            "duplicate work not prevented",
+            "duplicate_work_not_prevented",
+            "no duplicate work",
+        ),
+    }
+    matched = [
+        key for key, phrases in markers.items()
+        if any(phrase in text for phrase in phrases)
+    ]
+    if len(matched) != 1:
+        return {
+            "complete": False,
+            "downgraded": True,
+            "reasons": [
+                "work-issue report must state exactly one duplicate-work "
+                "outcome (duplicate PR/branch/commit prevented, or "
+                "duplicate work not prevented)"
+            ],
+        }
+    return {
+        "complete": True,
+        "downgraded": False,
+        "outcome": matched[0],
+        "reasons": [],
+    }

review_proofs.py

@@ -3624,9 +3624,12 @@ def assess_work_issue_mode_isolation(report_text: str) -> dict:
 
 def assess_work_issue_final_report(report_text: str) -> dict:
     """#139: composite verifier for work-issue final reports."""
+    from issue_work_duplicate_gate import assess_work_issue_duplicate_report
+
     checks = {
         "workflow_source": assess_work_issue_workflow_source(report_text),
         "mode_isolation": assess_work_issue_mode_isolation(report_text),
+        "duplicate_work_outcome": assess_work_issue_duplicate_report(report_text),
     }
 
     reasons = []

review_workflow_boundary.py

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

review_workflow_load.py

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

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

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

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

@@ -36,6 +36,44 @@ If available, load it first and report:
 
 If the canonical workflow cannot be loaded and the project requires it, stop and produce a recovery handoff only.
 
+## 0A. Workflow-load and session boundary anchor (#403)
+
+The MCP gate is the authority — not local file viewing.
+
+Before any reviewer mutation:
+
+1. Record pre-review commands with `gitea_record_pre_review_command` when they
+   are not automatically classified (inventory/diagnostic commands may be
+   recorded explicitly for proof).
+2. Call `gitea_load_review_workflow` to establish workflow hash proof **and**
+   session boundary state in the same in-process session proof.
+3. Do not claim the workflow was loaded by reading
+   `skills/llm-project-workflow/workflows/review-merge-pr.md` as a local file;
+   that narrative does not satisfy the validator.
+
+Allowed before workflow load (classify as `read_only_inventory` or
+`diagnostic`):
+
+* `gitea_whoami`, `gitea_resolve_task_capability`, `gitea_list_prs`,
+  `gitea_view_pr`, `gitea_get_runtime_context`
+* `git fetch` / `git remote update` for inventory
+* `git status`, `git worktree list` (read-only)
+
+Boundary violations (block downstream reviewer mutations even after load):
+
+* validation commands (`pytest`, `python -m unittest`) in the main checkout
+* local profile/credential/config inspection (`profiles.json`, `gitea-mcp.json`,
+  `.env`, keychain dumps)
+* MCP repair (`pkill`, MCP config edits)
+* git mutations before workflow load
+
+Final reports must include a structured **Workflow-load helper result** block
+copied from `gitea_load_review_workflow`, including at minimum:
+
+* `workflow_hash`
+* `final_report_schema_hash`
+* `boundary_status` (`clean` or `violation`)
+
 ## 1. Start with live identity, profile, runtime, and capability checks
 
 Prove:

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

@@ -297,6 +297,36 @@ Report:
 
 Do not create another branch/PR for the same issue unless the project explicitly allows taking over or updating existing work and exact capability is proven.
 
+### 10A. Duplicate-work gate phases (#400)
+
+Before any file edits, prove duplicate-work clearance with
+`gitea_assess_work_issue_duplicate` or `gitea_lock_issue` (which runs the same
+gate). The gate checks live:
+
+* open PRs linked to the issue (head branch or Closes/Fixes reference),
+* remote branches matching `issue-<number>`,
+* active claim leases from structured heartbeats.
+
+Re-check immediately before:
+
+* `gitea_commit_files` (commit),
+* branch push,
+* `gitea_create_pr` (PR creation).
+
+If a concurrent open PR appears after work begins:
+
+* before commit/push → stop and preserve local work without pushing,
+* after commit but before push → stop without pushing,
+* after push but before PR creation → stop and produce a reconciliation
+  handoff instead of opening a PR.
+
+Final reports must state exactly one duplicate-work outcome:
+
+* `duplicate PR prevented`
+* `duplicate branch prevented`
+* `duplicate commit prevented`
+* `duplicate work not prevented`
+
 ## 11. Claim or lock the issue before implementation
 
 Claim/lock the issue before implementation if the project provides a claim/lock mechanism.
@@ -717,6 +747,12 @@ Use only precise categories:
 * External-state mutations:
 * Read-only diagnostics:
 
+Issue-lock file (`/tmp/gitea_issue_lock.json`) read/write/delete is always an
+external-state mutation. Never claim `External-state mutations: none` after
+seeding, restoring, or removing that file. Manual lock seeding is not a normal
+recovery path (#447); use `gitea_lock_issue` or the #442 adoption recovery path
+instead. Link broader redesign: #438.
+
 `git fetch`, `git remote update`, and any command that updates refs must be listed under `Git ref mutations`, not read-only diagnostics.
 
 If `git reset --hard`, checkout, clean, worktree add/remove, merge simulation, merge abort, or similar commands occurred, report them under `Worktree/index mutations`.

tests/test_agent_temp_artifacts.py

@@ -80,7 +80,10 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
     def tearDown(self):
         self._env_patcher.stop()
 
-    @patch("mcp_server.api_get_all", return_value=[])
+    @patch(
+        "mcp_server.issue_duplicate_context_fetcher",
+        return_value=([], [], {"status": "not_claimed"}),
+    )
     @patch("mcp_server._auth", return_value="token x")
     @patch("mcp_server._resolve", return_value=("h", "o", "r"))
     @patch("mcp_server.ISSUE_LOCK_FILE", new_callable=lambda: tempfile.mktemp())

tests/test_commit_files_capability.py

@@ -76,7 +76,14 @@ class CommitFilesCapabilityBase(unittest.TestCase):
         with open(self.config_path, "w", encoding="utf-8") as fh:
             fh.write(json.dumps(CONFIG))
 
+        self._dup_fetcher_patcher = patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            return_value=([], [], {"status": "not_claimed"}),
+        )
+        self._dup_fetcher_patcher.start()
+
     def tearDown(self):
+        self._dup_fetcher_patcher.stop()
         self._remotes.stop()
         mcp_server._IDENTITY_CACHE.clear()
         mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (

tests/test_commit_payloads.py

@@ -67,6 +67,15 @@ class TestCommitPayloads(unittest.TestCase):
         self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
 
         self.lock_file_path = "/tmp/gitea_issue_lock.json"
+        import issue_lock_provenance
+
+        work_lease = {
+            "operation_type": "author_issue_work",
+            "issue_number": 263,
+            "branch": "feat/issue-263-native-commit-payloads",
+            "claimant": {"username": "test-user", "profile": "test-author"},
+            "expires_at": "2999-01-01T00:00:00Z",
+        }
         self.lock_data = {
             "issue_number": 263,
             "branch_name": "feat/issue-263-native-commit-payloads",
@@ -74,6 +83,11 @@ class TestCommitPayloads(unittest.TestCase):
             "org": "Example-Org",
             "repo": "Example-Repo",
             "worktree_path": self.locked_worktree_path,
+            "work_lease": work_lease,
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue",
+                claimant=work_lease.get("claimant"),
+            ),
         }
         with open(self.lock_file_path, "w", encoding="utf-8") as fh:
             fh.write(json.dumps(self.lock_data))
@@ -84,7 +98,14 @@ class TestCommitPayloads(unittest.TestCase):
         mcp_server._preflight_whoami_called = True
         mcp_server._preflight_capability_called = True
 
+        self._dup_fetcher_patcher = patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            return_value=([], [], {"status": "not_claimed"}),
+        )
+        self._dup_fetcher_patcher.start()
+
     def tearDown(self):
+        self._dup_fetcher_patcher.stop()
         self._remotes.stop()
         mcp_server._IDENTITY_CACHE.clear()
 

tests/test_issue_lock_provenance.py

@@ -0,0 +1,130 @@
+"""Tests for issue-lock provenance and external-state disclosure (#447)."""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_lock_provenance as ilp  # noqa: E402
+from final_report_validator import assess_final_report_validator  # noqa: E402
+
+
+def _sanctioned_lock(**overrides):
+    work_lease = {
+        "operation_type": "author_issue_work",
+        "issue_number": 447,
+        "branch": "feat/issue-447-lock-provenance",
+        "claimant": {"username": "jcwalker3", "profile": "prgs-author"},
+        "expires_at": "2999-01-01T00:00:00Z",
+    }
+    data = {
+        "issue_number": 447,
+        "branch_name": "feat/issue-447-lock-provenance",
+        "work_lease": work_lease,
+        "lock_provenance": ilp.build_sanctioned_lock_provenance(
+            tool="gitea_lock_issue",
+            claimant=work_lease["claimant"],
+        ),
+    }
+    data.update(overrides)
+    return data
+
+
+class TestLockProvenanceForCreatePr(unittest.TestCase):
+    def test_sanctioned_lock_passes(self):
+        result = ilp.assess_lock_file_for_create_pr(_sanctioned_lock())
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_manual_seed_without_provenance_blocked(self):
+        result = ilp.assess_lock_file_for_create_pr(
+            {"issue_number": 420, "branch_name": "feat/x", "work_lease": {}}
+        )
+        self.assertTrue(result["block"])
+        self.assertIn("lock_provenance", result["reasons"][0])
+
+    def test_operator_override_requires_reason(self):
+        result = ilp.assess_lock_file_for_create_pr(
+            _sanctioned_lock(
+                lock_provenance=ilp.build_sanctioned_lock_provenance(
+                    tool="operator_override",
+                    source=ilp.SOURCE_OPERATOR_OVERRIDE,
+                )
+            )
+        )
+        self.assertTrue(result["block"])
+
+
+class TestExternalStateReportRules(unittest.TestCase):
+    def test_seed_with_external_none_blocked(self):
+        report = (
+            "Restored /tmp/gitea_issue_lock.json to unblock PR creation.\n"
+            "- External-state mutations: none\n"
+        )
+        result = ilp.assess_issue_lock_external_state_report(report)
+        self.assertTrue(result["block"])
+
+    def test_seed_with_disclosure_passes(self):
+        report = (
+            "Restored /tmp/gitea_issue_lock.json after MCP restart.\n"
+            "- External-state mutations: wrote /tmp/gitea_issue_lock.json\n"
+        )
+        result = ilp.assess_issue_lock_external_state_report(report)
+        self.assertTrue(result["proven"])
+
+    def test_remove_claimed_as_cleanup_only_blocked(self):
+        report = (
+            "rm /tmp/gitea_issue_lock.json after PR creation.\n"
+            "- Cleanup mutations: lock removed\n"
+            "- External-state mutations: none\n"
+        )
+        result = ilp.assess_issue_lock_external_state_report(report)
+        self.assertTrue(result["block"])
+
+    def test_manual_lock_pr_without_override_blocked(self):
+        report = (
+            "Programmatically seeded gitea_issue_lock.json then gitea_create_pr.\n"
+            "PR #444 created.\n"
+        )
+        result = ilp.assess_manual_lock_pr_without_override(report)
+        self.assertTrue(result["block"])
+
+    def test_author_reviewer_same_run_blocked(self):
+        report = (
+            "gitea_create_pr opened PR #444.\n"
+            "Submitted approve review on PR #444.\n"
+        )
+        result = ilp.assess_author_reviewer_same_run_report(report)
+        self.assertTrue(result["block"])
+
+
+class TestFinalReportValidatorIntegration(unittest.TestCase):
+    def test_work_issue_blocks_hidden_lock_mutation(self):
+        report = (
+            "## Controller Handoff\n"
+            "- Task: work issue #420\n"
+            "- External-state mutations: none\n"
+            "Restored /tmp/gitea_issue_lock.json before PR creation.\n"
+        )
+        result = assess_final_report_validator(report, "work_issue")
+        rule_ids = {f["rule_id"] for f in result["findings"]}
+        self.assertIn("shared.issue_lock_external_state", rule_ids)
+        self.assertTrue(result["blocked"])
+
+    def test_review_pr_blocks_create_and_approve(self):
+        report = (
+            "## Controller Handoff\n"
+            "- Task: review PR #444\n"
+            "- Review decision: approve\n"
+            "Created PR #444 via gitea_create_pr earlier in this run.\n"
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        rule_ids = {f["rule_id"] for f in result["findings"]}
+        self.assertIn("shared.author_reviewer_same_run", rule_ids)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_work_duplicate_gate.py

@@ -0,0 +1,251 @@
+"""Tests for early duplicate-work detection (#400)."""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_work_duplicate_gate as dup_gate
+import mcp_server
+from issue_work_duplicate_gate import (
+    OUTCOME_DUPLICATE_BRANCH_PREVENTED,
+    OUTCOME_DUPLICATE_COMMIT_PREVENTED,
+    OUTCOME_DUPLICATE_PR_PREVENTED,
+    OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
+    PHASE_COMMIT,
+    PHASE_CREATE_PR,
+    PHASE_LOCK,
+    assess_work_issue_duplicate_gate,
+    assess_work_issue_duplicate_report,
+)
+
+
+class TestDuplicateGateAssessment(unittest.TestCase):
+    def test_clear_issue_passes(self):
+        result = assess_work_issue_duplicate_gate(
+            400,
+            open_prs=[],
+            branch_names=["feat/other-issue-99"],
+            claim_entry={"status": "not_claimed"},
+            locked_branch="feat/issue-400-duplicate-work-preflight",
+            phase=PHASE_LOCK,
+        )
+        self.assertFalse(result["block"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
+
+    def test_open_pr_blocks(self):
+        prs = [{
+            "number": 397,
+            "title": "feat: handoff",
+            "body": "Closes #395",
+            "head": {"ref": "feat/issue-395-proof-backed-review-handoff"},
+        }]
+        result = assess_work_issue_duplicate_gate(
+            395,
+            open_prs=prs,
+            branch_names=[],
+            phase=PHASE_LOCK,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED)
+
+    def test_conflicting_remote_branch_blocks(self):
+        result = assess_work_issue_duplicate_gate(
+            395,
+            open_prs=[],
+            branch_names=["feat/issue-395-proof-backed-handoff-claims"],
+            locked_branch="feat/issue-395-new-attempt",
+            phase=PHASE_LOCK,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_BRANCH_PREVENTED)
+
+    def test_active_claim_on_other_branch_blocks(self):
+        result = assess_work_issue_duplicate_gate(
+            398,
+            open_prs=[],
+            branch_names=[],
+            claim_entry={
+                "status": "active",
+                "latest_heartbeat": {
+                    "branch": "feat/issue-398-validation-cwd-proof",
+                },
+            },
+            locked_branch="feat/issue-398-other-branch",
+            phase=PHASE_LOCK,
+        )
+        self.assertTrue(result["block"])
+
+    def test_commit_phase_maps_to_commit_outcome(self):
+        prs = [{
+            "number": 411,
+            "title": "x",
+            "body": "Closes #398",
+            "head": {"ref": "feat/issue-398-validation-cwd-proof"},
+        }]
+        result = assess_work_issue_duplicate_gate(
+            398,
+            open_prs=prs,
+            branch_names=[],
+            locked_branch="feat/issue-398-alt",
+            phase=PHASE_COMMIT,
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_COMMIT_PREVENTED)
+
+    def test_stale_claim_does_not_block_by_status_alone(self):
+        result = assess_work_issue_duplicate_gate(
+            400,
+            open_prs=[],
+            branch_names=[],
+            claim_entry={"status": "reclaimable", "reasons": ["stale"]},
+            locked_branch="feat/issue-400-duplicate-work-preflight",
+            phase=PHASE_LOCK,
+        )
+        self.assertFalse(result["block"])
+
+
+class TestDuplicateReportOutcome(unittest.TestCase):
+    def test_requires_exactly_one_outcome(self):
+        bad = assess_work_issue_duplicate_report("work finished")
+        self.assertFalse(bad["complete"])
+
+        good = assess_work_issue_duplicate_report(
+            "Duplicate work not prevented for issue #400."
+        )
+        self.assertTrue(good["complete"])
+        self.assertEqual(good["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
+
+
+class TestInjectableDuplicateFetcher(unittest.TestCase):
+    @patch("mcp_server.get_auth_header", return_value="token x")
+    def test_lock_issue_uses_injected_fetcher(self, _auth):
+        seen = {}
+
+        def fetcher(h, o, r, auth, issue_number):
+            seen["issue_number"] = issue_number
+            return [], [], {"status": "not_claimed"}
+
+        with patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            side_effect=fetcher,
+        ), patch(
+            "mcp_server.issue_lock_worktree.read_worktree_git_state",
+            return_value={
+                "current_branch": "master",
+                "porcelain_status": "",
+                "base_equivalent": True,
+            },
+        ), patch.dict(os.environ, {
+            "GITEA_ALLOWED_OPERATIONS": "gitea.issue.comment",
+        }, clear=True):
+            with patch.object(mcp_server, "ISSUE_LOCK_FILE", tempfile.mktemp()):
+                mcp_server.gitea_lock_issue(
+                    issue_number=400,
+                    branch_name="feat/issue-400-duplicate-work-preflight",
+                    remote="prgs",
+                )
+        self.assertEqual(seen["issue_number"], 400)
+
+
+class TestMcpDuplicateRecheck(unittest.TestCase):
+    def setUp(self):
+        self._dir = tempfile.TemporaryDirectory()
+        self.lock_path = os.path.join(self._dir.name, "gitea_issue_lock.json")
+        self._lock_patch = patch.object(
+            mcp_server, "ISSUE_LOCK_FILE", self.lock_path
+        )
+        self._lock_patch.start()
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
+                     "repo": "Example-Repo"},
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def tearDown(self):
+        patch.stopall()
+        self._dir.cleanup()
+
+    def _write_lock(self, issue_number=400, branch="feat/issue-400-x"):
+        with open(self.lock_path, "w", encoding="utf-8") as fh:
+            json.dump({
+                "issue_number": issue_number,
+                "branch_name": branch,
+                "remote": "prgs",
+            }, fh)
+
+    @patch("mcp_server._assess_issue_duplicate_gate")
+    @patch("mcp_server.get_profile", return_value={
+        "profile_name": "test-author",
+        "allowed_operations": ["gitea.read", "gitea.repo.commit"],
+        "forbidden_operations": [],
+        "audit_label": "test-author",
+    })
+    @patch("mcp_server.get_auth_header", return_value="token x")
+    def test_commit_files_blocked_on_recheck(self, _auth, _profile, mock_gate):
+        self._write_lock()
+        mock_gate.return_value = {
+            "block": True,
+            "reasons": ["open PR #412 already covers issue #400"],
+            "outcome": OUTCOME_DUPLICATE_COMMIT_PREVENTED,
+            "safe_next_action": "stop",
+        }
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", resolved_role="author")
+        with patch(
+            "mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+            return_value=(True, []),
+        ):
+            result = mcp_server.gitea_commit_files(
+                files=[{
+                    "operation": "create",
+                    "path": "a.txt",
+                    "content_plain": "hi",
+                }],
+                message="test",
+                remote="prgs",
+            )
+        self.assertFalse(result["success"])
+        self.assertIn("duplicate_gate", result)
+
+    @patch("mcp_server._assess_issue_duplicate_gate")
+    @patch("mcp_server.get_profile", return_value={
+        "profile_name": "test-author",
+        "allowed_operations": ["gitea.read", "gitea.pr.create"],
+        "forbidden_operations": [],
+        "audit_label": "test-author",
+    })
+    @patch("mcp_server.get_auth_header", return_value="token x")
+    def test_create_pr_returns_handoff_on_duplicate(self, _auth, _profile, mock_gate):
+        self._write_lock()
+        mock_gate.return_value = {
+            "block": True,
+            "reasons": ["open PR #412 already covers issue #400"],
+            "outcome": OUTCOME_DUPLICATE_PR_PREVENTED,
+            "safe_next_action": "reconciliation handoff",
+        }
+        mcp_server.record_preflight_check("whoami")
+        mcp_server.record_preflight_check("capability", resolved_role="author")
+        with patch(
+            "mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+            return_value=(True, []),
+        ):
+            result = mcp_server.gitea_create_pr(
+                title="feat: x (Closes #400)",
+                head="feat/issue-400-x",
+                base="master",
+                body="Closes #400",
+                remote="prgs",
+            )
+        self.assertFalse(result["success"])
+        self.assertIsNone(result.get("number"))
+        self.assertIn("duplicate_gate", result)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_llm_agent_sha.py

@@ -22,6 +22,7 @@ from unittest.mock import patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
+import mcp_server  # noqa: E402
 from mcp_server import (  # noqa: E402
     gitea_check_pr_eligibility,
     gitea_merge_pr,
@@ -130,6 +131,7 @@ class TestShaCannotBypassSelfReview(unittest.TestCase):
         ]
         from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
         init_review_decision_lock("prgs", "review_pr")
+        mcp_server.gitea_load_review_workflow()
         gitea_mark_final_review_decision(9, "approve", remote="prgs")
         env = self._env(SHA_WOULD_BE_REVIEWER, "reviewer")
         with patch.dict(os.environ, env, clear=True):

tests/test_mcp_server.py

@@ -55,6 +55,12 @@ _NO_BLOCKER_FEEDBACK = {
 }
 
 
+def _init_reviewer_session(remote="prgs"):
+    """Seed review decision lock and required workflow-load proof (#389)."""
+    init_review_decision_lock(remote, "review_pr")
+    mcp_server.gitea_load_review_workflow()
+
+
 def _mark_request_changes_ready(pr_number=8, **kwargs):
     """Mark a request_changes decision ready with the #332 duplicate-
     suppression feedback fetch stubbed to 'no existing blocker'."""
@@ -101,17 +107,36 @@ ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
 
 
 def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
+    import issue_lock_provenance
+
+    work_lease = {
+        "operation_type": "author_issue_work",
+        "issue_number": issue_number,
+        "branch": branch_name,
+        "claimant": {"username": "test-user", "profile": "test-author"},
+        "expires_at": "2999-01-01T00:00:00Z",
+    }
     record = {
         "issue_number": issue_number,
         "branch_name": branch_name,
         "remote": "dadeschools",
         "org": "Scaled-Tech-Consulting",
         "repo": "Gitea-Tools",
+        "work_lease": work_lease,
+        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+            tool="gitea_lock_issue",
+            claimant=work_lease.get("claimant"),
+        ),
     }
     record.update(overrides)
     return record
 
 
+def _clear_duplicate_context_fetcher(*_args, **_kwargs):
+    """Default injectable duplicate-work context for lock/create_pr tests."""
+    return [], [], {"status": "not_claimed"}
+
+
 # ---------------------------------------------------------------------------
 # Create Issue
 # ---------------------------------------------------------------------------
@@ -166,13 +191,17 @@ class TestCreateIssue(unittest.TestCase):
 # ---------------------------------------------------------------------------
 class TestCreatePR(unittest.TestCase):
 
+    @patch(
+        "mcp_server.issue_duplicate_context_fetcher",
+        return_value=([], [], {"status": "not_claimed"}),
+    )
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []))
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
-    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
+    def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
         lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
         mock_open.return_value.__enter__.return_value.read.return_value = lock_json
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
@@ -187,13 +216,17 @@ class TestCreatePR(unittest.TestCase):
         self.assertEqual(payload["base"], "main")
         self.assertIn("Closes #123", payload["title"])
 
+    @patch(
+        "mcp_server.issue_duplicate_context_fetcher",
+        return_value=([], [], {"status": "not_claimed"}),
+    )
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []))
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("os.path.exists", return_value=True)
     @patch("builtins.open")
-    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
+    def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher):
         lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
         mock_open.return_value.__enter__.return_value.read.return_value = lock_json
         mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
@@ -512,6 +545,9 @@ class TestViewPR(unittest.TestCase):
 class TestMergePR(unittest.TestCase):
     """Gated merge workflow (#16). gitea_merge_pr is the only merge path."""
 
+    def setUp(self):
+        mcp_server.gitea_load_review_workflow()
+
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
         return {
             "user": {"login": author},
@@ -984,8 +1020,7 @@ class TestReviewPR(unittest.TestCase):
             {"login": "jcwalker3"}, # /api/v1/user (submit eligibility)
             {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1
         ]
-        from mcp_server import init_review_decision_lock
+        _init_reviewer_session("prgs")
-        init_review_decision_lock("prgs", "review_pr")
         gitea_mark_final_review_decision(1, "approve", remote="prgs")
         result = gitea_review_pr(
             pr_number=1,
@@ -1653,7 +1688,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
         }
 
     def setUp(self):
-        init_review_decision_lock("prgs", "review_pr")
+        _init_reviewer_session("prgs")
 
     def _env(self):
         return patch.dict(os.environ, {
@@ -1748,7 +1783,7 @@ class TestSubmitPrReview(unittest.TestCase):
     """Gated review-mutation tool (#15)."""
 
     def setUp(self):
-        init_review_decision_lock("prgs", "review_pr")
+        _init_reviewer_session("prgs")
         gitea_mark_final_review_decision(8, "approve", remote="prgs")
 
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
@@ -2141,7 +2176,7 @@ class TestSubmitPrReview(unittest.TestCase):
                 os.remove(spoof_path)
 
     def test_mark_final_decision_rejects_remote_mismatch(self):
-        init_review_decision_lock("prgs", "review_pr")
+        _init_reviewer_session("prgs")
         r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools")
         self.assertFalse(r["marked_ready"])
         self.assertTrue(any("does not match locked remote" in x for x in r["reasons"]))
@@ -2223,6 +2258,7 @@ if __name__ == "__main__":
 class TestTrackerHygieneCleanup(unittest.TestCase):
 
     def setUp(self):
+        mcp_server.gitea_load_review_workflow()
         self.mock_api = patch("mcp_server.api_request").start()
         self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
         patch("gitea_audit.audit_enabled", return_value=True).start()
@@ -3044,8 +3080,14 @@ class TestIssueLocking(unittest.TestCase):
     def setUp(self):
         self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
         self._env_patcher.start()
+        self._dup_fetcher_patcher = patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            return_value=([], [], {"status": "not_claimed"}),
+        )
+        self.mock_dup_fetcher = self._dup_fetcher_patcher.start()
 
     def tearDown(self):
+        self._dup_fetcher_patcher.stop()
         self._env_patcher.stop()
         if os.path.exists(ISSUE_LOCK_FILE):
             os.remove(ISSUE_LOCK_FILE)
@@ -3054,10 +3096,8 @@ class TestIssueLocking(unittest.TestCase):
         "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_success(self, _auth, mock_api, _git_state):
+    def test_lock_issue_success(self, _auth, _git_state):
-        mock_api.return_value = [] # no open PRs
         res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertTrue(res["success"])
         self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
@@ -3082,50 +3122,48 @@ class TestIssueLocking(unittest.TestCase):
         "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_reused_by_open_pr_branch(self, _auth, mock_api, _git_state):
+    def test_lock_issue_reused_by_open_pr_branch(self, _auth, _git_state):
-        mock_api.return_value = [{
+        self.mock_dup_fetcher.return_value = ([{
             "number": 200,
             "head": {"ref": "feat/issue-196-boundary"},
             "title": "Some PR",
-            "body": "No closes ref"
+            "body": "No closes ref",
-        }]
+        }], [], {"status": "not_claimed"})
         with self.assertRaises(ValueError) as ctx:
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("already tied to an open PR", str(ctx.exception))
+        self.assertIn("open PR #200 already covers issue", str(ctx.exception))
 
     @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_reused_by_open_pr_closes_ref(self, _auth, mock_api, _git_state):
+    def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, _git_state):
-        mock_api.return_value = [{
+        self.mock_dup_fetcher.return_value = ([{
             "number": 200,
             "head": {"ref": "feat/other-branch"},
             "title": "Some PR",
-            "body": "fixes #196"
+            "body": "fixes #196",
-        }]
+        }], [], {"status": "not_claimed"})
         with self.assertRaises(ValueError) as ctx:
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("already tied to an open PR", str(ctx.exception))
+        self.assertIn("open PR #200 already covers issue", str(ctx.exception))
 
     @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_reused_by_remote_branch(self, _auth, mock_api, _git_state):
+    def test_lock_issue_reused_by_remote_branch(self, _auth, _git_state):
-        mock_api.side_effect = [
+        self.mock_dup_fetcher.return_value = (
             [],
-            [{"name": "feat/issue-196-existing-work"}],
+            ["feat/issue-196-existing-work"],
-        ]
+            {"status": "not_claimed"},
+        )
         with self.assertRaises(ValueError) as ctx:
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
-        self.assertIn("already has matching branch", str(ctx.exception))
+        self.assertIn("remote branch(es) already match issue pattern", str(ctx.exception))
 
     def test_lock_issue_blocks_active_same_operation_lease(self):
         with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
@@ -3284,6 +3322,28 @@ class TestIssueLocking(unittest.TestCase):
                 )
         self.assertIn("does not match locked worktree", str(ctx.exception))
 
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_create_pr_manual_lock_seed_blocked(self, _auth, _role):
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump(
+                _sample_issue_lock(
+                    issue_number=447,
+                    branch_name="feat/issue-447-lock-provenance",
+                    lock_provenance=None,
+                ),
+                f,
+            )
+        with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
+            with self.assertRaises(RuntimeError) as ctx:
+                gitea_create_pr(
+                    title="feat: lock provenance Closes #447",
+                    head="feat/issue-447-lock-provenance",
+                    remote="prgs",
+                )
+        self.assertIn("lock provenance", str(ctx.exception).lower())
+
     @patch("mcp_server.api_request")
     @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
            return_value=(True, []))

tests/test_permission_reports.py

@@ -230,6 +230,7 @@ class TestEligibilityDenialReport(PermissionReportBase):
             return PR_PAYLOAD
         mock_api.side_effect = fake_api
         mcp_server.init_review_decision_lock("prgs", "review_pr")
+        mcp_server.gitea_load_review_workflow()
         mcp_server.gitea_mark_final_review_decision(42, "approve", remote="prgs")
         with patch.dict(os.environ, self._env("author-profile")):
             res = mcp_server.gitea_submit_pr_review(
@@ -272,6 +273,7 @@ class TestReviewCommentPathUsesCanonicalOp(PermissionReportBase):
             return PR_PAYLOAD
         mock_api.side_effect = fake_api
         mcp_server.init_review_decision_lock("prgs", "review_pr")
+        mcp_server.gitea_load_review_workflow()
         mcp_server.gitea_mark_final_review_decision(42, "comment", remote="prgs")
         with patch.dict(os.environ, self._env("author-profile")):
             res = mcp_server.gitea_submit_pr_review(

tests/test_review_proofs.py

@@ -2346,6 +2346,7 @@ class TestWorkIssueFinalReport(unittest.TestCase):
             "- Safe next action: open PR",
             "- Next: open PR",
             "- Safety statement: no review/merge",
+            "- Duplicate work outcome: duplicate work not prevented",
         ])
 
     def test_complete_work_issue_report_earns_a(self):

tests/test_review_workflow_boundary.py

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

tests/test_review_workflow_load.py

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

tests/test_webui_queue_dashboard.py

@@ -12,12 +12,14 @@ from starlette.testclient import TestClient
 from webui.app import create_app
 from webui.queue_loader import (
     PaginationMeta,
+    QueueSnapshot,
     _classify_issue,
     _classify_pr,
     _extract_linked_issue,
     load_queue_snapshot,
     snapshot_to_dict,
 )
+from webui.queue_views import render_queue_page
 
 _RECENT = datetime.now(timezone.utc).isoformat()
 _STALE = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
@@ -210,5 +212,77 @@ class TestQueueRoutes(unittest.TestCase):
         self.assertEqual(len(snapshot.prs), 0)
 
 
+def _empty_pagination() -> PaginationMeta:
+    return PaginationMeta(
+        page=1,
+        per_page=50,
+        returned_count=0,
+        has_more=False,
+        is_final_page=True,
+        inventory_complete=True,
+        pages_fetched=1,
+    )
+
+
+def _empty_fetch(*_args, **_kwargs):
+    return [], _empty_pagination()
+
+
+class TestQueueFailClosedUx(unittest.TestCase):
+    """Regression tests for #458 fail-closed empty-state copy."""
+
+    def test_fail_closed_view_suppresses_empty_queue_copy(self):
+        snapshot = QueueSnapshot(
+            project_id="gitea-tools",
+            repo_label="Scaled-Tech-Consulting/Gitea-Tools",
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+            fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
+        )
+        html = render_queue_page(snapshot)
+        self.assertIn("Queue unavailable", html)
+        self.assertIn("Not loaded", html)
+        self.assertNotIn("No open items.", html)
+        self.assertIn("pagination:</strong> unavailable", html)
+
+    def test_fail_closed_route_does_not_show_empty_queue_copy(self):
+        client = TestClient(create_app())
+        snapshot = load_queue_snapshot(
+            fetch_prs=_empty_fetch,
+            fetch_issues=_empty_fetch,
+        )
+        snapshot = QueueSnapshot(
+            project_id=snapshot.project_id,
+            repo_label=snapshot.repo_label,
+            prs=(),
+            issues=(),
+            pr_pagination=None,
+            issue_pagination=None,
+            fetch_error="Gitea credentials unavailable for gitea.prgs.cc",
+        )
+        with mock.patch("webui.app.load_queue_snapshot", return_value=snapshot):
+            response = client.get("/queue")
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("Queue unavailable", response.text)
+        self.assertNotIn("No open items.", response.text)
+
+    def test_successful_empty_inventory_shows_empty_copy(self):
+        snapshot = load_queue_snapshot(
+            fetch_prs=_empty_fetch,
+            fetch_issues=_empty_fetch,
+        )
+        self.assertIsNone(snapshot.fetch_error)
+        self.assertEqual(len(snapshot.prs), 0)
+        self.assertEqual(len(snapshot.issues), 0)
+        self.assertTrue(snapshot.pr_pagination.inventory_complete)
+
+        html = render_queue_page(snapshot)
+        self.assertNotIn("Queue unavailable", html)
+        self.assertEqual(html.count("No open items."), 2)
+        self.assertIn("pagination (complete)", html)
+
+
 if __name__ == "__main__":
     unittest.main()

webui/queue_views.py

@@ -38,7 +38,13 @@ def _queue_table(
     title: str,
     items: tuple[QueueItem, ...],
     columns: tuple[tuple[str, str], ...],
+    fetch_failed: bool = False,
 ) -> str:
+    if fetch_failed:
+        return (
+            f"<h3>{html.escape(title)}</h3>"
+            "<p class='muted'>Not loaded — queue fetch did not complete.</p>"
+        )
     if not items:
         return f"<h3>{html.escape(title)}</h3><p class='muted'>No open items.</p>"
 
@@ -73,9 +79,11 @@ def render_queue_page(snapshot: QueueSnapshot) -> str:
             f"{html.escape(snapshot.fetch_error)}</p></div>"
         )
 
+    fetch_failed = bool(snapshot.fetch_error)
     pr_section = _queue_table(
         title="Open pull requests",
         items=snapshot.prs,
+        fetch_failed=fetch_failed,
         columns=(
             ("number", "#"),
             ("title", "Title"),
@@ -88,6 +96,7 @@ def render_queue_page(snapshot: QueueSnapshot) -> str:
     issue_section = _queue_table(
         title="Open issues",
         items=snapshot.issues,
+        fetch_failed=fetch_failed,
         columns=(
             ("number", "#"),
             ("title", "Title"),