feat: gate author work on early duplicate-work detection (Closes #400)

Add author_duplicate_work_gate and enforce it at claim, lock, and PR
creation. Expose gitea_assess_author_duplicate_work for pre-commit/push
checks and extend work-issue final-report verification.

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

author_duplicate_work_gate.py

@@ -0,0 +1,198 @@
+"""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,22 +115,6 @@ _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",
@@ -886,54 +870,6 @@ 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,
     *,
@@ -971,7 +907,6 @@ _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,13 +498,12 @@ 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
@@ -1051,6 +1050,76 @@ 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,
@@ -1113,48 +1182,17 @@ 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"
+    _enforce_author_duplicate_work_gate(
-
+        issue_number,
-    try:
+        "lock",
-        prs = api_get_all(url, auth)
+        h=h,
-    except Exception as e:
+        o=o,
-        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
+        r=r,
-
+        auth=auth,
-    for pr in prs:
+        exclude_branch_name=branch_name,
-        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,
@@ -1275,6 +1313,17 @@ 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()
@@ -1291,7 +1340,6 @@ 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}
@@ -1824,7 +1872,6 @@ 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 = (
@@ -1850,11 +1897,6 @@ 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,
@@ -1865,10 +1907,7 @@ def check_review_decision_gate(
     repo: str | None = None,
 ) -> list[str]:
     """Fail closed unless validation completed and the final decision is ready."""
-    reasons = list(_review_workflow_load_gate_reasons())
+    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(
@@ -2224,7 +2263,6 @@ 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,
@@ -2240,10 +2278,6 @@ 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(
@@ -2428,13 +2462,6 @@ 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}
@@ -3018,7 +3045,6 @@ 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,
@@ -3036,10 +3062,6 @@ 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:
@@ -4776,8 +4798,6 @@ _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): "
@@ -5701,8 +5721,6 @@ 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:
@@ -5711,80 +5729,6 @@ 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.
@@ -6109,6 +6053,14 @@ 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"}):
@@ -6190,6 +6142,65 @@ 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",
@@ -6840,16 +6851,6 @@ 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,18 +3622,33 @@ 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 verdict in ("missing", "incomplete"):
+        if result.get("block"):
+            downgraded = True
+            reasons.extend(result.get("reasons") or [])
+        elif verdict in ("missing", "incomplete"):
             downgraded = True
             reasons.extend(result.get("reasons") or [])
         elif result.get("downgraded") or not result.get("complete", True):
@@ -3641,6 +3656,9 @@ 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

@@ -1,259 +0,0 @@
-"""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

@@ -1,217 +0,0 @@
-"""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,14 +63,8 @@ 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,44 +36,6 @@ 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,6 +277,24 @@ 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

@@ -0,0 +1,102 @@
+"""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,7 +22,6 @@ 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,
@@ -131,7 +130,6 @@ 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,6 +95,12 @@ 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,12 +55,6 @@ _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'."""
@@ -518,9 +512,6 @@ 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},
@@ -993,7 +984,8 @@ class TestReviewPR(unittest.TestCase):
             {"login": "jcwalker3"}, # /api/v1/user (submit eligibility)
             {"user": {"login": "jcwalker3"}, "state": "open", "head": {"sha": "abc1234"}, "mergeable": True}, # /pulls/1
         ]
-        _init_reviewer_session("prgs")
+        from mcp_server import init_review_decision_lock
+        init_review_decision_lock("prgs", "review_pr")
         gitea_mark_final_review_decision(1, "approve", remote="prgs")
         result = gitea_review_pr(
             pr_number=1,
@@ -1661,7 +1653,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
         }
 
     def setUp(self):
-        _init_reviewer_session("prgs")
+        init_review_decision_lock("prgs", "review_pr")
 
     def _env(self):
         return patch.dict(os.environ, {
@@ -1756,7 +1748,7 @@ class TestSubmitPrReview(unittest.TestCase):
     """Gated review-mutation tool (#15)."""
 
     def setUp(self):
-        _init_reviewer_session("prgs")
+        init_review_decision_lock("prgs", "review_pr")
         gitea_mark_final_review_decision(8, "approve", remote="prgs")
 
     def _pr(self, author, state="open", sha="abc123", mergeable=True):
@@ -2149,7 +2141,7 @@ class TestSubmitPrReview(unittest.TestCase):
                 os.remove(spoof_path)
 
     def test_mark_final_decision_rejects_remote_mismatch(self):
-        _init_reviewer_session("prgs")
+        init_review_decision_lock("prgs", "review_pr")
         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"]))
@@ -2231,7 +2223,6 @@ 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,7 +230,6 @@ 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(
@@ -273,7 +272,6 @@ 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

@@ -1,138 +0,0 @@
-"""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

@@ -1,151 +0,0 @@
-"""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()