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

@@ -490,42 +490,6 @@ def _rule_reviewer_validation_failure_history(
     ]
 
 
-def _rule_reviewer_validation_cwd_proof(
-    report_text: str,
-    *,
-    validation_session: dict | None = None,
-) -> list[dict[str, str]]:
-    from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report
-
-    session = validation_session or {}
-    claims = (
-        session.get("validation_ran")
-        or session.get("command")
-        or session.get("baseline_validation_ran")
-    )
-    if not claims and "validation command:" not in (report_text or "").lower():
-        return []
-
-    result = assess_validation_cwd_proof_report(
-        report_text,
-        validation_session=session,
-    )
-    if result.get("proven") or not result.get("claims_validation"):
-        return []
-    severity = "block" if result.get("violations") else "downgrade"
-    return [
-        validator_finding(
-            "reviewer.validation_cwd_proof",
-            severity,
-            "Validation cwd/HEAD proof",
-            reason,
-            result.get("safe_next_action")
-            or "document pwd, HEAD SHA, and explicit cwd before validation",
-        )
-        for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
-    ]
-
-
 def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
     text = report_text or ""
     if not _BARE_PYTEST_RE.search(text):
@@ -935,7 +899,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_git_fetch_readonly,
         _rule_reviewer_validation_command,
         _rule_reviewer_validation_failure_history,
-        _rule_reviewer_validation_cwd_proof,
         _rule_reviewer_validation_structured,
         _rule_reviewer_linked_issue,
         _rule_reviewer_baseline_on_failure,

gitea_mcp_server.py

@@ -503,6 +503,7 @@ 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
@@ -1049,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,
@@ -1111,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"
-
-    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)"
-            )
+    _enforce_author_duplicate_work_gate(
+        issue_number,
+        "lock",
+        h=h,
+        o=o,
+        r=r,
+        auth=auth,
+        exclude_branch_name=branch_name,
+    )
 
     work_lease = _build_author_issue_work_lease(
         issue_number=issue_number,
@@ -1273,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()
@@ -1289,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}
@@ -6003,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"}):
@@ -6084,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",

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

reviewer_validation_cwd_proof.py

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

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

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

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_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
@@ -213,12 +219,6 @@ def test_validation_failure_history_verifier_exported():
     assert callable(assess_validation_failure_history_report)
 
 
-def test_validation_cwd_proof_verifier_exported():
-    from review_proofs import assess_validation_cwd_proof_report
-
-    assert callable(assess_validation_cwd_proof_report)
-
-
 def test_prior_blocker_skip_verifier_exported():
     from review_proofs import assess_prior_blocker_skip_proof
 

tests/test_reviewer_validation_cwd_proof.py

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