feat: add precise validation status vocabulary for reviewer reports (Closes #406)

Introduce validation_status_vocabulary with proof-path binding for baseline-
equivalent, merge-simulation-resolved, and transient-pass labels. Wire the
assessor into final_report_validator and document the taxonomy in the
review-merge workflow.

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

author_duplicate_work_gate.py

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

final_report_validator.py

@@ -20,6 +20,7 @@ from review_proofs import (
     assess_review_mutation_final_report,
     assess_validation_report,
 )
+from validation_status_vocabulary import assess_validation_status_vocabulary
 
 FINAL_REPORT_TASK_KINDS = frozenset({
     "review_pr",
@@ -598,6 +599,34 @@ def _rule_reviewer_main_checkout_baseline(report_text: str) -> list[dict[str, st
     ]
 
 
+def _rule_reviewer_validation_status_vocabulary(
+    report_text: str,
+    *,
+    action_log: list[dict] | None = None,
+) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not re.search(
+        r"validation status|pr-head validation status|official validation status",
+        text,
+        re.IGNORECASE,
+    ):
+        return []
+    result = assess_validation_status_vocabulary(
+        text,
+        command_log=action_log,
+    )
+    if not result.get("block"):
+        return []
+    return _findings_from_reasons(
+        "reviewer.validation_status_vocabulary",
+        result.get("reasons") or [],
+        field="Validation status",
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or "use a validation status that matches the proof path executed",
+    )
+
+
 def _rule_reviewer_main_checkout_path(report_text: str) -> list[dict[str, str]]:
     text = report_text or ""
     if "baseline worktree path" not in text.lower():
@@ -902,6 +931,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_validation_structured,
         _rule_reviewer_linked_issue,
         _rule_reviewer_baseline_on_failure,
+        _rule_reviewer_validation_status_vocabulary,
         _rule_reviewer_main_checkout_baseline,
         _rule_reviewer_main_checkout_path,
         _rule_reviewer_already_landed_eligible,

gitea_mcp_server.py

@@ -503,7 +503,6 @@ import issue_lock_worktree  # noqa: E402
 import already_landed_reconcile  # noqa: E402
 import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
-import author_duplicate_work_gate  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import reconciler_profile  # noqa: E402
 import reconciliation_workflow  # noqa: E402
@@ -1050,76 +1049,6 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
-def _list_repo_branch_names(h: str, o: str, r: str, auth: str, *, limit: int = 200) -> list[str]:
-    branches = api_get_all(f"{repo_api_url(h, o, r)}/branches", auth, limit=limit)
-    return [_branch_entry_name(branch) for branch in branches]
-
-
-def _gather_author_duplicate_work_context(
-    issue_number: int,
-    *,
-    h: str,
-    o: str,
-    r: str,
-    auth: str,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    base = repo_api_url(h, o, r)
-    issue = api_request("GET", f"{base}/issues/{issue_number}", auth)
-    comments = api_request("GET", f"{base}/issues/{issue_number}/comments", auth) or []
-    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
-    branch_names = _list_repo_branch_names(h, o, r, auth)
-    if exclude_branch_name:
-        branch_names = [
-            name for name in branch_names
-            if name != exclude_branch_name
-        ]
-    return {
-        "issue": issue,
-        "comments": comments,
-        "open_prs": open_prs,
-        "branch_names": branch_names,
-    }
-
-
-def _enforce_author_duplicate_work_gate(
-    issue_number: int,
-    stage: str,
-    *,
-    h: str,
-    o: str,
-    r: str,
-    auth: str,
-    allow_stale_takeover: bool = False,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    """Fail closed when duplicate work is detected (#400)."""
-    ctx = _gather_author_duplicate_work_context(
-        issue_number,
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=exclude_branch_name,
-    )
-    result = author_duplicate_work_gate.classify_and_assess(
-        ctx["issue"],
-        stage=stage,
-        comments=ctx["comments"],
-        open_prs=ctx["open_prs"],
-        branch_names=ctx["branch_names"],
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    assessment = result.get("duplicate_work") or {}
-    if assessment.get("block"):
-        reasons = "; ".join(assessment.get("reasons") or ["duplicate work detected"])
-        raise RuntimeError(
-            f"Author duplicate-work gate (#400) blocked at stage '{stage}': "
-            f"{reasons} (fail closed)"
-        )
-    return result
-
-
 @mcp.tool()
 def gitea_lock_issue(
     issue_number: int,
@@ -1182,16 +1111,47 @@ def gitea_lock_issue(
             issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
         )
 
+    # 2. Check if the issue already has an open PR (reuse protection)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
-    _enforce_author_duplicate_work_gate(
+    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
-        issue_number,
+
-        "lock",
+    try:
-        h=h,
+        prs = api_get_all(url, auth)
-        o=o,
+    except Exception as e:
-        r=r,
+        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
-        auth=auth,
+
-        exclude_branch_name=branch_name,
+    for pr in prs:
+        pr_head = pr.get("head", {}).get("ref", "")
+        pr_title = pr.get("title", "")
+        pr_body = pr.get("body", "")
+
+        if expected_pattern in pr_head:
+            raise ValueError(
+                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
+            )
+
+        patterns = [
+            f"closes #{issue_number}",
+            f"fixes #{issue_number}",
+        ]
+        text_to_check = f"{pr_title} {pr_body}".lower()
+        if any(p in text_to_check for p in patterns):
+            raise ValueError(
+                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
+            )
+
+    branch_url = f"{repo_api_url(h, o, r)}/branches"
+    try:
+        branches = api_get_all(branch_url, auth)
+    except Exception as e:
+        raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
+    for branch in branches:
+        name = _branch_entry_name(branch)
+        if expected_pattern in name:
+            raise ValueError(
+                f"Issue #{issue_number} already has matching branch '{name}' "
+                "(fail closed)"
             )
 
     work_lease = _build_author_issue_work_lease(
@@ -1313,17 +1273,6 @@ def gitea_create_pr(
             f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
         )
 
-    auth = _auth(h)
-    _enforce_author_duplicate_work_gate(
-        int(locked_issue),
-        "create_pr",
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=locked_branch,
-    )
-
     # Check for forbidden terms anywhere in title/body
     forbidden_terms = ["equivalent", "related", "same as"]
     text_to_check = f"{title} {body}".lower()
@@ -1340,6 +1289,7 @@ def gitea_create_pr(
             f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
         )
 
+    auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls"
     payload = {"title": title, "body": body, "head": head, "base": base}
     meta = {"title": title, "head": head, "base": base}
@@ -6053,14 +6003,6 @@ def gitea_mark_issue(
         )
 
     if action == "start":
-        _enforce_author_duplicate_work_gate(
-            issue_number,
-            "claim",
-            h=h,
-            o=o,
-            r=r,
-            auth=auth,
-        )
         with _audited("label_issue", host=h, remote=remote, org=o, repo=r,
                       issue_number=issue_number,
                       request_metadata={"op": "add", "label": "status:in-progress"}):
@@ -6142,65 +6084,6 @@ def gitea_post_heartbeat(
     )
 
 
-@mcp.tool()
-def gitea_assess_author_duplicate_work(
-    issue_number: int,
-    stage: str,
-    remote: str = "dadeschools",
-    host: str | None = None,
-    org: str | None = None,
-    repo: str | None = None,
-    allow_stale_takeover: bool = False,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    """Read-only: assess duplicate-work risk before author mutations (#400).
-
-    Call at claim, lock, worktree, edit, commit, push, and create_pr stages.
-    """
-    read_block = _profile_operation_gate("gitea.read")
-    if read_block:
-        return {
-            "success": False,
-            "reasons": read_block,
-            "permission_report": _permission_block_report("gitea.read"),
-        }
-    h, o, r = _resolve(remote, host, org, repo)
-    auth = _auth(h)
-    ctx = _gather_author_duplicate_work_context(
-        issue_number,
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=exclude_branch_name,
-    )
-    result = author_duplicate_work_gate.classify_and_assess(
-        ctx["issue"],
-        stage=stage,
-        comments=ctx["comments"],
-        open_prs=ctx["open_prs"],
-        branch_names=ctx["branch_names"],
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    assessment = result.get("duplicate_work") or {}
-    return {
-        "success": True,
-        "performed": False,
-        "issue_number": issue_number,
-        "stage": stage,
-        "allowed": assessment.get("allowed"),
-        "block": assessment.get("block"),
-        "eligibility_class": assessment.get("eligibility_class"),
-        "outcome": assessment.get("outcome"),
-        "linked_open_pr": assessment.get("linked_open_pr"),
-        "matching_branches": assessment.get("matching_branches"),
-        "claim_status": assessment.get("claim_status"),
-        "reasons": assessment.get("reasons"),
-        "safe_next_action": assessment.get("safe_next_action"),
-        "claim": result.get("claim"),
-    }
-
-
 @mcp.tool()
 def gitea_reconcile_issue_claims(
     state: str = "open",

review_proofs.py

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

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

@@ -568,6 +568,28 @@ If the cause is unknown, do not erase the earlier failure with plain
 `gitea_validate_review_final_report` rejects reports that omit known earlier
 validation failures when `validation_session.observed_failures` is supplied.
 
+## 21B. Validation status taxonomy (#406)
+
+When the final report summarizes how validation concluded, use one of these
+**validation status** labels (distinct from per-command pass/fail entries):
+
+* `passed` — raw PR-head validation passed on the unmodified head.
+* `failed` — raw PR-head validation failed and no allowed resolution path
+  was proven.
+* `baseline-equivalent failure accepted` — only when a clean baseline
+  worktree under `branches/` proves matching failure signatures on the target
+  branch (baseline path, target SHA, exact commands, failure lists, and
+  `failure signatures match: true`).
+* `raw-head failure resolved by merge simulation` — raw PR-head validation
+  failed, but merge simulation into the current target passed cleanly; report
+  merge simulation under `Worktree/index mutations` with full #317 proof.
+* `passed after transient failure investigation` — a later run passed after an
+  earlier failure in the same session; document the failure history (#396).
+
+Do not use `baseline-equivalent failure accepted` when only merge simulation
+resolved the failure. Do not use bare `passed` when raw PR-head validation
+failed unless one of the resolution statuses above applies.
+
 ## 22. Baseline validation rule
 
 Do not run tests in the main checkout.

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

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

tests/test_author_duplicate_work_gate.py

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

tests/test_llm_workflow_split.py

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

tests/test_validation_status_vocabulary.py

@@ -0,0 +1,198 @@
+"""Tests for validation status vocabulary (#406)."""
+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 validation_status_vocabulary import (  # noqa: E402
+    STATUS_BASELINE_EQUIVALENT,
+    STATUS_FAILED,
+    STATUS_MERGE_SIM_RESOLVED,
+    STATUS_PASSED,
+    STATUS_TRANSIENT_PASS,
+    assess_validation_status_vocabulary,
+)
+
+
+def _handoff(**extra):
+    fields = {
+        "Task": "review PR #386",
+        "Validation status": STATUS_PASSED,
+        "Raw PR-head validation result": "passed",
+        "Merge simulation result": "not run",
+        "Baseline worktree used": "none",
+    }
+    fields.update(extra)
+    lines = ["## Controller Handoff", ""]
+    lines.extend(f"- {key}: {value}" for key, value in fields.items())
+    return "\n".join(lines)
+
+
+class TestValidationStatusVocabulary(unittest.TestCase):
+    def test_raw_head_pass_status(self):
+        report = _handoff()
+        result = assess_validation_status_vocabulary(report)
+        self.assertFalse(result["block"])
+        self.assertEqual(result["status_claimed"], STATUS_PASSED)
+
+    def test_raw_head_failure_with_baseline_match(self):
+        report = _handoff(
+            **{
+                "Validation status": STATUS_BASELINE_EQUIVALENT,
+                "Raw PR-head validation result": "failed",
+                "Baseline worktree used": "branches/baseline-master-pr386",
+                "Baseline target SHA": "a" * 40,
+                "Baseline failures": "test_foo failed",
+                "PR failures": "test_foo failed",
+                "Failure signatures match": "true",
+            }
+        )
+        result = assess_validation_status_vocabulary(report)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["baseline_proof_complete"])
+
+    def test_baseline_equivalent_without_baseline_proof_blocked(self):
+        report = _handoff(
+            **{
+                "Validation status": STATUS_BASELINE_EQUIVALENT,
+                "Raw PR-head validation result": "failed",
+            }
+        )
+        result = assess_validation_status_vocabulary(report)
+        self.assertTrue(result["block"])
+        self.assertIn("baseline-equivalent", result["reasons"][0])
+
+    def test_merge_simulation_resolution_passes(self):
+        report = "\n".join([
+            _handoff(
+                **{
+                    "Validation status": STATUS_MERGE_SIM_RESOLVED,
+                    "Raw PR-head validation result": "failed",
+                    "Merge simulation result": "passed",
+                }
+            ),
+            "Worktree/index mutations: merge simulation in branches/review-pr386",
+            "Worktree path: branches/review-pr386",
+            "Pre-simulation clean status: clean",
+            "Merge result: clean merge",
+            "Abort command: git merge --abort",
+            "Post-abort clean status: clean",
+        ])
+        command_log = [
+            {"command": "git merge --no-commit prgs/master"},
+            {"command": "git merge --abort"},
+        ]
+        result = assess_validation_status_vocabulary(
+            report, command_log=command_log
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["merge_simulation_passed"])
+
+    def test_merge_simulation_failure_stays_failed(self):
+        report = _handoff(
+            **{
+                "Validation status": STATUS_FAILED,
+                "Raw PR-head validation result": "failed",
+                "Merge simulation result": "failed",
+            }
+        )
+        result = assess_validation_status_vocabulary(report)
+        self.assertFalse(result["block"])
+
+    def test_failed_status_with_passing_merge_sim_blocked(self):
+        report = "\n".join([
+            _handoff(
+                **{
+                    "Validation status": STATUS_FAILED,
+                    "Raw PR-head validation result": "failed",
+                    "Merge simulation result": "passed",
+                }
+            ),
+            "Worktree/index mutations: merge simulation",
+            "Worktree path: branches/review-pr386",
+            "Pre-simulation clean status: clean",
+            "Merge result: clean",
+            "Abort command: git merge --abort",
+            "Post-abort clean status: clean",
+        ])
+        result = assess_validation_status_vocabulary(
+            report,
+            command_log=[{"command": "git merge --no-commit prgs/master"}],
+        )
+        self.assertTrue(result["block"])
+
+    def test_transient_failure_then_pass(self):
+        report = _handoff(
+            **{
+                "Validation status": STATUS_TRANSIENT_PASS,
+                "Raw PR-head validation result": "passed",
+                "Transient validation failure history": (
+                    "first run failed with infra flake; rerun passed"
+                ),
+            }
+        )
+        result = assess_validation_status_vocabulary(report)
+        self.assertFalse(result["block"])
+
+    def test_transient_pass_without_history_blocked(self):
+        report = _handoff(
+            **{
+                "Validation status": STATUS_TRANSIENT_PASS,
+                "Raw PR-head validation result": "passed",
+            }
+        )
+        result = assess_validation_status_vocabulary(report)
+        self.assertTrue(result["block"])
+
+    def test_bare_passed_after_raw_failure_blocked(self):
+        report = _handoff(
+            **{
+                "Validation status": STATUS_PASSED,
+                "Raw PR-head validation result": "failed",
+            }
+        )
+        result = assess_validation_status_vocabulary(report)
+        self.assertTrue(result["block"])
+
+    def test_wrong_baseline_label_when_merge_sim_used_blocked(self):
+        report = "\n".join([
+            _handoff(
+                **{
+                    "Validation status": STATUS_BASELINE_EQUIVALENT,
+                    "Raw PR-head validation result": "failed",
+                    "Merge simulation result": "passed",
+                }
+            ),
+            "Worktree/index mutations: merge simulation",
+            "Worktree path: branches/review-pr386",
+            "Pre-simulation clean status: clean",
+            "Merge result: clean",
+            "Abort command: git merge --abort",
+            "Post-abort clean status: clean",
+        ])
+        result = assess_validation_status_vocabulary(
+            report,
+            command_log=[{"command": "git merge --no-commit prgs/master"}],
+        )
+        self.assertTrue(result["block"])
+        joined = " ".join(result["reasons"]).lower()
+        self.assertTrue(
+            "misleading" in joined or "baseline-equivalent" in joined
+        )
+
+    def test_final_report_validator_integration_blocks_misleading_label(self):
+        report = _handoff(
+            **{
+                "Validation status": STATUS_BASELINE_EQUIVALENT,
+                "Raw PR-head validation result": "failed",
+            }
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        blocked_ids = {f["rule_id"] for f in result["findings"]}
+        self.assertIn("reviewer.validation_status_vocabulary", blocked_ids)
+
+
+if __name__ == "__main__":
+    unittest.main()

validation_status_vocabulary.py

@@ -0,0 +1,205 @@
+"""Precise validation-status vocabulary for reviewer final reports (#406)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from reviewer_merge_simulation import assess_merge_simulation_report
+
+STATUS_PASSED = "passed"
+STATUS_FAILED = "failed"
+STATUS_BASELINE_EQUIVALENT = "baseline-equivalent failure accepted"
+STATUS_MERGE_SIM_RESOLVED = "raw-head failure resolved by merge simulation"
+STATUS_TRANSIENT_PASS = "passed after transient failure investigation"
+
+ALLOWED_VALIDATION_STATUSES = frozenset({
+    STATUS_PASSED,
+    STATUS_FAILED,
+    STATUS_BASELINE_EQUIVALENT,
+    STATUS_MERGE_SIM_RESOLVED,
+    STATUS_TRANSIENT_PASS,
+})
+
+_STATUS_FIELD_RE = re.compile(
+    r"^\s*[-*]?\s*(?:validation status|pr-head validation status|"
+    r"official validation status)\s*:\s*(.+?)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_RAW_HEAD_RESULT_RE = re.compile(
+    r"^\s*[-*]?\s*raw pr-head validation result\s*:\s*(.+?)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_MERGE_SIM_RESULT_RE = re.compile(
+    r"^\s*[-*]?\s*merge simulation result\s*:\s*(.+?)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_BASELINE_WORKTREE_USED_RE = re.compile(
+    r"^\s*[-*]?\s*baseline (?:validation )?worktree(?: used)?\s*:\s*(.+?)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_BASELINE_TARGET_SHA_RE = re.compile(
+    r"^\s*[-*]?\s*baseline target sha\s*:\s*([0-9a-f]{7,40})\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_FAILURE_SIGNATURE_RE = re.compile(
+    r"failure signatures match\s*:\s*(true|yes)",
+    re.IGNORECASE,
+)
+_BASELINE_FAILURES_RE = re.compile(
+    r"baseline failures\s*:",
+    re.IGNORECASE,
+)
+_TRANSIENT_HISTORY_RE = re.compile(
+    r"(?:transient validation failure|earlier validation failure|"
+    r"prior failure|failure history|failed then passed)",
+    re.IGNORECASE,
+)
+_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
+
+
+def _first_match(pattern: re.Pattern[str], text: str) -> str:
+    match = pattern.search(text or "")
+    return (match.group(1).strip() if match else "")
+
+
+def _normalize_status_label(raw: str) -> str:
+    text = (raw or "").strip().lower()
+    for status in ALLOWED_VALIDATION_STATUSES:
+        if text == status.lower():
+            return status
+    return raw.strip()
+
+
+def _baseline_proof_complete(text: str, baseline_proof: dict | None) -> bool:
+    proof = baseline_proof or {}
+    worktree = (
+        (proof.get("worktree_path") or "").strip()
+        or _first_match(_BASELINE_WORKTREE_USED_RE, text)
+    ).lower()
+    if not worktree or worktree in {"none", "n/a", "not used", "not applicable"}:
+        return False
+    if "branches/" not in worktree and not worktree.startswith("branches/"):
+        return False
+    target_sha = (proof.get("baseline_target_sha") or "").strip()
+    if not target_sha:
+        target_sha = _first_match(_BASELINE_TARGET_SHA_RE, text)
+    if not _FULL_SHA.match(target_sha or ""):
+        return False
+    if proof.get("failure_signatures_match") is True:
+        return True
+    if _FAILURE_SIGNATURE_RE.search(text) and _BASELINE_FAILURES_RE.search(text):
+        return True
+    return False
+
+
+def _merge_simulation_passed(text: str, command_log: list | None) -> bool:
+    merge_result = _first_match(_MERGE_SIM_RESULT_RE, text).lower()
+    if merge_result in {"passed", "pass", "clean", "succeeded", "success"}:
+        sim = assess_merge_simulation_report(text, command_log=command_log)
+        return sim.get("proven") and not sim.get("block")
+    if "pass" in merge_result and "fail" not in merge_result:
+        sim = assess_merge_simulation_report(text, command_log=command_log)
+        return sim.get("proven") and not sim.get("block")
+    return False
+
+
+def _raw_head_failed(text: str) -> bool:
+    raw = _first_match(_RAW_HEAD_RESULT_RE, text).lower()
+    if raw in {"failed", "fail", "failure"}:
+        return True
+    if "fail" in raw and "pass" not in raw:
+        return True
+    return bool(re.search(r"\bfailed\b.*pr-head validation", text, re.IGNORECASE))
+
+
+def _raw_head_passed(text: str) -> bool:
+    raw = _first_match(_RAW_HEAD_RESULT_RE, text).lower()
+    return raw in {"passed", "pass", "success"}
+
+
+def assess_validation_status_vocabulary(
+    report_text: str,
+    *,
+    command_log: list | None = None,
+    baseline_proof: dict | None = None,
+) -> dict[str, Any]:
+    """Bind validation-status labels to the proof path that actually ran (#406)."""
+    text = report_text or ""
+    reasons: list[str] = []
+    status_raw = _first_match(_STATUS_FIELD_RE, text)
+    status = _normalize_status_label(status_raw) if status_raw else ""
+
+    if status_raw and status not in ALLOWED_VALIDATION_STATUSES:
+        reasons.append(
+            f"unknown validation status {status_raw!r}; use one of "
+            f"{sorted(ALLOWED_VALIDATION_STATUSES)}"
+        )
+
+    if status == STATUS_BASELINE_EQUIVALENT:
+        if not _baseline_proof_complete(text, baseline_proof):
+            reasons.append(
+                "baseline-equivalent failure accepted requires baseline "
+                "worktree path, baseline target SHA, and matching failure "
+                "signatures (#406)"
+            )
+
+    if status == STATUS_MERGE_SIM_RESOLVED:
+        if not _raw_head_failed(text):
+            reasons.append(
+                "raw-head failure resolved by merge simulation requires "
+                "raw PR-head validation result: failed (#406)"
+            )
+        if not _merge_simulation_passed(text, command_log):
+            reasons.append(
+                "raw-head failure resolved by merge simulation requires "
+                "passing merge simulation with worktree/index mutation proof "
+                "(#317/#406)"
+            )
+
+    if status == STATUS_TRANSIENT_PASS:
+        if not _TRANSIENT_HISTORY_RE.search(text):
+            reasons.append(
+                "passed after transient failure investigation requires "
+                "documented earlier validation failure history (#396/#406)"
+            )
+
+    if status == STATUS_PASSED and _raw_head_failed(text):
+        reasons.append(
+            "validation status passed contradicts raw PR-head validation "
+            "failure; use a precise status (#406)"
+        )
+
+    if status == STATUS_BASELINE_EQUIVALENT and _merge_simulation_passed(
+        text, command_log
+    ) and not _baseline_proof_complete(text, baseline_proof):
+        reasons.append(
+            "baseline-equivalent failure accepted is misleading when only "
+            "merge simulation resolved the failure; use "
+            "'raw-head failure resolved by merge simulation' (#406)"
+        )
+
+    if status == STATUS_FAILED and _merge_simulation_passed(text, command_log):
+        reasons.append(
+            "validation status failed contradicts passing merge simulation; "
+            "report the precise resolution status (#406)"
+        )
+
+    block = bool(reasons)
+    return {
+        "block": block,
+        "proven": not block,
+        "status_claimed": status or None,
+        "raw_status_label": status_raw or None,
+        "raw_head_failed": _raw_head_failed(text),
+        "raw_head_passed": _raw_head_passed(text),
+        "merge_simulation_passed": _merge_simulation_passed(text, command_log),
+        "baseline_proof_complete": _baseline_proof_complete(text, baseline_proof),
+        "reasons": reasons,
+        "safe_next_action": (
+            "use a validation status that matches the proof path executed "
+            "(baseline worktree, merge simulation, or transient history)"
+            if reasons
+            else "proceed"
+        ),
+    }