feat: require validation cwd and HEAD proof in reviewer reports (Closes #398)

Add assess_validation_cwd_proof_report so reviewer final reports must prove
pwd, git rev-parse HEAD, git status, and expected PR head SHA before
claiming validation ran in a branches/ worktree. Wire the verifier into
final_report_validator and gitea_validate_review_final_report.

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,
-    }

docs/gitea-dual-namespace-deployment.md

@@ -23,7 +23,7 @@ launched with exactly one static execution profile:
 |-----------------------------|----------------|-------------|
 | `gitea-author` | an author profile | implement issues, push branches, open PRs, comment |
 | `gitea-reviewer` | a reviewer profile | review, approve/request changes, merge |
-| `gitea-reconciler` | a reconciler profile | close already-landed open PRs after ancestry proof (#304 profile; #310 close tool) |
+| `gitea-reconciler` | a reconciler profile | close already-landed open PRs after ancestry proof (#310) |
 
 Properties:
 

docs/gitea-execution-profiles.md

@@ -226,12 +226,11 @@ explicit operator-directed closure of a contaminated PR). If `close_pr` ever
 resolves as unknown, agents must fail closed rather than fall back to the
 edit path.
 
-## Reconciler profile for already-landed open PRs (#304 / #310)
+## Reconciler profile for already-landed open PRs (#310)
 
-Normal author and reviewer profiles must not gain broad `gitea.pr.close`
+Normal author and reviewer profiles must not gain broad PR-close authority.
-authority. Already-landed open PRs (head SHA is an ancestor of the target
+Already-landed open PRs (head SHA is an ancestor of the target branch) need a
-branch) need a dedicated reconciler profile such as `prgs-reconciler` with a
+dedicated reconciler profile such as `prgs-reconciler` with:
-narrow operation set:
 
 - `gitea.read`
 - `gitea.pr.comment`
@@ -239,14 +238,8 @@ narrow operation set:
 - `gitea.issue.close`
 - `gitea.pr.close`
 
-Forbidden on reconciler profiles: `gitea.pr.approve`, `gitea.pr.merge`,
+Use the `gitea-reconciler` MCP namespace (static profile launch) and the
-`gitea.pr.review`, `gitea.pr.create`, `gitea.branch.push`, and
+`gitea_reconcile_already_landed_pr` tool. The resolver task is
-`gitea.repo.commit`.
-
-Launch a static `gitea-reconciler` MCP namespace with
-`GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
-`reconciler_profile.assess_reconciler_profile` (#304). Use the
-`gitea_reconcile_already_landed_pr` tool (#310). The resolver task is
 `reconcile_already_landed_pr`. PR close is allowed only after live PR fetch,
 fresh target-branch fetch, recorded target SHA, and ancestor proof. PRs whose
 heads are not already landed cannot be closed through this path.

final_report_validator.py

@@ -456,35 +456,37 @@ def _rule_reviewer_git_fetch_readonly(
     return []
 
 
-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_failure_history import (
+    from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report
-        assess_validation_failure_history_report,
-    )
 
     session = validation_session or {}
-    observed = session.get("observed_failures") or []
+    claims = (
-    if not observed:
+        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_failure_history_report(
+    result = assess_validation_cwd_proof_report(
         report_text,
         validation_session=session,
     )
-    if result.get("proven"):
+    if result.get("proven") or not result.get("claims_validation"):
         return []
     severity = "block" if result.get("violations") else "downgrade"
     return [
         validator_finding(
-            "reviewer.validation_failure_history",
+            "reviewer.validation_cwd_proof",
             severity,
-            "Validation failure history",
+            "Validation cwd/HEAD proof",
             reason,
             result.get("safe_next_action")
-            or "document every observed validation failure before claiming pass",
+            or "document pwd, HEAD SHA, and explicit cwd before validation",
         )
         for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
     ]
@@ -898,7 +900,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_mutation_categories,
         _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,9 +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
 import review_merge_state_machine  # noqa: E402
 import native_mcp_preference  # noqa: E402
@@ -1050,76 +1048,6 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
-def _list_repo_branch_names(h: str, o: str, r: str, auth: str, *, limit: int = 200) -> list[str]:
-    branches = api_get_all(f"{repo_api_url(h, o, r)}/branches", auth, limit=limit)
-    return [_branch_entry_name(branch) for branch in branches]
-
-
-def _gather_author_duplicate_work_context(
-    issue_number: int,
-    *,
-    h: str,
-    o: str,
-    r: str,
-    auth: str,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    base = repo_api_url(h, o, r)
-    issue = api_request("GET", f"{base}/issues/{issue_number}", auth)
-    comments = api_request("GET", f"{base}/issues/{issue_number}/comments", auth) or []
-    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
-    branch_names = _list_repo_branch_names(h, o, r, auth)
-    if exclude_branch_name:
-        branch_names = [
-            name for name in branch_names
-            if name != exclude_branch_name
-        ]
-    return {
-        "issue": issue,
-        "comments": comments,
-        "open_prs": open_prs,
-        "branch_names": branch_names,
-    }
-
-
-def _enforce_author_duplicate_work_gate(
-    issue_number: int,
-    stage: str,
-    *,
-    h: str,
-    o: str,
-    r: str,
-    auth: str,
-    allow_stale_takeover: bool = False,
-    exclude_branch_name: str | None = None,
-) -> dict:
-    """Fail closed when duplicate work is detected (#400)."""
-    ctx = _gather_author_duplicate_work_context(
-        issue_number,
-        h=h,
-        o=o,
-        r=r,
-        auth=auth,
-        exclude_branch_name=exclude_branch_name,
-    )
-    result = author_duplicate_work_gate.classify_and_assess(
-        ctx["issue"],
-        stage=stage,
-        comments=ctx["comments"],
-        open_prs=ctx["open_prs"],
-        branch_names=ctx["branch_names"],
-        allow_stale_takeover=allow_stale_takeover,
-    )
-    assessment = result.get("duplicate_work") or {}
-    if assessment.get("block"):
-        reasons = "; ".join(assessment.get("reasons") or ["duplicate work detected"])
-        raise RuntimeError(
-            f"Author duplicate-work gate (#400) blocked at stage '{stage}': "
-            f"{reasons} (fail closed)"
-        )
-    return result
-
-
 @mcp.tool()
 def gitea_lock_issue(
     issue_number: int,
@@ -1182,17 +1110,48 @@ def gitea_lock_issue(
             issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
         )
 
+    # 2. Check if the issue already has an open PR (reuse protection)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
-    _enforce_author_duplicate_work_gate(
+    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(
         issue_number=issue_number,
@@ -1285,7 +1244,7 @@ def gitea_create_pr(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote, worktree_path=worktree_path)
+    verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196) ──
@@ -1313,17 +1272,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 +1288,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}
@@ -3514,32 +3463,16 @@ def gitea_delete_branch(
         repo: Override the repository name.
 
     Returns:
-        dict with 'success' and 'message'; on a permission block,
+        dict with 'success' and 'message'.
-        'success'/'performed' False, 'reasons', and a structured
-        'permission_report' (#142) with no API call made.
     """
-    gate_reasons = _profile_operation_gate("gitea.branch.delete")
-    if gate_reasons:
-        return {
-            "success": False,
-            "performed": False,
-            "required_permission": "gitea.branch.delete",
-            "reasons": gate_reasons,
-            "permission_report": _permission_block_report("gitea.branch.delete"),
-        }
-
     verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     import urllib.parse
     encoded_branch = urllib.parse.quote(branch, safe="")
     url = f"{repo_api_url(h, o, r)}/branches/{encoded_branch}"
-    request_metadata = {
-        "branch": branch,
-        "required_permission": "gitea.branch.delete",
-    }
     with _audited("delete_branch", host=h, remote=remote, org=o, repo=r,
-                  target_branch=branch, request_metadata=request_metadata):
+                  target_branch=branch, request_metadata={"branch": branch}):
         api_request("DELETE", url, auth)
     return {"success": True, "message": f"Remote branch '{branch}' deleted."}
 
@@ -4539,15 +4472,12 @@ def gitea_create_issue_comment(
 def _role_kind(allowed, forbidden) -> str:
     """Classify the active profile from its normalized operations.
 
-    'reconciler' can close already-landed PRs without review/author powers;
     'author' can create PRs / push branches; 'reviewer' can approve/merge;
     'reconciler' can close already-landed PRs via ``gitea.pr.close`` without
     review/author powers; 'mixed' can do both (a config smell — the
     reviewer-deadlock invariant forbids it in v2 configs); 'limited' can do
     neither.
     """
-    if reconciler_profile.is_reconciler_profile(allowed, forbidden):
-        return "reconciler"
     def can(op):
         return gitea_config.check_operation(op, allowed, forbidden)[0]
     review = can("gitea.pr.approve") or can("gitea.pr.merge")
@@ -5716,10 +5646,6 @@ def gitea_get_runtime_context(
         "preflight_block_reasons": preflight["preflight_block_reasons"],
         "preflight_workspace": preflight.get("preflight_workspace"),
         "session_capabilities": session_capabilities,
-        "reconciler_profile_assessment": reconciler_profile.assess_reconciler_profile(
-            allowed, forbidden
-        ),
-        "role_kind": _role_kind(allowed, forbidden),
         "shell_health": native_mcp_preference.shell_health_status(),
     }
 
@@ -6053,14 +5979,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 +6060,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",

reconciler_profile.py

@@ -1,94 +0,0 @@
-"""Reconciler profile model for already-landed PR closure (#304).
-
-Defines the narrowly scoped operation set for a dedicated reconciler profile
-such as ``prgs-reconciler``. Close MCP tooling and ancestry gates ship in the
-#310 stack; this module validates profile shape only.
-"""
-
-from __future__ import annotations
-
-import gitea_config
-
-RECONCILER_REQUIRED_OPERATIONS = (
-    "gitea.read",
-    "gitea.pr.close",
-)
-
-RECONCILER_RECOMMENDED_OPERATIONS = (
-    "gitea.pr.comment",
-    "gitea.issue.comment",
-    "gitea.issue.close",
-)
-
-RECONCILER_FORBIDDEN_OPERATIONS = (
-    "gitea.pr.approve",
-    "gitea.pr.merge",
-    "gitea.pr.review",
-    "gitea.pr.create",
-    "gitea.branch.push",
-    "gitea.repo.commit",
-)
-
-
-def _normalized_allowed(allowed: list[str]) -> set[str]:
-    normalized: set[str] = set()
-    for entry in allowed or []:
-        try:
-            normalized.add(gitea_config.normalize_operation(entry))
-        except gitea_config.ConfigError:
-            continue
-    return normalized
-
-
-def _forbidden_in_allowed(allowed: list[str], ops: tuple[str, ...]) -> list[str]:
-    """Return forbidden ops that are explicitly listed in *allowed*."""
-    allowed_n = _normalized_allowed(allowed)
-    present: list[str] = []
-    for op in ops:
-        try:
-            if gitea_config.normalize_operation(op) in allowed_n:
-                present.append(op)
-        except gitea_config.ConfigError:
-            continue
-    return present
-
-
-def is_reconciler_profile(allowed: list[str], forbidden: list[str]) -> bool:
-    """Return True when *allowed*/*forbidden* describe a reconciler profile."""
-    def can(op: str) -> bool:
-        return gitea_config.check_operation(op, allowed, forbidden)[0]
-
-    if not can("gitea.pr.close"):
-        return False
-    if _forbidden_in_allowed(allowed, RECONCILER_FORBIDDEN_OPERATIONS):
-        return False
-    return True
-
-
-def assess_reconciler_profile(allowed: list[str], forbidden: list[str]) -> dict:
-    """Validate reconciler profile operations (read-only, fail closed)."""
-    allowed = list(allowed or [])
-    forbidden = list(forbidden or [])
-    reasons: list[str] = []
-
-    for op in RECONCILER_REQUIRED_OPERATIONS:
-        ok, _ = gitea_config.check_operation(op, allowed, forbidden)
-        if not ok:
-            reasons.append(f"missing required operation {op}")
-
-    for op in _forbidden_in_allowed(allowed, RECONCILER_FORBIDDEN_OPERATIONS):
-        reasons.append(f"forbidden operation must not be allowed: {op}")
-
-    missing_recommended = [
-        op for op in RECONCILER_RECOMMENDED_OPERATIONS
-        if not gitea_config.check_operation(op, allowed, forbidden)[0]
-    ]
-
-    return {
-        "is_reconciler_profile": is_reconciler_profile(allowed, forbidden),
-        "valid": not reasons and is_reconciler_profile(allowed, forbidden),
-        "reasons": reasons,
-        "missing_recommended_operations": missing_recommended,
-        "required_operations": list(RECONCILER_REQUIRED_OPERATIONS),
-        "forbidden_operations": list(RECONCILER_FORBIDDEN_OPERATIONS),
-    }

review_proofs.py

@@ -1735,11 +1735,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         if report_text
         else {"proven": True, "block": False, "reasons": []}
     )
-    proof_backed_handoff = (
-        assess_proof_backed_handoff_report(report_text)
-        if report_text and _PROOF_BACKED_HANDOFF_HINT.search(report_text)
-        else {"proven": True, "block": False, "reasons": []}
-    )
     if baseline_validation is None and report_text:
         baseline_validation = assess_reviewer_baseline_validation_proof(
             report_text=report_text,
@@ -1942,11 +1937,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "reviewer baseline validation proof missing or failed (#325)"
         )
         downgrade_reasons.extend(baseline_validation.get("reasons", []))
-    if not proof_backed_handoff.get("proven"):
-        downgrade_reasons.append(
-            "proof-sensitive handoff claims lack command/tool evidence (#395)"
-        )
-        downgrade_reasons.extend(proof_backed_handoff.get("reasons", []))
     if not already_landed_classification.get("proven"):
         downgrade_reasons.append(
             "already-landed classification wording missing or invalid (#295)"
@@ -3622,33 +3612,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 +3631,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 {
@@ -5116,12 +5088,6 @@ _PR_QUEUE_CLEANUP_REPORT_HINT = re.compile(
     re.I,
 )
 
-_PROOF_BACKED_HANDOFF_HINT = re.compile(
-    r"task:\s*review-merge-pr|task mode:\s*review-merge-pr|"
-    r"review-merge-pr|controller handoff",
-    re.I,
-)
-
 
 def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
     """#390: validate PR-only cleanup final reports (one PR per run)."""
@@ -5521,11 +5487,9 @@ def assess_validation_integrity_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
-def assess_validation_failure_history_report(report_text, **kwargs):
+def assess_validation_cwd_proof_report(report_text, **kwargs):
-    """#396: final reports must account for transient validation failures."""
+    """#398: validation commands require explicit worktree cwd and HEAD proof."""
-    from reviewer_validation_failure_history import (
+    from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report as _assess
-        assess_validation_failure_history_report as _assess,
-    )
 
     return _assess(report_text, **kwargs)
 
@@ -5609,10 +5573,3 @@ def assess_worktree_ownership_report(report_text, **kwargs):
     from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess
 
     return _assess(report_text, **kwargs)
-
-
-def assess_proof_backed_handoff_report(report_text, **kwargs):
-    """#395: proof-sensitive review handoff claims require explicit evidence."""
-    from reviewer_proof_backed_handoff import assess_proof_backed_handoff_report as _assess
-
-    return _assess(report_text, **kwargs)

reviewer_proof_backed_handoff.py

@@ -1,217 +0,0 @@
-"""Proof-backed reviewer handoff claim verifier (#395)."""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-_PAGINATION_FINALITY = re.compile(
-    r"has_more\s*:\s*false|is_final_page\s*:\s*true|"
-    r"inventory_complete\s*:\s*true|pages_fetched|total_count\s*:",
-    re.I,
-)
-_PAGE_SIZE_ONLY = re.compile(
-    r"(?:less|fewer) than (?:the )?(?:default )?page[- ]?size|"
-    r"under (?:the )?50[- ]?(?:item )?limit|"
-    r"returned \d+ (?:open )?prs?.*less than 50",
-    re.I,
-)
-_INVENTORY_COMPLETE_CLAIM = re.compile(
-    r"\b(?:inventory (?:is )?complete|inventory exhaustive|"
-    r"complete (?:pr )?inventory)\b",
-    re.I,
-)
-_SKIP_CLAIM = re.compile(
-    r"\b(?:earlier prs? skipped|skipped (?:earlier )?pr|"
-    r"skip(?:ped)? pr #?\d+|non-mergeable|prior request.changes)\b",
-    re.I,
-)
-_CONFLICT_PROOF = re.compile(
-    r"merge simulation|git merge --no-commit|conflicting files|"
-    r"conflict proof|merge_exit|non-mergeable",
-    re.I,
-)
-_BASELINE_CLAIM = re.compile(
-    r"\b(?:baseline (?:validation|worktree|comparison)|"
-    r"same as master|pre-existing (?:on )?master|"
-    r"failure signatures match)\b",
-    re.I,
-)
-_BASELINE_PATH = re.compile(r"baseline worktree path\s*:\s*(\S+)", re.I)
-_BASELINE_SHA = re.compile(r"baseline (?:target )?sha\s*:\s*([0-9a-f]{7,40})", re.I)
-_BASELINE_DIRTY_BEFORE = re.compile(
-    r"baseline.*dirty before|dirty before.*baseline", re.I
-)
-_BASELINE_DIRTY_AFTER = re.compile(
-    r"baseline.*dirty after|dirty after.*baseline", re.I
-)
-_BASELINE_COMMAND = re.compile(
-    r"baseline.*(?:validation )?command|pytest.*baseline", re.I
-)
-_BASELINE_RESULT = re.compile(
-    r"baseline.*(?:validation )?result|baseline_exit|baseline failures", re.I
-)
-_MASTER_INTEGRATION = re.compile(
-    r"\b(?:merged master into|merge(?:d)? (?:remote[- ]tracking )?branch.*master|"
-    r"master integration|integrated master|rebase.*master)\b",
-    re.I,
-)
-_CLEANUP_CLAIM = re.compile(
-    r"\b(?:worktree(?:s)? (?:were )?cleaned|cleanup (?:result|mutations)|"
-    r"removed (?:session[- ]owned )?worktree|git worktree remove)\b",
-    re.I,
-)
-_WORKTREE_LIST = re.compile(r"git worktree list|worktree list proof", re.I)
-_PROOF_SOURCE = re.compile(
-    r"proof source\s*:\s*(command|mcp metadata|prior blocker|not checked)",
-    re.I,
-)
-_HANDOFF_SECTION = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M)
-
-
-def _handoff_fields(report_text: str) -> dict[str, str]:
-    text = report_text or ""
-    match = _HANDOFF_SECTION.search(text)
-    if not match:
-        return {}
-    fields: dict[str, str] = {}
-    for line in text[match.end() :].splitlines():
-        stripped = line.strip().lstrip("-*").strip()
-        if ":" not in stripped:
-            continue
-        key, value = stripped.split(":", 1)
-        fields[key.strip().lower()] = value.strip()
-    return fields
-
-
-def _command_text(entry: Any) -> str:
-    if isinstance(entry, dict):
-        return str(entry.get("command") or entry.get("tool") or "").strip()
-    return str(entry or "").strip()
-
-
-def _action_log_has(action_log: list | None, pattern: re.Pattern[str]) -> bool:
-    for entry in action_log or []:
-        blob = " ".join(
-            filter(
-                None,
-                [
-                    _command_text(entry),
-                    str(entry.get("result") or "") if isinstance(entry, dict) else "",
-                    str(entry.get("reason") or "") if isinstance(entry, dict) else "",
-                ],
-            )
-        )
-        if pattern.search(blob):
-            return True
-    return False
-
-
-def assess_proof_backed_handoff_report(
-    report_text: str,
-    *,
-    action_log: list | None = None,
-    inventory_session: dict | None = None,
-    baseline_session: dict | None = None,
-    skip_session: list[dict] | None = None,
-) -> dict[str, Any]:
-    """Require explicit command/tool evidence for proof-sensitive review claims (#395)."""
-    text = report_text or ""
-    fields = _handoff_fields(text)
-    reasons: list[str] = []
-    inventory = dict(inventory_session or {})
-    baseline = dict(baseline_session or {})
-    skips = list(skip_session or [])
-
-    pagination_field = fields.get("inventory pagination proof", "")
-    if _INVENTORY_COMPLETE_CLAIM.search(text) or _PAGE_SIZE_ONLY.search(text):
-        has_meta = bool(
-            inventory.get("inventory_complete")
-            or inventory.get("pagination_complete")
-            or _PAGINATION_FINALITY.search(pagination_field)
-            or _PAGINATION_FINALITY.search(text)
-            or _action_log_has(action_log, re.compile(r"gitea_list_prs", re.I))
-        )
-        if _PAGE_SIZE_ONLY.search(text) and not has_meta:
-            reasons.append(
-                "inventory completeness claimed from page-size assumption only; "
-                "require has_more=false, is_final_page=true, or inventory_complete=true"
-            )
-        elif _INVENTORY_COMPLETE_CLAIM.search(text) and not has_meta:
-            reasons.append(
-                "inventory complete claim lacks explicit pagination metadata proof"
-            )
-
-    if _SKIP_CLAIM.search(text) or fields.get("earlier prs skipped", "").lower() not in {
-        "",
-        "none",
-        "n/a",
-    }:
-        skip_proof = bool(
-            skips
-            or _CONFLICT_PROOF.search(text)
-            or _action_log_has(
-                action_log, re.compile(r"git merge --no-commit|gitea_get_pr_review_feedback", re.I)
-            )
-        )
-        if not skip_proof:
-            reasons.append(
-                "earlier PR skip claim lacks command/tool conflict or blocker proof"
-            )
-
-    if _BASELINE_CLAIM.search(text) or fields.get("baseline worktree used", "").lower() == "true":
-        baseline_ok = bool(
-            baseline.get("complete")
-            or (
-                _BASELINE_PATH.search(text)
-                and _BASELINE_SHA.search(text)
-                and (_BASELINE_DIRTY_BEFORE.search(text) or baseline.get("clean_before"))
-                and (_BASELINE_COMMAND.search(text) or baseline.get("command"))
-                and (_BASELINE_RESULT.search(text) or baseline.get("result"))
-                and (_BASELINE_DIRTY_AFTER.search(text) or baseline.get("clean_after"))
-            )
-        )
-        if not baseline_ok:
-            reasons.append(
-                "baseline validation claim missing worktree path, target SHA, "
-                "dirty-before/after status, command, or result proof"
-            )
-
-    if _MASTER_INTEGRATION.search(text):
-        if not _action_log_has(
-            action_log,
-            re.compile(r"git merge.*master|git rebase.*master", re.I),
-        ) and not re.search(r"merge_exit\s*=\s*\d+|merge simulation", text, re.I):
-            reasons.append(
-                "master integration claim lacks exact merge/rebase command and result"
-            )
-
-    if _CLEANUP_CLAIM.search(text) or fields.get("cleanup mutations", "").lower() not in {
-        "",
-        "none",
-        "n/a",
-    }:
-        if not (_WORKTREE_LIST.search(text) or _action_log_has(action_log, _WORKTREE_LIST)):
-            reasons.append(
-                "cleanup claim lacks final git worktree list or equivalent proof"
-            )
-
-    if re.search(r"\blive proof\b", text, re.I) and not _PROOF_SOURCE.search(text):
-        if not action_log:
-            reasons.append(
-                "live proof wording requires proof source classification "
-                "(command, MCP metadata, prior blocker, or not checked)"
-            )
-
-    proven = not reasons
-    return {
-        "proven": proven,
-        "block": not proven,
-        "reasons": reasons,
-        "safe_next_action": (
-            "cite explicit command/tool evidence or structured MCP pagination metadata "
-            "for each proof-sensitive claim"
-            if not proven
-            else "proceed"
-        ),
-    }

reviewer_validation_cwd_proof.py

@@ -0,0 +1,233 @@
+"""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"
+        ),
+    }

reviewer_validation_failure_history.py

@@ -1,198 +0,0 @@
-"""Transient validation failure history verifier (#396).
-
-Reviewer sessions may observe validation failures that later pass on rerun.
-Final reports must document every failure observed during the session, not
-only the last passing result.
-"""
-
-from __future__ import annotations
-
-import re
-from typing import Any
-
-_SECTION_RE = re.compile(
-    r"validation failure history",
-    re.IGNORECASE,
-)
-_FAILURE_ENTRY_RE = re.compile(
-    r"(?:failure\s*(?:#|entry)?\s*\d+|validation failure)\s*:",
-    re.IGNORECASE,
-)
-_COMMAND_RE = re.compile(
-    r"(?:command|validation command)\s*:\s*(.+)",
-    re.IGNORECASE,
-)
-_FAILING_TEST_RE = re.compile(
-    r"(?:failing test|failure|error)\s*:\s*(.+)",
-    re.IGNORECASE,
-)
-_CAUSE_RE = re.compile(
-    r"(?:suspected cause|cause)\s*:\s*(.+)",
-    re.IGNORECASE,
-)
-_REPRODUCED_RE = re.compile(
-    r"(?:reproduced|reproduces)\s*:\s*(yes|no|unknown)",
-    re.IGNORECASE,
-)
-_BASELINE_RE = re.compile(
-    r"(?:on baseline master|baseline master|exists on baseline)\s*:\s*(yes|no|unknown|not checked)",
-    re.IGNORECASE,
-)
-_PR_CAUSED_RE = re.compile(
-    r"(?:pr[- ]caused|pr caused)\s*:\s*(yes|no|unknown|not proven)",
-    re.IGNORECASE,
-)
-_TRANSIENT_STATUS_RE = re.compile(
-    r"(?:passed after transient failure investigation|transient failure investigation)",
-    re.IGNORECASE,
-)
-_PLAIN_PASS_RE = re.compile(
-    r"validation\s*:\s*(?:pass|passed|strong|ok|green)\b",
-    re.IGNORECASE,
-)
-_ENV_CLEANUP_RE = re.compile(
-    r"(?:environmental cleanup|state cleaned|what changed between runs)\s*:",
-    re.IGNORECASE,
-)
-_UNKNOWN_CAUSE_RE = re.compile(
-    r"(?:suspected cause|cause)\s*:\s*(?:unknown|unexplained|not determined)",
-    re.IGNORECASE,
-)
-
-
-def _failure_documented_in_text(text: str, failure: dict[str, Any]) -> bool:
-    """Return True when *failure* appears documented in free-form report text."""
-    command = (failure.get("command") or "").strip()
-    failing = (failure.get("failing_test") or failure.get("error") or "").strip()
-    if command and command not in text:
-        return False
-    if failing and failing not in text:
-        return False
-    return bool(command or failing)
-
-
-def _section_has_structured_fields(text: str) -> bool:
-    if not _SECTION_RE.search(text):
-        return False
-    section_start = _SECTION_RE.search(text).start()
-    section = text[section_start:]
-    has_command = bool(_COMMAND_RE.search(section))
-    has_failure = bool(_FAILING_TEST_RE.search(section))
-    return has_command and has_failure
-
-
-def assess_validation_failure_history_report(
-    report_text: str,
-    *,
-    validation_session: dict | None = None,
-) -> dict[str, Any]:
-    """Require final reports to account for transient validation failures (#396)."""
-    text = report_text or ""
-    session = dict(validation_session or {})
-    reasons: list[str] = []
-    violations: list[str] = []
-
-    observed = list(session.get("observed_failures") or [])
-    final_status = (session.get("final_validation_status") or "").strip().lower()
-    cause_unknown = any(
-        (f.get("suspected_cause") or "").strip().lower() in {
-            "unknown", "unexplained", "not determined", ""
-        }
-        and not (f.get("pr_caused") or "").strip().lower() in {"yes", "no"}
-        for f in observed
-    ) or bool(session.get("cause_unknown"))
-
-    if not observed:
-        return {
-            "proven": True,
-            "block": False,
-            "reasons": [],
-            "violations": [],
-            "observed_failure_count": 0,
-            "safe_next_action": "proceed",
-        }
-
-    documented = _section_has_structured_fields(text)
-    if not documented:
-        for failure in observed:
-            if _failure_documented_in_text(text, failure):
-                documented = True
-                break
-
-    if not documented:
-        violations.append(
-            "session observed validation failure(s) but final report omits "
-            "Validation failure history"
-        )
-        reasons.append(
-            "every validation failure observed during the session must appear "
-            "in a Validation failure history section"
-        )
-
-    if documented and not _SECTION_RE.search(text):
-        reasons.append(
-            "failure details must appear under an explicit "
-            "'Validation failure history' heading"
-        )
-
-    for idx, failure in enumerate(observed, start=1):
-        prefix = f"failure #{idx}"
-        if not _failure_documented_in_text(text, failure) and documented:
-            reasons.append(
-                f"{prefix}: command and failing test/error not documented in report"
-            )
-        command = (failure.get("command") or "").strip()
-        failing = (failure.get("failing_test") or failure.get("error") or "").strip()
-        if not command:
-            reasons.append(f"{prefix}: missing command in session failure record")
-        if not failing:
-            reasons.append(
-                f"{prefix}: missing failing test or error in session failure record"
-            )
-
-    plain_pass = bool(_PLAIN_PASS_RE.search(text))
-    transient_wording = bool(_TRANSIENT_STATUS_RE.search(text))
-    final_passed = final_status in {
-        "passed",
-        "pass",
-        "passed_after_transient_failure_investigation",
-    }
-
-    if (final_passed or plain_pass) and observed:
-        if cause_unknown and not transient_wording:
-            violations.append(
-                "unknown transient failure cause cannot be erased as plain 'passed'"
-            )
-            reasons.append(
-                "when cause is unknown, use status "
-                "'passed after transient failure investigation'"
-            )
-        elif not transient_wording and final_status != "passed_after_transient_failure_investigation":
-            if not _ENV_CLEANUP_RE.search(text):
-                reasons.append(
-                    "later passing rerun must document what changed between runs "
-                    "or environmental cleanup performed"
-                )
-
-    if session.get("environmental_contamination") and not _ENV_CLEANUP_RE.search(text):
-        reasons.append(
-            "environmental /tmp state contamination must document cleanup or "
-            "what changed before the passing rerun"
-        )
-
-    proven = not reasons and not violations
-    return {
-        "proven": proven,
-        "block": bool(violations) or not proven,
-        "reasons": reasons,
-        "violations": violations,
-        "observed_failure_count": len(observed),
-        "documented": documented,
-        "safe_next_action": (
-            "add Validation failure history with command, failing test, cause, "
-            "reproduction, baseline comparison, and PR-caused evidence; use "
-            "'passed after transient failure investigation' when appropriate"
-            if not proven
-            else "proceed"
-        ),
-    }

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

@@ -92,22 +92,3 @@ Verifier: `review_proofs.assess_queue_status_report()`.
 Narrative final report and controller handoff must agree on eligibility class,
 candidate/reviewed head SHA, mutation state, worktree usage, review decision,
 terminal review mutation, merge result, and linked issue status.
-
-### Proof-backed claims (#395)
-
-Proof-sensitive claims must cite explicit command/tool evidence in the report
-or structured MCP metadata — not narrative alone:
-
-- **Inventory complete:** `has_more=false`, `is_final_page=true`,
-  `inventory_complete=true`, and/or `total_count` from `gitea_list_prs`.
-- **Skipped earlier PRs:** merge-simulation command output, conflict file list,
-  or live `gitea_get_pr_review_feedback` / mergeability fields.
-- **Baseline validation:** baseline worktree path, baseline target SHA,
-  dirty-before/after, exact command, exact result.
-- **Master integration:** exact `git merge` / `git rebase` command and exit status.
-- **Cleanup:** final `git worktree list` (or remove commands) proving session
-  worktrees were removed.
-
-When a claim relies on prior-session blocker state or MCP metadata only, label
-the proof source explicitly (`command`, `MCP metadata`, `prior blocker`,
-`not checked`). Do not use `live proof` without that classification.

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

@@ -538,36 +538,6 @@ Official validation integrity status must be one of:
 
 Do not invent a softer status.
 
-## 21A. Validation failure history rule
-
-Every validation failure observed during the review session must appear in the
-final report, even if a later rerun passes.
-
-Before claiming `Validation: passed`, list every failed validation command from
-the session under `Validation failure history`.
-
-For each failure, report:
-
-* command
-* failing test or error
-* suspected cause
-* whether it reproduced
-* whether it exists on baseline master
-* evidence for whether it is or is not PR-caused
-
-If a later rerun passes, also report:
-
-* what changed between runs
-* whether environmental state was cleaned (for example `/tmp` lock files)
-* why the pass is trustworthy
-
-If the cause is unknown, do not erase the earlier failure with plain
-`Validation: passed`. Use a status such as
-`passed after transient failure investigation`.
-
-`gitea_validate_review_final_report` rejects reports that omit known earlier
-validation failures when `validation_session.observed_failures` is supplied.
-
 ## 22. Baseline validation rule
 
 Do not run tests in the main checkout.
@@ -604,6 +574,28 @@ 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.
@@ -1126,7 +1118,7 @@ Controller Handoff:
 * Baseline worktree path:
 * 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,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_delete_branch_capability.py

@@ -1,217 +0,0 @@
-"""Tests for gitea_delete_branch capability gate (Issue #408).
-
-``gitea_delete_branch`` requires the exact ``gitea.branch.delete`` operation:
-without it the delete fails closed (no preflight, no auth lookup, no API call,
-structured permission report). With it, deletion proceeds through existing
-preflight and audit unchanged.
-"""
-import json
-import os
-import sys
-import tempfile
-import unittest
-from unittest.mock import patch
-
-sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
-
-import mcp_server
-from mcp_server import gitea_delete_branch
-
-FAKE_AUTH = "token fake"
-
-AUTHOR_NO_DELETE = {
-    "profile_name": "prgs-author",
-    "allowed_operations": [
-        "gitea.read", "gitea.pr.create", "gitea.pr.comment",
-        "gitea.branch.push", "gitea.issue.comment",
-    ],
-    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
-    "audit_label": "prgs-author",
-}
-
-AUTHOR_WITH_DELETE = {
-    "profile_name": "prgs-author-deleter",
-    "allowed_operations": AUTHOR_NO_DELETE["allowed_operations"] + [
-        "gitea.branch.delete",
-    ],
-    "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
-    "audit_label": "prgs-author-deleter",
-}
-
-CONFIG = {
-    "version": 2,
-    "contexts": {
-        "ctx": {
-            "enabled": True,
-            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
-        }
-    },
-    "profiles": {
-        "author-no-delete": {
-            "enabled": True,
-            "context": "ctx",
-            "role": "author",
-            "username": "author-user",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
-            "allowed_operations": AUTHOR_NO_DELETE["allowed_operations"],
-            "forbidden_operations": [],
-            "execution_profile": "author-no-delete",
-        },
-        "author-with-delete": {
-            "enabled": True,
-            "context": "ctx",
-            "role": "author",
-            "username": "deleter-user",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
-            "allowed_operations": AUTHOR_WITH_DELETE["allowed_operations"],
-            "forbidden_operations": [],
-            "execution_profile": "author-with-delete",
-        },
-        "reviewer-profile": {
-            "enabled": True,
-            "context": "ctx",
-            "role": "reviewer",
-            "username": "reviewer-user",
-            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
-            "allowed_operations": [
-                "gitea.read", "gitea.pr.review", "gitea.pr.merge",
-            ],
-            "forbidden_operations": [
-                "gitea.branch.delete", "gitea.branch.push", "gitea.pr.create",
-            ],
-            "execution_profile": "reviewer-profile",
-        },
-    },
-    "rules": {"allow_runtime_switching": False},
-}
-
-
-class TestDeleteBranchToolGate(unittest.TestCase):
-
-    def setUp(self):
-        self._preflight_snapshot = (
-            mcp_server._preflight_whoami_called,
-            mcp_server._preflight_capability_called,
-        )
-        mcp_server._preflight_whoami_called = False
-        mcp_server._preflight_capability_called = False
-        self._remotes = patch.dict(mcp_server.REMOTES, {
-            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
-                     "repo": "Example-Repo"},
-        })
-        self._remotes.start()
-        mcp_server._IDENTITY_CACHE.clear()
-        patch("gitea_config.load_config", return_value={}).start()
-        patch(
-            "gitea_config.is_runtime_switching_enabled", return_value=False
-        ).start()
-        patch("gitea_audit.audit_enabled", return_value=False).start()
-        self.mock_api = patch("mcp_server.api_request").start()
-        self.mock_auth = patch(
-            "mcp_server.get_auth_header", return_value=FAKE_AUTH
-        ).start()
-
-    def tearDown(self):
-        patch.stopall()
-        mcp_server._IDENTITY_CACHE.clear()
-        mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (
-            self._preflight_snapshot
-        )
-
-    def _set_profile(self, profile):
-        patch("mcp_server.get_profile", return_value=profile).start()
-
-    def test_blocked_without_delete_capability(self):
-        self._set_profile(AUTHOR_NO_DELETE)
-        res = gitea_delete_branch(branch="feat/branch", remote="prgs")
-        self.assertFalse(res["success"])
-        self.assertFalse(res["performed"])
-        self.assertEqual(res["required_permission"], "gitea.branch.delete")
-        self.assertTrue(res["reasons"])
-        self.assertEqual(
-            res["permission_report"]["missing_permission"],
-            "gitea.branch.delete",
-        )
-        self.mock_api.assert_not_called()
-        self.mock_auth.assert_not_called()
-
-    def test_allowed_delete_proceeds(self):
-        self._set_profile(AUTHOR_WITH_DELETE)
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check("capability", resolved_role="author")
-        res = gitea_delete_branch(branch="feat/branch", remote="prgs")
-        self.assertTrue(res["success"])
-        self.assertIn("deleted", res["message"])
-        delete_calls = [
-            c for c in self.mock_api.call_args_list if c.args[0] == "DELETE"
-        ]
-        self.assertTrue(delete_calls)
-
-    def test_allowed_delete_audited_with_capability_proof(self):
-        self._set_profile(AUTHOR_WITH_DELETE)
-        patch("gitea_audit.audit_enabled", return_value=True).start()
-        mock_write = patch("gitea_audit.write_event").start()
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check("capability", resolved_role="author")
-        self.mock_api.return_value = {}
-        gitea_delete_branch(branch="feat/branch", remote="prgs")
-        mock_write.assert_called()
-        event = mock_write.call_args[0][0]
-        self.assertEqual(event["action"], "delete_branch")
-        self.assertEqual(
-            event["request_metadata"]["required_permission"],
-            "gitea.branch.delete",
-        )
-
-
-class TestDeleteBranchResolverParity(unittest.TestCase):
-    def setUp(self):
-        self._remotes = patch.dict(mcp_server.REMOTES, {
-            "prgs": {"host": "gitea.example.com", "org": "Example-Org",
-                     "repo": "Example-Repo"},
-        })
-        self._remotes.start()
-        mcp_server._IDENTITY_CACHE.clear()
-        self._dir = tempfile.TemporaryDirectory()
-        self.config_path = os.path.join(self._dir.name, "profiles.json")
-        with open(self.config_path, "w", encoding="utf-8") as fh:
-            fh.write(json.dumps(CONFIG))
-        patch(
-            "gitea_config.is_runtime_switching_enabled", return_value=False
-        ).start()
-        patch("gitea_audit.audit_enabled", return_value=False).start()
-        self.mock_api = patch("mcp_server.api_request").start()
-        patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
-
-    def tearDown(self):
-        patch.stopall()
-        mcp_server._IDENTITY_CACHE.clear()
-        self._dir.cleanup()
-
-    def _env(self, profile: str) -> dict:
-        return {
-            "GITEA_MCP_CONFIG": self.config_path,
-            "GITEA_MCP_PROFILE": profile,
-            "GITEA_TOKEN_AUTHOR": "author-pass",
-            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
-        }
-
-    def test_reviewer_resolver_denial_blocks_raw_tool(self):
-        with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
-            resolve = mcp_server.gitea_resolve_task_capability(
-                task="delete_branch", remote="prgs")
-            self.assertFalse(resolve["allowed_in_current_session"])
-            res = gitea_delete_branch(branch="feat/branch", remote="prgs")
-        self.assertFalse(res["success"])
-        self.assertEqual(
-            res["permission_report"]["missing_permission"],
-            resolve["required_operation_permission"],
-        )
-        delete_calls = [
-            c for c in self.mock_api.call_args_list if c.args[0] == "DELETE"
-        ]
-        self.assertFalse(delete_calls)
-
-
-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
@@ -213,10 +207,10 @@ def test_validation_integrity_verifier_exported():
     assert callable(assess_validation_integrity_report)
 
 
-def test_validation_failure_history_verifier_exported():
+def test_validation_cwd_proof_verifier_exported():
-    from review_proofs import assess_validation_failure_history_report
+    from review_proofs import assess_validation_cwd_proof_report
 
-    assert callable(assess_validation_failure_history_report)
+    assert callable(assess_validation_cwd_proof_report)
 
 
 def test_prior_blocker_skip_verifier_exported():
@@ -261,12 +255,6 @@ def test_inventory_worktree_verifier_exported():
     assert callable(assess_inventory_worktree_report)
 
 
-def test_proof_backed_handoff_verifier_exported():
-    from review_proofs import assess_proof_backed_handoff_report
-
-    assert callable(assess_proof_backed_handoff_report)
-
-
 def test_infra_stop_handoff_verifier_exported():
     from review_proofs import assess_infra_stop_handoff_report
 

tests/test_mcp_server.py

@@ -1005,19 +1005,9 @@ class TestReviewPR(unittest.TestCase):
 # ---------------------------------------------------------------------------
 class TestDeleteBranch(unittest.TestCase):
 
-    DELETE_PROFILE = {
-        "profile_name": "test-deleter",
-        "allowed_operations": ["gitea.read", "gitea.branch.delete"],
-        "forbidden_operations": [],
-        "audit_label": "test-deleter",
-    }
-
-    @patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
-    def test_delete_branch(self, _auth, mock_api, _profile):
+    def test_delete_branch(self, _auth, mock_api):
-        mcp_server.record_preflight_check("whoami")
-        mcp_server.record_preflight_check("capability", resolved_role="author")
         mock_api.return_value = {}
         result = gitea_delete_branch(branch="feat/branch")
         self.assertTrue(result["success"])

tests/test_reconciler_profile.py

@@ -1,87 +0,0 @@
-"""Tests for reconciler profile model (#304)."""
-
-import os
-import sys
-import unittest
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-import gitea_mcp_server as mcp_server
-import migrate_profiles
-import reconciler_profile
-
-
-PRGS_RECONCILER_ALLOWED = [
-    "gitea.read",
-    "gitea.pr.comment",
-    "gitea.issue.comment",
-    "gitea.issue.close",
-    "gitea.pr.close",
-]
-PRGS_RECONCILER_FORBIDDEN = [
-    "gitea.pr.approve",
-    "gitea.pr.merge",
-    "gitea.pr.create",
-    "gitea.branch.push",
-]
-
-
-class TestReconcilerProfileModel(unittest.TestCase):
-    def test_prgs_reconciler_shape_valid(self):
-        result = reconciler_profile.assess_reconciler_profile(
-            PRGS_RECONCILER_ALLOWED,
-            PRGS_RECONCILER_FORBIDDEN,
-        )
-        self.assertTrue(result["is_reconciler_profile"])
-        self.assertTrue(result["valid"])
-        self.assertEqual(result["reasons"], [])
-
-    def test_author_profile_not_reconciler(self):
-        allowed = [
-            "gitea.read",
-            "gitea.pr.create",
-            "gitea.branch.push",
-            "gitea.issue.comment",
-        ]
-        forbidden = ["gitea.pr.approve", "gitea.pr.merge"]
-        self.assertFalse(
-            reconciler_profile.is_reconciler_profile(allowed, forbidden)
-        )
-
-    def test_reviewer_profile_not_reconciler(self):
-        allowed = [
-            "gitea.read",
-            "gitea.pr.review",
-            "gitea.pr.approve",
-            "gitea.pr.merge",
-        ]
-        forbidden = ["gitea.pr.create", "gitea.branch.push"]
-        self.assertFalse(
-            reconciler_profile.is_reconciler_profile(allowed, forbidden)
-        )
-
-    def test_broad_close_on_author_invalid(self):
-        allowed = PRGS_RECONCILER_ALLOWED + ["gitea.pr.create"]
-        result = reconciler_profile.assess_reconciler_profile(
-            allowed,
-            PRGS_RECONCILER_FORBIDDEN,
-        )
-        self.assertFalse(result["valid"])
-        self.assertFalse(result["is_reconciler_profile"])
-
-    def test_role_kind_detects_reconciler(self):
-        role = mcp_server._role_kind(
-            PRGS_RECONCILER_ALLOWED,
-            PRGS_RECONCILER_FORBIDDEN,
-        )
-        self.assertEqual(role, "reconciler")
-
-    def test_migrate_infer_role_reconciler(self):
-        self.assertEqual(
-            migrate_profiles.infer_role("prgs-reconciler", "prgs-reconciler"),
-            "reconciler",
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_reviewer_proof_backed_handoff.py

@@ -1,117 +0,0 @@
-"""Tests for proof-backed reviewer handoff verifier (#395)."""
-import sys
-import unittest
-from pathlib import Path
-
-sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
-
-from reviewer_proof_backed_handoff import (  # noqa: E402
-    assess_proof_backed_handoff_report,
-)
-
-
-def _good_report() -> str:
-    return "\n".join([
-        "Inventory pagination proof: is_final_page: true, has_more: false, total_count: 11",
-        "Earlier PRs skipped: #372 non-mergeable — merge simulation conflicts in review_proofs.py",
-        "Baseline worktree path: branches/baseline-master-pr383",
-        "Baseline target SHA: 4243b60ca32d79f1ee5e6b0f10d7570e9dea2937",
-        "Baseline dirty before validation: false",
-        "Baseline validation command: venv/bin/python -m pytest tests/ -q",
-        "Baseline validation result: 1 failed (test_clean_reviewer_report_passes)",
-        "Baseline dirty after validation: false",
-        "Cleanup mutations: git worktree list showed session worktrees removed",
-        "## Controller Handoff",
-        "- Inventory pagination proof: inventory_complete: true, total_count: 11",
-        "- Earlier PRs skipped: #372 conflict proof via merge simulation",
-        "- Baseline worktree used: true",
-        "- Baseline worktree path: branches/baseline-master-pr383",
-        "- Cleanup mutations: git worktree remove branches/review-pr383",
-    ])
-
-
-class TestProofBackedHandoff(unittest.TestCase):
-    def test_fully_proof_backed_report_passes(self):
-        result = assess_proof_backed_handoff_report(
-            _good_report(),
-            action_log=[{"command": "gitea_list_prs", "result": "inventory_complete=true"}],
-            baseline_session={
-                "complete": True,
-                "clean_before": True,
-                "clean_after": True,
-                "command": "pytest",
-                "result": "passed",
-            },
-        )
-        self.assertTrue(result["proven"], result["reasons"])
-
-    def test_inventory_page_size_assumption_blocks(self):
-        report = (
-            "Inventory is complete because 13 results are less than page size 50."
-        )
-        result = assess_proof_backed_handoff_report(report)
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("page-size" in r.lower() for r in result["reasons"]))
-
-    def test_skip_without_conflict_proof_blocks(self):
-        report = "\n".join([
-            "Earlier PRs skipped: #372, #374",
-            "## Controller Handoff",
-            "- Earlier PRs skipped: #372, #374",
-        ])
-        result = assess_proof_backed_handoff_report(report)
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("skip" in r.lower() for r in result["reasons"]))
-
-    def test_baseline_without_command_blocks(self):
-        report = "Baseline validation passed; same as master failures."
-        result = assess_proof_backed_handoff_report(report)
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("baseline" in r.lower() for r in result["reasons"]))
-
-    def test_cleanup_without_worktree_list_blocks(self):
-        report = "Worktrees were cleaned after merge."
-        result = assess_proof_backed_handoff_report(report)
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("cleanup" in r.lower() for r in result["reasons"]))
-
-    def test_cleanup_with_worktree_list_passes(self):
-        report = "Cleanup: git worktree list confirmed removal."
-        result = assess_proof_backed_handoff_report(
-            report,
-            action_log=[{"command": "git worktree list"}],
-        )
-        self.assertTrue(result["proven"], result["reasons"])
-
-    def test_master_integration_requires_command(self):
-        report = "Merged master into the review worktree before validation."
-        result = assess_proof_backed_handoff_report(report)
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("master integration" in r.lower() for r in result["reasons"]))
-
-    def test_master_integration_with_action_log_passes(self):
-        report = "Merged master into review worktree; merge_exit=0"
-        result = assess_proof_backed_handoff_report(
-            report,
-            action_log=[{"command": "git merge --no-commit prgs/master"}],
-        )
-        self.assertTrue(result["proven"], result["reasons"])
-
-    def test_live_proof_requires_source_classification(self):
-        report = "Live proof shows inventory complete."
-        result = assess_proof_backed_handoff_report(report)
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("proof source" in r.lower() for r in result["reasons"]))
-
-
-class TestExport(unittest.TestCase):
-    def test_review_proofs_reexport(self):
-        from review_proofs import assess_proof_backed_handoff_report as exported
-
-        self.assertTrue(callable(exported))
-        result = exported(_good_report(), inventory_session={"inventory_complete": True})
-        self.assertTrue(result["proven"], result["reasons"])
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_reviewer_validation_cwd_proof.py

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

tests/test_reviewer_validation_failure_history.py

@@ -1,147 +0,0 @@
-"""Tests for transient validation failure history verifier (#396)."""
-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_failure_history import (  # noqa: E402
-    assess_validation_failure_history_report,
-)
-
-
-def _full_history_report() -> str:
-    return "\n".join([
-        "Validation failure history",
-        "Failure #1:",
-        "Command: venv/bin/python -m pytest tests/test_worktrees.py -q",
-        "Failing test: tests/test_worktrees.py::TestWorktreeStart::test_rejects_untraceable_branches",
-        "Suspected cause: stale /tmp/gitea_issue_lock.json from prior session",
-        "Reproduced: no",
-        "On baseline master: no",
-        "PR-caused: no",
-        "What changed between runs: removed /tmp/gitea_issue_lock.json",
-        "Environmental cleanup: lock file removed before rerun",
-        "Validation: passed after transient failure investigation",
-        "Final validation command: venv/bin/python -m pytest tests/ -q",
-        "Final result: 1497 passed, 6 skipped",
-    ])
-
-
-class TestValidationFailureHistory(unittest.TestCase):
-    def test_no_failures_observed_passes(self):
-        result = assess_validation_failure_history_report(
-            "Validation: passed",
-            validation_session={"observed_failures": []},
-        )
-        self.assertTrue(result["proven"])
-
-    def test_omitted_failure_blocks(self):
-        result = assess_validation_failure_history_report(
-            "Validation: passed\n1497 passed, 6 skipped",
-            validation_session={
-                "observed_failures": [{
-                    "command": "pytest tests/test_worktrees.py -q",
-                    "failing_test": (
-                        "tests/test_worktrees.py::TestWorktreeStart::"
-                        "test_rejects_untraceable_branches"
-                    ),
-                }],
-                "final_validation_status": "passed",
-            },
-        )
-        self.assertFalse(result["proven"])
-        self.assertTrue(result["block"])
-
-    def test_full_history_with_transient_investigation_passes(self):
-        result = assess_validation_failure_history_report(
-            _full_history_report(),
-            validation_session={
-                "observed_failures": [{
-                    "command": (
-                        "venv/bin/python -m pytest tests/test_worktrees.py -q"
-                    ),
-                    "failing_test": (
-                        "tests/test_worktrees.py::TestWorktreeStart::"
-                        "test_rejects_untraceable_branches"
-                    ),
-                    "suspected_cause": "stale /tmp/gitea_issue_lock.json",
-                    "reproduced": "no",
-                    "on_baseline_master": "no",
-                    "pr_caused": "no",
-                }],
-                "final_validation_status": "passed_after_transient_failure_investigation",
-                "environmental_contamination": True,
-            },
-        )
-        self.assertTrue(result["proven"], result["reasons"])
-
-    def test_baseline_equivalent_failure_documented(self):
-        report = "\n".join([
-            "Validation failure history",
-            "Command: pytest tests/test_example.py -q",
-            "Failing test: tests/test_example.py::test_flaky",
-            "Suspected cause: pre-existing master failure",
-            "On baseline master: yes",
-            "PR-caused: no",
-            "What changed between runs: none; failure matches baseline",
-            "Validation: passed after transient failure investigation",
-        ])
-        result = assess_validation_failure_history_report(
-            report,
-            validation_session={
-                "observed_failures": [{
-                    "command": "pytest tests/test_example.py -q",
-                    "failing_test": "tests/test_example.py::test_flaky",
-                    "on_baseline_master": "yes",
-                    "pr_caused": "no",
-                }],
-                "final_validation_status": "passed_after_transient_failure_investigation",
-            },
-        )
-        self.assertTrue(result["proven"], result["reasons"])
-
-    def test_unknown_cause_plain_pass_blocks(self):
-        result = assess_validation_failure_history_report(
-            "Validation: passed",
-            validation_session={
-                "observed_failures": [{
-                    "command": "pytest tests/ -q",
-                    "failing_test": "tests/test_foo.py::test_bar",
-                    "suspected_cause": "unknown",
-                }],
-                "final_validation_status": "passed",
-                "cause_unknown": True,
-            },
-        )
-        self.assertFalse(result["proven"])
-        self.assertTrue(any("transient" in r.lower() for r in result["reasons"]))
-
-    def test_final_report_validator_rejects_omitted_failure(self):
-        result = assess_final_report_validator(
-            "Validation: passed",
-            "review_pr",
-            validation_session={
-                "observed_failures": [{
-                    "command": "pytest tests/ -q",
-                    "failing_test": "tests/test_foo.py::test_bar",
-                }],
-            },
-        )
-        self.assertTrue(result["blocked"] or result["downgraded"])
-        self.assertTrue(
-            any(
-                f.get("rule_id") == "reviewer.validation_failure_history"
-                for f in result.get("findings") or []
-            )
-        )
-
-    def test_exported_from_review_proofs(self):
-        from review_proofs import assess_validation_failure_history_report as exported
-
-        self.assertTrue(callable(exported))
-
-
-if __name__ == "__main__":
-    unittest.main()