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

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

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

author_duplicate_work_gate.py

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

author_mutation_worktree.py

@@ -0,0 +1,105 @@
+"""Branches-only author mutation worktree guard (#274).
+
+Author/coder mutations must run from a session-owned worktree under the
+project's ``branches/`` directory, never from the stable control checkout.
+"""
+
+from __future__ import annotations
+
+import os
+
+BASE_BRANCHES = frozenset({"master", "main", "dev"})
+
+
+def _normalize_path(path: str) -> str:
+    return (path or "").replace("\\", "/").rstrip("/")
+
+
+def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
+    """True when *path* resolves inside ``<project_root>/branches/``."""
+    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(os.path.realpath(project_root))
+        real = _normalize_path(os.path.realpath(path))
+        if real.startswith(f"{root}/"):
+            rel = real[len(root) + 1 :]
+            return rel == "branches" or rel.startswith("branches/")
+    return False
+
+
+def resolve_mutation_workspace(
+    worktree_path: str | None,
+    project_root: str,
+    *,
+    active_worktree_env: str | None = None,
+    author_worktree_env: str | None = None,
+) -> str:
+    """Resolve the workspace path inspected before author mutations."""
+    for candidate in (worktree_path, active_worktree_env, author_worktree_env):
+        text = (candidate or "").strip()
+        if text:
+            return os.path.realpath(os.path.abspath(text))
+    return os.path.realpath(project_root)
+
+
+def assess_author_mutation_worktree(
+    *,
+    workspace_path: str,
+    project_root: str,
+    current_branch: str | None = None,
+    base_branches: frozenset[str] | None = None,
+) -> dict:
+    """Fail closed when author mutations are not rooted under ``branches/``."""
+    bases = base_branches or BASE_BRANCHES
+    reasons: list[str] = []
+    root = os.path.realpath(project_root)
+    workspace = os.path.realpath(workspace_path)
+    branch = (current_branch or "").strip()
+
+    under_branches = is_path_under_branches(workspace, root)
+    if not under_branches:
+        if workspace == root:
+            reasons.append(
+                "author mutation blocked: workspace is the stable control checkout; "
+                "create or switch to a session-owned worktree under branches/"
+            )
+        else:
+            reasons.append(
+                f"author mutation blocked: workspace '{workspace}' is not under "
+                f"'{root}/branches/'; create a branches/<task> worktree first"
+            )
+
+    if not under_branches and workspace == root and branch and branch not in bases:
+        reasons.append(
+            f"control checkout drift: branch '{branch}' is not a stable base "
+            f"branch ({'/'.join(sorted(bases))})"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "project_root": root,
+        "workspace_path": workspace,
+        "under_branches": is_path_under_branches(workspace, root),
+        "current_branch": branch or None,
+    }
+
+
+def format_author_mutation_worktree_error(assessment: dict) -> str:
+    """Single RuntimeError message for MCP preflight gates."""
+    workspace = assessment.get("workspace_path") or "(unknown)"
+    root = assessment.get("project_root") or "(unknown)"
+    reasons = "; ".join(assessment.get("reasons") or ["unknown branches-only violation"])
+    return (
+        f"Branches-only mutation guard (#274): {reasons}. "
+        f"project root: {root}; workspace: {workspace}. "
+        "Create a session-owned worktree under branches/ before mutating."
+    )

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 (#310) |
+| `gitea-reconciler` | a reconciler profile | close already-landed open PRs after ancestry proof (#304 profile; #310 close tool) |
 
 Properties:
 

docs/gitea-execution-profiles.md

@@ -226,11 +226,12 @@ 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 (#310)
+## Reconciler profile for already-landed open PRs (#304 / #310)
 
-Normal author and reviewer profiles must not gain broad PR-close authority.
-Already-landed open PRs (head SHA is an ancestor of the target branch) need a
-dedicated reconciler profile such as `prgs-reconciler` with:
+Normal author and reviewer profiles must not gain broad `gitea.pr.close`
+authority. Already-landed open PRs (head SHA is an ancestor of the target
+branch) need a dedicated reconciler profile such as `prgs-reconciler` with a
+narrow operation set:
 
 - `gitea.read`
 - `gitea.pr.comment`
@@ -238,8 +239,14 @@ dedicated reconciler profile such as `prgs-reconciler` with:
 - `gitea.issue.close`
 - `gitea.pr.close`
 
-Use the `gitea-reconciler` MCP namespace (static profile launch) and the
-`gitea_reconcile_already_landed_pr` tool. The resolver task is
+Forbidden on reconciler profiles: `gitea.pr.approve`, `gitea.pr.merge`,
+`gitea.pr.review`, `gitea.pr.create`, `gitea.branch.push`, and
+`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.

docs/llm-workflow-runbooks.md

@@ -273,7 +273,20 @@ Never create a parallel branch or PR for the same issue unless the old branch
 is proven abandoned and the takeover is recorded.
 
 Gitea-Tools lease gates: `gitea_lock_issue` (fail-closed before author
-mutations), `status:in-progress`, and claim comments. Full portable wording:
+mutations), `status:in-progress`, and claim comments. `gitea_lock_issue`
+records an `author_issue_work` lease in the issue-lock payload with issue
+number, optional PR number, branch, worktree path, claimant identity/profile,
+created timestamp, expiry timestamp, and last heartbeat timestamp. An active
+same-issue/same-operation lease blocks duplicate work. An expired lease still
+blocks takeover until a recovery review records why the prior work is abandoned,
+completed, or unsafe to continue.
+
+Remote branches matching the issue number are also treated as active work unless
+the recovery review proves the branch is abandoned or superseded. Never delete
+or clean up a branch when it has an active lease, dirty worktree, open PR, or is
+the only copy of unmerged work.
+
+Full portable wording:
 [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
 
 ## Global LLM Worktree Rule

docs/wiki/Workflow.md

@@ -17,6 +17,10 @@
 0. Work Selection Rule — verify a work lease before any mutations (open PRs,
    issue-linked PRs, branches, worktrees, dirty worktrees, active leases/
    handoffs, merged-PR completion). Stop if another session owns the lease.
+   `gitea_lock_issue` records an operation-scoped `author_issue_work` lease
+   with issue, branch, worktree, claimant, created, expiry, and heartbeat
+   fields; active or expired same-operation leases require recovery review
+   before takeover.
 0b. Global LLM Worktree Rule — main checkout on `master`/`main`/`dev` only;
    mutate only from a `branches/` worktree after proving root, cwd, branch,
    stable main-checkout branch, and session worktree path (no exceptions).
@@ -31,4 +35,4 @@
 Wiki publication, credential provisioning, and cross-repo mirror operations are
 operator-confirmed actions — see [Runbooks](Runbooks.md).
 
-Full Gitea-specific runbooks: `docs/llm-workflow-runbooks.md` in the repository.
+Full Gitea-specific runbooks: `docs/llm-workflow-runbooks.md` in the repository.

final_report_validator.py

@@ -92,7 +92,8 @@ _ALREADY_LANDED_GATE_FIRED_RE = re.compile(
     re.IGNORECASE,
 )
 _ELIGIBLE_REVIEW_RE = re.compile(
-    r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr",
+    r"eligible for (?:review|merge)|ready for (?:review|merge)|"
+    r"(?:next|oldest)\s+eligible\s+pr",
     re.IGNORECASE,
 )
 _APPROVED_STATE_RE = re.compile(r"\bapproved?\b", re.IGNORECASE)
@@ -121,11 +122,37 @@ _RECONCILE_STALE_FIELDS = (
     "scratch worktree used",
     "review decision",
     "merge result",
+    "issue lock proof",
+    "claim/comment status",
+    "workspace mutations",
 )
+# Issue #307: reconciliation handoffs must carry the full canonical schema.
 _RECONCILE_REQUIRED_FIELDS = (
+    "Task",
+    "Repo",
+    "Role/profile",
+    "Identity",
     "Selected PR",
-    "Eligibility class",
+    "PR live state",
+    "Candidate head SHA",
+    "Target branch",
+    "Target branch SHA",
+    "Ancestor proof",
+    "Linked issue",
     "Linked issue live status",
+    "Eligibility class",
+    "Capabilities proven",
+    "Missing capabilities",
+    "PR comments posted",
+    "Issue comments posted",
+    "PRs closed",
+    "Issues closed",
+    "Git ref mutations",
+    "Worktree mutations",
+    "MCP/Gitea mutations",
+    "External-state mutations",
+    "Read-only diagnostics",
+    "Blocker",
     "Safe next action",
     "No review/merge confirmation",
 )
@@ -414,19 +441,55 @@ def _rule_reviewer_git_fetch_readonly(
                 "classify git fetch under Git ref mutations, not read-only diagnostics",
             )
         ]
-    if not git_ref_reported and _GIT_FETCH_RE.search(text):
+    if not git_ref_reported:
+        # Issue #307: an observed git fetch (report text or action log)
+        # must be listed under Git ref mutations.
         return [
             validator_finding(
                 "reviewer.git_fetch_readonly",
                 "downgrade",
                 "Git ref mutations",
-                "git fetch mentioned without Git ref mutation classification",
+                "git fetch occurred without Git ref mutation classification",
                 "record git fetch under Git ref mutations",
             )
         ]
     return []
 
 
+def _rule_reviewer_validation_failure_history(
+    report_text: str,
+    *,
+    validation_session: dict | None = None,
+) -> list[dict[str, str]]:
+    from reviewer_validation_failure_history import (
+        assess_validation_failure_history_report,
+    )
+
+    session = validation_session or {}
+    observed = session.get("observed_failures") or []
+    if not observed:
+        return []
+
+    result = assess_validation_failure_history_report(
+        report_text,
+        validation_session=session,
+    )
+    if result.get("proven"):
+        return []
+    severity = "block" if result.get("violations") else "downgrade"
+    return [
+        validator_finding(
+            "reviewer.validation_failure_history",
+            severity,
+            "Validation failure history",
+            reason,
+            result.get("safe_next_action")
+            or "document every observed validation failure before claiming pass",
+        )
+        for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
+    ]
+
+
 def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]:
     text = report_text or ""
     if not _BARE_PYTEST_RE.search(text):
@@ -668,10 +731,18 @@ def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]
     return []
 
 
-def _rule_reconcile_stale_author_fields(report_text: str) -> list[dict[str, str]]:
+def _rule_reconcile_stale_author_fields(
+    report_text: str,
+    *,
+    session_pr_opened: bool = False,
+) -> list[dict[str, str]]:
     text = (report_text or "").lower()
     findings = []
     for field in _RECONCILE_STALE_FIELDS:
+        if field == "pr number opened" and session_pr_opened:
+            # Issue #307: allowed only when a PR was actually opened
+            # in this session.
+            continue
         if f"{field}:" in text:
             findings.append(
                 validator_finding(
@@ -827,6 +898,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_structured,
         _rule_reviewer_linked_issue,
         _rule_reviewer_baseline_on_failure,
@@ -932,6 +1004,8 @@ def assess_final_report_validator(
     mutations_observed: bool = False,
     local_edits: bool = False,
     issue_filing_lock: dict | None = None,
+    session_pr_opened: bool = False,
+    validation_session: dict | None = None,
 ) -> dict[str, Any]:
     """Validate final-report text against task-specific proof rules (#327).
 
@@ -986,6 +1060,8 @@ def assess_final_report_validator(
         "action_log": action_log,
         "mutations_observed": mutations_observed,
         "local_edits": local_edits,
+        "session_pr_opened": session_pr_opened,
+        "validation_session": validation_session,
     }
 
     for rule in _RULES_BY_TASK.get(normalized_kind, ()):

gitea_mcp_server.py

@@ -21,6 +21,7 @@ import json
 import functools
 import contextlib
 import subprocess
+from datetime import datetime, timedelta, timezone
 
 
 # Mutation-authority record (#199, refs #194). Deliberately in-process, NOT a
@@ -397,6 +398,28 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
             _preflight_resolved_role = resolved_role
 
 
+def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
+    """#274: author mutations must run from a branches/ session worktree."""
+    if _preflight_resolved_role == "reviewer":
+        return
+    workspace = author_mutation_worktree.resolve_mutation_workspace(
+        worktree_path,
+        PROJECT_ROOT,
+        active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
+        author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
+    )
+    git_state = issue_lock_worktree.read_worktree_git_state(workspace)
+    assessment = author_mutation_worktree.assess_author_mutation_worktree(
+        workspace_path=workspace,
+        project_root=PROJECT_ROOT,
+        current_branch=git_state.get("current_branch"),
+    )
+    if assessment["block"]:
+        raise RuntimeError(
+            author_mutation_worktree.format_author_mutation_worktree_error(assessment)
+        )
+
+
 def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
     """Verify that identity and capability were verified prior to session edits."""
     global _preflight_reviewer_violation_files
@@ -426,32 +449,33 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
                 "file edits before mutation (fail closed). "
                 f"{_format_preflight_workspace_details(details)}"
             )
-        return
-
-    if _preflight_whoami_violation:
-        raise RuntimeError(
-            "Pre-flight order violation: Workspace file edits occurred before "
-            f"gitea_whoami verification (fail closed). Offending files: "
-            f"{_format_preflight_files(_preflight_whoami_violation_files)}"
-        )
-    if _preflight_capability_violation:
-        raise RuntimeError(
-            "Pre-flight order violation: Workspace file edits occurred before "
-            f"gitea_resolve_task_capability verification (fail closed). Offending files: "
-            f"{_format_preflight_files(_preflight_capability_violation_files)}"
-        )
-
-    if _preflight_resolved_role == "reviewer":
-        current = _get_workspace_porcelain()
-        baseline = _preflight_capability_baseline_porcelain or ""
-        reviewer_delta = _new_tracked_changes_since(baseline, current)
-        _preflight_reviewer_violation_files = reviewer_delta
-        if reviewer_delta:
+    else:
+        if _preflight_whoami_violation:
             raise RuntimeError(
-                "Reviewer role violation: Reviewer profile is forbidden from modifying "
-                "tracked workspace files (fail closed). Offending files: "
-                f"{_format_preflight_files(reviewer_delta)}"
+                "Pre-flight order violation: Workspace file edits occurred before "
+                f"gitea_whoami verification (fail closed). Offending files: "
+                f"{_format_preflight_files(_preflight_whoami_violation_files)}"
             )
+        if _preflight_capability_violation:
+            raise RuntimeError(
+                "Pre-flight order violation: Workspace file edits occurred before "
+                f"gitea_resolve_task_capability verification (fail closed). Offending files: "
+                f"{_format_preflight_files(_preflight_capability_violation_files)}"
+            )
+
+        if _preflight_resolved_role == "reviewer":
+            current = _get_workspace_porcelain()
+            baseline = _preflight_capability_baseline_porcelain or ""
+            reviewer_delta = _new_tracked_changes_since(baseline, current)
+            _preflight_reviewer_violation_files = reviewer_delta
+            if reviewer_delta:
+                raise RuntimeError(
+                    "Reviewer role violation: Reviewer profile is forbidden from modifying "
+                    "tracked workspace files (fail closed). Offending files: "
+                    f"{_format_preflight_files(reviewer_delta)}"
+                )
+
+    _enforce_branches_only_author_mutation(worktree_path)
 
 from mcp.server.fastmcp import FastMCP  # noqa: E402
 
@@ -477,8 +501,12 @@ import review_proofs  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 import already_landed_reconcile  # noqa: E402
+import author_mutation_worktree  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
+import author_duplicate_work_gate  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
+import reconciler_profile  # noqa: E402
+import reconciliation_workflow  # noqa: E402
 import review_merge_state_machine  # noqa: E402
 import native_mcp_preference  # noqa: E402
 
@@ -486,6 +514,132 @@ import native_mcp_preference  # noqa: E402
 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
 # consumed by gitea_create_pr and scripts/worktree-start.
 ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
+WORK_LEASE_TTL_HOURS = 4
+AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
+VALID_WORK_LEASE_OPERATIONS = frozenset({
+    AUTHOR_ISSUE_WORK_LEASE,
+    "review_pr_work",
+    "fix_pr_changes",
+    "cleanup_branch_work",
+    "issue_filing_work",
+    "recovery_work",
+})
+
+
+def _work_lease_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _work_lease_timestamp(value: datetime) -> str:
+    return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
+
+
+def _parse_work_lease_timestamp(value: str | None) -> datetime | None:
+    text = (value or "").strip()
+    if not text:
+        return None
+    try:
+        return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(timezone.utc)
+    except ValueError:
+        return None
+
+
+def _load_existing_issue_lock() -> dict | None:
+    if not os.path.exists(ISSUE_LOCK_FILE):
+        return None
+    try:
+        with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
+            data = json.load(f)
+        return data if isinstance(data, dict) else None
+    except Exception:
+        return None
+
+
+def _work_lease_claimant(host: str | None) -> dict:
+    profile = get_profile()
+    username = _IDENTITY_CACHE.get(host) if host else None
+    if username is None and host and not _preflight_in_test_mode():
+        username = _authenticated_username(host)
+    return {
+        "username": username,
+        "profile": profile.get("profile_name"),
+    }
+
+
+def _build_author_issue_work_lease(
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    host: str | None,
+) -> dict:
+    created = _work_lease_now()
+    expires = created + timedelta(hours=WORK_LEASE_TTL_HOURS)
+    return {
+        "operation_type": AUTHOR_ISSUE_WORK_LEASE,
+        "issue_number": issue_number,
+        "pr_number": None,
+        "branch": branch_name,
+        "worktree_path": worktree_path,
+        "claimant": _work_lease_claimant(host),
+        "created_at": _work_lease_timestamp(created),
+        "expires_at": _work_lease_timestamp(expires),
+        "last_heartbeat_at": _work_lease_timestamp(created),
+    }
+
+
+def _active_work_lease_block(
+    existing_lock: dict | None,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    operation_type: str,
+) -> str | None:
+    if not existing_lock:
+        return None
+    existing_issue = existing_lock.get("issue_number")
+    existing_branch = existing_lock.get("branch_name")
+    existing_worktree = existing_lock.get("worktree_path")
+    lease = existing_lock.get("work_lease")
+    existing_operation = (
+        lease.get("operation_type")
+        if isinstance(lease, dict)
+        else AUTHOR_ISSUE_WORK_LEASE
+    )
+    if existing_issue != issue_number or existing_operation != operation_type:
+        return None
+
+    same_owner = (
+        existing_branch == branch_name
+        and os.path.realpath(str(existing_worktree or "")) == os.path.realpath(worktree_path)
+    )
+    expires_at = (
+        _parse_work_lease_timestamp(lease.get("expires_at"))
+        if isinstance(lease, dict)
+        else None
+    )
+    if expires_at and expires_at <= _work_lease_now():
+        return (
+            f"Issue #{issue_number} has an expired {operation_type} lease on "
+            f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
+            "Recovery review is required before takeover (fail closed)"
+        )
+    if same_owner:
+        return None
+    return (
+        f"Issue #{issue_number} already has an active {operation_type} lease on "
+        f"branch '{existing_branch}' from worktree '{existing_worktree}' "
+        "(fail closed)"
+    )
+
+
+def _branch_entry_name(branch: dict | str) -> str:
+    if isinstance(branch, str):
+        return branch
+    if not isinstance(branch, dict):
+        return ""
+    return str(branch.get("name") or branch.get("ref") or "")
 
 
 def _reveal_endpoints() -> bool:
@@ -896,6 +1050,76 @@ def gitea_create_issue(
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
+def _list_repo_branch_names(h: str, o: str, r: str, auth: str, *, limit: int = 200) -> list[str]:
+    branches = api_get_all(f"{repo_api_url(h, o, r)}/branches", auth, limit=limit)
+    return [_branch_entry_name(branch) for branch in branches]
+
+
+def _gather_author_duplicate_work_context(
+    issue_number: int,
+    *,
+    h: str,
+    o: str,
+    r: str,
+    auth: str,
+    exclude_branch_name: str | None = None,
+) -> dict:
+    base = repo_api_url(h, o, r)
+    issue = api_request("GET", f"{base}/issues/{issue_number}", auth)
+    comments = api_request("GET", f"{base}/issues/{issue_number}/comments", auth) or []
+    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
+    branch_names = _list_repo_branch_names(h, o, r, auth)
+    if exclude_branch_name:
+        branch_names = [
+            name for name in branch_names
+            if name != exclude_branch_name
+        ]
+    return {
+        "issue": issue,
+        "comments": comments,
+        "open_prs": open_prs,
+        "branch_names": branch_names,
+    }
+
+
+def _enforce_author_duplicate_work_gate(
+    issue_number: int,
+    stage: str,
+    *,
+    h: str,
+    o: str,
+    r: str,
+    auth: str,
+    allow_stale_takeover: bool = False,
+    exclude_branch_name: str | None = None,
+) -> dict:
+    """Fail closed when duplicate work is detected (#400)."""
+    ctx = _gather_author_duplicate_work_context(
+        issue_number,
+        h=h,
+        o=o,
+        r=r,
+        auth=auth,
+        exclude_branch_name=exclude_branch_name,
+    )
+    result = author_duplicate_work_gate.classify_and_assess(
+        ctx["issue"],
+        stage=stage,
+        comments=ctx["comments"],
+        open_prs=ctx["open_prs"],
+        branch_names=ctx["branch_names"],
+        allow_stale_takeover=allow_stale_takeover,
+    )
+    assessment = result.get("duplicate_work") or {}
+    if assessment.get("block"):
+        reasons = "; ".join(assessment.get("reasons") or ["duplicate work detected"])
+        raise RuntimeError(
+            f"Author duplicate-work gate (#400) blocked at stage '{stage}': "
+            f"{reasons} (fail closed)"
+        )
+    return result
+
+
 @mcp.tool()
 def gitea_lock_issue(
     issue_number: int,
@@ -933,6 +1157,16 @@ def gitea_lock_issue(
     resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
         worktree_path, PROJECT_ROOT
     )
+    active_lease_block = _active_work_lease_block(
+        _load_existing_issue_lock(),
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=resolved_worktree,
+        operation_type=AUTHOR_ISSUE_WORK_LEASE,
+    )
+    if active_lease_block:
+        raise RuntimeError(active_lease_block)
+
     git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
     verify_preflight_purity(remote, worktree_path=resolved_worktree)
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
@@ -948,36 +1182,24 @@ def gitea_lock_issue(
             issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
         )
 
-    # 2. Check if the issue already has an open PR (reuse protection)
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
-    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
-
-    try:
-        prs = api_get_all(url, auth)
-    except Exception as e:
-        raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
-
-    for pr in prs:
-        pr_head = pr.get("head", {}).get("ref", "")
-        pr_title = pr.get("title", "")
-        pr_body = pr.get("body", "")
-
-        if expected_pattern in pr_head:
-            raise ValueError(
-                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
-            )
-
-        patterns = [
-            f"closes #{issue_number}",
-            f"fixes #{issue_number}",
-        ]
-        text_to_check = f"{pr_title} {pr_body}".lower()
-        if any(p in text_to_check for p in patterns):
-            raise ValueError(
-                f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
-            )
+    _enforce_author_duplicate_work_gate(
+        issue_number,
+        "lock",
+        h=h,
+        o=o,
+        r=r,
+        auth=auth,
+        exclude_branch_name=branch_name,
+    )
 
+    work_lease = _build_author_issue_work_lease(
+        issue_number=issue_number,
+        branch_name=branch_name,
+        worktree_path=resolved_worktree,
+        host=h,
+    )
     data = {
         "issue_number": issue_number,
         "branch_name": branch_name,
@@ -985,6 +1207,7 @@ def gitea_lock_issue(
         "org": o,
         "repo": r,
         "worktree_path": resolved_worktree,
+        "work_lease": work_lease,
     }
 
     try:
@@ -1005,6 +1228,7 @@ def gitea_lock_issue(
         "issue_number": issue_number,
         "branch_name": branch_name,
         "worktree_path": resolved_worktree,
+        "work_lease": work_lease,
     }
     if agent_artifacts:
         result["warnings"] = [
@@ -1061,7 +1285,7 @@ def gitea_create_pr(
     )
     if blocked:
         return blocked
-    verify_preflight_purity(remote)
+    verify_preflight_purity(remote, worktree_path=worktree_path)
     h, o, r = _resolve(remote, host, org, repo)
 
     # ── Issue Lock Validation (Issue #194 / #196) ──
@@ -1089,6 +1313,17 @@ def gitea_create_pr(
             f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
         )
 
+    auth = _auth(h)
+    _enforce_author_duplicate_work_gate(
+        int(locked_issue),
+        "create_pr",
+        h=h,
+        o=o,
+        r=r,
+        auth=auth,
+        exclude_branch_name=locked_branch,
+    )
+
     # Check for forbidden terms anywhere in title/body
     forbidden_terms = ["equivalent", "related", "same as"]
     text_to_check = f"{title} {body}".lower()
@@ -1105,7 +1340,6 @@ def gitea_create_pr(
             f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
         )
 
-    auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls"
     payload = {"title": title, "body": body, "head": head, "base": base}
     meta = {"title": title, "head": head, "base": base}
@@ -3280,16 +3514,32 @@ def gitea_delete_branch(
         repo: Override the repository name.
 
     Returns:
-        dict with 'success' and 'message'.
+        dict with 'success' and 'message'; on a permission block,
+        '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={"branch": branch}):
+                  target_branch=branch, request_metadata=request_metadata):
         api_request("DELETE", url, auth)
     return {"success": True, "message": f"Remote branch '{branch}' deleted."}
 
@@ -3429,6 +3679,178 @@ def gitea_reconcile_merged_cleanups(
     return {"success": True, "performed": True, **report}
 
 
+@mcp.tool()
+def gitea_assess_already_landed_reconciliation(
+    pr_number: int,
+    target_branch: str = "master",
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only: assess already-landed reconciliation for one open PR (#301).
+
+    Fetches the PR live, refreshes the target branch, checks ancestor proof,
+    and returns a capability plan for the active profile. Does not review,
+    approve, request changes, merge, or mutate Gitea state.
+
+    Args:
+        pr_number: Open PR to assess.
+        target_branch: Branch used for ancestry proof (default master).
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with eligibility proof, capability plan, and safe next action.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth)
+
+    assessment = reconciliation_workflow.assess_open_pr_reconciliation(
+        pr=pr,
+        project_root=PROJECT_ROOT,
+        remote=remote,
+        target_branch=target_branch,
+    )
+
+    profile = get_profile()
+    capabilities = reconciliation_workflow.profile_reconciliation_capabilities(
+        profile.get("allowed_operations"),
+        profile.get("forbidden_operations"),
+    )
+    plan = reconciliation_workflow.resolve_reconciliation_plan(
+        assessment=assessment,
+        capabilities=capabilities,
+    )
+
+    return {
+        "success": True,
+        "performed": False,
+        "pr_number": pr_number,
+        "assessment": assessment,
+        "capabilities": capabilities,
+        "plan": plan,
+        "workflow_source": "skills/llm-project-workflow/workflows/reconcile-landed-pr.md",
+        "task_mode": "reconcile-landed-pr",
+        "review_merge_allowed": False,
+    }
+
+
+@mcp.tool()
+def gitea_scan_already_landed_open_prs(
+    target_branch: str = "master",
+    limit: int = 50,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only: list open PRs already landed on the target branch (#301).
+
+    Traverses open PR inventory with pagination proof, assesses each candidate,
+    and returns PRs classified as ``ALREADY_LANDED_RECONCILE_REQUIRED``. Does
+    not review, approve, merge, or mutate Gitea state.
+
+    Args:
+        target_branch: Branch used for ancestry proof (default master).
+        limit: Max open PRs to inspect across pages.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with candidates, pagination metadata, and target branch SHA proof.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    url = f"{repo_api_url(h, o, r)}/pulls?state=open"
+
+    open_prs: list[dict] = []
+    current_page = 1
+    pages_fetched = 0
+    last_meta: dict = {}
+    per_page = min(50, max(1, int(limit)))
+    while pages_fetched < 100 and len(open_prs) < limit:
+        raw_page, last_meta = api_fetch_page(
+            url, auth, page=current_page, limit=per_page
+        )
+        pages_fetched += 1
+        remaining = limit - len(open_prs)
+        open_prs.extend(raw_page[:remaining])
+        if last_meta["is_final_page"] or len(open_prs) >= limit:
+            break
+        current_page += 1
+
+    pagination = {
+        "page": 1,
+        "per_page": per_page,
+        "returned_count": len(open_prs),
+        "has_more": False,
+        "next_page": None,
+        "is_final_page": bool(last_meta.get("is_final_page")),
+        "pages_fetched": pages_fetched,
+        "inventory_complete": bool(last_meta.get("is_final_page")),
+        "total_count": len(open_prs) if last_meta.get("is_final_page") else None,
+    }
+
+    target_fetch = reconciliation_workflow.fetch_target_branch(
+        PROJECT_ROOT, remote, target_branch
+    )
+
+    candidates: list[dict] = []
+    for pr in open_prs:
+        assessment = reconciliation_workflow.assess_open_pr_reconciliation(
+            pr=pr,
+            project_root=PROJECT_ROOT,
+            remote=remote,
+            target_branch=target_branch,
+            target_fetch=target_fetch,
+        )
+        if assessment.get("eligibility_class") == (
+            reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED
+        ):
+            candidates.append(assessment)
+
+    return {
+        "success": True,
+        "performed": False,
+        "target_branch": target_branch,
+        "target_branch_sha": target_fetch.get("target_branch_sha"),
+        "git_ref_mutations": (
+            [target_fetch["git_fetch_command"]]
+            if target_fetch.get("git_fetch_command")
+            else []
+        ),
+        "candidates": candidates,
+        "candidate_count": len(candidates),
+        "pagination": pagination,
+        "task_mode": "reconcile-landed-pr",
+        "review_merge_allowed": False,
+    }
+
+
 @mcp.tool()
 def gitea_reconcile_already_landed_pr(
     pr_number: int,
@@ -3608,7 +4030,6 @@ def gitea_reconcile_already_landed_pr(
 
     return result
 
-
 @mcp.tool()
 def gitea_close_issue(
     issue_number: int,
@@ -4118,12 +4539,15 @@ 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")
@@ -4391,6 +4815,28 @@ _PROJECT_SKILLS = {
             "Do not merge unless the operator explicitly authorizes it.",
         ],
     },
+    "gitea-reconcile-landed-pr": {
+        "description": "Reconcile open PRs whose heads are already on the "
+                       "target branch without review or merge.",
+        "when_to_use": "An open PR blocks the review queue but its head SHA "
+                       "is already an ancestor of master; use the dedicated "
+                       "reconciliation path instead of review/merge.",
+        "required_operations": ["gitea.read"],
+        "status": "available",
+        "notes": "Load skills/llm-project-workflow/workflows/reconcile-landed-pr.md "
+                 "first. PR close requires exact gitea.pr.close; review/merge "
+                 "capabilities must not be used on this path.",
+        "steps": [
+            "Resolve task: gitea_resolve_task_capability(task='reconcile_landed_pr').",
+            "Load canonical workflow via mcp_get_skill_guide or "
+            "skills/llm-project-workflow/workflows/reconcile-landed-pr.md.",
+            "Scan queue: gitea_scan_already_landed_open_prs (pagination proof).",
+            "Assess one PR: gitea_assess_already_landed_reconciliation.",
+            "Follow the plan outcome: full close only with gitea.pr.close, "
+            "comment-then-stop when close is missing, recovery handoff otherwise.",
+            "Never approve, request changes, or merge on this path.",
+        ],
+    },
     "gitea-pr-merge": {
         "description": "Gated merge of an approved pull request.",
         "when_to_use": "Reviewer/merger profile with explicit operator "
@@ -4915,6 +5361,7 @@ _RUNTIME_CAPABILITY_TASKS = (
     "create_issue",
     "comment_issue",
     "create_pr",
+    "work_issue",
     "review_pr",
     "merge_pr",
     "close_issue",
@@ -4967,6 +5414,7 @@ def _build_runtime_task_capabilities(
         "create_issue": "can_create_issues",
         "comment_issue": "can_comment_on_issues",
         "create_pr": "can_author_prs",
+        "work_issue": "can_work_issues",
         "review_pr": "can_review_prs",
         "merge_pr": "can_merge_prs",
         "close_issue": "can_close_issues",
@@ -5119,6 +5567,37 @@ def gitea_get_shell_health() -> dict:
     return native_mcp_preference.shell_health_status()
 
 
+@mcp.tool()
+def gitea_validate_review_final_report(
+    report_text: str,
+    review_decision_lock: dict | None = None,
+    linked_issue_lock: dict | None = None,
+    validation_report: dict | None = None,
+    action_log: list[dict] | None = None,
+    mutations_observed: bool = False,
+    local_edits: bool = False,
+    validation_session: dict | None = None,
+) -> dict:
+    """Read-only: machine-validate reviewer final report schema before output (#391).
+
+    Composes the composable final-report validator with review-specific schema
+    gates (legacy fields, mutation categories, merge/issue claims, blocked
+    handoff replay, and narrative/handoff consistency).
+    """
+    from review_final_report_schema import assess_review_final_report_schema
+
+    return assess_review_final_report_schema(
+        report_text,
+        review_decision_lock=review_decision_lock,
+        linked_issue_lock=linked_issue_lock,
+        validation_report=validation_report,
+        action_log=action_log,
+        mutations_observed=mutations_observed,
+        local_edits=local_edits,
+        validation_session=validation_session,
+    )
+
+
 @mcp.tool()
 def gitea_get_runtime_context(
     remote: str = "dadeschools",
@@ -5237,6 +5716,10 @@ 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(),
     }
 
@@ -5570,6 +6053,14 @@ def gitea_mark_issue(
         )
 
     if action == "start":
+        _enforce_author_duplicate_work_gate(
+            issue_number,
+            "claim",
+            h=h,
+            o=o,
+            r=r,
+            auth=auth,
+        )
         with _audited("label_issue", host=h, remote=remote, org=o, repo=r,
                       issue_number=issue_number,
                       request_metadata={"op": "add", "label": "status:in-progress"}):
@@ -5651,6 +6142,65 @@ def gitea_post_heartbeat(
     )
 
 
+@mcp.tool()
+def gitea_assess_author_duplicate_work(
+    issue_number: int,
+    stage: str,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    allow_stale_takeover: bool = False,
+    exclude_branch_name: str | None = None,
+) -> dict:
+    """Read-only: assess duplicate-work risk before author mutations (#400).
+
+    Call at claim, lock, worktree, edit, commit, push, and create_pr stages.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    ctx = _gather_author_duplicate_work_context(
+        issue_number,
+        h=h,
+        o=o,
+        r=r,
+        auth=auth,
+        exclude_branch_name=exclude_branch_name,
+    )
+    result = author_duplicate_work_gate.classify_and_assess(
+        ctx["issue"],
+        stage=stage,
+        comments=ctx["comments"],
+        open_prs=ctx["open_prs"],
+        branch_names=ctx["branch_names"],
+        allow_stale_takeover=allow_stale_takeover,
+    )
+    assessment = result.get("duplicate_work") or {}
+    return {
+        "success": True,
+        "performed": False,
+        "issue_number": issue_number,
+        "stage": stage,
+        "allowed": assessment.get("allowed"),
+        "block": assessment.get("block"),
+        "eligibility_class": assessment.get("eligibility_class"),
+        "outcome": assessment.get("outcome"),
+        "linked_open_pr": assessment.get("linked_open_pr"),
+        "matching_branches": assessment.get("matching_branches"),
+        "claim_status": assessment.get("claim_status"),
+        "reasons": assessment.get("reasons"),
+        "safe_next_action": assessment.get("safe_next_action"),
+        "claim": result.get("claim"),
+    }
+
+
 @mcp.tool()
 def gitea_reconcile_issue_claims(
     state: str = "open",

reconciler_profile.py

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

reconciliation_workflow.py

@@ -0,0 +1,292 @@
+"""Already-landed PR reconciliation workflow helpers (#301).
+
+Read-only assessment and capability planning for reconciling open PRs whose
+heads are already ancestors of the target branch. Does not invoke review or
+merge paths.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from typing import Any
+
+import gitea_config
+from merged_cleanup_reconcile import extract_linked_issue, is_head_ancestor_of_ref
+
+ELIGIBILITY_ALREADY_LANDED = "ALREADY_LANDED_RECONCILE_REQUIRED"
+ELIGIBILITY_NOT_LANDED = "NOT_ALREADY_LANDED"
+ELIGIBILITY_STALE_TARGET = "TARGET_BRANCH_UNVERIFIED"
+ELIGIBILITY_PR_NOT_OPEN = "PR_NOT_OPEN"
+
+OUTCOME_FULL_RECONCILE = "FULL_RECONCILE_CLOSE_ALLOWED"
+OUTCOME_PARTIAL_COMMENT = "PARTIAL_RECONCILE_COMMENT_THEN_STOP"
+OUTCOME_RECOVERY_HANDOFF = "RECOVERY_HANDOFF_ONLY"
+OUTCOME_NOT_LANDED = "NOT_LANDED_NO_ACTION"
+OUTCOME_GATE_NOT_PROVEN = "GATE_NOT_PROVEN"
+
+RECONCILE_WORKFLOW_MARKERS = (
+    "workflows/reconcile-landed-pr.md",
+    "reconcile-landed-pr.md",
+)
+
+RECONCILE_TASK_MARKERS = (
+    "reconcile-landed-pr",
+    "reconcile already-landed",
+    "reconcile_already_landed",
+)
+
+
+def fetch_target_branch(project_root: str, remote: str, branch: str) -> dict[str, Any]:
+    """Fetch *branch* from *remote* and return the resolved SHA."""
+    ref = f"{remote}/{branch}"
+    fetch = subprocess.run(
+        ["git", "-C", project_root, "fetch", remote, branch],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if fetch.returncode != 0:
+        return {
+            "success": False,
+            "target_branch": branch,
+            "target_ref": ref,
+            "target_branch_sha": None,
+            "reasons": [
+                f"git fetch {remote} {branch} failed: "
+                f"{(fetch.stderr or fetch.stdout or '').strip()}"
+            ],
+        }
+
+    rev = subprocess.run(
+        ["git", "-C", project_root, "rev-parse", ref],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if rev.returncode != 0:
+        return {
+            "success": False,
+            "target_branch": branch,
+            "target_ref": ref,
+            "target_branch_sha": None,
+            "reasons": [
+                f"git rev-parse {ref} failed: "
+                f"{(rev.stderr or rev.stdout or '').strip()}"
+            ],
+        }
+
+    return {
+        "success": True,
+        "target_branch": branch,
+        "target_ref": ref,
+        "target_branch_sha": (rev.stdout or "").strip(),
+        "reasons": [],
+        "git_fetch_command": f"git fetch {remote} {branch}",
+    }
+
+
+def profile_reconciliation_capabilities(
+    allowed: list[str] | None,
+    forbidden: list[str] | None,
+) -> dict[str, bool]:
+    """Map active profile operations to reconciliation capabilities."""
+    allowed = allowed or []
+    forbidden = forbidden or []
+
+    def can(op: str) -> bool:
+        return gitea_config.check_operation(op, allowed, forbidden)[0]
+
+    return {
+        "read": can("gitea.read"),
+        "comment_pr": can("gitea.pr.comment"),
+        "comment_issue": can("gitea.issue.comment"),
+        "close_pr": can("gitea.pr.close"),
+        "close_issue": can("gitea.issue.close"),
+        "review_pr": can("gitea.pr.review") or can("gitea.pr.approve"),
+        "merge_pr": can("gitea.pr.merge"),
+    }
+
+
+def assess_open_pr_reconciliation(
+    *,
+    pr: dict[str, Any],
+    project_root: str,
+    remote: str,
+    target_branch: str,
+    target_fetch: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Assess whether an open PR is eligible for already-landed reconciliation."""
+    pr_number = int(pr.get("number") or 0)
+    pr_state = (pr.get("state") or "").strip().lower()
+    head = pr.get("head") or {}
+    head_sha = head.get("sha") if isinstance(head, dict) else None
+    head_ref = head.get("ref") if isinstance(head, dict) else None
+    base = pr.get("base") or {}
+    base_ref = base.get("ref") if isinstance(base, dict) else None
+    title = pr.get("title") or ""
+    body = pr.get("body") or ""
+
+    fetch_result = target_fetch or fetch_target_branch(
+        project_root, remote, target_branch
+    )
+    linked_issue = extract_linked_issue(title, body)
+
+    result: dict[str, Any] = {
+        "pr_number": pr_number,
+        "pr_state": pr_state,
+        "candidate_head_sha": head_sha,
+        "head_ref": head_ref,
+        "base_ref": base_ref or target_branch,
+        "target_branch": target_branch,
+        "target_branch_sha": fetch_result.get("target_branch_sha"),
+        "linked_issue": linked_issue,
+        "git_ref_mutations": [],
+        "reasons": [],
+        "review_merge_allowed": False,
+    }
+    if fetch_result.get("git_fetch_command"):
+        result["git_ref_mutations"].append(fetch_result["git_fetch_command"])
+
+    if pr_state != "open":
+        result["eligibility_class"] = ELIGIBILITY_PR_NOT_OPEN
+        result["ancestor_proof"] = None
+        result["reconciliation_allowed"] = False
+        result["reasons"].append(f"PR #{pr_number} state is {pr_state!r}, not open")
+        return result
+
+    if not fetch_result.get("success"):
+        result["eligibility_class"] = ELIGIBILITY_STALE_TARGET
+        result["ancestor_proof"] = None
+        result["reconciliation_allowed"] = False
+        result["reasons"].extend(fetch_result.get("reasons") or [])
+        return result
+
+    target_ref = fetch_result.get("target_ref") or f"{remote}/{target_branch}"
+    ancestor = is_head_ancestor_of_ref(project_root, head_sha, target_ref)
+    result["ancestor_proof"] = ancestor
+
+    if ancestor is None:
+        result["eligibility_class"] = ELIGIBILITY_STALE_TARGET
+        result["reconciliation_allowed"] = False
+        result["reasons"].append(
+            f"ancestor check failed for head {head_sha!r} against {target_ref}"
+        )
+        return result
+
+    if ancestor:
+        result["eligibility_class"] = ELIGIBILITY_ALREADY_LANDED
+        result["reconciliation_allowed"] = True
+        return result
+
+    result["eligibility_class"] = ELIGIBILITY_NOT_LANDED
+    result["reconciliation_allowed"] = False
+    result["reasons"].append(
+        f"PR head {head_sha} is not an ancestor of {target_ref}"
+    )
+    return result
+
+
+def resolve_reconciliation_plan(
+    *,
+    assessment: dict[str, Any],
+    capabilities: dict[str, bool] | None,
+) -> dict[str, Any]:
+    """Map eligibility + profile capabilities to a reconciliation outcome."""
+    caps = capabilities or {}
+    reasons: list[str] = []
+
+    if not assessment.get("reconciliation_allowed"):
+        outcome = OUTCOME_NOT_LANDED
+        if assessment.get("eligibility_class") in {
+            ELIGIBILITY_STALE_TARGET,
+            ELIGIBILITY_PR_NOT_OPEN,
+        }:
+            outcome = OUTCOME_GATE_NOT_PROVEN
+        reasons.extend(assessment.get("reasons") or [])
+        return {
+            "outcome": outcome,
+            "close_pr_allowed": False,
+            "comment_pr_allowed": False,
+            "close_issue_allowed": False,
+            "comment_issue_allowed": False,
+            "review_merge_allowed": False,
+            "missing_capabilities": _missing_reconciliation_capabilities(caps),
+            "safe_next_action": (
+                "Do not reconcile via review/merge; repair missing proof first."
+            ),
+            "reasons": reasons,
+        }
+
+    close_pr = bool(caps.get("close_pr"))
+    comment_pr = bool(caps.get("comment_pr"))
+    close_issue = bool(caps.get("close_issue"))
+    comment_issue = bool(caps.get("comment_issue"))
+
+    if caps.get("review_pr") or caps.get("merge_pr"):
+        reasons.append(
+            "reconciliation path must not use review/merge capabilities"
+        )
+
+    if close_pr:
+        outcome = OUTCOME_FULL_RECONCILE
+        safe_next_action = (
+            "Run reconciliation close via exact gitea.pr.close after proof; "
+            "do not approve or merge."
+        )
+    elif comment_pr:
+        outcome = OUTCOME_PARTIAL_COMMENT
+        safe_next_action = (
+            "Post one reconciliation comment with ancestor proof, then stop "
+            "for an authorized close profile."
+        )
+    else:
+        outcome = OUTCOME_RECOVERY_HANDOFF
+        safe_next_action = (
+            "Produce a recovery handoff naming missing gitea.pr.close and/or "
+            "gitea.pr.comment; do not loop through review/merge."
+        )
+        reasons.append("gitea.pr.close is not available in the active profile")
+
+    return {
+        "outcome": outcome,
+        "close_pr_allowed": close_pr,
+        "comment_pr_allowed": comment_pr,
+        "close_issue_allowed": close_issue,
+        "comment_issue_allowed": comment_issue,
+        "review_merge_allowed": False,
+        "missing_capabilities": _missing_reconciliation_capabilities(caps),
+        "safe_next_action": safe_next_action,
+        "reasons": reasons,
+    }
+
+
+def _missing_reconciliation_capabilities(caps: dict[str, bool]) -> list[str]:
+    missing = []
+    if not caps.get("read"):
+        missing.append("gitea.read")
+    if not caps.get("close_pr"):
+        missing.append("gitea.pr.close")
+    if not caps.get("comment_pr"):
+        missing.append("gitea.pr.comment")
+    return missing
+
+
+def assess_reconcile_workflow_source(report_text: str) -> dict[str, Any]:
+    """Reconciliation reports must cite the canonical workflow file."""
+    lower = (report_text or "").lower()
+    reasons = []
+    if not any(marker in lower for marker in RECONCILE_WORKFLOW_MARKERS):
+        reasons.append(
+            "reconciliation report missing workflow source "
+            "(workflows/reconcile-landed-pr.md)"
+        )
+    if not any(marker in lower for marker in RECONCILE_TASK_MARKERS):
+        reasons.append(
+            "reconciliation report missing task mode declaration "
+            "(reconcile-landed-pr)"
+        )
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }

review_final_report_schema.py

@@ -0,0 +1,321 @@
+"""Reviewer final-report schema verification before session output (#391).
+
+Composes the composable ``assess_final_report_validator`` (#327) with
+review-specific schema gates required before a reviewer session completes.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from final_report_validator import (
+    assess_final_report_validator,
+    validator_finding,
+    _handoff_fields,
+)
+
+_LEGACY_STALE_FIELDS = (
+    "pinned reviewed head",
+    "scratch worktree used",
+    "workspace mutations",
+)
+
+_REVIEWED_HEAD_RE = re.compile(
+    r"(?:pinned reviewed head|reviewed head sha)\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE,
+)
+_VALIDATION_PASS_RE = re.compile(
+    r"validation\s*:\s*(?:pass|passed|strong|ok|green)",
+    re.IGNORECASE,
+)
+_MERGED_CLAIM_RE = re.compile(
+    r"\b(?:merged|merge result\s*:\s*merged|pr\s+#\d+\s+merged)\b",
+    re.IGNORECASE,
+)
+_MERGE_RESULT_RE = re.compile(r"merge result\s*:", re.IGNORECASE)
+_ANCESTRY_PROOF_RE = re.compile(
+    r"(?:ancestor proof|target branch sha|merge commit)",
+    re.IGNORECASE,
+)
+_ISSUE_CLOSED_RE = re.compile(
+    r"(?:linked issue(?:\s+live)?\s+status\s*:\s*closed|issue\s+#\d+\s+closed)",
+    re.IGNORECASE,
+)
+_LIVE_ISSUE_PROOF_RE = re.compile(
+    r"gitea_view_issue|(?:issue|linked issue).{0,40}fetched live|live fetch proof",
+    re.IGNORECASE,
+)
+_FULL_SUITE_PASS_RE = re.compile(
+    r"(?:full suite passed|full test suite passed|\d+\s+passed,\s*0\s+failed)",
+    re.IGNORECASE,
+)
+_TESTS_IGNORED_RE = re.compile(
+    r"(?:ignored tests|tests?\s+ignored|skipped tests|not run|tests?\s+skipped)",
+    re.IGNORECASE,
+)
+_BLOCKED_HANDOFF_RE = re.compile(
+    r"(?:blocker\s*:|recovery handoff|infra_stop|capability stop|mutation blocked)",
+    re.IGNORECASE,
+)
+_REPLAY_COMMAND_RE = re.compile(
+    r"(?:gitea_submit_pr_review|gitea_merge_pr|gitea_review_pr|"
+    r"submitted\s+['\"]approve['\"]|replay approve|replay merge)",
+    re.IGNORECASE,
+)
+_PENDING_RE = re.compile(r"\bPENDING\b", re.IGNORECASE)
+_APPROVED_RE = re.compile(
+    r"(?:review decision\s*:\s*approve|submitted\s+approve|official review submitted)",
+    re.IGNORECASE,
+)
+_DRY_RUN_RE = re.compile(r"dry[- ]run", re.IGNORECASE)
+_FINDING_READY_RE = re.compile(
+    r"(?:finding ready|ready to submit|not yet submitted)",
+    re.IGNORECASE,
+)
+_NARRATIVE_APPROVE_RE = re.compile(
+    r"(?:^|\n)\s*(?:##\s+)?(?:summary|verdict|recommendation)\b[^\n]*\bapprove\b",
+    re.IGNORECASE | re.MULTILINE,
+)
+_HANDOFF_DECISION_RE = re.compile(
+    r"review decision\s*:\s*(\w+)",
+    re.IGNORECASE,
+)
+
+
+def _rule_legacy_stale_fields(report_text: str) -> list[dict[str, str]]:
+    fields = _handoff_fields(report_text)
+    findings: list[dict[str, str]] = []
+    for stale in _LEGACY_STALE_FIELDS:
+        if stale in fields and fields[stale].lower() not in {"", "none", "n/a"}:
+            findings.append(
+                validator_finding(
+                    "reviewer.legacy_stale_field",
+                    "block",
+                    stale.title(),
+                    f"legacy or stale handoff field '{stale}' must not appear in "
+                    "canonical reviewer final reports",
+                    "remove stale fields; use Candidate head SHA and precise "
+                    "mutation categories instead",
+                )
+            )
+    return findings
+
+
+def _rule_reviewed_head_without_validation(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _REVIEWED_HEAD_RE.search(text):
+        return []
+    if _VALIDATION_PASS_RE.search(text):
+        return []
+    return [
+        validator_finding(
+            "reviewer.reviewed_head_without_validation",
+            "block",
+            "Pinned reviewed head",
+            "reviewed/pinned head SHA reported without validation pass proof",
+            "document validation command, cwd, and pass result before pinning head SHA",
+        )
+    ]
+
+
+def _rule_merged_without_proof(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _MERGED_CLAIM_RE.search(text):
+        return []
+    if _MERGE_RESULT_RE.search(text) and _ANCESTRY_PROOF_RE.search(text):
+        return []
+    return [
+        validator_finding(
+            "reviewer.merged_without_proof",
+            "block",
+            "Merge result",
+            "merged outcome claimed without merge result and target ancestry proof",
+            "include Merge result, target branch SHA, and ancestor proof before claiming merged",
+        )
+    ]
+
+
+def _rule_issue_closed_without_live_proof(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _ISSUE_CLOSED_RE.search(text):
+        return []
+    fields = _handoff_fields(text)
+    status_value = fields.get("linked issue live status", "")
+    readonly_value = fields.get("read-only diagnostics", "")
+    proof_blob = f"{status_value} {readonly_value}".lower()
+    if _LIVE_ISSUE_PROOF_RE.search(proof_blob):
+        return []
+    return [
+        validator_finding(
+            "reviewer.issue_closed_without_live_proof",
+            "block",
+            "Linked issue",
+            "issue closed claimed without live linked-issue verification proof",
+            "fetch linked issue live (gitea_view_issue) and record Linked issue live status",
+        )
+    ]
+
+
+def _rule_full_suite_pass_ignored_tests(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not (_FULL_SUITE_PASS_RE.search(text) and _TESTS_IGNORED_RE.search(text)):
+        return []
+    if re.search(r"explicitly\s+(?:ignored|skipped)", text, re.IGNORECASE):
+        return []
+    return [
+        validator_finding(
+            "reviewer.full_suite_pass_ignored_tests",
+            "block",
+            "Validation",
+            "full suite pass claimed while tests were ignored or skipped without "
+            "explicit disclosure",
+            "state which tests were ignored/skipped or downgrade validation verdict",
+        )
+    ]
+
+
+def _rule_blocked_handoff_replay(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _BLOCKED_HANDOFF_RE.search(text):
+        return []
+    if not _REPLAY_COMMAND_RE.search(text):
+        return []
+    return [
+        validator_finding(
+            "reviewer.blocked_handoff_replay",
+            "block",
+            "Safe next action",
+            "blocked recovery handoff includes direct approve or merge replay commands",
+            "restart the full workflow after the blocker clears; do not replay mutations",
+        )
+    ]
+
+
+def _rule_narrative_handoff_drift(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    narrative_approve = bool(_NARRATIVE_APPROVE_RE.search(text))
+    decision_match = _HANDOFF_DECISION_RE.search(text)
+    if not narrative_approve or not decision_match:
+        return []
+    handoff_decision = decision_match.group(1).strip().lower()
+    if handoff_decision in {"approve", "approved"}:
+        return []
+    return [
+        validator_finding(
+            "reviewer.narrative_handoff_drift",
+            "block",
+            "Review decision",
+            f"narrative approves PR but controller handoff says '{handoff_decision}'",
+            "align narrative summary with controller handoff Review decision field",
+        )
+    ]
+
+
+def _rule_review_state_ambiguous(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    findings: list[dict[str, str]] = []
+    if _PENDING_RE.search(text) and _APPROVED_RE.search(text):
+        findings.append(
+            validator_finding(
+                "reviewer.pending_vs_approved",
+                "block",
+                "Review decision",
+                "report conflates PENDING review state with APPROVED outcome",
+                "distinguish official submitted review from pending or ready-to-submit state",
+            )
+        )
+    if _DRY_RUN_RE.search(text) and _APPROVED_RE.search(text):
+        if "dry-run only" not in text.lower():
+            findings.append(
+                validator_finding(
+                    "reviewer.dry_run_vs_submitted",
+                    "block",
+                    "Review mutations",
+                    "dry-run language mixed with official approve/submitted claims",
+                    "mark dry-run explicitly or document the live review mutation proof",
+                )
+            )
+    if _FINDING_READY_RE.search(text) and _APPROVED_RE.search(text):
+        findings.append(
+            validator_finding(
+                "reviewer.finding_ready_vs_submitted",
+                "downgrade",
+                "Review decision",
+                "finding-ready wording combined with submitted-approve claims",
+                "state whether review was submitted live or only prepared for submission",
+            )
+        )
+    return findings
+
+
+_SCHEMA_RULES = (
+    _rule_legacy_stale_fields,
+    _rule_reviewed_head_without_validation,
+    _rule_merged_without_proof,
+    _rule_issue_closed_without_live_proof,
+    _rule_full_suite_pass_ignored_tests,
+    _rule_blocked_handoff_replay,
+    _rule_narrative_handoff_drift,
+    _rule_review_state_ambiguous,
+)
+
+
+def _merge_validator_results(
+    base: dict[str, Any],
+    extra_findings: list[dict[str, str]],
+) -> dict[str, Any]:
+    findings = list(base.get("findings") or []) + extra_findings
+    blocked = base.get("blocked") or any(f["severity"] == "block" for f in extra_findings)
+    downgraded = base.get("downgraded") or any(
+        f["severity"] == "downgrade" for f in extra_findings
+    )
+    grade = base.get("grade", "A")
+    if blocked:
+        grade = "blocked"
+    elif downgraded and grade == "A":
+        grade = "downgraded"
+    reasons = [f"{f['rule_id']}: {f['reason']}" for f in findings]
+    safe_next = base.get("safe_next_action") or "none"
+    if extra_findings:
+        safe_next = extra_findings[0].get("safe_next_action", safe_next)
+    return {
+        **base,
+        "grade": grade,
+        "blocked": blocked,
+        "downgraded": downgraded,
+        "findings": findings,
+        "reasons": reasons,
+        "safe_next_action": safe_next,
+        "complete": grade == "A",
+        "schema_rules_applied": len(_SCHEMA_RULES),
+    }
+
+
+def assess_review_final_report_schema(
+    report_text: str,
+    *,
+    review_decision_lock: dict | None = None,
+    linked_issue_lock: dict | None = None,
+    validation_report: dict | None = None,
+    action_log: list[dict] | None = None,
+    mutations_observed: bool = False,
+    local_edits: bool = False,
+    validation_session: dict | None = None,
+) -> dict[str, Any]:
+    """Validate reviewer final report text before session completion (#391)."""
+    base = assess_final_report_validator(
+        report_text,
+        "review_pr",
+        review_decision_lock=review_decision_lock,
+        linked_issue_lock=linked_issue_lock,
+        validation_report=validation_report,
+        action_log=action_log,
+        mutations_observed=mutations_observed,
+        local_edits=local_edits,
+        validation_session=validation_session,
+    )
+    extra: list[dict[str, str]] = []
+    for rule in _SCHEMA_RULES:
+        extra.extend(rule(report_text))
+    return _merge_validator_results(base, extra)

review_proofs.py

@@ -18,6 +18,11 @@ import hashlib
 import re
 
 import issue_duplicate_gate
+from reconciliation_workflow import (
+    RECONCILE_TASK_MARKERS,
+    RECONCILE_WORKFLOW_MARKERS,
+    assess_reconcile_workflow_source,
+)
 from reviewer_worktree import assess_reviewer_worktree_proof
 
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
@@ -1710,6 +1715,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         if report_text
         else {"proven": True, "block": False, "reasons": [], "violations": []}
     )
+    reconcile_inventory = (
+        assess_reconcile_inventory_report(report_text)
+        if report_text and _RECONCILE_INVENTORY_HINT.search(report_text)
+        else {"proven": True, "block": False, "reasons": []}
+    )
     reconcile_linked_issue = (
         assess_reconcile_linked_issue_report(report_text)
         if report_text and _RECONCILE_LINKED_ISSUE_HINT.search(report_text)
@@ -1725,6 +1735,11 @@ 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,
@@ -1737,6 +1752,17 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "reasons": [],
             "violations": [],
         }
+    if report_text:
+        already_landed_classification = assess_already_landed_classification_report(
+            report_text
+        )
+    else:
+        already_landed_classification = {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "violations": [],
+        }
 
     contamination_status = contamination.get("status", "unknown")
     checkout_proven = bool(checkout_proof.get("proven"))
@@ -1886,6 +1912,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "queue-status report violates loaded workflow proof gates (#339)"
         )
         downgrade_reasons.extend(queue_status_report.get("reasons", []))
+    if not reconcile_inventory.get("proven"):
+        downgrade_reasons.append(
+            "reconciliation PR inventory lacks pagination proof (#308)"
+        )
+        downgrade_reasons.extend(reconcile_inventory.get("reasons", []))
     if not reconcile_linked_issue.get("proven"):
         downgrade_reasons.append(
             "reconciliation linked issue status lacks live fetch proof (#300)"
@@ -1911,6 +1942,16 @@ 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)"
+        )
+        downgrade_reasons.extend(already_landed_classification.get("reasons", []))
 
     merge_allowed = (
         identity_eligible
@@ -1999,6 +2040,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "queue_status_violations": list(
             queue_status_report.get("violations") or []
         ),
+        "reconcile_inventory_proven": bool(reconcile_inventory.get("proven")),
+        "reconcile_inventory_violations": list(
+            reconcile_inventory.get("reasons") or []
+        ),
         "reconcile_linked_issue_proven": bool(reconcile_linked_issue.get("proven")),
         "reconcile_linked_issue_violations": list(
             reconcile_linked_issue.get("reasons") or []
@@ -2021,6 +2066,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "baseline_validation_violations": list(
             baseline_validation.get("violations") or []
         ),
+        "already_landed_classification_proven": bool(
+            already_landed_classification.get("proven")
+        ),
+        "already_landed_classification_violations": list(
+            already_landed_classification.get("reasons") or []
+        ),
     }
 
 
@@ -2287,6 +2338,30 @@ CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = (
     ("Terminal review mutation", ("terminal review mutation",)),
 )
 
+WORK_ISSUE_WORKFLOW_MARKERS = (
+    "workflows/work-issue.md",
+    "work-issue.md",
+)
+
+WORK_ISSUE_TASK_MARKERS = (
+    "work-issue",
+    "work issue",
+)
+
+WORK_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = (
+    ("Selected PR", ("selected pr",)),
+    ("Review decision", ("review decision",)),
+    ("Merge result", ("merge result",)),
+    ("Merge preflight", ("merge preflight",)),
+    ("Already-landed gate", ("already-landed gate",)),
+    ("Pinned reviewed head", ("pinned reviewed head",)),
+    ("Terminal review mutation", ("terminal review mutation",)),
+    ("Duplicate search terms", ("duplicate search terms",)),
+    ("Issues created", ("issues created",)),
+    ("Issues skipped as duplicates", ("issues skipped as duplicates",)),
+    ("Requested issue task", ("requested issue task",)),
+)
+
 
 def _handoff_section_lines(report_text):
     """Return the lines of the Controller Handoff section, or None."""
@@ -3479,6 +3554,122 @@ def assess_create_issue_workflow_source(report_text: str) -> dict:
     }
 
 
+def assess_work_issue_workflow_source(report_text: str) -> dict:
+    """#139: work-issue reports must cite the canonical workflow file."""
+    lower = (report_text or "").lower()
+    reasons = []
+    if not any(marker in lower for marker in WORK_ISSUE_WORKFLOW_MARKERS):
+        reasons.append(
+            "work-issue report missing workflow source "
+            "(workflows/work-issue.md)"
+        )
+    if not any(marker in lower for marker in WORK_ISSUE_TASK_MARKERS):
+        reasons.append(
+            "work-issue report missing task mode declaration (work-issue)"
+        )
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+def assess_work_issue_mode_isolation(report_text: str) -> dict:
+    """#139: reject review-merge/create-issue handoff fields in work-issue runs."""
+    section = _handoff_section_lines(report_text)
+    reasons = []
+    if section is None:
+        return {
+            "complete": True,
+            "downgraded": False,
+            "reasons": [],
+        }
+
+    labels = []
+    for line in section:
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" in stripped:
+            labels.append(stripped.split(":", 1)[0].strip().lower())
+
+    for name, aliases in WORK_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS:
+        if any(
+            label.startswith(alias)
+            for label in labels
+            for alias in aliases
+        ):
+            reasons.append(
+                f"work-issue handoff must not include cross-mode field "
+                f"'{name}'"
+            )
+
+    legacy_workspace = any(
+        label.startswith("workspace mutations") for label in labels
+    )
+    precise_categories = any(
+        label.startswith("file edits by author")
+        for label in labels
+    )
+    if legacy_workspace and not precise_categories:
+        reasons.append(
+            "work-issue handoff must use precise mutation categories "
+            "instead of legacy 'Workspace mutations'"
+        )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+def assess_work_issue_duplicate_prevention_report(report_text, **kwargs):
+    """#400: work-issue reports must classify duplicate-work prevention."""
+    from author_duplicate_work_gate import (
+        assess_work_issue_duplicate_prevention_report as _assess,
+    )
+
+    return _assess(report_text, **kwargs)
+
+
+def assess_work_issue_final_report(report_text: str) -> dict:
+    """#139: composite verifier for work-issue final reports."""
+    checks = {
+        "workflow_source": assess_work_issue_workflow_source(report_text),
+        "mode_isolation": assess_work_issue_mode_isolation(report_text),
+        "duplicate_prevention": assess_work_issue_duplicate_prevention_report(
+            report_text
+        ),
+    }
+
+    reasons = []
+    downgraded = False
+    for name, result in checks.items():
+        verdict = result.get("verdict")
+        if result.get("block"):
+            downgraded = True
+            reasons.extend(result.get("reasons") or [])
+        elif verdict in ("missing", "incomplete"):
+            downgraded = True
+            reasons.extend(result.get("reasons") or [])
+        elif result.get("downgraded") or not result.get("complete", True):
+            downgraded = True
+            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 {
+        "grade": grade,
+        "downgraded": downgraded,
+        "checks": checks,
+        "reasons": reasons,
+        "complete": not downgraded,
+    }
+
+
 def assess_create_issue_mode_isolation(report_text: str) -> dict:
     """#337: reject work-issue/review-merge handoff fields in create-issue runs."""
     section = _handoff_section_lines(report_text)
@@ -4895,12 +5086,25 @@ def assess_final_report_validator(report_text, task_kind, **kwargs):
     return _validate(report_text, task_kind, **kwargs)
 
 
+def assess_review_final_report_schema(report_text, **kwargs):
+    """#391: reviewer final-report schema verification before session output."""
+    from review_final_report_schema import assess_review_final_report_schema as _assess
+
+    return _assess(report_text, **kwargs)
+
+
 _QUEUE_STATUS_REPORT_HINT = re.compile(
     r"queue[- ]status|selected pr:\s*none|no pr selected|"
     r"queue[- ]status[- ]only",
     re.I,
 )
 
+_RECONCILE_INVENTORY_HINT = re.compile(
+    r"already[- ]landed reconciliation|reconcile[- ]landed|"
+    r"open pr inventory|inventory pagination proof",
+    re.I,
+)
+
 _RECONCILE_LINKED_ISSUE_HINT = re.compile(
     r"already[- ]landed|reconcil|linked issue(?:\s+live)?\s+status",
     re.I,
@@ -4912,6 +5116,12 @@ _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)."""
@@ -5311,6 +5521,24 @@ def assess_validation_integrity_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
+def assess_validation_failure_history_report(report_text, **kwargs):
+    """#396: final reports must account for transient validation failures."""
+    from reviewer_validation_failure_history import (
+        assess_validation_failure_history_report as _assess,
+    )
+
+    return _assess(report_text, **kwargs)
+
+
+def assess_already_landed_classification_report(report_text, **kwargs):
+    """#295: already-landed PRs are reconciliation-only, not review eligible."""
+    from reviewer_already_landed_classification import (
+        assess_already_landed_classification_report as _assess,
+    )
+
+    return _assess(report_text, **kwargs)
+
+
 def assess_prior_blocker_skip_proof(report_text, **kwargs):
     """#318: require live blocker proof before skipping earlier open PRs."""
     from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess
@@ -5334,6 +5562,13 @@ def assess_validation_worktree_edit_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
+def assess_reconcile_inventory_report(report_text, **kwargs):
+    """#308: prove PR inventory pagination in reconciliation reports."""
+    from reviewer_reconcile_inventory import assess_reconcile_inventory_report as _assess
+
+    return _assess(report_text, **kwargs)
+
+
 def assess_reconcile_linked_issue_report(report_text, **kwargs):
     """#300: prove linked issue status was fetched live in reconciliation handoffs."""
     from reviewer_reconcile_linked_issue import assess_reconcile_linked_issue_report as _assess
@@ -5374,3 +5609,10 @@ 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_already_landed_classification.py

@@ -0,0 +1,143 @@
+"""Already-landed PR classification verifier for reviewer reports (#295)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED"
+
+_LANDED_CONTEXT_RE = re.compile(
+    r"already[- ]landed|ancestor proof\s*:\s*passed|"
+    r"eligibility class\s*:\s*already_landed",
+    re.IGNORECASE,
+)
+_ELIGIBILITY_CLASS_LINE_RE = re.compile(
+    r"eligibility class\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_INELIGIBLE_SELECTION_RE = re.compile(
+    r"\b(?:next|oldest)\s+eligible\s+pr\b",
+    re.IGNORECASE,
+)
+_REVIEW_ELIGIBLE_RE = re.compile(
+    r"\beligible for (?:review|merge)\b|\bready for (?:review|merge)\b",
+    re.IGNORECASE,
+)
+_PINNED_REVIEWED_HEAD_RE = re.compile(
+    r"\bpinned reviewed head\s*:",
+    re.IGNORECASE,
+)
+_CANDIDATE_HEAD_RE = re.compile(
+    r"\bcandidate head sha\s*:",
+    re.IGNORECASE,
+)
+_REVIEWED_HEAD_NONE_RE = re.compile(
+    r"\breviewed head sha\s*:\s*none\b",
+    re.IGNORECASE,
+)
+_VALIDATION_PROOF_RE = re.compile(
+    r"(?:validation passed|pytest.*passed|diff review passed|"
+    r"validated on pinned head)",
+    re.IGNORECASE,
+)
+_QUEUE_BLOCKED_RE = re.compile(
+    r"queue blocked by already[- ]landed|"
+    r"reconciliation(?:-only)?\s+(?:next step|required)",
+    re.IGNORECASE,
+)
+
+
+def _landed_context(report_text: str, eligibility_class: str | None) -> bool:
+    if (eligibility_class or "").strip().upper() == ALREADY_LANDED_ELIGIBILITY_CLASS:
+        return True
+    return bool(_LANDED_CONTEXT_RE.search(report_text or ""))
+
+
+def assess_already_landed_classification_report(
+    report_text: str,
+    *,
+    eligibility_class: str | None = None,
+    selected_pr_already_landed: bool | None = None,
+) -> dict[str, Any]:
+    """#295: already-landed PRs are reconciliation-only, not review eligible."""
+    text = report_text or ""
+    reasons: list[str] = []
+
+    landed = _landed_context(text, eligibility_class)
+    if selected_pr_already_landed is True:
+        landed = True
+    if not landed:
+        return {
+            "proven": True,
+            "block": False,
+            "landed_context": False,
+            "reasons": [],
+            "safe_next_action": "proceed",
+        }
+
+    if _INELIGIBLE_SELECTION_RE.search(text):
+        reasons.append(
+            "already-landed PR must not be described as oldest/next eligible PR; "
+            "use 'Oldest open PR requiring action' and "
+            f"'Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}'"
+        )
+
+    if _REVIEW_ELIGIBLE_RE.search(text):
+        reasons.append(
+            "already-landed PR must not be described as eligible for review or merge"
+        )
+
+    class_match = _ELIGIBILITY_CLASS_LINE_RE.search(text)
+    if class_match:
+        declared = class_match.group(1).strip().upper()
+        if declared != ALREADY_LANDED_ELIGIBILITY_CLASS:
+            reasons.append(
+                "already-landed PR eligibility class must be "
+                f"{ALREADY_LANDED_ELIGIBILITY_CLASS}"
+            )
+    elif selected_pr_already_landed is True:
+        reasons.append(
+            f"already-landed selected PR must declare Eligibility class: "
+            f"{ALREADY_LANDED_ELIGIBILITY_CLASS}"
+        )
+
+    if _PINNED_REVIEWED_HEAD_RE.search(text):
+        if not _VALIDATION_PROOF_RE.search(text):
+            reasons.append(
+                "already-landed pre-review report must use Candidate head SHA, "
+                "not Pinned reviewed head, unless validation and diff review passed"
+            )
+
+    if selected_pr_already_landed is True and not _CANDIDATE_HEAD_RE.search(text):
+        if _PINNED_REVIEWED_HEAD_RE.search(text) or "head sha" in text.lower():
+            reasons.append(
+                "already-landed ancestry check must report Candidate head SHA"
+            )
+
+    if selected_pr_already_landed is True and not _REVIEWED_HEAD_NONE_RE.search(text):
+        if _PINNED_REVIEWED_HEAD_RE.search(text) and not _VALIDATION_PROOF_RE.search(text):
+            reasons.append(
+                "already-landed report must state 'Reviewed head SHA: none' "
+                "when validation did not run"
+            )
+
+    if selected_pr_already_landed is True and not _QUEUE_BLOCKED_RE.search(text):
+        reasons.append(
+            "already-landed queue blocker must state reconciliation requirement "
+            "or queue blocked wording"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "landed_context": True,
+        "reasons": reasons,
+        "safe_next_action": (
+            "classify as ALREADY_LANDED_RECONCILE_REQUIRED, use Candidate head SHA, "
+            "and stop normal review"
+            if not proven
+            else "proceed"
+        ),
+    }

reviewer_proof_backed_handoff.py

@@ -0,0 +1,217 @@
+"""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_reconcile_inventory.py

@@ -0,0 +1,162 @@
+"""Reconciliation PR inventory pagination proof verifier (#308)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_COMPLETE_SCAN_CLAIM_RE = re.compile(
+    r"\b(?:all already[- ]landed(?:\s+prs?)?(?:\s+were)?\s+found|"
+    r"complete queue scan|all open prs? (?:were )?(?:found|checked|inventoried|listed)|"
+    r"inventory (?:is )?complete|exhaustive (?:pr )?inventory|"
+    r"no additional already[- ]landed|scanned (?:the )?(?:full )?open pr queue)\b",
+    re.I,
+)
+
+_FINALITY_RE = re.compile(
+    r"is_final_page\s*:\s*true|has_more\s*:\s*false|no next page|final[- ]page|"
+    r"pagination_complete\s*:\s*true|inventory pagination proof\s*:\s*(?!assumed|none\b).+",
+    re.I,
+)
+
+_REQUESTED_PAGE_SIZE_RE = re.compile(
+    r"requested page size\s*:\s*(\d+)|page[- ]?size\s*(?:=|:)\s*(\d+)",
+    re.I,
+)
+_PAGES_FETCHED_RE = re.compile(
+    r"pages? fetched\s*:\s*(\d+)|page count\s*:\s*(\d+)",
+    re.I,
+)
+_RETURNED_PER_PAGE_RE = re.compile(
+    r"returned pr count(?: per page)?\s*:|open pr count per page|"
+    r"returned \d+ open prs? per page|per[- ]page counts?\s*:",
+    re.I,
+)
+_EXACT_LIMIT_RETURN_RE = re.compile(
+    r"returned\s+(\d+)\s+open prs?|listed\s+(\d+)\s+open prs?",
+    re.I,
+)
+
+_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile(
+    r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|"
+    r"default (?:gitea )?page[- ]?size|assumed complete",
+    re.I,
+)
+
+
+def _int_from_groups(match: re.Match[str] | None) -> int | None:
+    if not match:
+        return None
+    for group in match.groups():
+        if group:
+            return int(group)
+    return None
+
+
+def _pagination_proven(text: str, session: dict) -> bool:
+    if session.get("pagination_complete") or session.get("inventory_complete"):
+        return True
+    if _FINALITY_RE.search(text):
+        return True
+    pages = session.get("pages_fetched")
+    if isinstance(pages, int) and pages >= 1 and session.get("is_final_page"):
+        return True
+    return False
+
+
+def _metadata_present(text: str, session: dict) -> list[str]:
+    missing: list[str] = []
+    has_page_size = bool(
+        _REQUESTED_PAGE_SIZE_RE.search(text)
+        or session.get("requested_page_size")
+    )
+    has_pages_fetched = bool(
+        _PAGES_FETCHED_RE.search(text) or session.get("pages_fetched")
+    )
+    has_returned_counts = bool(
+        _RETURNED_PER_PAGE_RE.search(text) or session.get("returned_per_page")
+    )
+    if not has_page_size:
+        missing.append("requested page size")
+    if not has_pages_fetched:
+        missing.append("page count fetched")
+    if not has_returned_counts:
+        missing.append("returned PR count per page")
+    if not _pagination_proven(text, session):
+        missing.append("final-page/no-next-page proof")
+    return missing
+
+
+def assess_reconcile_inventory_report(
+    report_text: str,
+    *,
+    inventory_session: dict | None = None,
+) -> dict[str, Any]:
+    """Validate PR inventory pagination proof in reconciliation reports (#308)."""
+    text = report_text or ""
+    session = dict(inventory_session or {})
+    reasons: list[str] = []
+
+    claims_scan = bool(_COMPLETE_SCAN_CLAIM_RE.search(text))
+    lists_open_prs = bool(
+        re.search(r"open prs? listed|listed open prs?|pr inventory", text, re.I)
+    )
+
+    if not claims_scan and not lists_open_prs and not session.get("inventory_required"):
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "inventory_claimed": False,
+            "safe_next_action": "proceed",
+        }
+
+    if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text) and not _pagination_proven(text, session):
+        reasons.append(
+            "reconciliation inventory assumed complete from page-size guess "
+            "without final-page proof (#308)"
+        )
+
+    if claims_scan and not _pagination_proven(text, session):
+        reasons.append(
+            "complete queue scan or all already-landed claim requires "
+            "pagination finality proof (#308)"
+        )
+
+    missing = _metadata_present(text, session)
+    if (claims_scan or lists_open_prs) and missing:
+        reasons.append(
+            "reconciliation inventory missing pagination metadata: "
+            + ", ".join(missing)
+        )
+
+    requested = session.get("requested_page_size")
+    returned = session.get("returned_page_size")
+    if isinstance(requested, int) and isinstance(returned, int):
+        if returned >= requested > 0 and not _pagination_proven(text, session):
+            reasons.append(
+                "exactly-full first reconciliation page without final-page proof (#308)"
+            )
+    else:
+        for match in _EXACT_LIMIT_RETURN_RE.finditer(text):
+            count = _int_from_groups(match)
+            if count in {10, 20, 50} and not _pagination_proven(text, session):
+                reasons.append(
+                    "exactly-full first reconciliation page without final-page proof (#308)"
+                )
+                break
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": list(dict.fromkeys(reasons)),
+        "inventory_claimed": claims_scan or lists_open_prs,
+        "pagination_proven": _pagination_proven(text, session),
+        "safe_next_action": (
+            "include requested page size, pages fetched, per-page counts, and "
+            "final-page/no-next-page proof before claiming complete scan"
+            if reasons
+            else "proceed"
+        ),
+    }

reviewer_validation_failure_history.py

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

role_session_router.py

@@ -76,6 +76,7 @@ AUTHOR_TASKS = frozenset({
     "delete_branch",
     "work_issue",
     "work-issue",
+    "reconcile_landed_pr",
 })
 
 RECONCILER_TASKS = frozenset({
@@ -104,6 +105,7 @@ TASK_REQUIRED_ROLE = {
     "approve_pr": "reviewer",
     "work_issue": "author",
     "work-issue": "author",
+    "reconcile_landed_pr": "author",
     "reconcile_already_landed_pr": "reconciler",
     "reconcile_already_landed": "reconciler",
     "reconcile-landed-pr": "reconciler",

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

@@ -91,4 +91,23 @@ 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.
+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,6 +538,36 @@ 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.
@@ -1096,6 +1126,7 @@ Controller Handoff:
 * Baseline worktree path:
 * Files reviewed:
 * Validation:
+* Validation failure history:
 * Official validation integrity status:
 * Terminal review mutation:
 * Review decision:

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

@@ -83,6 +83,20 @@ Examples:
 
 If capability cannot be proven, stop and produce a recovery handoff only.
 
+### 2A. Pre-task role routing (#139)
+
+Before issue selection or any mutation, resolve the composite author task:
+
+* `gitea_route_task_session(task_type="work-issue", remote=…)` — must return
+  `route_result: allowed_current_session` and `downstream_allowed: true`
+* `gitea_resolve_task_capability(task="work_issue", remote=…)` — must show
+  `required_role_kind: author` and `allowed_in_current_session: true`
+
+Hyphen (`work-issue`) and underscore (`work_issue`) aliases are equivalent.
+If routing returns `ambiguous_task_stop`, `wrong_role_stop`, or
+`route_to_author_session`, stop and produce a recovery handoff only — do not
+fall back to reviewer tools or guess a different profile.
+
 ## 3. Stop immediately on blocked infrastructure
 
 If any of the following appears, stop immediately:
@@ -263,6 +277,24 @@ Do not select an issue based only on memory from a previous session.
 
 Before claiming or working on an issue, check whether there is already an open PR, branch, or active claim for that issue.
 
+Run `gitea_assess_author_duplicate_work` at these stages and stop when `block` is true:
+
+* `claim` — before `gitea_mark_issue`
+* `lock` — before `gitea_lock_issue`
+* `worktree` / `edit` — before creating a worktree or editing files
+* `commit` — immediately before `git commit`
+* `push` — immediately before `git push`
+* `create_pr` — immediately before `gitea_create_pr` (also enforced server-side)
+
+`gitea_mark_issue`, `gitea_lock_issue`, and `gitea_create_pr` enforce the same gate server-side and fail closed.
+
+If a concurrent open PR appears after work begins:
+
+* before commit or push — stop and preserve local work without pushing
+* after push but before PR creation — produce a reconciliation handoff instead of opening a PR
+
+Final reports must name the duplicate-work outcome (`duplicate PR prevented`, `duplicate branch prevented`, `duplicate commit prevented`, `duplicate push prevented`, `duplicate work not prevented`, or `reconciliation handoff`).
+
 If an open PR already exists for the issue, do not implement duplicate work.
 
 Classify the issue as:

task_capability_map.py

@@ -104,6 +104,22 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
+    "work_issue": {
+        "permission": "gitea.pr.create",
+        "role": "author",
+    },
+    "work-issue": {
+        "permission": "gitea.pr.create",
+        "role": "author",
+    },
+    "reconcile_landed_pr": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "reconcile-landed-pr": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
     "reconcile_already_landed_pr": {
         "permission": "gitea.pr.close",
         "role": "reconciler",

tests/test_author_duplicate_work_gate.py

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

tests/test_author_mutation_worktree.py

@@ -0,0 +1,109 @@
+"""Tests for branches-only author mutation worktree guard (#274)."""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import author_mutation_worktree as amw  # noqa: E402
+
+
+class TestPathUnderBranches(unittest.TestCase):
+    ROOT = "/repo/Gitea-Tools"
+
+    def test_branches_subdirectory_passes(self):
+        self.assertTrue(
+            amw.is_path_under_branches(f"{self.ROOT}/branches/issue-274", self.ROOT)
+        )
+
+    def test_control_checkout_fails(self):
+        self.assertFalse(amw.is_path_under_branches(self.ROOT, self.ROOT))
+
+    def test_sibling_directory_fails(self):
+        self.assertFalse(
+            amw.is_path_under_branches("/repo/other-checkout", self.ROOT)
+        )
+
+
+class TestAssessAuthorMutationWorktree(unittest.TestCase):
+    ROOT = "/repo/Gitea-Tools"
+
+    def test_branches_worktree_passes(self):
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=f"{self.ROOT}/branches/issue-274",
+            project_root=self.ROOT,
+            current_branch="feat/issue-274-branches-only-worktrees",
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+    def test_control_checkout_blocks(self):
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=self.ROOT,
+            project_root=self.ROOT,
+            current_branch="master",
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertIn("control checkout", result["reasons"][0])
+
+    def test_drifted_control_branch_blocks(self):
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=self.ROOT,
+            project_root=self.ROOT,
+            current_branch="feat/some-task",
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(any("drift" in r for r in result["reasons"]))
+
+    def test_branches_worktree_as_project_root_passes(self):
+        """MCP launched from a branches/ worktree uses that path as PROJECT_ROOT."""
+        worktree_root = f"{self.ROOT}/branches/issue-274"
+        result = amw.assess_author_mutation_worktree(
+            workspace_path=worktree_root,
+            project_root=worktree_root,
+            current_branch="feat/issue-274-branches-only-worktrees",
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["block"])
+
+
+class TestPreflightIntegration(unittest.TestCase):
+    def test_verify_preflight_blocks_control_checkout_with_test_porcelain(self):
+        import mcp_server
+
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "author"
+        control_root = "/repo/Gitea-Tools"
+        with mock.patch.object(mcp_server, "PROJECT_ROOT", control_root):
+            with mock.patch.dict(
+                "os.environ",
+                {"GITEA_TEST_PORCELAIN": ""},
+                clear=False,
+            ):
+                with self.assertRaises(RuntimeError) as ctx:
+                    mcp_server.verify_preflight_purity()
+        self.assertIn("Branches-only mutation guard", str(ctx.exception))
+
+    def test_verify_preflight_allows_branches_worktree(self):
+        import mcp_server
+
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "author"
+        worktree = "/repo/Gitea-Tools/branches/issue-274"
+        with mock.patch.dict(
+            "os.environ",
+            {"GITEA_TEST_PORCELAIN": ""},
+            clear=False,
+        ):
+            mcp_server.verify_preflight_purity(worktree_path=worktree)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_commit_payloads.py

@@ -55,7 +55,15 @@ class TestCommitPayloads(unittest.TestCase):
         with open(self.config_path, "w", encoding="utf-8") as fh:
             fh.write(json.dumps(CONFIG))
 
-        self.locked_worktree_dir = tempfile.TemporaryDirectory()
+        repo_root = os.path.realpath(
+            os.path.join(os.path.dirname(__file__), os.pardir)
+        )
+        branches_root = os.path.join(repo_root, "branches")
+        os.makedirs(branches_root, exist_ok=True)
+        self.locked_worktree_dir = tempfile.TemporaryDirectory(
+            dir=branches_root,
+            prefix="test-commit-payloads-",
+        )
         self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
 
         self.lock_file_path = "/tmp/gitea_issue_lock.json"
@@ -94,6 +102,7 @@ class TestCommitPayloads(unittest.TestCase):
             "GITEA_MCP_PROFILE": profile,
             "GITEA_TOKEN_AUTHOR": "author-pass",
             "GITEA_TEST_PORCELAIN": "",
+            "GITEA_AUTHOR_WORKTREE": self.locked_worktree_path,
         }
 
     @patch("mcp_server.api_request")

tests/test_delete_branch_capability.py

@@ -0,0 +1,217 @@
+"""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_final_report_validator.py

@@ -57,26 +57,47 @@ def _review_handoff(**overrides):
     return text
 
 
-def _reconcile_handoff(**overrides):
+def _reconcile_handoff(drop=(), **overrides):
     lines = [
         "## Controller Handoff",
         "",
         "- Task: reconcile already-landed PR #99",
         "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
-        "- Role: reconciler",
-        "- Identity: sysadmin / prgs-author",
+        "- Role/profile: reconciler / prgs-reconciler",
+        "- Identity: sysadmin",
         "- Selected PR: #99",
-        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+        "- PR live state: open",
+        "- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Target branch: master",
+        "- Target branch SHA: 5e023dc71b0e2b813a0b1eee6b0f42630c47a95c",
+        "- Ancestor proof: candidate head is ancestor of target branch SHA",
         "- Linked issue: #98",
         "- Linked issue live status: not verified in this session",
+        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+        "- Capabilities proven: gitea.read, gitea.pr.comment",
+        "- Missing capabilities: gitea.pr.close",
+        "- PR comments posted: 1 reconciliation comment on PR #99",
+        "- Issue comments posted: none",
+        "- PRs closed: none",
+        "- Issues closed: none",
         "- File edits by reconciler: none",
         "- Git ref mutations: git fetch prgs master",
+        "- Worktree mutations: none",
+        "- MCP/Gitea mutations: PR comment posted",
+        "- External-state mutations: none",
+        "- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
         "- Reconciliation mutations: PR comment posted",
         "- Current status: reconciliation complete",
-        "- Safe next action: none",
+        "- Blocker: missing gitea.pr.close capability",
+        "- Safe next action: hand off to a profile with gitea.pr.close",
         "- No review/merge confirmation: confirmed",
     ]
-    text = "\n".join(lines)
+    kept = [
+        line
+        for line in lines
+        if not any(line.startswith(f"- {name}:") for name in drop)
+    ]
+    text = "\n".join(kept)
     for key, value in overrides.items():
         if f"- {key}:" in text:
             text = text.replace(f"- {key}:", f"- {key}: {value}")
@@ -194,6 +215,20 @@ class TestReviewPrRules(unittest.TestCase):
             )
         )
 
+    def test_already_landed_oldest_eligible_pr_blocks(self):
+        report = (
+            _review_handoff()
+            + "\nAlready landed.\nOldest eligible PR: PR #278."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.already_landed_eligible_wording"
+                for f in result["findings"]
+            )
+        )
+
     def test_already_landed_gate_blocks_review_terminal_states(self):
         cases = {
             "APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"),
@@ -242,7 +277,6 @@ class TestReviewPrRules(unittest.TestCase):
         )
         result = assess_final_report_validator(report, "review_pr")
         self.assertEqual(result["grade"], "A")
-
     def test_git_fetch_must_not_be_readonly_only(self):
         report = (
             _review_handoff()
@@ -307,6 +341,159 @@ class TestReconciliationRules(unittest.TestCase):
         )
 
 
+class TestCanonicalReconcileSchema(unittest.TestCase):
+    """Issue #307: reconciliation workflows emit only the canonical schema."""
+
+    def test_comment_only_reconciliation_passes(self):
+        result = assess_final_report_validator(
+            _reconcile_handoff(),
+            "reconcile_already_landed",
+        )
+        self.assertEqual(result["grade"], "A")
+        self.assertEqual(result["findings"], [])
+
+    def test_successful_close_reconciliation_passes(self):
+        report = _reconcile_handoff().replace(
+            "- Capabilities proven: gitea.read, gitea.pr.comment",
+            "- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close",
+        ).replace(
+            "- Missing capabilities: gitea.pr.close",
+            "- Missing capabilities: none",
+        ).replace(
+            "- PRs closed: none",
+            "- PRs closed: #99",
+        ).replace(
+            "- Blocker: missing gitea.pr.close capability",
+            "- Blocker: none",
+        )
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertEqual(result["grade"], "A")
+
+    def test_blocked_reconciliation_passes(self):
+        report = _reconcile_handoff().replace(
+            "- Capabilities proven: gitea.read, gitea.pr.comment",
+            "- Capabilities proven: gitea.read",
+        ).replace(
+            "- Missing capabilities: gitea.pr.close",
+            "- Missing capabilities: gitea.pr.close, gitea.pr.comment",
+        ).replace(
+            "- PR comments posted: 1 reconciliation comment on PR #99",
+            "- PR comments posted: none",
+        ).replace(
+            "- MCP/Gitea mutations: PR comment posted",
+            "- MCP/Gitea mutations: none",
+        ).replace(
+            "- Reconciliation mutations: PR comment posted",
+            "- Reconciliation mutations: none",
+        )
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertEqual(result["grade"], "A")
+
+    def test_missing_capabilities_field_required(self):
+        report = _reconcile_handoff(drop=("Missing capabilities",))
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertEqual(result["grade"], "downgraded")
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.controller_handoff"
+                and "Missing capabilities" in f["reason"]
+                for f in result["findings"]
+            )
+        )
+
+    def test_missing_ancestor_proof_and_candidate_sha_downgrade(self):
+        report = _reconcile_handoff(drop=("Ancestor proof", "Candidate head SHA"))
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertEqual(result["grade"], "downgraded")
+        reasons = " ".join(f["reason"] for f in result["findings"])
+        self.assertIn("Ancestor proof", reasons)
+        self.assertIn("Candidate head SHA", reasons)
+
+    def test_stale_issue_lock_proof_blocks(self):
+        report = _reconcile_handoff() + "\n- Issue lock proof: locked before diff"
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                and f["field"] == "issue lock proof"
+                for f in result["findings"]
+            )
+        )
+
+    def test_stale_claim_comment_status_blocks(self):
+        report = _reconcile_handoff() + "\n- Claim/comment status: claimed"
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                and f["field"] == "claim/comment status"
+                for f in result["findings"]
+            )
+        )
+
+    def test_stale_workspace_mutations_blocks_without_observed_mutations(self):
+        report = _reconcile_handoff() + "\n- Workspace mutations: None"
+        result = assess_final_report_validator(report, "reconcile_already_landed")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                and f["field"] == "workspace mutations"
+                for f in result["findings"]
+            )
+        )
+
+    def test_pr_number_opened_allowed_when_session_opened_pr(self):
+        report = _reconcile_handoff() + "\n- PR number opened: #12"
+        result = assess_final_report_validator(
+            report,
+            "reconcile_already_landed",
+            session_pr_opened=True,
+        )
+        self.assertFalse(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                and f["field"] == "pr number opened"
+                for f in result["findings"]
+            )
+        )
+
+    def test_pr_number_opened_still_blocked_without_session_proof(self):
+        report = _reconcile_handoff() + "\n- PR number opened: #12"
+        result = assess_final_report_validator(
+            report,
+            "reconcile_already_landed",
+            session_pr_opened=False,
+        )
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reconcile.stale_author_reviewer_fields"
+                and f["field"] == "pr number opened"
+                for f in result["findings"]
+            )
+        )
+
+    def test_action_log_fetch_requires_git_ref_mutation_entry(self):
+        report = _reconcile_handoff().replace(
+            "- Git ref mutations: git fetch prgs master",
+            "- Git ref mutations: none",
+        )
+        result = assess_final_report_validator(
+            report,
+            "reconcile_already_landed",
+            action_log=[{"command": "git fetch prgs master", "performed": True}],
+        )
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.git_fetch_readonly"
+                for f in result["findings"]
+            )
+        )
+
+
 class TestEntryPoint(unittest.TestCase):
     def test_unknown_task_kind_blocks(self):
         result = assess_final_report_validator("report", "unknown_mode")

tests/test_llm_workflow_split.py

@@ -95,11 +95,19 @@ 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
     assert "Do not merge your own PR" in text
     assert "## 21. PR creation rules" in text
+    assert "gitea_route_task_session" in text
+    assert "work-issue" in text
 
 
 def test_pr_queue_cleanup_workflow_contract():
@@ -139,6 +147,16 @@ def test_start_issue_template_references_work_issue_workflow():
     assert "workflows/work-issue.md" in text
 
 
+def test_reconcile_skill_registered_in_mcp_server():
+    from mcp_server import _PROJECT_SKILLS
+
+    assert "gitea-reconcile-landed-pr" in _PROJECT_SKILLS
+    steps = _PROJECT_SKILLS["gitea-reconcile-landed-pr"]["steps"]
+    joined = " ".join(steps).lower()
+    assert "gitea_scan_already_landed_open_prs" in joined
+    assert "gitea_assess_already_landed_reconciliation" in joined
+
+
 def test_pr_queue_cleanup_template_references_workflow():
     text = (SKILL_DIR / "templates" / "pr-queue-cleanup.md").read_text(
         encoding="utf-8"
@@ -165,12 +183,24 @@ def test_final_report_validator_exported():
     assert callable(assess_final_report_validator)
 
 
+def test_review_final_report_schema_verifier_exported():
+    from review_proofs import assess_review_final_report_schema
+
+    assert callable(assess_review_final_report_schema)
+
+
 def test_create_issue_final_report_verifier_exported():
     from review_proofs import assess_create_issue_final_report
 
     assert callable(assess_create_issue_final_report)
 
 
+def test_work_issue_final_report_verifier_exported():
+    from review_proofs import assess_work_issue_final_report
+
+    assert callable(assess_work_issue_final_report)
+
+
 def test_merge_simulation_verifier_exported():
     from review_proofs import assess_merge_simulation_report
 
@@ -183,6 +213,12 @@ def test_validation_integrity_verifier_exported():
     assert callable(assess_validation_integrity_report)
 
 
+def test_validation_failure_history_verifier_exported():
+    from review_proofs import assess_validation_failure_history_report
+
+    assert callable(assess_validation_failure_history_report)
+
+
 def test_prior_blocker_skip_verifier_exported():
     from review_proofs import assess_prior_blocker_skip_proof
 
@@ -201,6 +237,12 @@ def test_validation_worktree_edit_verifier_exported():
     assert callable(assess_validation_worktree_edit_report)
 
 
+def test_reconcile_inventory_verifier_exported():
+    from review_proofs import assess_reconcile_inventory_report
+
+    assert callable(assess_reconcile_inventory_report)
+
+
 def test_reconcile_linked_issue_verifier_exported():
     from review_proofs import assess_reconcile_linked_issue_report
 
@@ -219,6 +261,12 @@ 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
 
@@ -237,6 +285,12 @@ def test_worktree_ownership_verifier_exported():
     assert callable(assess_worktree_ownership_report)
 
 
+def test_already_landed_classification_verifier_exported():
+    from review_proofs import assess_already_landed_classification_report
+
+    assert callable(assess_already_landed_classification_report)
+
+
 def test_pr_queue_cleanup_verifier_exported():
     from review_proofs import assess_pr_queue_cleanup_report
 

tests/test_mcp_server.py

@@ -1005,9 +1005,19 @@ 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):
+    def test_delete_branch(self, _auth, mock_api, _profile):
+        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"])
@@ -3050,10 +3060,18 @@ class TestIssueLocking(unittest.TestCase):
         mock_api.return_value = [] # no open PRs
         res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertTrue(res["success"])
+        self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work")
+        self.assertEqual(res["work_lease"]["issue_number"], 196)
+        self.assertEqual(res["work_lease"]["branch"], "feat/issue-196-mutations")
+        self.assertIn("created_at", res["work_lease"])
+        self.assertIn("expires_at", res["work_lease"])
+        self.assertIn("last_heartbeat_at", res["work_lease"])
+        self.assertEqual(res["work_lease"]["claimant"]["profile"], "gitea-default")
         self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
         with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
             lock = json.load(f)
         self.assertIn("worktree_path", lock)
+        self.assertIn("work_lease", lock)
 
     def test_lock_issue_mismatch_branch_fails(self):
         with self.assertRaises(ValueError) as ctx:
@@ -3094,6 +3112,51 @@ class TestIssueLocking(unittest.TestCase):
             gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
         self.assertIn("already tied to an open PR", str(ctx.exception))
 
+    @patch(
+        "mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value=_clean_master_git_state_for_lock(),
+    )
+    @patch("mcp_server.api_get_all")
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    def test_lock_issue_reused_by_remote_branch(self, _auth, mock_api, _git_state):
+        mock_api.side_effect = [
+            [],
+            [{"name": "feat/issue-196-existing-work"}],
+        ]
+        with self.assertRaises(ValueError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("already has matching branch", str(ctx.exception))
+
+    def test_lock_issue_blocks_active_same_operation_lease(self):
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump({
+                "issue_number": 196,
+                "branch_name": "feat/issue-196-other-work",
+                "worktree_path": "/tmp/other-worktree",
+                "work_lease": {
+                    "operation_type": "author_issue_work",
+                    "expires_at": "2999-01-01T00:00:00Z",
+                },
+            }, f)
+        with self.assertRaises(RuntimeError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("already has an active author_issue_work lease", str(ctx.exception))
+
+    def test_lock_issue_blocks_expired_same_operation_lease_for_recovery(self):
+        with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
+            json.dump({
+                "issue_number": 196,
+                "branch_name": "feat/issue-196-other-work",
+                "worktree_path": "/tmp/other-worktree",
+                "work_lease": {
+                    "operation_type": "author_issue_work",
+                    "expires_at": "2000-01-01T00:00:00Z",
+                },
+            }, f)
+        with self.assertRaises(RuntimeError) as ctx:
+            gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
+        self.assertIn("Recovery review is required before takeover", str(ctx.exception))
+
     @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_lock_from_clean_scratch_worktree(self, _auth, _api):

tests/test_reconciler_profile.py

@@ -0,0 +1,87 @@
+"""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_reconciliation_mcp_tools.py

@@ -0,0 +1,106 @@
+"""Tests for reconciliation MCP assessment tools (#301)."""
+import sys
+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_assess_already_landed_reconciliation,
+    gitea_scan_already_landed_open_prs,
+)
+
+AUTHOR_PROFILE = {
+    "profile_name": "prgs-author",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.comment",
+        "gitea.issue.comment",
+        "gitea.issue.close",
+    ],
+    "forbidden_operations": ["gitea.pr.close"],
+    "audit_label": "prgs-author",
+}
+
+OPEN_PR = {
+    "number": 278,
+    "title": "Landed (Closes #263)",
+    "body": "",
+    "state": "open",
+    "head": {"ref": "feat/x", "sha": "a" * 40},
+    "base": {"ref": "master"},
+}
+
+
+class TestReconciliationMcpTools(unittest.TestCase):
+    def setUp(self):
+        self.mock_api = patch("mcp_server.api_request").start()
+        self.mock_auth = patch(
+            "mcp_server.get_auth_header", return_value="token test"
+        ).start()
+        patch("mcp_server.get_profile", return_value=AUTHOR_PROFILE).start()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def tearDown(self):
+        patch.stopall()
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def test_assess_returns_plan_without_mutations(self):
+        self.mock_api.return_value = OPEN_PR
+        with patch(
+            "mcp_server.reconciliation_workflow.assess_open_pr_reconciliation",
+            return_value={
+                "reconciliation_allowed": True,
+                "eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
+                "candidate_head_sha": OPEN_PR["head"]["sha"],
+                "target_branch_sha": "deadbeef",
+                "linked_issue": 263,
+                "review_merge_allowed": False,
+            },
+        ):
+            res = gitea_assess_already_landed_reconciliation(
+                pr_number=278, remote="prgs"
+            )
+        self.assertTrue(res["success"])
+        self.assertFalse(res["performed"])
+        self.assertEqual(res["plan"]["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP")
+        self.assertFalse(res["review_merge_allowed"])
+        patch_calls = [
+            c for c in self.mock_api.call_args_list if c.args[0] == "PATCH"
+        ]
+        self.assertEqual(patch_calls, [])
+
+    def test_scan_returns_candidates_with_pagination(self):
+        with patch(
+            "mcp_server.api_fetch_page",
+            return_value=([OPEN_PR], {
+                "page": 1,
+                "per_page": 50,
+                "is_final_page": True,
+                "has_more": False,
+                "next_page": None,
+            }),
+        ), patch(
+            "mcp_server.reconciliation_workflow.fetch_target_branch",
+            return_value={
+                "success": True,
+                "target_branch_sha": "deadbeef",
+                "git_fetch_command": "git fetch prgs master",
+            },
+        ), patch(
+            "mcp_server.reconciliation_workflow.assess_open_pr_reconciliation",
+            return_value={
+                "pr_number": 278,
+                "eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
+                "reconciliation_allowed": True,
+            },
+        ):
+            res = gitea_scan_already_landed_open_prs(remote="prgs", limit=50)
+        self.assertTrue(res["success"])
+        self.assertEqual(res["candidate_count"], 1)
+        self.assertTrue(res["pagination"]["inventory_complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reconciliation_workflow.py

@@ -0,0 +1,149 @@
+"""Tests for already-landed reconciliation workflow helpers (#301)."""
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import reconciliation_workflow
+from review_proofs import assess_reconcile_workflow_source
+
+
+class TestReconciliationAssessment(unittest.TestCase):
+    def test_already_landed_open_pr(self):
+        with patch(
+            "reconciliation_workflow.is_head_ancestor_of_ref",
+            return_value=True,
+        ):
+            assessment = reconciliation_workflow.assess_open_pr_reconciliation(
+                pr={
+                    "number": 278,
+                    "state": "open",
+                    "title": "Fix (Closes #263)",
+                    "body": "",
+                    "head": {"ref": "feat/x", "sha": "abc" * 13 + "a"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                    "git_fetch_command": "git fetch prgs master",
+                },
+            )
+        self.assertTrue(assessment["reconciliation_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+        )
+
+    def test_not_landed_pr(self):
+        with patch(
+            "reconciliation_workflow.is_head_ancestor_of_ref",
+            return_value=False,
+        ):
+            assessment = reconciliation_workflow.assess_open_pr_reconciliation(
+                pr={
+                    "number": 1,
+                    "state": "open",
+                    "title": "WIP",
+                    "body": "",
+                    "head": {"ref": "feat/y", "sha": "fff" * 13 + "f"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                },
+            )
+        self.assertFalse(assessment["reconciliation_allowed"])
+
+
+class TestReconciliationPlan(unittest.TestCase):
+    def test_full_close_when_close_pr_allowed(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": True,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+            },
+            capabilities={
+                "close_pr": True,
+                "comment_pr": True,
+                "close_issue": True,
+                "comment_issue": True,
+            },
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_FULL_RECONCILE,
+        )
+        self.assertTrue(plan["close_pr_allowed"])
+        self.assertFalse(plan["review_merge_allowed"])
+
+    def test_comment_then_stop_without_close(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": True,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+            },
+            capabilities={
+                "close_pr": False,
+                "comment_pr": True,
+            },
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_PARTIAL_COMMENT,
+        )
+
+    def test_recovery_handoff_without_comment_or_close(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": True,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_ALREADY_LANDED,
+            },
+            capabilities={"close_pr": False, "comment_pr": False},
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_RECOVERY_HANDOFF,
+        )
+
+    def test_not_landed_no_action(self):
+        plan = reconciliation_workflow.resolve_reconciliation_plan(
+            assessment={
+                "reconciliation_allowed": False,
+                "eligibility_class": reconciliation_workflow.ELIGIBILITY_NOT_LANDED,
+                "reasons": ["not ancestor"],
+            },
+            capabilities={"close_pr": True},
+        )
+        self.assertEqual(
+            plan["outcome"],
+            reconciliation_workflow.OUTCOME_NOT_LANDED,
+        )
+
+
+class TestReconcileWorkflowSource(unittest.TestCase):
+    def test_valid_report_passes(self):
+        report = (
+            "Task mode: reconcile-landed-pr\n"
+            "Workflow source: workflows/reconcile-landed-pr.md\n"
+        )
+        result = assess_reconcile_workflow_source(report)
+        self.assertTrue(result["complete"])
+
+    def test_missing_workflow_fails(self):
+        result = assess_reconcile_workflow_source("Task mode: reconcile-landed-pr")
+        self.assertFalse(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -121,6 +121,32 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertTrue(res["allowed_in_current_session"])
             self.assertFalse(res["different_mcp_namespace_required"])
 
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_work_issue_author_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("author-profile")):
+            for task in ("work_issue", "work-issue"):
+                res = mcp_server.gitea_resolve_task_capability(
+                    task=task, remote="prgs"
+                )
+                self.assertEqual(res["requested_task"], task)
+                self.assertEqual(
+                    res["required_operation_permission"], "gitea.pr.create"
+                )
+                self.assertEqual(res["required_role_kind"], "author")
+                self.assertTrue(res["allowed_in_current_session"])
+
+    @patch("mcp_server.api_request", return_value={"login": "author-user"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_resolve_work_issue_reviewer_profile_blocked(self, _auth, _api):
+        with patch.dict(os.environ, self._env("reviewer-profile")):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="work_issue", remote="prgs"
+            )
+            self.assertFalse(res["allowed_in_current_session"])
+            self.assertEqual(res["required_role_kind"], "author")
+            self.assertTrue(res["different_mcp_namespace_required"])
+
     @patch("mcp_server.api_request", return_value={"login": "author-user"})
     @patch("mcp_server.get_auth_header", return_value="token author-pass")
     def test_resolve_issue_comment_task(self, _auth, _api):

tests/test_review_final_report_schema.py

@@ -0,0 +1,158 @@
+"""Tests for reviewer final-report schema verification (#391)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_final_report_schema import assess_review_final_report_schema  # noqa: E402
+
+
+def _minimal_review_report(**overrides):
+    lines = [
+        "## Controller Handoff",
+        "",
+        "- Task: review PR #203",
+        "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+        "- Role: reviewer",
+        "- Identity: sysadmin / prgs-reviewer",
+        "- Issue/PR: #182 / PR #203",
+        "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Files changed: review_proofs.py",
+        "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
+        "- Mutations: review only",
+        "- File edits by reviewer: none",
+        "- Worktree/index mutations: none",
+        "- Git ref mutations: git fetch prgs master",
+        "- MCP/Gitea mutations: review submitted",
+        "- Review mutations: request_changes submitted",
+        "- Merge mutations: none",
+        "- Cleanup mutations: none",
+        "- External-state mutations: none",
+        "- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
+        "- Current status: PR open",
+        "- Blockers: none",
+        "- Next: await author",
+        "- Safety: no self-review; no self-merge",
+        "- Selected PR: #203",
+        "- Reviewer eligibility: eligible",
+        "- Candidate head SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
+        "- Worktree path: branches/review-203",
+        "- Worktree dirty: clean",
+        "- Unrelated local mutations: none",
+        "- Review decision: request_changes",
+        "- Merge result: not attempted",
+        "- Linked issue: #182",
+        "- Linked issue live status: open (gitea_view_issue)",
+        "- Cleanup status: none",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        prefix = f"- {key}:"
+        replaced = False
+        new_lines = []
+        for line in text.splitlines():
+            if line.strip().startswith(prefix):
+                new_lines.append(f"{prefix} {value}")
+                replaced = True
+            else:
+                new_lines.append(line)
+        text = "\n".join(new_lines)
+        if not replaced:
+            text += f"\n{prefix} {value}"
+    return text
+
+
+class TestReviewFinalReportSchema(unittest.TestCase):
+    def test_clean_report_passes(self):
+        result = assess_review_final_report_schema(
+            _minimal_review_report(),
+            linked_issue_lock={"issue_number": 182, "pr_number": 203},
+        )
+        self.assertFalse(result["blocked"])
+        self.assertIn(result["grade"], ("A", "downgraded"))
+
+    def test_legacy_pinned_reviewed_head_blocks(self):
+        report = _minimal_review_report(**{"Pinned reviewed head": "abc123"})
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.legacy_stale_field", rule_ids)
+
+    def test_reviewed_head_without_validation_blocks(self):
+        report = _minimal_review_report().replace(
+            "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
+            "- Validation: not run",
+        )
+        report += "\n- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.reviewed_head_without_validation", rule_ids)
+
+    def test_merged_without_proof_blocks(self):
+        report = _minimal_review_report() + "\n\nPR #203 merged successfully."
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.merged_without_proof", rule_ids)
+
+    def test_issue_closed_without_live_proof_blocks(self):
+        report = _minimal_review_report(
+            **{
+                "Linked issue live status": "closed",
+                "Read-only diagnostics": "gitea_view_pr only",
+            }
+        )
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.issue_closed_without_live_proof", rule_ids)
+
+    def test_full_suite_pass_with_ignored_tests_blocks(self):
+        report = (
+            _minimal_review_report()
+            + "\nFull test suite passed with 0 failed; some tests ignored."
+        )
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.full_suite_pass_ignored_tests", rule_ids)
+
+    def test_blocked_handoff_replay_blocks(self):
+        report = (
+            _minimal_review_report(**{"Blockers": "capability stop"})
+            + "\nSafe next action: run gitea_merge_pr to finish."
+        )
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.blocked_handoff_replay", rule_ids)
+
+    def test_narrative_handoff_drift_blocks(self):
+        report = (
+            _minimal_review_report()
+            + "\n## Summary\nVerdict: APPROVE this PR."
+        )
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.narrative_handoff_drift", rule_ids)
+
+    def test_pending_vs_approved_blocks(self):
+        report = _minimal_review_report(
+            **{"Review decision": "PENDING official review submitted approve"}
+        )
+        result = assess_review_final_report_schema(report)
+        rule_ids = [f["rule_id"] for f in result["findings"]]
+        self.assertIn("reviewer.pending_vs_approved", rule_ids)
+
+    def test_exported_from_review_proofs(self):
+        from review_proofs import assess_review_final_report_schema as exported
+
+        self.assertTrue(callable(exported))
+
+
+class TestMcpValidateReviewFinalReport(unittest.TestCase):
+    def test_mcp_tool_callable(self):
+        import gitea_mcp_server
+
+        result = gitea_mcp_server.gitea_validate_review_final_report(
+            _minimal_review_report()
+        )
+        self.assertIn("grade", result)
+        self.assertIn("findings", result)
+        self.assertFalse(result["blocked"])

tests/test_review_proofs.py

@@ -40,6 +40,9 @@ from review_proofs import (  # noqa: E402
     assess_create_issue_final_report,
     assess_create_issue_mode_isolation,
     assess_create_issue_workflow_source,
+    assess_work_issue_final_report,
+    assess_work_issue_mode_isolation,
+    assess_work_issue_workflow_source,
     assess_edited_pr_inventory_coverage,
     assess_empty_queue_report,
     assess_fresh_issue_selection,
@@ -2320,6 +2323,52 @@ class TestCreateIssueFinalReport(unittest.TestCase):
         self.assertTrue(result["downgraded"])
 
 
+class TestWorkIssueFinalReport(unittest.TestCase):
+    """#139: work-issue final-report verifier coverage."""
+
+    def _good_report(self):
+        return "\n".join([
+            "Task mode: work-issue",
+            "Workflow source: skills/llm-project-workflow/workflows/work-issue.md",
+            "## Controller Handoff",
+            "- Task: work-issue",
+            "- Repo: Scaled-Tech-Consulting/Gitea-Tools",
+            "- Role: author",
+            "- Identity: jcwalker3 / prgs-author",
+            "- Active profile: prgs-author",
+            "- Runtime context: prgs-author on prgs",
+            "- Selected issue: #139",
+            "- Branch name: feat/issue-139-role-aware-work-issue-routing",
+            "- PR number: not created in this session",
+            "- File edits by author: role_session_router.py",
+            "- Blockers: none",
+            "- Current status: implementing routing",
+            "- Safe next action: open PR",
+            "- Next: open PR",
+            "- Safety statement: no review/merge",
+        ])
+
+    def test_complete_work_issue_report_earns_a(self):
+        result = assess_work_issue_final_report(self._good_report())
+        self.assertEqual(result["grade"], "A")
+        self.assertFalse(result["downgraded"])
+
+    def test_missing_workflow_source_downgrades(self):
+        report = self._good_report().replace(
+            "workflows/work-issue.md", "SKILL.md only"
+        )
+        result = assess_work_issue_workflow_source(report)
+        self.assertTrue(result["downgraded"])
+
+    def test_review_field_in_handoff_downgrades(self):
+        report = self._good_report().replace(
+            "- Safety statement:",
+            "- Review decision: APPROVE\n- Safety statement:",
+        )
+        result = assess_work_issue_mode_isolation(report)
+        self.assertTrue(result["downgraded"])
+
+
 class TestValidationEnvironmentProof(unittest.TestCase):
     """Issue #311: exact validation command and execution environment."""
 

tests/test_reviewer_already_landed_classification.py

@@ -0,0 +1,81 @@
+import unittest
+
+from reviewer_already_landed_classification import (  # noqa: E402
+    ALREADY_LANDED_ELIGIBILITY_CLASS,
+    assess_already_landed_classification_report,
+)
+
+
+class TestAlreadyLandedClassification(unittest.TestCase):
+    def test_non_landed_report_passes(self):
+        result = assess_already_landed_classification_report(
+            "Selected the next eligible PR: PR #286."
+        )
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["landed_context"])
+
+    def test_oldest_eligible_pr_on_landed_blocks(self):
+        result = assess_already_landed_classification_report(
+            "PR is already landed.\n"
+            "Oldest eligible PR: PR #278."
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_next_eligible_pr_on_landed_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Ancestor proof: passed\n"
+            "Selected the next eligible PR: PR #278."
+        )
+        self.assertFalse(result["proven"])
+
+    def test_eligible_for_review_on_landed_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Already landed on master and eligible for review next."
+        )
+        self.assertFalse(result["proven"])
+
+    def test_proper_landed_wording_passes(self):
+        result = assess_already_landed_classification_report(
+            "Oldest open PR requiring action: PR #278\n"
+            f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n"
+            "Candidate head SHA: abc123\n"
+            "Reviewed head SHA: none\n"
+            "Queue blocked by already-landed PR requiring reconciliation",
+            selected_pr_already_landed=True,
+        )
+        self.assertTrue(result["proven"])
+
+    def test_pinned_reviewed_head_without_validation_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Already landed.\n"
+            "Pinned reviewed head: abc123def456",
+            selected_pr_already_landed=True,
+        )
+        self.assertFalse(result["proven"])
+
+    def test_wrong_eligibility_class_blocks(self):
+        result = assess_already_landed_classification_report(
+            "Already landed.\n"
+            "Eligibility class: REVIEW_ELIGIBLE",
+            selected_pr_already_landed=True,
+        )
+        self.assertFalse(result["proven"])
+
+    def test_missing_queue_blocker_blocks(self):
+        result = assess_already_landed_classification_report(
+            f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n"
+            "Candidate head SHA: abc123\n"
+            "Reviewed head SHA: none",
+            selected_pr_already_landed=True,
+        )
+        self.assertFalse(result["proven"])
+
+    def test_exported_from_review_proofs(self):
+        from review_proofs import assess_already_landed_classification_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_proof_backed_handoff.py

@@ -0,0 +1,117 @@
+"""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_reconcile_inventory.py

@@ -0,0 +1,102 @@
+"""Tests for reconciliation inventory pagination verifier (#308)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_reconcile_inventory import assess_reconcile_inventory_report  # noqa: E402
+
+
+def _good_report() -> str:
+    return "\n".join([
+        "## Already-landed reconciliation",
+        "Open PR inventory:",
+        "- Requested page size: 50",
+        "- Pages fetched: 2",
+        "- Returned PR count per page: page1=50, page2=6",
+        "- Inventory pagination proof: is_final_page: true, has_more: false",
+        "All already-landed PRs in the scanned open queue were found.",
+        "",
+        "## Controller Handoff",
+        "- Selected PR: #278",
+        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+    ])
+
+
+class TestReconcileInventoryPagination(unittest.TestCase):
+    def test_complete_metadata_passes(self):
+        result = assess_reconcile_inventory_report(_good_report())
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["pagination_proven"])
+
+    def test_no_inventory_claim_skips_checks(self):
+        report = "## Controller Handoff\n- Selected PR: #278"
+        result = assess_reconcile_inventory_report(report)
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["inventory_claimed"])
+
+    def test_first_page_below_limit_with_finality_passes(self):
+        report = "\n".join([
+            "Open PRs listed for reconciliation.",
+            "- Requested page size: 50",
+            "- Pages fetched: 1",
+            "- Returned PR count per page: page1=6",
+            "- Inventory pagination proof: is_final_page: true",
+        ])
+        result = assess_reconcile_inventory_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_exactly_full_first_page_without_finality_blocks(self):
+        report = "\n".join([
+            "Complete queue scan of open PRs.",
+            "- Requested page size: 50",
+            "- Pages fetched: 1",
+            "- Returned PR count per page: page1=50",
+            "Listed 50 open PRs on the first page.",
+        ])
+        result = assess_reconcile_inventory_report(
+            report,
+            inventory_session={"requested_page_size": 50, "returned_page_size": 50},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("full first" in r.lower() for r in result["reasons"]))
+
+    def test_missing_pagination_metadata_blocks(self):
+        report = "All open PRs were inventoried. Inventory is complete."
+        result = assess_reconcile_inventory_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("metadata" in r.lower() for r in result["reasons"]))
+
+    def test_default_page_size_assumption_blocks(self):
+        report = (
+            "Exhaustive PR inventory: fewer than the default Gitea page size returned."
+        )
+        result = assess_reconcile_inventory_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("page-size" in r.lower() or "page size" in r.lower()
+                            for r in result["reasons"]))
+
+    def test_session_pagination_complete_passes(self):
+        report = "All already-landed PRs were found after scanning open PRs."
+        result = assess_reconcile_inventory_report(
+            report,
+            inventory_session={
+                "pagination_complete": True,
+                "requested_page_size": 50,
+                "pages_fetched": 2,
+                "returned_per_page": [50, 3],
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_reconcile_inventory_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_validation_failure_history.py

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

tests/test_role_session_router.py

@@ -192,6 +192,31 @@ class TestRoleSessionRouter(unittest.TestCase):
             )
         self.assertEqual(result.get("number"), 999)
 
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_work_issue_task_under_author_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-author")):
+            for task_type in ("work_issue", "work-issue"):
+                route = mcp_server.gitea_route_task_session(
+                    task_type=task_type, remote="prgs"
+                )
+                self.assertEqual(
+                    route["route_result"], role_session_router.ROUTE_ALLOWED
+                )
+                self.assertTrue(route["downstream_allowed"])
+
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_work_issue_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-reviewer")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="work-issue", remote="prgs"
+            )
+        self.assertEqual(
+            route["route_result"], role_session_router.ROUTE_TO_AUTHOR
+        )
+        self.assertFalse(route["downstream_allowed"])
+
     @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
     @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
     def test_reviewer_task_under_reviewer_profile_allowed(self, _auth, _api):