feat: add workflow-load session boundary proof (Closes #403)

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

author_duplicate_work_gate.py

@@ -1,198 +0,0 @@
-"""Early duplicate-work detection for author work-issue flows (#400)."""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-from issue_claim_heartbeat import (
-    _linked_open_pr,
-    _matching_branch_names,
-    classify_issue_claim,
-)
-
-STAGES = (
-    "claim",
-    "lock",
-    "worktree",
-    "edit",
-    "commit",
-    "push",
-    "create_pr",
-)
-
-ELIGIBILITY_OPEN_PR_EXISTS = "OPEN_PR_EXISTS"
-ELIGIBILITY_DUPLICATE_BRANCH_EXISTS = "DUPLICATE_BRANCH_EXISTS"
-ELIGIBILITY_ACTIVE_CLAIM = "ACTIVE_CLAIM_BY_OTHER"
-ELIGIBILITY_CLEAR = "CLEAR"
-
-
-def assess_author_duplicate_work(
-    issue_number: int,
-    *,
-    stage: str,
-    open_prs: list[dict] | None = None,
-    branch_names: list[str] | None = None,
-    claim_entry: dict | None = None,
-    matching_branches: list[str] | None = None,
-    allow_stale_takeover: bool = False,
-) -> dict[str, Any]:
-    """Fail closed when duplicate work is detected before author mutations (#400)."""
-    stage_norm = (stage or "").strip().lower()
-    if stage_norm not in STAGES:
-        return {
-            "allowed": False,
-            "block": True,
-            "eligibility_class": "INVALID_STAGE",
-            "stage": stage_norm or None,
-            "reasons": [f"unknown duplicate-work stage {stage!r}"],
-            "safe_next_action": f"use one of: {', '.join(STAGES)}",
-        }
-
-    prs = list(open_prs or [])
-    linked_pr = _linked_open_pr(int(issue_number), prs)
-    branches = list(
-        matching_branches
-        if matching_branches is not None
-        else _matching_branch_names(int(issue_number), list(branch_names or []))
-    )
-
-    reasons: list[str] = []
-    eligibility = ELIGIBILITY_CLEAR
-
-    if linked_pr:
-        eligibility = ELIGIBILITY_OPEN_PR_EXISTS
-        reasons.append(
-            f"open PR #{linked_pr.get('number')} already covers issue "
-            f"#{issue_number}"
-        )
-
-    if branches and stage_norm in {"claim", "lock", "worktree", "edit"}:
-        if eligibility == ELIGIBILITY_CLEAR:
-            eligibility = ELIGIBILITY_DUPLICATE_BRANCH_EXISTS
-        reasons.append(
-            f"remote branch(es) already exist for issue #{issue_number}: "
-            f"{', '.join(branches)}"
-        )
-
-    entry = claim_entry or {}
-    claim_status = (entry.get("status") or "").strip()
-    if claim_status in {"active", "awaiting_review"} and stage_norm == "claim":
-        if entry.get("reclaimable") and allow_stale_takeover:
-            pass
-        elif claim_status == "active" and not entry.get("reclaimable"):
-            if eligibility == ELIGIBILITY_CLEAR:
-                eligibility = ELIGIBILITY_ACTIVE_CLAIM
-            reasons.append(
-                f"issue #{issue_number} has active claim "
-                f"(status={claim_status})"
-            )
-    elif claim_status == "awaiting_review" and stage_norm == "claim":
-        if not linked_pr:
-            reasons.append(
-                f"issue #{issue_number} is awaiting_review but no linked open PR "
-                "was supplied for duplicate-work proof"
-            )
-
-    allowed = not reasons
-    outcome = "duplicate_work_prevented" if not allowed else "clear"
-    if stage_norm == "create_pr" and not allowed:
-        outcome = "duplicate_pr_prevented"
-    elif stage_norm in {"commit", "push"} and not allowed:
-        outcome = "duplicate_push_prevented" if stage_norm == "push" else "duplicate_commit_prevented"
-
-    return {
-        "allowed": allowed,
-        "block": not allowed,
-        "eligibility_class": eligibility if not allowed else ELIGIBILITY_CLEAR,
-        "stage": stage_norm,
-        "linked_open_pr": linked_pr.get("number") if linked_pr else None,
-        "matching_branches": branches,
-        "claim_status": claim_status or None,
-        "outcome": outcome,
-        "reasons": reasons,
-        "safe_next_action": (
-            "stop without edits/commit/push/PR; produce reconciliation handoff "
-            "preserving local work only"
-            if not allowed
-            else "proceed"
-        ),
-    }
-
-
-def build_claim_entry_from_classification(classification: dict) -> dict:
-    """Map ``classify_issue_claim`` output to gate claim metadata."""
-    return {
-        "status": classification.get("status"),
-        "reclaimable": classification.get("reclaimable"),
-        "linked_open_pr": classification.get("linked_open_pr"),
-    }
-
-
-_DUPLICATE_OUTCOME_RE = re.compile(
-    r"(duplicate\s+(?:pr|branch|commit|push)\s+prevented|"
-    r"duplicate\s+work\s+not\s+prevented|reconciliation\s+handoff)",
-    re.IGNORECASE,
-)
-
-
-def assess_work_issue_duplicate_prevention_report(report_text: str) -> dict:
-    """#400: work-issue reports must state duplicate-work prevention outcome."""
-    text = report_text or ""
-    if "duplicate work" not in text.lower() and "duplicate pr" not in text.lower():
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": [],
-            "safe_next_action": "proceed",
-        }
-    if _DUPLICATE_OUTCOME_RE.search(text):
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": [],
-            "safe_next_action": "proceed",
-        }
-    return {
-        "proven": False,
-        "block": True,
-        "reasons": [
-            "duplicate-work discussion must name a prevention outcome "
-            "(duplicate PR/branch/commit/push prevented, or duplicate work "
-            "not prevented, or reconciliation handoff)"
-        ],
-        "safe_next_action": "state exact duplicate-work prevention class in final report",
-    }
-
-
-def classify_and_assess(
-    issue: dict,
-    *,
-    stage: str,
-    comments: list[dict] | None = None,
-    open_prs: list[dict] | None = None,
-    branch_names: list[str] | None = None,
-    allow_stale_takeover: bool = False,
-) -> dict[str, Any]:
-    """Combine claim classification with duplicate-work gate assessment."""
-    issue_number = int(issue.get("number") or 0)
-    claim = classify_issue_claim(
-        issue=issue,
-        comments=comments or [],
-        open_prs=open_prs or [],
-        branch_names=branch_names or [],
-    )
-    assessment = assess_author_duplicate_work(
-        issue_number,
-        stage=stage,
-        open_prs=open_prs,
-        branch_names=branch_names,
-        claim_entry=build_claim_entry_from_classification(claim),
-        matching_branches=claim.get("matching_branches"),
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    return {
-        "issue_number": issue_number,
-        "claim": claim,
-        "duplicate_work": assessment,
-    }

final_report_validator.py

@@ -115,6 +115,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 +886,54 @@ def _rule_reviewer_mutation_ledger(
     )
 
 
+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,
     *,
@@ -907,6 +971,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_already_landed_eligible,
         _rule_reviewer_already_landed_state,
         _rule_reviewer_target_branch_freshness,
+        _rule_reviewer_workflow_load_boundary,
         _rule_reviewer_mutation_ledger,
         _rule_reviewer_review_mutation,
     ],

gitea_mcp_server.py

@@ -498,12 +498,13 @@ 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 already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
-import author_duplicate_work_gate  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
@@ -1050,76 +1051,6 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
-def _list_repo_branch_names(h: str, o: str, r: str, auth: str, *, limit: int = 200) -> list[str]:
-    branches = api_get_all(f"{repo_api_url(h, o, r)}/branches", auth, limit=limit)
-    return [_branch_entry_name(branch) for branch in branches]
-
-
-def _gather_author_duplicate_work_context(
-    issue_number: int,
-    *,
-    h: str,
-    o: str,
-    r: str,
-    auth: str,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    base = repo_api_url(h, o, r)
-    issue = api_request("GET", f"{base}/issues/{issue_number}", auth)
-    comments = api_request("GET", f"{base}/issues/{issue_number}/comments", auth) or []
-    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
-    branch_names = _list_repo_branch_names(h, o, r, auth)
-    if exclude_branch_name:
-        branch_names = [
-            name for name in branch_names
-            if name != exclude_branch_name
-        ]
-    return {
-        "issue": issue,
-        "comments": comments,
-        "open_prs": open_prs,
-        "branch_names": branch_names,
-    }
-
-
-def _enforce_author_duplicate_work_gate(
-    issue_number: int,
-    stage: str,
-    *,
-    h: str,
-    o: str,
-    r: str,
-    auth: str,
-    allow_stale_takeover: bool = False,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    """Fail closed when duplicate work is detected (#400)."""
-    ctx = _gather_author_duplicate_work_context(
-        issue_number,
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=exclude_branch_name,
-    )
-    result = author_duplicate_work_gate.classify_and_assess(
-        ctx["issue"],
-        stage=stage,
-        comments=ctx["comments"],
-        open_prs=ctx["open_prs"],
-        branch_names=ctx["branch_names"],
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    assessment = result.get("duplicate_work") or {}
-    if assessment.get("block"):
-        reasons = "; ".join(assessment.get("reasons") or ["duplicate work detected"])
-        raise RuntimeError(
-            f"Author duplicate-work gate (#400) blocked at stage '{stage}': "
-            f"{reasons} (fail closed)"
-        )
-    return result
-
-
 @mcp.tool()
 def gitea_lock_issue(
     issue_number: int,
@@ -1182,17 +1113,48 @@ 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)
-    _enforce_author_duplicate_work_gate(
-        issue_number,
-        "lock",
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=branch_name,
-    )
+    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
+
+    try:
+        prs = api_get_all(url, auth)
+    except Exception as e:
+        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
+
+    for pr in prs:
+        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)"
+            )
 
     work_lease = _build_author_issue_work_lease(
         issue_number=issue_number,
@@ -1313,17 +1275,6 @@ def gitea_create_pr(
             f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
         )
 
-    auth = _auth(h)
-    _enforce_author_duplicate_work_gate(
-        int(locked_issue),
-        "create_pr",
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=locked_branch,
-    )
-
     # Check for forbidden terms anywhere in title/body
     forbidden_terms = ["equivalent", "related", "same as"]
     text_to_check = f"{title} {body}".lower()
@@ -1340,6 +1291,7 @@ 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)"
         )
 
+    auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls"
     payload = {"title": title, "body": body, "head": head, "base": base}
     meta = {"title": title, "head": head, "base": base}
@@ -1872,6 +1824,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 = (
@@ -1897,6 +1850,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,
@@ -1907,7 +1865,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(
@@ -2263,6 +2224,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,
@@ -2278,6 +2240,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(
@@ -2462,6 +2428,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}
@@ -3045,6 +3018,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,
@@ -3062,6 +3036,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:
@@ -4798,6 +4776,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): "
@@ -5721,6 +5701,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:
@@ -5729,6 +5711,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.
@@ -6053,14 +6109,6 @@ def gitea_mark_issue(
         )
 
     if action == "start":
-        _enforce_author_duplicate_work_gate(
-            issue_number,
-            "claim",
-            h=h,
-            o=o,
-            r=r,
-            auth=auth,
-        )
         with _audited("label_issue", host=h, remote=remote, org=o, repo=r,
                       issue_number=issue_number,
                       request_metadata={"op": "add", "label": "status:in-progress"}):
@@ -6142,65 +6190,6 @@ def gitea_post_heartbeat(
     )
 
 
-@mcp.tool()
-def gitea_assess_author_duplicate_work(
-    issue_number: int,
-    stage: str,
-    remote: str = "dadeschools",
-    host: str | None = None,
-    org: str | None = None,
-    repo: str | None = None,
-    allow_stale_takeover: bool = False,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    """Read-only: assess duplicate-work risk before author mutations (#400).
-
-    Call at claim, lock, worktree, edit, commit, push, and create_pr stages.
-    """
-    read_block = _profile_operation_gate("gitea.read")
-    if read_block:
-        return {
-            "success": False,
-            "reasons": read_block,
-            "permission_report": _permission_block_report("gitea.read"),
-        }
-    h, o, r = _resolve(remote, host, org, repo)
-    auth = _auth(h)
-    ctx = _gather_author_duplicate_work_context(
-        issue_number,
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=exclude_branch_name,
-    )
-    result = author_duplicate_work_gate.classify_and_assess(
-        ctx["issue"],
-        stage=stage,
-        comments=ctx["comments"],
-        open_prs=ctx["open_prs"],
-        branch_names=ctx["branch_names"],
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    assessment = result.get("duplicate_work") or {}
-    return {
-        "success": True,
-        "performed": False,
-        "issue_number": issue_number,
-        "stage": stage,
-        "allowed": assessment.get("allowed"),
-        "block": assessment.get("block"),
-        "eligibility_class": assessment.get("eligibility_class"),
-        "outcome": assessment.get("outcome"),
-        "linked_open_pr": assessment.get("linked_open_pr"),
-        "matching_branches": assessment.get("matching_branches"),
-        "claim_status": assessment.get("claim_status"),
-        "reasons": assessment.get("reasons"),
-        "safe_next_action": assessment.get("safe_next_action"),
-        "claim": result.get("claim"),
-    }
-
-
 @mcp.tool()
 def gitea_reconcile_issue_claims(
     state: str = "open",
@@ -6851,6 +6840,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)

review_proofs.py

@@ -3622,33 +3622,18 @@ def assess_work_issue_mode_isolation(report_text: str) -> dict:
     }
 
 
-def assess_work_issue_duplicate_prevention_report(report_text, **kwargs):
-    """#400: work-issue reports must classify duplicate-work prevention."""
-    from author_duplicate_work_gate import (
-        assess_work_issue_duplicate_prevention_report as _assess,
-    )
-
-    return _assess(report_text, **kwargs)
-
-
 def assess_work_issue_final_report(report_text: str) -> dict:
     """#139: composite verifier for work-issue final reports."""
     checks = {
         "workflow_source": assess_work_issue_workflow_source(report_text),
         "mode_isolation": assess_work_issue_mode_isolation(report_text),
-        "duplicate_prevention": assess_work_issue_duplicate_prevention_report(
-            report_text
-        ),
     }
 
     reasons = []
     downgraded = False
     for name, result in checks.items():
         verdict = result.get("verdict")
-        if result.get("block"):
-            downgraded = True
-            reasons.extend(result.get("reasons") or [])
-        elif verdict in ("missing", "incomplete"):
+        if verdict in ("missing", "incomplete"):
             downgraded = True
             reasons.extend(result.get("reasons") or [])
         elif result.get("downgraded") or not result.get("complete", True):
@@ -3656,9 +3641,6 @@ def assess_work_issue_final_report(report_text: str) -> dict:
             reasons.extend(
                 f"{name}: {r}" for r in (result.get("reasons") or [])
             )
-        elif result.get("proven") is False:
-            downgraded = True
-            reasons.extend(result.get("reasons") or [])
 
     grade = "A" if not downgraded else "downgraded"
     return {

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

@@ -277,24 +277,6 @@ Do not select an issue based only on memory from a previous session.
 
 Before claiming or working on an issue, check whether there is already an open PR, branch, or active claim for that issue.
 
-Run `gitea_assess_author_duplicate_work` at these stages and stop when `block` is true:
-
-* `claim` — before `gitea_mark_issue`
-* `lock` — before `gitea_lock_issue`
-* `worktree` / `edit` — before creating a worktree or editing files
-* `commit` — immediately before `git commit`
-* `push` — immediately before `git push`
-* `create_pr` — immediately before `gitea_create_pr` (also enforced server-side)
-
-`gitea_mark_issue`, `gitea_lock_issue`, and `gitea_create_pr` enforce the same gate server-side and fail closed.
-
-If a concurrent open PR appears after work begins:
-
-* before commit or push — stop and preserve local work without pushing
-* after push but before PR creation — produce a reconciliation handoff instead of opening a PR
-
-Final reports must name the duplicate-work outcome (`duplicate PR prevented`, `duplicate branch prevented`, `duplicate commit prevented`, `duplicate push prevented`, `duplicate work not prevented`, or `reconciliation handoff`).
-
 If an open PR already exists for the issue, do not implement duplicate work.
 
 Classify the issue as:

tests/test_author_duplicate_work_gate.py

@@ -1,102 +0,0 @@
-"""Tests for early author duplicate-work gate (#400)."""
-import sys
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from author_duplicate_work_gate import (  # noqa: E402
-    ELIGIBILITY_OPEN_PR_EXISTS,
-    assess_author_duplicate_work,
-    assess_work_issue_duplicate_prevention_report,
-)
-
-
-def _open_pr(number: int = 397, issue: int = 395) -> dict:
-    return {
-        "number": number,
-        "head": {"ref": f"feat/issue-{issue}-example"},
-        "title": f"feat: example (Closes #{issue})",
-        "body": f"Closes #{issue}",
-    }
-
-
-class TestAuthorDuplicateWorkGate(unittest.TestCase):
-    def test_clear_when_no_duplicates(self):
-        result = assess_author_duplicate_work(
-            400,
-            stage="claim",
-            open_prs=[],
-            branch_names=["master", "feat/issue-399-other"],
-        )
-        self.assertTrue(result["allowed"])
-        self.assertFalse(result["block"])
-
-    def test_open_pr_blocks_claim(self):
-        result = assess_author_duplicate_work(
-            395,
-            stage="claim",
-            open_prs=[_open_pr()],
-            branch_names=[],
-        )
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["eligibility_class"], ELIGIBILITY_OPEN_PR_EXISTS)
-
-    def test_matching_branch_blocks_lock_not_create_pr(self):
-        branches = ["feat/issue-400-early-duplicate-work-gate"]
-        lock = assess_author_duplicate_work(
-            400,
-            stage="lock",
-            open_prs=[],
-            branch_names=branches,
-        )
-        self.assertFalse(lock["allowed"])
-
-        create_pr = assess_author_duplicate_work(
-            400,
-            stage="create_pr",
-            open_prs=[],
-            branch_names=branches,
-            matching_branches=[],
-        )
-        self.assertTrue(create_pr["allowed"])
-
-    def test_open_pr_blocks_create_pr_stage(self):
-        result = assess_author_duplicate_work(
-            395,
-            stage="create_pr",
-            open_prs=[_open_pr()],
-            branch_names=["feat/issue-395-proof-backed-review-handoff"],
-        )
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["outcome"], "duplicate_pr_prevented")
-
-    def test_push_stage_blocks_on_concurrent_pr(self):
-        result = assess_author_duplicate_work(
-            395,
-            stage="push",
-            open_prs=[_open_pr()],
-            branch_names=[],
-        )
-        self.assertFalse(result["allowed"])
-        self.assertEqual(result["outcome"], "duplicate_push_prevented")
-
-    def test_duplicate_prevention_report_requires_outcome(self):
-        bad = assess_work_issue_duplicate_prevention_report(
-            "Duplicate work detected for issue #395."
-        )
-        self.assertFalse(bad["proven"])
-
-        good = assess_work_issue_duplicate_prevention_report(
-            "Duplicate PR prevented; reconciliation handoff produced."
-        )
-        self.assertTrue(good["proven"])
-
-    def test_exported_from_review_proofs(self):
-        from review_proofs import assess_work_issue_duplicate_prevention_report as exported
-
-        self.assertTrue(callable(exported))
-
-
-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_llm_workflow_split.py

@@ -95,12 +95,6 @@ def test_create_issue_workflow_contract():
     assert "## 9. Duplicate search before mutation" in text
 
 
-def test_author_duplicate_work_gate_exported():
-    from review_proofs import assess_work_issue_duplicate_prevention_report
-
-    assert callable(assess_work_issue_duplicate_prevention_report)
-
-
 def test_work_issue_workflow_contract():
     text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8")
     assert "canonical: true" in text

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'."""
@@ -512,6 +518,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 +993,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_review_decision_lock("prgs", "review_pr")
+        _init_reviewer_session("prgs")
         gitea_mark_final_review_decision(1, "approve", remote="prgs")
         result = gitea_review_pr(
             pr_number=1,
@@ -1653,7 +1661,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 +1756,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 +2149,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 +2231,7 @@ if __name__ == "__main__":
 class TestTrackerHygieneCleanup(unittest.TestCase):
 
     def setUp(self):
+        mcp_server.gitea_load_review_workflow()
         self.mock_api = patch("mcp_server.api_request").start()
         self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
         patch("gitea_audit.audit_enabled", return_value=True).start()

tests/test_permission_reports.py

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