fix: resolve conflicts for PR #374 against latest master

gitea_mcp_server.py

@@ -500,6 +500,7 @@ import review_proofs  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 import author_mutation_worktree  # noqa: E402
+import issue_claim_heartbeat  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 
 
@@ -5170,6 +5171,41 @@ def gitea_audit_config() -> dict:
     return report
 
 
+def _post_structured_issue_comment(
+    *,
+    issue_number: int,
+    body: str,
+    remote: str,
+    host: str | None,
+    org: str | None,
+    repo: str | None,
+    audit_op: str = "create_issue_comment",
+) -> dict:
+    """Post an issue-thread comment after permission gates already passed."""
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
+    with _audited(
+        audit_op,
+        host=h,
+        remote=remote,
+        org=o,
+        repo=r,
+        issue_number=issue_number,
+        request_metadata={"body_chars": len(body)},
+    ):
+        data = api_request("POST", api, auth, {"body": body})
+    result = {
+        "success": True,
+        "performed": True,
+        "comment_id": data["id"],
+        "issue_number": issue_number,
+    }
+    if _reveal_endpoints():
+        result["url"] = data.get("html_url")
+    return result
+
+
 @mcp.tool()
 def gitea_mark_issue(
     issue_number: int,
@@ -5179,6 +5215,8 @@ def gitea_mark_issue(
     org: str | None = None,
     repo: str | None = None,
     worktree_path: str | None = None,
+    branch_name: str | None = None,
+    profile: str | None = None,
 ) -> dict:
     """Claim or release an issue via the status:in-progress label.
 
@@ -5228,7 +5266,31 @@ def gitea_mark_issue(
                       request_metadata={"op": "add", "label": "status:in-progress"}):
             api_request("POST", f"{base}/issues/{issue_number}/labels", auth,
                          {"labels": [label_id]})
-        return {"success": True, "message": f"Issue #{issue_number} claimed."}
+        active_profile = (profile or get_profile().get("profile_name") or "unknown")
+        branch = (branch_name or "pending").strip() or "pending"
+        heartbeat_body = issue_claim_heartbeat.format_heartbeat_body(
+            kind="claim",
+            issue_number=issue_number,
+            branch=branch,
+            phase="claimed",
+            profile=active_profile,
+            next_action="create worktree and begin implementation",
+        )
+        heartbeat = _post_structured_issue_comment(
+            issue_number=issue_number,
+            body=heartbeat_body,
+            remote=remote,
+            host=host,
+            org=org,
+            repo=repo,
+            audit_op="claim_heartbeat",
+        )
+        return {
+            "success": True,
+            "message": f"Issue #{issue_number} claimed.",
+            "heartbeat_posted": heartbeat.get("success", False),
+            "heartbeat_comment_id": heartbeat.get("comment_id"),
+        }
     else:
         with _audited("unlabel_issue", host=h, remote=remote, org=o, repo=r,
                       issue_number=issue_number,
@@ -5238,6 +5300,217 @@ def gitea_mark_issue(
         return {"success": True, "message": f"Issue #{issue_number} released."}
 
 
+@mcp.tool()
+def gitea_post_heartbeat(
+    issue_number: int,
+    branch: str,
+    phase: str,
+    pr: str = "none",
+    next_action: str = "none",
+    blocker: str = "none",
+    profile: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Post a structured progress heartbeat on a claimed issue (#268)."""
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission("post_heartbeat"))
+    if blocked:
+        return blocked
+    verify_preflight_purity(remote)
+    active_profile = profile or get_profile().get("profile_name")
+    body = issue_claim_heartbeat.format_heartbeat_body(
+        kind="progress",
+        issue_number=issue_number,
+        branch=branch,
+        phase=phase,
+        profile=active_profile,
+        pr=pr,
+        next_action=next_action,
+        blocker=blocker,
+    )
+    return _post_structured_issue_comment(
+        issue_number=issue_number,
+        body=body,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+        audit_op="progress_heartbeat",
+    )
+
+
+@mcp.tool()
+def gitea_reconcile_issue_claims(
+    state: str = "open",
+    stale_after_hours: int = 24,
+    heartbeat_lease_minutes: int = 30,
+    limit: int = 100,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only inventory of issue claims and heartbeat lease status (#268)."""
+    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)
+    base = repo_api_url(h, o, r)
+    issues = api_get_all(f"{base}/issues?state={state}&type=issues", auth, limit=limit)
+    open_prs = api_get_all(f"{base}/pulls?state=open", auth)
+    branches = api_get_all(f"{base}/branches", auth, limit=limit)
+    branch_names = [b.get("name") for b in branches if b.get("name")]
+
+    comments_by_issue: dict[int, list[dict]] = {}
+    for issue in issues:
+        if not issue_claim_heartbeat.issue_has_in_progress_label(issue):
+            continue
+        number = int(issue["number"])
+        api = f"{base}/issues/{number}/comments"
+        comments_by_issue[number] = api_request("GET", api, auth) or []
+
+    reclaim_after_minutes = max(heartbeat_lease_minutes * 2, 60)
+    inventory = issue_claim_heartbeat.build_claim_inventory(
+        issues=issues,
+        comments_by_issue=comments_by_issue,
+        open_prs=open_prs,
+        branch_names=branch_names,
+        heartbeat_lease_minutes=heartbeat_lease_minutes,
+        reclaim_after_minutes=reclaim_after_minutes,
+    )
+    inventory["cleanup_plan"] = issue_claim_heartbeat.build_cleanup_plan(inventory)
+    inventory["success"] = True
+    inventory["performed"] = False
+    return inventory
+
+
+@mcp.tool()
+def gitea_cleanup_stale_claims(
+    dry_run: bool = True,
+    execute_confirmed: bool = False,
+    heartbeat_lease_minutes: int = 30,
+    limit: int = 100,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Propose or execute stale-claim cleanup for phantom/reclaimable issues (#268)."""
+    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"),
+        }
+
+    inventory = gitea_reconcile_issue_claims(
+        heartbeat_lease_minutes=heartbeat_lease_minutes,
+        limit=limit,
+        remote=remote,
+        host=host,
+        org=org,
+        repo=repo,
+    )
+    if not inventory.get("success"):
+        return inventory
+
+    plan = [
+        entry
+        for entry in inventory.get("cleanup_plan") or []
+        if entry.get("action") == "remove_status_in_progress_and_comment"
+    ]
+    report = {
+        "success": True,
+        "dry_run": dry_run,
+        "performed": False,
+        "planned_actions": plan,
+        "inventory_counts": inventory.get("counts"),
+    }
+    if dry_run:
+        return report
+
+    if not execute_confirmed:
+        raise ValueError(
+            "execute_confirmed must be True when dry_run=False (fail closed)"
+        )
+
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission("cleanup_stale_claims"))
+    if blocked:
+        return blocked
+    verify_preflight_purity(remote)
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    base = repo_api_url(h, o, r)
+    labels = api_request("GET", f"{base}/labels?limit=100", auth)
+    label_id = next(
+        (lb["id"] for lb in labels if lb.get("name") == "status:in-progress"),
+        None,
+    )
+    if label_id is None:
+        raise RuntimeError("Label 'status:in-progress' not found")
+
+    actions: list[dict] = []
+    active_profile = get_profile().get("profile_name")
+    for entry in plan:
+        issue_number = int(entry["issue_number"])
+        with _audited(
+            "unlabel_issue",
+            host=h,
+            remote=remote,
+            org=o,
+            repo=r,
+            issue_number=issue_number,
+            request_metadata={"op": "cleanup_stale_claim", "label": "status:in-progress"},
+        ):
+            api_request(
+                "DELETE",
+                f"{base}/issues/{issue_number}/labels/{label_id}",
+                auth,
+            )
+        cleanup_body = issue_claim_heartbeat.format_heartbeat_body(
+            kind="cleanup",
+            issue_number=issue_number,
+            branch="none",
+            phase="stale-claim-cleanup",
+            profile=active_profile,
+            next_action="issue reclaimable by queue",
+            blocker=entry.get("status") or "stale",
+        )
+        comment = _post_structured_issue_comment(
+            issue_number=issue_number,
+            body=cleanup_body,
+            remote=remote,
+            host=host,
+            org=org,
+            repo=repo,
+            audit_op="cleanup_heartbeat",
+        )
+        actions.append(
+            {
+                "issue_number": issue_number,
+                "label_removed": True,
+                "cleanup_comment_id": comment.get("comment_id"),
+            }
+        )
+
+    report["performed"] = True
+    report["actions"] = actions
+    return report
+
+
 @mcp.tool()
 def gitea_list_labels(
     remote: str = "dadeschools",

issue_claim_heartbeat.py

@@ -0,0 +1,295 @@
+"""Issue claim heartbeat leases and stale-claim reconciliation (#268).
+
+Structured issue-thread comments prove live ownership beyond the
+``status:in-progress`` label alone. Queue inventory can classify claims as
+active, stale, reclaimable, PR-backed, or phantom.
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+MARKER = "<!-- gitea-issue-claim-heartbeat:v1 -->"
+IN_PROGRESS_LABEL = "status:in-progress"
+
+_KIND_CLAIM = "claim"
+_KIND_PROGRESS = "progress"
+_KIND_CLEANUP = "cleanup"
+
+_FIELD_RE = re.compile(
+    r"^\s*-\s*([a-z_]+)\s*:\s*(.+?)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_ISSUE_REF_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
+
+
+def _parse_timestamp(value: str | None) -> datetime | None:
+    if not value:
+        return None
+    text = value.strip()
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    try:
+        parsed = datetime.fromisoformat(text)
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
+def format_heartbeat_body(
+    *,
+    kind: str,
+    issue_number: int,
+    branch: str,
+    phase: str,
+    profile: str | None = None,
+    pr: str = "none",
+    next_action: str = "none",
+    blocker: str = "none",
+) -> str:
+    """Return a structured, machine-parseable issue comment body."""
+    profile_value = (profile or "unknown").strip() or "unknown"
+    lines = [
+        MARKER,
+        "**Issue claim heartbeat**",
+        f"- kind: {kind}",
+        f"- issue: #{issue_number}",
+        f"- branch: {branch}",
+        f"- phase: {phase}",
+        f"- profile: {profile_value}",
+        f"- pr: {pr}",
+        f"- blocker: {blocker}",
+        f"- next_action: {next_action}",
+    ]
+    return "\n".join(lines)
+
+
+def parse_heartbeat_comment(body: str) -> dict[str, Any] | None:
+    """Parse one structured heartbeat comment, or None when not a heartbeat."""
+    text = body or ""
+    if MARKER not in text:
+        return None
+    fields: dict[str, str] = {}
+    for match in _FIELD_RE.finditer(text):
+        fields[match.group(1).strip().lower()] = match.group(2).strip()
+    if not fields:
+        return None
+    issue_raw = fields.get("issue", "")
+    issue_digits = re.sub(r"[^\d]", "", issue_raw)
+    issue_number = int(issue_digits) if issue_digits.isdigit() else None
+    return {
+        "kind": fields.get("kind"),
+        "issue_number": issue_number,
+        "branch": fields.get("branch"),
+        "phase": fields.get("phase"),
+        "profile": fields.get("profile"),
+        "pr": fields.get("pr"),
+        "blocker": fields.get("blocker"),
+        "next_action": fields.get("next_action"),
+        "raw_fields": fields,
+    }
+
+
+def extract_issue_heartbeats(
+    comments: list[dict],
+    *,
+    issue_number: int | None = None,
+) -> list[dict]:
+    """Return parsed heartbeats newest-last, optionally filtered to *issue_number*."""
+    heartbeats: list[dict] = []
+    for comment in comments or []:
+        parsed = parse_heartbeat_comment(comment.get("body") or "")
+        if not parsed:
+            continue
+        if issue_number is not None and parsed.get("issue_number") != issue_number:
+            continue
+        heartbeats.append(
+            {
+                **parsed,
+                "comment_id": comment.get("id"),
+                "author": (comment.get("user") or {}).get("login")
+                or comment.get("author"),
+                "created_at": comment.get("created_at"),
+                "updated_at": comment.get("updated_at"),
+            }
+        )
+    return heartbeats
+
+
+def issue_has_in_progress_label(issue: dict) -> bool:
+    labels = issue.get("labels") or []
+    for label in labels:
+        name = label if isinstance(label, str) else label.get("name")
+        if name == IN_PROGRESS_LABEL:
+            return True
+    return False
+
+
+def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None:
+    pattern = f"issue-{issue_number}"
+    closes = f"closes #{issue_number}"
+    fixes = f"fixes #{issue_number}"
+    for pr in open_prs or []:
+        head = (pr.get("head") or {}).get("ref") or ""
+        text = f"{pr.get('title', '')} {pr.get('body', '')}".lower()
+        if pattern in head.lower():
+            return pr
+        if closes in text or fixes in text:
+            return pr
+    return None
+
+
+def _matching_branch_names(issue_number: int, branch_names: list[str]) -> list[str]:
+    pattern = f"issue-{issue_number}"
+    return [name for name in branch_names if pattern in (name or "").lower()]
+
+
+def classify_issue_claim(
+    *,
+    issue: dict,
+    comments: list[dict],
+    open_prs: list[dict] | None = None,
+    branch_names: list[str] | None = None,
+    now: datetime | None = None,
+    heartbeat_lease_minutes: int = 30,
+    reclaim_after_minutes: int = 60,
+) -> dict[str, Any]:
+    """Classify one issue's claim state from labels, heartbeats, PRs, and branches."""
+    issue_number = int(issue.get("number") or 0)
+    open_prs = open_prs or []
+    branch_names = branch_names or []
+    current = now or datetime.now(timezone.utc)
+
+    has_label = issue_has_in_progress_label(issue)
+    heartbeats = extract_issue_heartbeats(comments, issue_number=issue_number)
+    latest = heartbeats[-1] if heartbeats else None
+    latest_at = _parse_timestamp(
+        (latest or {}).get("updated_at") or (latest or {}).get("created_at")
+    )
+    age_minutes = None
+    if latest_at:
+        age_minutes = int((current - latest_at).total_seconds() // 60)
+
+    linked_pr = _linked_open_pr(issue_number, open_prs)
+    matching_branches = _matching_branch_names(issue_number, branch_names)
+
+    if not has_label:
+        status = "not_claimed"
+        reasons = ["issue lacks status:in-progress label"]
+    elif linked_pr:
+        status = "awaiting_review"
+        reasons = [f"open PR #{linked_pr.get('number')} covers this issue"]
+    elif not heartbeats:
+        status = "phantom"
+        reasons = [
+            "status:in-progress label present without structured claim heartbeat"
+        ]
+    elif latest_at is None:
+        status = "phantom"
+        reasons = ["heartbeat comments present but timestamps could not be parsed"]
+    elif age_minutes is not None and age_minutes <= heartbeat_lease_minutes:
+        status = "active"
+        reasons = [f"heartbeat age {age_minutes}m within {heartbeat_lease_minutes}m lease"]
+    elif matching_branches and age_minutes is not None and age_minutes <= reclaim_after_minutes:
+        status = "active"
+        reasons = [
+            f"matching branch(es) {', '.join(matching_branches)} with heartbeat "
+            f"age {age_minutes}m"
+        ]
+    elif age_minutes is not None and age_minutes > reclaim_after_minutes and not matching_branches:
+        status = "reclaimable"
+        reasons = [
+            f"no heartbeat within {reclaim_after_minutes}m and no matching branch"
+        ]
+    elif age_minutes is not None and age_minutes > heartbeat_lease_minutes:
+        status = "stale"
+        reasons = [
+            f"heartbeat age {age_minutes}m exceeds {heartbeat_lease_minutes}m lease"
+        ]
+    else:
+        status = "active"
+        reasons = ["claim has structured heartbeat proof"]
+
+    return {
+        "issue_number": issue_number,
+        "title": issue.get("title"),
+        "status": status,
+        "has_in_progress_label": has_label,
+        "heartbeat_count": len(heartbeats),
+        "latest_heartbeat": latest,
+        "heartbeat_age_minutes": age_minutes,
+        "linked_open_pr": linked_pr.get("number") if linked_pr else None,
+        "matching_branches": matching_branches,
+        "reasons": reasons,
+        "reclaimable": status == "reclaimable",
+        "stale": status in {"stale", "phantom", "reclaimable"},
+    }
+
+
+def build_claim_inventory(
+    *,
+    issues: list[dict],
+    comments_by_issue: dict[int, list[dict]],
+    open_prs: list[dict],
+    branch_names: list[str],
+    now: datetime | None = None,
+    heartbeat_lease_minutes: int = 30,
+    reclaim_after_minutes: int = 60,
+) -> dict[str, Any]:
+    """Build a queue inventory report for in-progress issue claims."""
+    entries: list[dict] = []
+    for issue in issues or []:
+        if not issue_has_in_progress_label(issue):
+            continue
+        number = int(issue["number"])
+        entry = classify_issue_claim(
+            issue=issue,
+            comments=comments_by_issue.get(number, []),
+            open_prs=open_prs,
+            branch_names=branch_names,
+            now=now,
+            heartbeat_lease_minutes=heartbeat_lease_minutes,
+            reclaim_after_minutes=reclaim_after_minutes,
+        )
+        entries.append(entry)
+
+    counts: dict[str, int] = {}
+    for entry in entries:
+        counts[entry["status"]] = counts.get(entry["status"], 0) + 1
+
+    return {
+        "entries": entries,
+        "counts": counts,
+        "heartbeat_lease_minutes": heartbeat_lease_minutes,
+        "reclaim_after_minutes": reclaim_after_minutes,
+        "in_progress_total": len(entries),
+    }
+
+
+def build_cleanup_plan(inventory: dict[str, Any]) -> list[dict]:
+    """Return stale/reclaimable/phantom claims that may be cleaned up."""
+    plan: list[dict] = []
+    for entry in inventory.get("entries") or []:
+        if entry.get("status") in {"reclaimable", "phantom"}:
+            plan.append(
+                {
+                    "issue_number": entry["issue_number"],
+                    "status": entry["status"],
+                    "action": "remove_status_in_progress_and_comment",
+                    "reasons": list(entry.get("reasons") or []),
+                }
+            )
+        elif entry.get("status") == "stale" and not entry.get("linked_open_pr"):
+            plan.append(
+                {
+                    "issue_number": entry["issue_number"],
+                    "status": entry["status"],
+                    "action": "report_only",
+                    "reasons": list(entry.get("reasons") or []),
+                }
+            )
+    return plan

review_proofs.py

@@ -1052,6 +1052,15 @@ def _prs_from_list_response(list_prs_response: list | dict | None) -> list | Non
     return None
 
 
+# ── Linked-issue consistency (Issue #314) ────────────────────────────────────
+#
+# Stale linked-issue text from a previous PR must not leak into a final
+# report: the "Linked issue status" field must match the issue(s) the
+# selected PR actually links, verified live this session, and no other
+# "Issue #N" mention may appear unless it is a verified linked issue.
+
+_ISSUE_MENTION_RE = re.compile(r"\bissue\s+#(\d+)", re.IGNORECASE)
+
 # ── Workspace-vs-worktree mutation consistency (Issue #313) ──────────────────
 #
 # A worktree reset/checkout/clean is a workspace mutation even when the
@@ -1092,6 +1101,68 @@ def _report_labeled_fields(report_text):
     return fields
 
 
+def _issue_numbers_mentioned(text):
+    """Return the set of ints referenced as 'Issue #N' in *text*."""
+    return {int(n) for n in _ISSUE_MENTION_RE.findall(text or "")}
+
+
+def assess_linked_issue_consistency(report_text, *, selected_pr=None,
+                                    linked_issues=None):
+    """#314: linked-issue status must match the selected PR, live-verified.
+
+    *linked_issues* is the list of issue numbers the selected PR was
+    live-verified to link this session, or None when no live verification
+    happened. Without live proof the report may only say the status was
+    not verified; with proof, the ``Linked issue status`` field must name
+    every linked issue and no stale issue number may appear anywhere in
+    the report.
+
+    Returns {'complete', 'downgraded', 'reasons'}; fails closed.
+    """
+    fields = _report_labeled_fields(report_text)
+    status_value = fields.get("linked issue status")
+
+    reasons = []
+
+    if status_value is None:
+        reasons.append(
+            "final report missing 'Linked issue status' field for the "
+            "selected PR"
+        )
+    elif linked_issues is None:
+        if not status_value.strip().lower().startswith("not verified"):
+            reasons.append(
+                "linked issue status claimed without live proof; report "
+                "'Linked issue status: not verified in this session' or "
+                "verify the linked issue live"
+            )
+    else:
+        allowed = set(linked_issues)
+        status_mentions = _issue_numbers_mentioned(status_value)
+        for missing in sorted(allowed - status_mentions):
+            reasons.append(
+                f"linked issue #{missing} of selected PR "
+                f"#{selected_pr} not reported in 'Linked issue status'"
+            )
+        for stale in sorted(status_mentions - allowed):
+            reasons.append(
+                f"'Linked issue status' names issue #{stale}, which is "
+                f"not a live-verified linked issue of selected PR "
+                f"#{selected_pr}"
+            )
+        for stale in sorted(_issue_numbers_mentioned(report_text) - allowed):
+            reasons.append(
+                f"stale issue #{stale} mentioned in final report but not "
+                f"live-verified as linked to selected PR #{selected_pr}"
+            )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
 def _field_claims_none(fields, label):
     """True when *label* is present and its value is empty or 'none...'."""
     value = fields.get(label)
@@ -1171,6 +1242,102 @@ def assess_workspace_mutation_consistency(report_text, observed_commands=None):
     }
 
 
+# ── Already-landed report wording (Issue #298) ───────────────────────────────
+#
+# An already-landed PR is reconciliation-only: it must never be called
+# eligible, its head SHA must never be called reviewed, and the legacy
+# eligible/reviewed/scratch-worktree handoff fields must not appear.
+# Reviewed-head claims in any report require validation + diff review to
+# have actually passed.
+
+ALREADY_LANDED_FORBIDDEN_PHRASES = (
+    "oldest eligible pr",
+    "next eligible pr",
+    "pinned reviewed head",
+    "scratch worktree used",
+)
+
+ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED"
+
+
+def assess_already_landed_report_wording(report_text, *, already_landed,
+                                         review_validated=False):
+    """#298: reject eligible/reviewed wording for already-landed PRs.
+
+    *already_landed* states whether the already-landed gate fired for the
+    selected PR. *review_validated* states whether validation and diff
+    review actually passed this session; only then may a reviewed head
+    SHA be populated in a non-already-landed report.
+
+    Returns {'complete', 'downgraded', 'reasons'}; fails closed.
+    """
+    text_lower = (report_text or "").lower()
+    fields = _report_labeled_fields(report_text)
+    reasons = []
+
+    reviewed_head = fields.get("reviewed head sha")
+    reviewed_head_populated = bool(
+        reviewed_head and not reviewed_head.strip().lower().startswith("none")
+    )
+    pinned_head_present = "pinned reviewed head" in fields
+
+    if already_landed:
+        for phrase in ALREADY_LANDED_FORBIDDEN_PHRASES:
+            if phrase in text_lower:
+                reasons.append(
+                    f"already-landed report must not use '{phrase}'; the PR "
+                    "is reconciliation-only, not review/merge eligible"
+                )
+
+        eligibility = fields.get("eligibility class", "")
+        if ALREADY_LANDED_ELIGIBILITY_CLASS not in eligibility.upper():
+            reasons.append(
+                "already-landed report missing 'Eligibility class: "
+                f"{ALREADY_LANDED_ELIGIBILITY_CLASS}'"
+            )
+
+        for required in ("oldest open pr requiring action",
+                         "candidate head sha"):
+            if required not in fields:
+                reasons.append(
+                    f"already-landed report missing '{required}' field"
+                )
+
+        if "reviewed head sha" not in fields:
+            reasons.append(
+                "already-landed report must state 'Reviewed head SHA: none'"
+            )
+        elif reviewed_head_populated:
+            reasons.append(
+                "already-landed report must not claim a reviewed head SHA; "
+                "no validation or diff review passed for this PR"
+            )
+
+        worktree_used = fields.get("review worktree used", "")
+        if worktree_used.strip().lower() not in ("false", "no"):
+            reasons.append(
+                "already-landed report must state 'Review worktree used: "
+                "false'"
+            )
+    elif not review_validated:
+        if pinned_head_present:
+            reasons.append(
+                "'Pinned reviewed head' populated before validation and "
+                "diff review passed"
+            )
+        if reviewed_head_populated:
+            reasons.append(
+                "reviewed head SHA populated before validation and diff "
+                "review passed"
+            )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
 def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
                          justification=None):
     """Assess reviewer/author role separation for blind queue workflows.
@@ -1351,6 +1518,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
                        report_text=None, review_decision_lock=None,
                        controller_handoff=None, capability_proof=None,
                        sweep_proof=None, worktree_proof=None,
+                       validation_environment=None,
                        queue_ordering=None,
                        baseline_validation=None, project_root=None):
     """Required behavior 6 + acceptance criteria: one report, distinct proofs.
@@ -1379,6 +1547,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         )
 
     empty_queue_report = assess_empty_queue_report(report_text)
+    if validation_environment is None and report_text:
+        validation_env_proof = assess_validation_environment_proof(
+            report_text=report_text,
+        )
+    elif validation_environment is not None:
+        validation_env_proof = assess_validation_environment_proof(
+            report_text=report_text,
+            validation_environment=validation_environment,
+        )
+    else:
+        validation_env_proof = {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "violations": [],
+        }
     if queue_ordering is None and report_text:
         queue_ordering_proof = assess_queue_ordering_proof(
             report_text=report_text,
@@ -1546,6 +1730,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "empty-queue report missing or failed trust-gate proof (#198)"
         )
         downgrade_reasons.extend(empty_queue_report.get("reasons", []))
+    if not validation_env_proof.get("proven"):
+        downgrade_reasons.append(
+            "validation environment proof missing or failed (#311)"
+        )
+        downgrade_reasons.extend(validation_env_proof.get("reasons", []))
     if not queue_ordering_proof.get("proven"):
         downgrade_reasons.append(
             "queue ordering proof missing or failed (#321)"
@@ -1577,11 +1766,13 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         # #179: no merge without a proven final live-state recheck.
         and live_state_proven
         and worktree_proven
+        and validation_env_proof.get("proven")
         and queue_ordering_proof.get("proven")
         and baseline_validation.get("proven")
     )
 
     violations = []
+    violations.extend(validation_env_proof.get("violations", []))
     violations.extend(queue_ordering_proof.get("violations", []))
     violations.extend(baseline_validation.get("violations", []))
     if merge_performed and not merge_allowed:
@@ -1635,6 +1826,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             else True
         ),
         "empty_queue_trust_gate_status": empty_queue_report.get("status"),
+        "validation_environment_proven": bool(
+            validation_env_proof.get("proven")
+        ),
+        "validation_environment_violations": list(
+            validation_env_proof.get("violations") or []
+        ),
         "queue_ordering_proven": bool(queue_ordering_proof.get("proven")),
         "queue_ordering_policy": queue_ordering_proof.get("policy"),
         "queue_ordering_violations": list(
@@ -3251,6 +3448,192 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict:
     }
 
 
+# ---------------------------------------------------------------------------
+# Validation environment proof (#311)
+# ---------------------------------------------------------------------------
+
+_VALIDATION_CLAIM_RE = re.compile(
+    r"(?:validation\s+command|pytest\s+(?:executed|passed|failed)|"
+    r"\d+\s+passed|\d+\s+failed|tests?\s+passed)",
+    re.IGNORECASE,
+)
+
+_VALIDATION_CMD_LINE_RE = re.compile(
+    r"validation\s+command\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+
+_VALIDATION_CWD_LINE_RE = re.compile(
+    r"working\s+directory\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+
+_WHICH_PYTEST_LINE_RE = re.compile(
+    r"which\s+pytest\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+
+_PYTEST_VERSION_LINE_RE = re.compile(
+    r"pytest\s+--version\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+
+_BARE_PYTEST_CMD_RE = re.compile(
+    r"^(?:python\s+-m\s+)?pytest(?:\s|$)",
+    re.IGNORECASE,
+)
+
+_VAGUE_PYTEST_CLAIM_RE = re.compile(
+    r"pytest\s+(?:executed|passed|failed)\b",
+    re.IGNORECASE,
+)
+
+
+def _parse_validation_command(report_text: str) -> str:
+    for line in (report_text or "").splitlines():
+        match = _VALIDATION_CMD_LINE_RE.search(line)
+        if match:
+            return match.group(1).strip().rstrip(".")
+    return ""
+
+
+def _parse_validation_cwd(report_text: str) -> str:
+    for line in (report_text or "").splitlines():
+        match = _VALIDATION_CWD_LINE_RE.search(line)
+        if match:
+            return match.group(1).strip().rstrip(",.;")
+    return ""
+
+
+def _is_bare_pytest_command(command: str) -> bool:
+    text = (command or "").strip()
+    if not text:
+        return False
+    if "/" in text or "\\" in text:
+        return False
+    return bool(_BARE_PYTEST_CMD_RE.match(text))
+
+
+def assess_validation_environment_proof(
+    *,
+    report_text: str | None = None,
+    validation_environment: dict | None = None,
+) -> dict:
+    """#311: reviewer reports must include exact validation command/environment."""
+    text = report_text or ""
+    lower = text.lower()
+    env = validation_environment or {}
+    reasons: list[str] = []
+    violations: list[str] = []
+
+    claims_validation = bool(_VALIDATION_CLAIM_RE.search(text))
+    if not claims_validation and not env.get("command"):
+        return {
+            "proven": True,
+            "block": False,
+            "claims_validation": False,
+            "reasons": [],
+            "violations": [],
+            "safe_next_action": "proceed",
+        }
+
+    command = (env.get("command") or _parse_validation_command(text)).strip()
+    working_directory = (
+        env.get("working_directory") or env.get("cwd")
+        or _parse_validation_cwd(text)
+    ).strip()
+
+    if not command:
+        reasons.append(
+            "validation result claimed without exact validation command (#311)"
+        )
+    elif _VAGUE_PYTEST_CLAIM_RE.search(text) and command == "":
+        violations.append(
+            "vague pytest summary without exact validation command (#311)"
+        )
+
+    if not working_directory:
+        reasons.append(
+            "validation result claimed without working directory (#311)"
+        )
+
+    has_result = any(
+        token in lower
+        for token in ("passed", "failed", "skipped", "error")
+    ) or any(
+        env.get(key) is not None
+        for key in ("passed", "failed", "skipped", "result")
+    )
+    if not has_result:
+        reasons.append(
+            "validation result claimed without pass/fail result evidence (#311)"
+        )
+
+    if command and _is_bare_pytest_command(command):
+        which_pytest = (
+            (env.get("which_pytest") or env.get("executable_path") or "")
+            .strip()
+        )
+        if not which_pytest:
+            for line in text.splitlines():
+                match = _WHICH_PYTEST_LINE_RE.search(line)
+                if match:
+                    which_pytest = match.group(1).strip()
+                    break
+        if not which_pytest:
+            reasons.append(
+                "bare pytest command requires which-pytest/executable proof (#311)"
+            )
+
+        pytest_version = (env.get("pytest_version") or "").strip()
+        if not pytest_version:
+            for line in text.splitlines():
+                match = _PYTEST_VERSION_LINE_RE.search(line)
+                if match:
+                    pytest_version = match.group(1).strip()
+                    break
+        if not pytest_version:
+            reasons.append(
+                "bare pytest command requires pytest --version proof (#311)"
+            )
+
+        venv_resolved = env.get("venv_resolved")
+        if venv_resolved is None:
+            venv_resolved = any(
+                marker in lower
+                for marker in (
+                    "project venv",
+                    "venv/bin/pytest",
+                    "resolves to the project venv",
+                    "venv resolved: yes",
+                )
+            )
+        if not venv_resolved and which_pytest and "venv" not in which_pytest:
+            reasons.append(
+                "bare pytest command requires proof whether executable "
+                "resolves to the project venv (#311)"
+            )
+
+    proven = not reasons and not violations
+    return {
+        "proven": proven,
+        "block": bool(violations),
+        "claims_validation": True,
+        "command": command or None,
+        "working_directory": working_directory or None,
+        "reasons": reasons,
+        "violations": violations,
+        "safe_next_action": (
+            "report exact validation command, working directory, result, and "
+            "for bare pytest also which-pytest, pytest --version, and venv proof"
+            if not proven
+            else "proceed"
+        ),
+    }
+
+
+
+
 # ---------------------------------------------------------------------------
 # Reviewer queue ordering proof (#321)
 # ---------------------------------------------------------------------------
@@ -3581,6 +3964,279 @@ def assess_reviewer_baseline_validation_proof(
     }
 
 
+# ---------------------------------------------------------------------------
+# Already-landed review gate (#292)
+# ---------------------------------------------------------------------------
+
+ALREADY_LANDED_STATE = "ALREADY_LANDED_RECONCILE_REQUIRED"
+
+ALREADY_LANDED_HANDOFF_FIELDS = (
+    "PR number/title",
+    "candidate head SHA",
+    "target branch",
+    "target branch SHA",
+    "ancestor proof",
+    "linked issue status",
+    "recommended reconciliation action",
+    "capability proof for close/comment mutations",
+)
+
+_FORBIDDEN_LANDED_STATES_RE = re.compile(
+    r"review decision\s*:\s*approved?\b|"
+    r"merge result\s*:\s*merged\b|"
+    r"\bready[_ ]to[_ ]merge\b|"
+    r"\bmerge[d]?\s*:\s*success\b",
+    re.IGNORECASE,
+)
+
+
+def assess_already_landed_review_gate(
+    *,
+    pr_number: int | None,
+    candidate_head_sha: str | None,
+    target_branch: str | None,
+    target_branch_sha: str | None,
+    head_is_ancestor_of_target: bool | None,
+    live_head_sha: str | None = None,
+    real_blocker: bool = False,
+    mergeable: bool | None = None,
+) -> dict:
+    """#292: hard gate before any review mutation on an already-landed PR.
+
+    Runs after PR selection and head pinning, before approval, request-
+    changes, or the merge API. *head_is_ancestor_of_target* is the result
+    of an ancestry check of *candidate_head_sha* against the freshly
+    fetched target branch at *target_branch_sha* (e.g. ``git merge-base
+    --is-ancestor``). ``None`` means the check was not run — the gate then
+    fails closed.
+
+    *live_head_sha*, when provided, must equal *candidate_head_sha*; a
+    changed head invalidates the pinned ancestry result until re-checked.
+
+    *mergeable* is accepted for caller convenience but never consulted:
+    conflict/mergeability handling is a separate gate.
+    """
+    del mergeable  # ancestry gate only; mergeability is a separate gate
+    reasons: list[str] = []
+
+    if not isinstance(pr_number, int) or pr_number <= 0:
+        reasons.append("PR number missing or invalid (#292)")
+    if not _FULL_SHA.match((candidate_head_sha or "").strip()):
+        reasons.append(
+            "candidate head SHA is not a full 40-hex commit SHA (#292)"
+        )
+    if not (target_branch or "").strip():
+        reasons.append("target branch missing (#292)")
+    if not _FULL_SHA.match((target_branch_sha or "").strip()):
+        reasons.append(
+            "target branch SHA is not a full 40-hex commit SHA; fetch the "
+            "target branch and record its SHA before the ancestry check "
+            "(#292)"
+        )
+    if live_head_sha is not None and candidate_head_sha and (
+        live_head_sha.strip() != candidate_head_sha.strip()
+    ):
+        reasons.append(
+            "PR head changed since the ancestry check was pinned; re-fetch "
+            "and re-run the already-landed gate (#292)"
+        )
+    if head_is_ancestor_of_target is None:
+        reasons.append(
+            "ancestry of the PR head against the target branch was not "
+            "checked; the already-landed gate must run before any review "
+            "mutation (#292)"
+        )
+
+    if reasons:
+        return {
+            "state": "GATE_NOT_PROVEN",
+            "approve_allowed": False,
+            "request_changes_allowed": False,
+            "merge_api_allowed": False,
+            "reconciliation_handoff_required": False,
+            "required_handoff_fields": (),
+            "reasons": reasons,
+            "safe_next_action": (
+                "fetch the target branch, pin the live PR head, run the "
+                "ancestry check, then re-run this gate"
+            ),
+        }
+
+    if head_is_ancestor_of_target:
+        return {
+            "state": ALREADY_LANDED_STATE,
+            "approve_allowed": False,
+            "request_changes_allowed": bool(real_blocker),
+            "merge_api_allowed": False,
+            "reconciliation_handoff_required": True,
+            "required_handoff_fields": ALREADY_LANDED_HANDOFF_FIELDS,
+            "reasons": [
+                f"PR #{pr_number} head {candidate_head_sha} is already an "
+                f"ancestor of {target_branch} @ {target_branch_sha}; the PR "
+                "is reconciliation-only (#292)"
+            ],
+            "safe_next_action": (
+                "stop before review mutation and emit an "
+                f"{ALREADY_LANDED_STATE} reconciliation handoff; any PR/issue "
+                "close or comment requires exact capability proof"
+            ),
+        }
+
+    return {
+        "state": "NORMAL_REVIEW_CANDIDATE",
+        "approve_allowed": True,
+        "request_changes_allowed": True,
+        "merge_api_allowed": True,
+        "reconciliation_handoff_required": False,
+        "required_handoff_fields": (),
+        "reasons": [],
+        "safe_next_action": "proceed with the normal review workflow gates",
+    }
+
+
+def assess_already_landed_report_state(
+    report_text: str | None,
+    *,
+    gate_fired: bool,
+) -> dict:
+    """#292: reject APPROVED/MERGED/READY_TO_MERGE after the gate fired.
+
+    *gate_fired* comes from the session's own gate result, so a report
+    that omits the already-landed markers entirely still fails closed —
+    unlike the text-only #327 wording rules this does not depend on the
+    report admitting the PR was already landed.
+    """
+    if not gate_fired:
+        return {"complete": True, "block": False, "reasons": []}
+
+    text = report_text or ""
+    reasons: list[str] = []
+
+    if _FORBIDDEN_LANDED_STATES_RE.search(text):
+        reasons.append(
+            "report claims APPROVED/MERGED/READY_TO_MERGE although the "
+            "already-landed gate fired (#292)"
+        )
+    if ALREADY_LANDED_STATE.lower() not in text.lower():
+        reasons.append(
+            f"report must state {ALREADY_LANDED_STATE} when the "
+            "already-landed gate fired (#292)"
+        )
+
+    return {
+        "complete": not reasons,
+        "block": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Full-suite failure approval gate (#323)
+# ---------------------------------------------------------------------------
+
+
+def assess_full_suite_failure_approval_gate(
+    *,
+    review_decision: str,
+    full_suite_passed: bool | None,
+    baseline_validation: dict | None = None,
+    baseline_proof: dict | None = None,
+    report_text: str | None = None,
+    project_root: str | None = None,
+) -> dict:
+    """#323: approving a PR whose full suite fails requires baseline proof.
+
+    Default action on a full-suite failure is REQUEST_CHANGES. Approval is
+    allowed only when the #325 baseline proof passes in full (branches/
+    worktree, pinned SHAs, matching failure signatures) and the extra #323
+    fields prove the failures are pre-existing: ``pr_suite_command``,
+    ``baseline_suite_command``, ``new_tests_passed``, and a non-empty
+    ``unrelated_explanation``.
+
+    *review_decision* is the decision the workflow intends to submit; only
+    ``approve`` can be blocked. ``block`` is True when an approval was
+    requested that the proofs do not allow.
+    """
+    decision = (review_decision or "").strip().lower().replace("-", "_")
+    wants_approval = decision == "approve"
+    reasons: list[str] = []
+
+    if full_suite_passed is True:
+        return {
+            "approve_allowed": True,
+            "block": False,
+            "default_action": "proceed",
+            "reasons": [],
+        }
+
+    default_action = "request_changes"
+
+    if full_suite_passed is None:
+        reasons.append(
+            "full-suite result not proven; approval requires a recorded "
+            "full-suite command and result (#323)"
+        )
+    else:
+        if baseline_validation is None:
+            baseline_validation = assess_reviewer_baseline_validation_proof(
+                baseline_proof=baseline_proof,
+                report_text=report_text,
+                project_root=project_root,
+            )
+        proof = baseline_proof or {}
+
+        if not proof:
+            reasons.append(
+                "full suite failed but no baseline proof was provided; "
+                "default action is REQUEST_CHANGES (#323)"
+            )
+        if not baseline_validation.get("proven"):
+            reasons.append(
+                "baseline validation proof missing or failed (#323/#325)"
+            )
+            reasons.extend(baseline_validation.get("reasons", []))
+
+        if proof:
+            worktree = (proof.get("worktree_path") or "").strip()
+            if not _path_under_branches(worktree, project_root):
+                reasons.append(
+                    "baseline comparison must run in a clean baseline "
+                    "worktree under branches/, not the main checkout (#323)"
+                )
+            if not (proof.get("pr_suite_command") or "").strip():
+                reasons.append(
+                    "PR full-suite command/result missing from baseline "
+                    "proof (#323)"
+                )
+            if not (proof.get("baseline_suite_command") or "").strip():
+                reasons.append(
+                    "baseline full-suite command/result missing from "
+                    "baseline proof (#323)"
+                )
+            if proof.get("new_tests_passed") is not True:
+                reasons.append(
+                    "proof that new/changed tests passed is missing (#323)"
+                )
+            if not (proof.get("unrelated_explanation") or "").strip():
+                reasons.append(
+                    "explanation why failures are unrelated to the PR is "
+                    "missing (#323)"
+                )
+            if proof.get("failure_signatures_match") is not True:
+                reasons.append(
+                    "PR and baseline failure signatures do not match; "
+                    "request changes (#323)"
+                )
+
+    approve_allowed = not reasons
+    return {
+        "approve_allowed": approve_allowed,
+        "block": wants_approval and not approve_allowed,
+        "default_action": "proceed" if approve_allowed else default_action,
+        "reasons": reasons,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Identity disclosure (#305)
 # ---------------------------------------------------------------------------
@@ -4059,6 +4715,20 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
     }
 
 
+def assess_merge_simulation_report(report_text, **kwargs):
+    """#317: classify local merge simulation as worktree/index mutations."""
+    from reviewer_merge_simulation import assess_merge_simulation_report as _assess
+
+    return _assess(report_text, **kwargs)
+
+
+def assess_validation_integrity_report(report_text, **kwargs):
+    """#316: separate official PR-head validation from diagnostic experiments."""
+    from reviewer_validation_integrity import assess_validation_integrity_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
@@ -4071,3 +4741,26 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs):
     from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess
 
     return _assess(report_text, **kwargs)
+
+
+def assess_validation_worktree_edit_report(report_text, **kwargs):
+    """#315: block edits in PR validation worktrees; require diagnostic separation."""
+    from reviewer_validation_worktree_edits import (
+        assess_validation_worktree_edit_report as _assess,
+    )
+
+    return _assess(report_text, **kwargs)
+
+
+def assess_mutation_categories_report(report_text, **kwargs):
+    """#319: require precise mutation categories in reviewer controller handoffs."""
+    from reviewer_mutation_categories import assess_mutation_categories_report as _assess
+
+    return _assess(report_text, **kwargs)
+
+
+def assess_worktree_ownership_report(report_text, **kwargs):
+    """#312: prove session-owned or safe-reuse worktree before reset/validation."""
+    from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess
+
+    return _assess(report_text, **kwargs)

reviewer_merge_simulation.py

@@ -0,0 +1,148 @@
+"""Merge-simulation worktree mutation verifier for reviewer reports (#317)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_WORKTREE_INDEX_FIELD_RE = re.compile(
+    r"^\s*worktree/index mutations\s*:\s*(.+?)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_READONLY_DIAG_RE = re.compile(
+    r"read[- ]only diagnostics\s*:\s*(.+)$",
+    re.IGNORECASE | re.MULTILINE,
+)
+_WORKTREE_PATH_RE = re.compile(
+    r"(?:worktree path|diagnostic worktree|simulation worktree)\s*:",
+    re.IGNORECASE,
+)
+_PRE_CLEAN_RE = re.compile(
+    r"(?:pre[- ]simulation|before simulation|clean before).*(?:clean|status)",
+    re.IGNORECASE,
+)
+_MERGE_RESULT_RE = re.compile(
+    r"(?:merge result|simulation result|conflict status|merge conflict)",
+    re.IGNORECASE,
+)
+_ABORT_RE = re.compile(
+    r"(?:merge --abort|abort/reset command|abort command|git merge --abort)",
+    re.IGNORECASE,
+)
+_POST_CLEAN_RE = re.compile(
+    r"(?:post[- ]abort|after abort|clean after).*(?:clean|status)",
+    re.IGNORECASE,
+)
+
+
+def _command_text(entry: Any) -> str:
+    if isinstance(entry, dict):
+        return str(entry.get("command") or "").strip()
+    return str(entry or "").strip()
+
+
+def _git_subcommand_line(command: str) -> str | None:
+    tokens = command.split()
+    if not tokens or tokens[0] != "git":
+        return None
+    return " ".join(tokens[1:])
+
+
+def merge_simulation_commands(command_log: list | None) -> list[str]:
+    """Return logged commands that perform local merge simulation (#317)."""
+    out: list[str] = []
+    for entry in command_log or []:
+        command = _command_text(entry)
+        rest = _git_subcommand_line(command)
+        if rest is None:
+            continue
+        lower = command.lower()
+        if rest.startswith("merge"):
+            if "--no-commit" in lower or (
+                "--no-ff" in lower and "merge" in lower and "--commit" not in lower
+            ):
+                out.append(command)
+            elif "--abort" in lower:
+                out.append(command)
+    return out
+
+
+def assess_merge_simulation_report(
+    report_text: str,
+    *,
+    command_log: list | None = None,
+) -> dict[str, Any]:
+    """Reject merge simulation classified as read-only; require worktree proof (#317)."""
+    text = report_text or ""
+    lower = text.lower()
+    reasons: list[str] = []
+    simulation_commands = merge_simulation_commands(command_log)
+    has_abort = any("--abort" in c.lower() for c in simulation_commands)
+    has_simulation = any(
+        "--no-commit" in c.lower() or (
+            "--no-ff" in c.lower() and "--abort" not in c.lower()
+        )
+        for c in simulation_commands
+    )
+
+    if not simulation_commands:
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "simulation_commands": [],
+            "safe_next_action": "proceed",
+        }
+
+    field = _WORKTREE_INDEX_FIELD_RE.search(text)
+    value = (field.group(1).strip().lower() if field else "") or ""
+    if not value or value in {"none", "n/a", "no"}:
+        reasons.append(
+            "merge simulation commands ran but report lacks "
+            "'Worktree/index mutations' entry"
+        )
+    elif "merge" not in value and "simulation" not in value:
+        if not any(cmd.lower() in lower for cmd in simulation_commands):
+            reasons.append(
+                "Worktree/index mutations must document merge simulation commands"
+            )
+
+    readonly = _READONLY_DIAG_RE.search(text)
+    if readonly:
+        readonly_value = readonly.group(1)
+        for command in simulation_commands:
+            if command.lower() in readonly_value.lower():
+                reasons.append(
+                    "merge simulation may not be listed under read-only diagnostics"
+                )
+        if has_simulation and "merge simulation" in readonly_value.lower():
+            reasons.append(
+                "merge simulation may not be classified as read-only diagnostics"
+            )
+
+    if has_simulation:
+        if not _WORKTREE_PATH_RE.search(text):
+            reasons.append("merge simulation report missing worktree path")
+        if not _PRE_CLEAN_RE.search(text):
+            reasons.append("merge simulation report missing pre-simulation clean status")
+        if not _MERGE_RESULT_RE.search(text):
+            reasons.append("merge simulation report missing merge result/conflict status")
+    if has_abort or has_simulation:
+        if has_abort and not _ABORT_RE.search(text):
+            reasons.append("merge simulation report missing abort/reset command proof")
+        if has_abort and not _POST_CLEAN_RE.search(text):
+            reasons.append("merge simulation report missing post-abort clean status")
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "simulation_commands": simulation_commands,
+        "safe_next_action": (
+            "classify merge simulation under Worktree/index mutations with full "
+            "pre/merge/abort/post-clean proof; never list as read-only diagnostics"
+            if reasons
+            else "proceed"
+        ),
+    }

reviewer_mutation_categories.py

@@ -0,0 +1,130 @@
+"""Precise mutation category verifier for reviewer controller handoffs (#319)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_CONTROLLER_HANDOFF_RE = re.compile(r"controller handoff", re.IGNORECASE)
+_WORKSPACE_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*workspace\s+mutations\s*:",
+    re.IGNORECASE | re.MULTILINE,
+)
+_VAGUE_MUTATIONS_NONE_RE = re.compile(
+    r"^\s*[-*]?\s*mutations\s*:\s*none\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+
+_REQUIRED_CATEGORIES = (
+    "file edits by reviewer",
+    "worktree/index mutations",
+    "git ref mutations",
+)
+_WORKTREE_FIELD_ALIASES = (
+    "worktree/index mutations",
+    "worktree mutations",
+)
+
+
+def _has_category(text: str, category: str) -> bool:
+    pattern = re.compile(
+        rf"^\s*[-*]?\s*{re.escape(category)}\s*:",
+        re.IGNORECASE | re.MULTILINE,
+    )
+    return bool(pattern.search(text))
+
+
+def _has_worktree_category(text: str) -> bool:
+    return any(_has_category(text, alias) for alias in _WORKTREE_FIELD_ALIASES)
+
+
+def _normalize_for_consistency_check(text: str) -> str:
+    """Map #319 precise labels to #313 field names for consistency delegation."""
+    normalized = text
+    if _has_category(text, "worktree/index mutations") and not _has_category(
+        text, "worktree mutations"
+    ):
+        normalized = re.sub(
+            r"(worktree/index mutations\s*:)",
+            "Worktree mutations:",
+            normalized,
+            flags=re.IGNORECASE,
+        )
+    return normalized
+
+
+def assess_mutation_categories_report(
+    report_text: str,
+    *,
+    handoff_session: dict | None = None,
+    observed_commands: list | None = None,
+) -> dict[str, Any]:
+    """Require precise mutation categories in reviewer controller handoffs (#319)."""
+    text = report_text or ""
+    session = dict(handoff_session or {})
+    reasons: list[str] = []
+    lower = text.lower()
+
+    is_controller = bool(
+        session.get("controller_handoff")
+        or _CONTROLLER_HANDOFF_RE.search(text)
+    )
+    if not is_controller and not session.get("require_categories"):
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "controller_handoff": False,
+            "safe_next_action": "proceed",
+        }
+
+    if _WORKSPACE_MUTATIONS_RE.search(text):
+        reasons.append(
+            "controller handoffs must not use legacy 'Workspace mutations'; "
+            "use precise mutation category fields"
+        )
+
+    if _VAGUE_MUTATIONS_NONE_RE.search(text):
+        reasons.append(
+            "controller handoffs must not use vague 'Mutations: none'; "
+            "report each precise category separately"
+        )
+
+    missing = [
+        category
+        for category in _REQUIRED_CATEGORIES
+        if category != "worktree/index mutations" and not _has_category(text, category)
+    ]
+    if not _has_worktree_category(text):
+        missing.append("worktree/index mutations")
+
+    if missing:
+        reasons.append(
+            "controller handoff missing precise mutation categories: "
+            + ", ".join(missing)
+        )
+
+    commands = observed_commands or session.get("observed_commands")
+    if commands:
+        from review_proofs import assess_workspace_mutation_consistency
+
+        consistency = assess_workspace_mutation_consistency(
+            _normalize_for_consistency_check(text),
+            commands,
+        )
+        if not consistency.get("complete"):
+            reasons.extend(consistency.get("reasons") or [])
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "controller_handoff": True,
+        "safe_next_action": (
+            "replace Workspace mutations with file edits, worktree/index, git ref, "
+            "and other precise mutation category fields"
+            if reasons
+            else "proceed"
+        ),
+    }

reviewer_validation_integrity.py

@@ -0,0 +1,148 @@
+"""PR-head vs diagnostic validation integrity verifier (#316)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_DIAGNOSTIC_LABEL_RE = re.compile(
+    r"diagnostic local experiment|not pr-head validation|diagnostic-only validation",
+    re.IGNORECASE,
+)
+_OFFICIAL_VALIDATION_RE = re.compile(
+    r"(?:official validation|pr-head validation|validation integrity)",
+    re.IGNORECASE,
+)
+_OFFICIAL_COMMAND_RE = re.compile(
+    r"(?:official validation command|pr-head validation command|validation command)",
+    re.IGNORECASE,
+)
+_OFFICIAL_RESULT_RE = re.compile(
+    r"(?:official validation result|pr-head validation result|validation result|validation integrity status)",
+    re.IGNORECASE,
+)
+_DIRTY_AFTER_RE = re.compile(
+    r"(?:worktree dirty after (?:official )?validation|dirty (?:state )?after (?:official )?validation|"
+    r"validation worktree dirty after)",
+    re.IGNORECASE,
+)
+_DIAGNOSTIC_EDIT_RE = re.compile(
+    r"(?:diagnostic edit(?:ed)? path|file edits by reviewer|diagnostic local edit)",
+    re.IGNORECASE,
+)
+_DIAGNOSTIC_COMMAND_RE = re.compile(
+    r"(?:diagnostic (?:validation )?command|diagnostic test run|diagnostic result)",
+    re.IGNORECASE,
+)
+_SUGGESTED_FIX_RE = re.compile(
+    r"(?:diagnostic results? (?:were )?used only for suggested fix|suggested fix only|"
+    r"not used as official validation)",
+    re.IGNORECASE,
+)
+_INTEGRITY_STATUS_RE = re.compile(
+    r"validation integrity status\s*:\s*(passed|failed|not run|contaminated)",
+    re.IGNORECASE,
+)
+_POST_EDIT_AS_OFFICIAL_RE = re.compile(
+    r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit|"
+    r"full suite passed after diagnostic edit)",
+    re.IGNORECASE,
+)
+
+
+def _performed_edits(action_log: list[dict] | None) -> list[str]:
+    paths: list[str] = []
+    for entry in action_log or []:
+        if entry.get("gated_rejected") or entry.get("performed") is False:
+            continue
+        path = entry.get("path")
+        if path and entry.get("kind", "file_edit") in {"file_edit", "edit", "write"}:
+            paths.append(str(path))
+    return paths
+
+
+def assess_validation_integrity_report(
+    report_text: str,
+    *,
+    validation_session: dict | None = None,
+    action_log: list[dict] | None = None,
+) -> dict[str, Any]:
+    """Separate official PR-head validation from diagnostic edited-worktree tests (#316)."""
+    text = report_text or ""
+    session = dict(validation_session or {})
+    reasons: list[str] = []
+
+    diagnostic_edits = list(session.get("diagnostic_edits") or [])
+    if not diagnostic_edits:
+        diagnostic_edits = _performed_edits(action_log)
+    diagnostic_ran = bool(
+        session.get("diagnostic_validation_ran")
+        or session.get("diagnostic_runs")
+    )
+    official_ran = bool(session.get("official_validation_ran", True))
+    official_result = (session.get("official_result") or "").strip().lower()
+    diagnostic_result = (session.get("diagnostic_result") or "").strip().lower()
+    dirty_after = session.get("worktree_dirty_after_official")
+    contaminated = bool(session.get("contaminated"))
+
+    if official_ran:
+        if not _OFFICIAL_VALIDATION_RE.search(text) and not _OFFICIAL_COMMAND_RE.search(text):
+            reasons.append("report missing official PR-head validation section")
+        if not _OFFICIAL_COMMAND_RE.search(text) and "validation:" not in text.lower():
+            reasons.append("report missing official validation command")
+        if not _OFFICIAL_RESULT_RE.search(text):
+            reasons.append("report missing official validation result or integrity status")
+        if dirty_after is not None and not _DIRTY_AFTER_RE.search(text):
+            reasons.append(
+                "report missing worktree dirty state after official validation"
+            )
+        elif dirty_after is None and official_ran and not _DIRTY_AFTER_RE.search(text):
+            reasons.append(
+                "report missing worktree dirty state after official validation"
+            )
+
+    if diagnostic_edits or diagnostic_ran:
+        if not _DIAGNOSTIC_LABEL_RE.search(text):
+            reasons.append(
+                "post-edit diagnostic runs must be labeled "
+                "'Diagnostic local experiment — not PR-head validation'"
+            )
+        if diagnostic_edits and not _DIAGNOSTIC_EDIT_RE.search(text):
+            reasons.append("report missing diagnostic edit path")
+        if diagnostic_ran and not _DIAGNOSTIC_COMMAND_RE.search(text):
+            reasons.append("report missing diagnostic command/result")
+        if not _SUGGESTED_FIX_RE.search(text):
+            reasons.append(
+                "report must state diagnostic results were used only for suggested fix"
+            )
+        if _POST_EDIT_AS_OFFICIAL_RE.search(text):
+            reasons.append(
+                "post-edit diagnostic results cannot be reported as official PR-head validation"
+            )
+        if diagnostic_result == "pass" and official_result == "fail":
+            if "request_changes" not in text.lower() and "failed" not in text.lower():
+                reasons.append(
+                    "request-changes must cite unmodified PR-head failure, not diagnostic pass"
+                )
+
+    if contaminated:
+        status = _INTEGRITY_STATUS_RE.search(text)
+        if not status or "contaminated" not in status.group(1).lower():
+            reasons.append(
+                "contaminated validation must report integrity status "
+                "'contaminated — recovery required'"
+            )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "diagnostic_edits": diagnostic_edits,
+        "safe_next_action": (
+            "separate official PR-head validation from diagnostic experiments; "
+            "report dirty-after-validation state"
+            if reasons
+            else "proceed"
+        ),
+    }

reviewer_validation_worktree_edits.py

@@ -0,0 +1,184 @@
+"""PR validation worktree read-only edit verifier (#315)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_VALIDATION_WORKTREE_RE = re.compile(
+    r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)",
+    re.IGNORECASE,
+)
+_DIAGNOSTIC_WORKTREE_RE = re.compile(
+    r"branches/(?:diagnostic[\w/-]*|scratch[\w/-]*)",
+    re.IGNORECASE,
+)
+_FILE_EDITS_NONE_RE = re.compile(
+    r"file edits by reviewer\s*:\s*none\b",
+    re.IGNORECASE,
+)
+_FILE_EDITS_FIELD_RE = re.compile(
+    r"file edits by reviewer\s*:\s*(.+)",
+    re.IGNORECASE,
+)
+_VALIDATION_WORKTREE_PATH_RE = re.compile(
+    r"(?:validation worktree path|review worktree path)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_DIAGNOSTIC_WORKTREE_PATH_RE = re.compile(
+    r"(?:diagnostic (?:scratch )?worktree path|diagnostic worktree)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_DIAGNOSTIC_LABEL_RE = re.compile(
+    r"diagnostic local experiment|not pr-head validation|diagnostic-only",
+    re.IGNORECASE,
+)
+_OFFICIAL_VALIDATION_RE = re.compile(
+    r"(?:official (?:pr-head )?validation|pr-head validation result)",
+    re.IGNORECASE,
+)
+_POST_EDIT_OFFICIAL_RE = re.compile(
+    r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit)",
+    re.IGNORECASE,
+)
+
+_PERFORMED_EDIT_KINDS = frozenset({"file_edit", "edit", "write", "edited", "created", "wrote"})
+
+
+def _normalize_path(path: str) -> str:
+    return (path or "").replace("\\", "/").strip()
+
+
+def _is_validation_worktree(path: str) -> bool:
+    return bool(_VALIDATION_WORKTREE_RE.search(_normalize_path(path)))
+
+
+def _is_diagnostic_worktree(path: str) -> bool:
+    normalized = _normalize_path(path)
+    return bool(
+        _DIAGNOSTIC_WORKTREE_RE.search(normalized)
+        or "diagnostic" in normalized.lower()
+    )
+
+
+def _performed_edits(action_log: list[dict] | None) -> list[dict[str, str]]:
+    edits: list[dict[str, str]] = []
+    for entry in action_log or []:
+        if entry.get("gated_rejected") or entry.get("performed") is False:
+            continue
+        kind = (entry.get("kind") or entry.get("action") or "file_edit").strip().lower()
+        if kind not in _PERFORMED_EDIT_KINDS:
+            continue
+        path = (entry.get("path") or "").strip()
+        if not path:
+            continue
+        worktree = _normalize_path(str(entry.get("worktree_path") or entry.get("cwd") or ""))
+        edits.append({"path": path, "worktree_path": worktree})
+    return edits
+
+
+def _extract_validation_path(text: str, session: dict) -> str:
+    path = _normalize_path(str(session.get("validation_worktree_path") or ""))
+    if path:
+        return path
+    match = _VALIDATION_WORKTREE_PATH_RE.search(text or "")
+    return _normalize_path(match.group(1)) if match else ""
+
+
+def _extract_diagnostic_path(text: str, session: dict) -> str:
+    path = _normalize_path(str(session.get("diagnostic_worktree_path") or ""))
+    if path:
+        return path
+    match = _DIAGNOSTIC_WORKTREE_PATH_RE.search(text or "")
+    return _normalize_path(match.group(1)) if match else ""
+
+
+def assess_validation_worktree_edit_report(
+    report_text: str,
+    *,
+    validation_session: dict | None = None,
+    action_log: list[dict] | None = None,
+) -> dict[str, Any]:
+    """Block edits in PR validation worktrees; require diagnostic separation (#315)."""
+    text = report_text or ""
+    session = dict(validation_session or {})
+    reasons: list[str] = []
+
+    validation_path = _extract_validation_path(text, session)
+    diagnostic_path = _extract_diagnostic_path(text, session)
+    validation_edits = list(session.get("validation_worktree_edits") or [])
+    diagnostic_edits = list(session.get("diagnostic_edits") or [])
+    edits = _performed_edits(action_log)
+
+    if not validation_edits:
+        for entry in edits:
+            wt = entry.get("worktree_path") or validation_path
+            if wt and _is_validation_worktree(wt) and not _is_diagnostic_worktree(wt):
+                validation_edits.append(entry["path"])
+
+    if not diagnostic_edits:
+        for entry in edits:
+            wt = entry.get("worktree_path") or ""
+            if wt and _is_diagnostic_worktree(wt):
+                diagnostic_edits.append(entry["path"])
+            elif entry["path"] and diagnostic_path and not validation_edits:
+                if _is_diagnostic_worktree(diagnostic_path):
+                    diagnostic_edits.append(entry["path"])
+
+    claimed_none = bool(_FILE_EDITS_NONE_RE.search(text))
+    any_edits = bool(validation_edits or diagnostic_edits or edits)
+
+    if validation_edits:
+        reasons.append(
+            "reviewer must not edit files in the PR validation worktree; "
+            f"observed edits: {', '.join(validation_edits)}"
+        )
+
+    if diagnostic_edits or session.get("diagnostic_experiment"):
+        if not diagnostic_path and not _DIAGNOSTIC_WORKTREE_PATH_RE.search(text):
+            reasons.append(
+                "diagnostic experiments require a separate diagnostic scratch worktree path"
+            )
+        if not _DIAGNOSTIC_LABEL_RE.search(text):
+            reasons.append(
+                "diagnostic experiments must be labeled "
+                "'Diagnostic local experiment — not PR-head validation'"
+            )
+        if not _FILE_EDITS_FIELD_RE.search(text) or claimed_none:
+            reasons.append(
+                "diagnostic edits must be reported under 'File edits by reviewer'"
+            )
+
+    if any_edits and claimed_none:
+        reasons.append(
+            "report claims 'File edits by reviewer: none' but reviewer file edits occurred"
+        )
+
+    if session.get("official_validation_after_edit") or _POST_EDIT_OFFICIAL_RE.search(text):
+        if validation_edits or (validation_path and diagnostic_edits):
+            reasons.append(
+                "post-edit test results cannot be reported as official PR-head validation"
+            )
+
+    if validation_edits and _OFFICIAL_VALIDATION_RE.search(text):
+        if not _DIAGNOSTIC_LABEL_RE.search(text):
+            reasons.append(
+                "validation worktree was edited; official PR-head validation is no longer valid"
+            )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "validation_worktree_path": validation_path or None,
+        "diagnostic_worktree_path": diagnostic_path or None,
+        "validation_worktree_edits": validation_edits,
+        "diagnostic_edits": diagnostic_edits,
+        "safe_next_action": (
+            "keep PR validation worktrees read-only; use a separate diagnostic "
+            "scratch worktree and report all reviewer file edits"
+            if reasons
+            else "proceed"
+        ),
+    }

reviewer_worktree_ownership.py

@@ -0,0 +1,207 @@
+"""Reviewer worktree ownership and safe-reuse proof verifier (#312)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_SESSION_OWNED_RE = re.compile(
+    r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)",
+    re.IGNORECASE,
+)
+_WORKTREE_PATH_RE = re.compile(
+    r"(?:review worktree path|worktree path|session-owned worktree)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_BRANCHES_PATH_RE = re.compile(r"/branches/|\bbranches/", re.IGNORECASE)
+_MAIN_CHECKOUT_RE = re.compile(
+    r"main checkout|not (?:the )?main checkout|outside branches/",
+    re.IGNORECASE,
+)
+_SAFE_REUSE_RE = re.compile(r"safe[- ]reuse proof", re.IGNORECASE)
+_REUSE_POLICY_RE = re.compile(
+    r"(?:project policy|policy) (?:allows|allowing) (?:reuse|reset)",
+    re.IGNORECASE,
+)
+_CLEAN_TRACKED_RE = re.compile(
+    r"(?:clean tracked state|tracked state\s*:\s*clean|no uncommitted tracked)",
+    re.IGNORECASE,
+)
+_CLEAN_UNTRACKED_RE = re.compile(
+    r"(?:clean untracked state|untracked state\s*:\s*clean|no untracked files)",
+    re.IGNORECASE,
+)
+_NOT_OTHER_SESSION_RE = re.compile(
+    r"not owned by (?:another|other) (?:active )?(?:task|session)",
+    re.IGNORECASE,
+)
+_BRANCH_BEFORE_RESET_RE = re.compile(
+    r"(?:branch/head before reset|head before reset|branch before reset)\s*:\s*(\S+)",
+    re.IGNORECASE,
+)
+_RESET_TARGET_RE = re.compile(
+    r"(?:reset target sha|reset target)\s*:\s*([0-9a-f]{7,40})",
+    re.IGNORECASE,
+)
+_DESTRUCTIVE_CMD_RE = re.compile(
+    r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
+    r"(?:reset\s+--hard|clean(?:\s+-[A-Za-z]+)*|checkout\s+(?!HEAD\s+--)|switch\s+)",
+    re.IGNORECASE,
+)
+_WORKSPACE_NONE_RE = re.compile(r"workspace mutations\s*:\s*none", re.IGNORECASE)
+_WORKTREE_MUTATION_RE = re.compile(
+    r"(?:worktree(?:/index)? mutations|destructive reset)\s*:\s*(?!none\b).+",
+    re.IGNORECASE,
+)
+_RESET_IN_REPORT_RE = re.compile(r"reset\s+--hard", re.IGNORECASE)
+
+
+def _command_text(entry) -> str:
+    if isinstance(entry, dict):
+        return str(entry.get("command") or "").strip()
+    return str(entry or "").strip()
+
+
+def _destructive_commands(command_log: list | None) -> list[str]:
+    return [
+        cmd
+        for cmd in (_command_text(entry) for entry in (command_log or []))
+        if cmd and _DESTRUCTIVE_CMD_RE.search(cmd)
+    ]
+
+
+def _extract_worktree_path(text: str, session: dict) -> str:
+    path = (session.get("worktree_path") or "").strip()
+    if path:
+        return path
+    match = _WORKTREE_PATH_RE.search(text or "")
+    return match.group(1).strip() if match else ""
+
+
+def _is_session_owned_path(path: str) -> bool:
+    normalized = (path or "").replace("\\", "/")
+    return bool(_SESSION_OWNED_RE.search(normalized))
+
+
+def _safe_reuse_fields_present(text: str) -> list[str]:
+    missing: list[str] = []
+    if not _WORKTREE_PATH_RE.search(text):
+        missing.append("exact worktree path")
+    if not _BRANCHES_PATH_RE.search(text):
+        missing.append("worktree inside branches/")
+    if not _MAIN_CHECKOUT_RE.search(text):
+        missing.append("worktree is not the main checkout")
+    if not _NOT_OTHER_SESSION_RE.search(text):
+        missing.append("worktree not owned by another active session")
+    if not _CLEAN_TRACKED_RE.search(text):
+        missing.append("clean tracked state")
+    if not _CLEAN_UNTRACKED_RE.search(text):
+        missing.append("clean untracked state")
+    if not _BRANCH_BEFORE_RESET_RE.search(text):
+        missing.append("branch/head before reset")
+    if not _RESET_TARGET_RE.search(text):
+        missing.append("reset target SHA")
+    if not _REUSE_POLICY_RE.search(text):
+        missing.append("explicit project policy allowing reuse/reset")
+    return missing
+
+
+def assess_worktree_ownership_report(
+    report_text: str,
+    *,
+    ownership_session: dict | None = None,
+    command_log: list | None = None,
+) -> dict[str, Any]:
+    """Prove reviewer worktree ownership before reset or validation (#312)."""
+    text = report_text or ""
+    session = dict(ownership_session or {})
+    reasons: list[str] = []
+
+    worktree_path = _extract_worktree_path(text, session)
+    destructive = _destructive_commands(command_log or session.get("command_log"))
+    if not destructive and session.get("destructive_reset"):
+        destructive = ["git reset --hard (session)"]
+
+    session_owned = bool(
+        session.get("session_owned")
+        or (worktree_path and _is_session_owned_path(worktree_path))
+    )
+    safe_reuse = bool(session.get("safe_reuse") or _SAFE_REUSE_RE.search(text))
+    main_checkout = bool(session.get("main_checkout"))
+    dirty_tracked = session.get("dirty_tracked")
+    dirty_untracked = session.get("dirty_untracked")
+    other_session_owned = bool(session.get("other_session_owned"))
+
+    if worktree_path or destructive or session.get("validation_ran"):
+        if not worktree_path:
+            reasons.append("report missing exact review worktree path")
+        elif main_checkout or "branches/" not in worktree_path.replace("\\", "/"):
+            reasons.append("review worktree must be inside branches/, not the main checkout")
+        elif not _BRANCHES_PATH_RE.search(worktree_path.replace("\\", "/")):
+            reasons.append("review worktree path must be under branches/")
+
+    if other_session_owned and not safe_reuse:
+        reasons.append(
+            "reused worktree appears owned by another active task/session; "
+            "safe-reuse proof required"
+        )
+
+    if dirty_tracked:
+        reasons.append(
+            "worktree has uncommitted tracked changes before reset or validation"
+        )
+    if dirty_untracked and safe_reuse:
+        reasons.append("safe-reuse proof requires clean untracked state")
+
+    if safe_reuse and not session_owned:
+        reasons.extend(
+            f"safe-reuse proof missing {field}"
+            for field in _safe_reuse_fields_present(text)
+        )
+
+    if destructive:
+        if not session_owned and not safe_reuse:
+            reasons.append(
+                "destructive worktree commands forbidden without session-owned "
+                "worktree or safe-reuse proof"
+            )
+        if _WORKSPACE_NONE_RE.search(text) and (
+            _RESET_IN_REPORT_RE.search(text) or any("reset" in c.lower() for c in destructive)
+        ):
+            reasons.append(
+                "final report must not claim 'Workspace mutations: none' when "
+                "git reset --hard occurred"
+            )
+        if not _WORKTREE_MUTATION_RE.search(text) and not _RESET_IN_REPORT_RE.search(text):
+            reasons.append(
+                "destructive reset must be reported under Worktree/index mutations "
+                "or destructive reset operations"
+            )
+
+    if (
+        worktree_path
+        and not session_owned
+        and not safe_reuse
+        and (destructive or session.get("validation_ran"))
+        and not _is_session_owned_path(worktree_path)
+    ):
+        reasons.append(
+            "reused branch-named worktree requires safe-reuse proof before reset or validation"
+        )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "worktree_path": worktree_path or None,
+        "session_owned": session_owned,
+        "safe_reuse": safe_reuse,
+        "destructive_commands": destructive,
+        "safe_next_action": (
+            "use a fresh session-owned review worktree under branches/review-pr<N>-* "
+            "or document full safe-reuse proof before destructive reset"
+            if reasons
+            else "proceed"
+        ),
+    }

task_capability_map.py

@@ -96,6 +96,18 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
+    "post_heartbeat": {
+        "permission": "gitea.issue.comment",
+        "role": "author",
+    },
+    "reconcile_issue_claims": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "cleanup_stale_claims": {
+        "permission": "gitea.issue.comment",
+        "role": "author",
+    },
 }
 
 # Issue-mutating MCP tools and their resolver task keys.

tests/test_agent_temp_artifacts.py

@@ -62,7 +62,24 @@ class TestPreflightWarnings(unittest.TestCase):
         self.assertIn("_encode_commit_payload.py", status["preflight_warnings"][0])
 
 
+# Issue-write tools are profile-gated (#69); gitea_lock_issue requires
+# gitea.issue.comment (see task_capability_map), so the gate must be
+# seeded exactly like tests/test_mcp_server.py::TestIssueLocking (#359).
+ISSUE_WRITE_ENV = {
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.issue.create,gitea.issue.close,gitea.issue.comment"
+    ),
+}
+
+
 class TestIssueLockArtifactWarning(unittest.TestCase):
+    def setUp(self):
+        self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
+        self._env_patcher.start()
+
+    def tearDown(self):
+        self._env_patcher.stop()
+
     @patch("mcp_server.api_get_all", return_value=[])
     @patch("mcp_server._auth", return_value="token x")
     @patch("mcp_server._resolve", return_value=("h", "o", "r"))
@@ -72,6 +89,9 @@ class TestIssueLockArtifactWarning(unittest.TestCase):
         mock_state.return_value = {
             "current_branch": "master",
             "porcelain_status": "?? _emit_payload.py\n",
+            "base_equivalent": True,
+            "inspected_git_root": "/scratch/wt",
+            "base_branch": "origin/master",
         }
         result = mcp_server.gitea_lock_issue(
             issue_number=261,

tests/test_already_landed_report_wording.py

@@ -0,0 +1,151 @@
+"""Tests for already-landed final-report wording validator (#298).
+
+An already-landed PR is reconciliation-only, never review/merge
+eligible, and a head SHA may not be called reviewed unless validation
+and diff review actually passed. Final reports and controller handoffs
+must use the reconciliation wording, not the legacy eligible/reviewed
+fields.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import assess_already_landed_report_wording  # noqa: E402
+
+
+GOOD_ALREADY_LANDED = (
+    "Controller Handoff\n"
+    "- Oldest open PR requiring action: PR #278\n"
+    "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n"
+    "- Candidate head SHA: 2a544b7\n"
+    "- Reviewed head SHA: none\n"
+    "- Review worktree used: false\n"
+)
+
+
+class TestAlreadyLandedWording(unittest.TestCase):
+    def test_correct_reconciliation_wording_passes(self):
+        result = assess_already_landed_report_wording(
+            GOOD_ALREADY_LANDED, already_landed=True
+        )
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["downgraded"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_oldest_eligible_wording_fails(self):
+        report = GOOD_ALREADY_LANDED + "- oldest eligible PR: PR #278\n"
+        result = assess_already_landed_report_wording(
+            report, already_landed=True
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("oldest eligible pr" in r.lower() for r in result["reasons"])
+        )
+
+    def test_next_eligible_wording_fails(self):
+        report = GOOD_ALREADY_LANDED + "- next eligible PR: PR #278\n"
+        result = assess_already_landed_report_wording(
+            report, already_landed=True
+        )
+        self.assertFalse(result["complete"])
+
+    def test_pinned_reviewed_head_fails(self):
+        report = GOOD_ALREADY_LANDED + "- Pinned reviewed head: 2a544b7\n"
+        result = assess_already_landed_report_wording(
+            report, already_landed=True
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("pinned reviewed head" in r.lower() for r in result["reasons"])
+        )
+
+    def test_scratch_worktree_field_fails(self):
+        report = GOOD_ALREADY_LANDED + "- Scratch worktree used: False\n"
+        result = assess_already_landed_report_wording(
+            report, already_landed=True
+        )
+        self.assertFalse(result["complete"])
+
+    def test_missing_eligibility_class_fails(self):
+        report = (
+            "Controller Handoff\n"
+            "- Oldest open PR requiring action: PR #278\n"
+            "- Candidate head SHA: 2a544b7\n"
+            "- Reviewed head SHA: none\n"
+            "- Review worktree used: false\n"
+        )
+        result = assess_already_landed_report_wording(
+            report, already_landed=True
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("eligibility class" in r.lower() for r in result["reasons"])
+        )
+
+    def test_populated_reviewed_head_sha_fails(self):
+        report = GOOD_ALREADY_LANDED.replace(
+            "- Reviewed head SHA: none\n",
+            "- Reviewed head SHA: 2a544b7\n",
+        )
+        result = assess_already_landed_report_wording(
+            report, already_landed=True
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("reviewed head" in r.lower() for r in result["reasons"])
+        )
+
+
+class TestNonAlreadyLandedReports(unittest.TestCase):
+    def test_normal_reviewed_report_passes(self):
+        report = (
+            "- Selected PR: 281\n"
+            "- Pinned reviewed head: abc1234\n"
+            "- Review decision: approve\n"
+        )
+        result = assess_already_landed_report_wording(
+            report, already_landed=False, review_validated=True
+        )
+        self.assertTrue(result["complete"])
+
+    def test_blocked_infra_report_with_pinned_head_fails(self):
+        report = (
+            "- Selected PR: 281\n"
+            "- Pinned reviewed head: abc1234\n"
+            "- Current status: blocked on infra_stop\n"
+        )
+        result = assess_already_landed_report_wording(
+            report, already_landed=False, review_validated=False
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("before validation" in r.lower() for r in result["reasons"])
+        )
+
+    def test_validation_failed_report_with_reviewed_sha_fails(self):
+        report = (
+            "- Selected PR: 281\n"
+            "- Reviewed head SHA: abc1234\n"
+            "- Validation: full suite failed\n"
+        )
+        result = assess_already_landed_report_wording(
+            report, already_landed=False, review_validated=False
+        )
+        self.assertFalse(result["complete"])
+
+    def test_validation_failed_report_with_reviewed_none_passes(self):
+        report = (
+            "- Selected PR: 281\n"
+            "- Reviewed head SHA: none\n"
+            "- Validation: full suite failed\n"
+        )
+        result = assess_already_landed_report_wording(
+            report, already_landed=False, review_validated=False
+        )
+        self.assertTrue(result["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_claim_heartbeat.py

@@ -0,0 +1,116 @@
+"""Tests for issue claim heartbeat leases (#268)."""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_claim_heartbeat as ich  # noqa: E402
+
+
+class TestHeartbeatFormatting(unittest.TestCase):
+    def test_format_and_parse_roundtrip(self):
+        body = ich.format_heartbeat_body(
+            kind="claim",
+            issue_number=268,
+            branch="feat/issue-268-heartbeat-leases",
+            phase="claimed",
+            profile="prgs-author",
+            next_action="implement module",
+        )
+        parsed = ich.parse_heartbeat_comment(body)
+        self.assertIsNotNone(parsed)
+        assert parsed is not None
+        self.assertEqual(parsed["issue_number"], 268)
+        self.assertEqual(parsed["branch"], "feat/issue-268-heartbeat-leases")
+        self.assertEqual(parsed["phase"], "claimed")
+
+
+class TestClaimClassification(unittest.TestCase):
+    NOW = datetime(2026, 7, 7, 12, 0, tzinfo=timezone.utc)
+
+    def _comment(self, *, minutes_ago: int, kind: str = "progress") -> dict:
+        ts = (self.NOW - timedelta(minutes=minutes_ago)).isoformat()
+        body = ich.format_heartbeat_body(
+            kind=kind,
+            issue_number=268,
+            branch="feat/issue-268-heartbeat-leases",
+            phase="implementation",
+            profile="prgs-author",
+        )
+        return {"id": 1, "body": body, "created_at": ts, "updated_at": ts}
+
+    def test_active_claim_within_lease(self):
+        issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
+        result = ich.classify_issue_claim(
+            issue=issue,
+            comments=[self._comment(minutes_ago=5)],
+            now=self.NOW,
+        )
+        self.assertEqual(result["status"], "active")
+
+    def test_stale_claim_without_branch(self):
+        issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
+        result = ich.classify_issue_claim(
+            issue=issue,
+            comments=[self._comment(minutes_ago=45)],
+            now=self.NOW,
+        )
+        self.assertEqual(result["status"], "stale")
+
+    def test_reclaimable_after_sixty_minutes_without_branch(self):
+        issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
+        result = ich.classify_issue_claim(
+            issue=issue,
+            comments=[self._comment(minutes_ago=90)],
+            now=self.NOW,
+        )
+        self.assertEqual(result["status"], "reclaimable")
+
+    def test_awaiting_review_when_open_pr_exists(self):
+        issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
+        open_prs = [
+            {
+                "number": 400,
+                "title": "feat",
+                "body": "Closes #268",
+                "head": {"ref": "feat/issue-268-heartbeat-leases"},
+            }
+        ]
+        result = ich.classify_issue_claim(
+            issue=issue,
+            comments=[self._comment(minutes_ago=120)],
+            open_prs=open_prs,
+            now=self.NOW,
+        )
+        self.assertEqual(result["status"], "awaiting_review")
+
+    def test_phantom_claim_without_heartbeat(self):
+        issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]}
+        result = ich.classify_issue_claim(
+            issue=issue,
+            comments=[{"id": 1, "body": "working on it"}],
+            now=self.NOW,
+        )
+        self.assertEqual(result["status"], "phantom")
+
+
+class TestInventory(unittest.TestCase):
+    def test_build_cleanup_plan_flags_phantom(self):
+        inventory = {
+            "entries": [
+                {"issue_number": 1, "status": "phantom", "reasons": ["no heartbeat"]},
+                {"issue_number": 2, "status": "active", "reasons": []},
+            ]
+        }
+        plan = ich.build_cleanup_plan(inventory)
+        self.assertEqual(len(plan), 1)
+        self.assertEqual(plan[0]["issue_number"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_write_tool_gates.py

@@ -172,6 +172,8 @@ class TestAllowedIssueWritesProceed(IssueWriteGateBase):
                 return [{"id": 10, "name": "status:in-progress"}]
             if method == "POST" and "/issues/9/labels" in url:
                 return [{"name": "status:in-progress"}]
+            if method == "POST" and "/issues/9/comments" in url:
+                return {"id": 501}
             if method == "GET" and url.endswith("/user"):
                 return {"login": "author-user"}
             return {}

tests/test_linked_issue_consistency.py

@@ -0,0 +1,118 @@
+"""Tests for linked-issue consistency verifier (#314).
+
+Reviewer reports must verify the linked issue live for the selected PR;
+stale issue numbers from a previous PR must not leak into the final
+report, and linked-issue status must never be claimed without live proof.
+"""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from review_proofs import assess_linked_issue_consistency  # noqa: E402
+
+
+class TestCorrectLinkedIssue(unittest.TestCase):
+    def test_matching_linked_issue_passes(self):
+        report = (
+            "Selected PR: 280\n"
+            "Linked issue status: Issue #275 open, closes on merge\n"
+        )
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275]
+        )
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["downgraded"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_multiple_linked_issues_all_reported_passes(self):
+        report = (
+            "Linked issue status: Issue #275 and Issue #276 both open\n"
+        )
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275, 276]
+        )
+        self.assertTrue(result["complete"])
+
+    def test_multiple_linked_issues_one_missing_fails(self):
+        report = "Linked issue status: Issue #275 open\n"
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275, 276]
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("276" in r for r in result["reasons"])
+        )
+
+
+class TestStaleIssueLeak(unittest.TestCase):
+    def test_stale_issue_in_status_field_fails(self):
+        # #314 observed behavior: PR #280 links Issue #275 but the report
+        # carries Issue #260 from the previous PR.
+        report = "Linked issue status: Issue #260 open\n"
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275]
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(any("260" in r for r in result["reasons"]))
+
+    def test_stale_issue_in_diagnostics_fails(self):
+        report = (
+            "Linked issue status: Issue #275 open\n"
+            "Read-only diagnostics: viewed Issue #260 status\n"
+        )
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275]
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("stale" in r.lower() and "260" in r for r in result["reasons"])
+        )
+
+    def test_selected_pr_number_mention_is_not_stale(self):
+        # "PR #280" mentions must not be confused with issue mentions.
+        report = (
+            "Selected PR: 280\n"
+            "Validation: ran tests at PR #280 head\n"
+            "Linked issue status: Issue #275 open\n"
+        )
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275]
+        )
+        self.assertTrue(result["complete"])
+
+
+class TestMissingProof(unittest.TestCase):
+    def test_status_claim_without_live_proof_fails(self):
+        report = "Linked issue status: Issue #275 open\n"
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=None
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("live proof" in r.lower() for r in result["reasons"])
+        )
+
+    def test_unverified_wording_without_proof_passes(self):
+        report = (
+            "Linked issue status: not verified in this session\n"
+        )
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=None
+        )
+        self.assertTrue(result["complete"])
+
+    def test_missing_field_fails(self):
+        report = "Selected PR: 280\n"
+        result = assess_linked_issue_consistency(
+            report, selected_pr=280, linked_issues=[275]
+        )
+        self.assertFalse(result["complete"])
+        self.assertTrue(
+            any("linked issue status" in r.lower() for r in result["reasons"])
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_llm_workflow_split.py

@@ -136,6 +136,18 @@ def test_create_issue_final_report_verifier_exported():
     assert callable(assess_create_issue_final_report)
 
 
+def test_merge_simulation_verifier_exported():
+    from review_proofs import assess_merge_simulation_report
+
+    assert callable(assess_merge_simulation_report)
+
+
+def test_validation_integrity_verifier_exported():
+    from review_proofs import assess_validation_integrity_report
+
+    assert callable(assess_validation_integrity_report)
+
+
 def test_prior_blocker_skip_verifier_exported():
     from review_proofs import assess_prior_blocker_skip_proof
 
@@ -146,3 +158,21 @@ def test_non_mergeable_skip_verifier_exported():
     from review_proofs import assess_non_mergeable_skip_proof
 
     assert callable(assess_non_mergeable_skip_proof)
+
+
+def test_validation_worktree_edit_verifier_exported():
+    from review_proofs import assess_validation_worktree_edit_report
+
+    assert callable(assess_validation_worktree_edit_report)
+
+
+def test_mutation_categories_verifier_exported():
+    from review_proofs import assess_mutation_categories_report
+
+    assert callable(assess_mutation_categories_report)
+
+
+def test_worktree_ownership_verifier_exported():
+    from review_proofs import assess_worktree_ownership_report
+
+    assert callable(assess_worktree_ownership_report)

tests/test_mcp_server.py

@@ -321,15 +321,17 @@ class TestMarkIssue(unittest.TestCase):
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     def test_start_adds_label(self, _auth, mock_api):
-        # First call: get labels; second call: add label
+        # labels, add label, post claim heartbeat comment
         mock_api.side_effect = [
             [{"id": 10, "name": "status:in-progress"}],
             [{"name": "status:in-progress"}],
+            {"id": 99},
         ]
         with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
             result = gitea_mark_issue(issue_number=5, action="start")
         self.assertTrue(result["success"])
         self.assertIn("claimed", result["message"])
+        self.assertTrue(result.get("heartbeat_posted"))
 
     @patch("mcp_server.api_request")
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)

tests/test_review_proofs.py

@@ -23,7 +23,11 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par
 from review_proofs import (  # noqa: E402
     ISSUE_SELECTION_CONTINUATION_EXPLICIT,
     ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
+    assess_already_landed_report_state,
+    assess_already_landed_review_gate,
     assess_author_pr_report,
+    assess_validation_environment_proof,
+    assess_full_suite_failure_approval_gate,
     assess_queue_ordering_proof,
     assess_reviewer_baseline_validation_proof,
     assess_capability_evidence,
@@ -2314,10 +2318,108 @@ class TestCreateIssueFinalReport(unittest.TestCase):
         self.assertTrue(result["downgraded"])
 
 
-class TestQueueOrderingProof(unittest.TestCase):
+class TestValidationEnvironmentProof(unittest.TestCase):
-    """Issue #321: reviewer queue selection must prove ordering."""
+    """Issue #311: exact validation command and execution environment."""
+
+    ROOT = "/repo/Gitea-Tools/branches/review-pr-280"
+
+    def test_venv_pytest_with_full_report_passes(self):
+        result = assess_validation_environment_proof(
+            report_text=(
+                "Validation command: venv/bin/pytest tests/ --ignore=branches\n"
+                f"Working directory: {self.ROOT}\n"
+                "Result: 1014 passed, 6 skipped"
+            ),
+        )
+        self.assertTrue(result["proven"])
+
+    def test_bare_pytest_without_path_proof_fails(self):
+        result = assess_validation_environment_proof(
+            report_text=(
+                "Validation command: pytest\n"
+                f"Working directory: {self.ROOT}\n"
+                "1014 passed, 6 skipped"
+            ),
+        )
+        self.assertFalse(result["proven"])
+
+    def test_bare_pytest_with_path_and_version_proof_passes(self):
+        result = assess_validation_environment_proof(
+            report_text=(
+                "Validation command: pytest tests/\n"
+                f"Working directory: {self.ROOT}\n"
+                "which pytest: /repo/Gitea-Tools/venv/bin/pytest\n"
+                "pytest --version: pytest 8.3.4\n"
+                "Resolves to the project venv: yes\n"
+                "1014 passed, 6 skipped"
+            ),
+        )
+        self.assertTrue(result["proven"])
+
+    def test_vague_pytest_summary_without_command_fails(self):
+        result = assess_validation_environment_proof(
+            report_text=(
+                f"Working directory: {self.ROOT}\n"
+                "pytest executed inside the worktree: 1014 passed, 6 skipped"
+            ),
+        )
+        self.assertFalse(result["proven"])
+
+    def test_structured_environment_proof_passes(self):
+        result = assess_validation_environment_proof(
+            validation_environment={
+                "command": "pytest tests/",
+                "working_directory": self.ROOT,
+                "which_pytest": "/repo/Gitea-Tools/venv/bin/pytest",
+                "pytest_version": "pytest 8.3.4",
+                "venv_resolved": True,
+                "passed": 1014,
+                "skipped": 6,
+            },
+        )
+        self.assertTrue(result["proven"])
+
+    def test_missing_working_directory_fails(self):
+        result = assess_validation_environment_proof(
+            report_text=(
+                "Validation command: venv/bin/pytest tests/\n"
+                "1014 passed, 6 skipped"
+            ),
+        )
+        self.assertFalse(result["proven"])
+
+    def test_build_final_report_downgrades_missing_environment_proof(self):
+        report = build_final_report(
+            checkout_proof=_good_checkout(),
+            inventory=_good_inventory(),
+            validation=_good_validation(),
+            contamination=_good_contamination(),
+            identity_eligible=True,
+            merge_performed=False,
+            issue_status_verified=True,
+            capability_evidence=_good_capability_evidence(),
+            sweep=_good_sweep(),
+            live_state=_good_live_state(),
+            role_boundary=_good_role_boundary(),
+            review_mutation=_good_review_mutation(),
+            controller_handoff=_good_handoff(),
+            capability_proof=_good_capability_proof(),
+            sweep_proof=_good_secret_sweep(),
+            worktree_proof={
+                "worktree_path": "/repo/branches/review-feat-issue-224",
+                "porcelain_status": "",
+                "pr_scope_files": ["docs/wiki/Repositories.md"],
+                "scratch_used": True,
+            },
+            report_text=(
+                f"Working directory: {self.ROOT}\n"
+                "pytest executed: 1014 passed, 6 skipped"
+            ),
+        )
+        self.assertNotEqual(report["grade"], "A")
+        self.assertFalse(report["validation_environment_proven"])
+
 
-    def test_next_eligible_claim_without_policy_fails(self):
         result = assess_queue_ordering_proof(
             report_text="Selected the next eligible PR: PR #286.",
         )
@@ -2539,5 +2641,306 @@ class TestReviewerBaselineValidationProof(unittest.TestCase):
         self.assertFalse(report["baseline_validation_proven"])
 
 
+class TestAlreadyLandedReviewGate(unittest.TestCase):
+    """Issue #292: PR head already on target blocks approval and merge."""
+
+    def _gate(self, **overrides):
+        kwargs = {
+            "pr_number": 278,
+            "candidate_head_sha": PINNED,
+            "target_branch": "master",
+            "target_branch_sha": OTHER,
+            "head_is_ancestor_of_target": True,
+            "live_head_sha": PINNED,
+        }
+        kwargs.update(overrides)
+        return assess_already_landed_review_gate(**kwargs)
+
+    def test_already_landed_blocks_approval_and_merge(self):
+        result = self._gate()
+        self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED")
+        self.assertFalse(result["approve_allowed"])
+        self.assertFalse(result["merge_api_allowed"])
+        self.assertFalse(result["request_changes_allowed"])
+        self.assertTrue(result["reconciliation_handoff_required"])
+
+    def test_already_landed_with_real_blocker_allows_request_changes(self):
+        result = self._gate(real_blocker=True)
+        self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED")
+        self.assertFalse(result["approve_allowed"])
+        self.assertFalse(result["merge_api_allowed"])
+        self.assertTrue(result["request_changes_allowed"])
+
+    def test_normal_candidate_passes_gate(self):
+        result = self._gate(head_is_ancestor_of_target=False)
+        self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE")
+        self.assertTrue(result["approve_allowed"])
+        self.assertTrue(result["merge_api_allowed"])
+        self.assertFalse(result["reconciliation_handoff_required"])
+
+    def test_non_mergeable_normal_pr_still_passes_this_gate(self):
+        # Mergeability/conflicts are separate gates; ancestry gate only
+        # decides already-landed vs normal candidate.
+        result = self._gate(
+            head_is_ancestor_of_target=False, mergeable=False)
+        self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE")
+        self.assertTrue(result["request_changes_allowed"])
+
+    def test_unchecked_ancestry_blocks_review_mutations(self):
+        result = self._gate(head_is_ancestor_of_target=None)
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["approve_allowed"])
+        self.assertFalse(result["merge_api_allowed"])
+
+    def test_changed_head_since_pin_blocks_until_recheck(self):
+        result = self._gate(
+            head_is_ancestor_of_target=False, live_head_sha=OTHER)
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["approve_allowed"])
+        self.assertTrue(any(
+            "head" in reason.lower() for reason in result["reasons"]
+        ))
+
+    def test_invalid_shas_block_gate(self):
+        result = self._gate(candidate_head_sha="deadbeef")
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["approve_allowed"])
+
+        result = self._gate(target_branch_sha=None)
+        self.assertEqual(result["state"], "GATE_NOT_PROVEN")
+
+    def test_already_landed_handoff_lists_required_fields(self):
+        result = self._gate()
+        for field in (
+            "PR number/title",
+            "candidate head SHA",
+            "target branch",
+            "target branch SHA",
+            "ancestor proof",
+            "linked issue status",
+            "recommended reconciliation action",
+            "capability proof for close/comment mutations",
+        ):
+            self.assertIn(field, result["required_handoff_fields"])
+
+
+class TestAlreadyLandedReportState(unittest.TestCase):
+    """Issue #292: reports cannot say APPROVED after the gate fired."""
+
+    GOOD_REPORT = (
+        "Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n"
+        "Candidate head SHA: " + PINNED + "\n"
+        "Review decision: none\n"
+        "Merge result: none\n"
+    )
+
+    def test_gate_fired_with_reconcile_state_passes(self):
+        result = assess_already_landed_report_state(
+            self.GOOD_REPORT, gate_fired=True)
+        self.assertTrue(result["complete"])
+        self.assertFalse(result["block"])
+
+    def test_gate_fired_with_approved_state_blocks(self):
+        report = self.GOOD_REPORT.replace(
+            "Review decision: none", "Review decision: APPROVED")
+        result = assess_already_landed_report_state(report, gate_fired=True)
+        self.assertTrue(result["block"])
+
+    def test_gate_fired_with_merged_state_blocks(self):
+        report = self.GOOD_REPORT.replace(
+            "Merge result: none", "Merge result: MERGED")
+        result = assess_already_landed_report_state(report, gate_fired=True)
+        self.assertTrue(result["block"])
+
+    def test_gate_fired_with_ready_to_merge_blocks(self):
+        result = assess_already_landed_report_state(
+            self.GOOD_REPORT + "Current status: READY_TO_MERGE\n",
+            gate_fired=True)
+        self.assertTrue(result["block"])
+
+    def test_gate_fired_without_reconcile_state_blocks(self):
+        result = assess_already_landed_report_state(
+            "Current status: review complete.\n", gate_fired=True)
+        self.assertTrue(result["block"])
+        self.assertTrue(any(
+            "already_landed_reconcile_required" in reason.lower()
+            for reason in result["reasons"]
+        ))
+
+    def test_gate_not_fired_reports_pass_untouched(self):
+        result = assess_already_landed_report_state(
+            "Review decision: APPROVED\nMerge result: MERGED\n",
+            gate_fired=False)
+        self.assertTrue(result["complete"])
+
+
+class TestFullSuiteFailureApprovalGate(unittest.TestCase):
+    """Issue #323: full-suite failure blocks approval without baseline proof."""
+
+    ROOT = "/repo/Gitea-Tools"
+
+    def _good_proof(self, **overrides):
+        proof = {
+            "worktree_path": f"{self.ROOT}/branches/baseline-master-pr280",
+            "baseline_target_sha": PINNED,
+            "pr_head_sha": OTHER,
+            "baseline_failures": ["tests/test_foo.py::test_bar"],
+            "pr_failures": ["tests/test_foo.py::test_bar"],
+            "failure_signatures_match": True,
+            "clean_before": True,
+            "clean_after": True,
+            "pr_suite_command": "venv/bin/pytest tests/",
+            "baseline_suite_command": "venv/bin/pytest tests/",
+            "new_tests_passed": True,
+            "unrelated_explanation": (
+                "failures touch agent_temp_artifacts only; PR changes "
+                "review_proofs handoff fields"
+            ),
+        }
+        proof.update(overrides)
+        return proof
+
+    def _good_report(self):
+        return (
+            "Full suite failed on PR worktree and on baseline.\n"
+            "- Validation command: venv/bin/pytest tests/\n"
+            f"- Working directory: {self.ROOT}/branches/review-pr-280\n"
+            "Failures are pre-existing on master.\n"
+        )
+
+    def test_full_suite_pass_allows_approval_without_baseline(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="approve",
+            full_suite_passed=True,
+            project_root=self.ROOT,
+        )
+        self.assertTrue(result["approve_allowed"])
+        self.assertFalse(result["block"])
+        self.assertEqual(result["default_action"], "proceed")
+
+    def test_full_suite_failure_defaults_to_request_changes(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="request_changes",
+            full_suite_passed=False,
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["block"])
+        self.assertEqual(result["default_action"], "request_changes")
+
+    def test_approval_with_failure_and_no_proof_blocks(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="approve",
+            full_suite_passed=False,
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["approve_allowed"])
+        self.assertTrue(result["block"])
+        self.assertEqual(result["default_action"], "request_changes")
+
+    def test_approval_with_matching_baseline_proof_allowed(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="approve",
+            full_suite_passed=False,
+            baseline_proof=self._good_proof(),
+            report_text=self._good_report(),
+            project_root=self.ROOT,
+        )
+        self.assertTrue(result["approve_allowed"])
+        self.assertFalse(result["block"])
+
+    def test_mismatched_failure_signatures_block_approval(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="approve",
+            full_suite_passed=False,
+            baseline_proof=self._good_proof(
+                failure_signatures_match=False,
+                pr_failures=["tests/test_other.py::test_other"],
+            ),
+            report_text=self._good_report(),
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["approve_allowed"])
+        self.assertTrue(result["block"])
+
+    def test_main_checkout_baseline_blocks_approval(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="approve",
+            full_suite_passed=False,
+            baseline_proof=self._good_proof(worktree_path=self.ROOT),
+            report_text=self._good_report(),
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["approve_allowed"])
+        self.assertTrue(result["block"])
+
+    def test_missing_new_tests_proof_blocks_approval(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="approve",
+            full_suite_passed=False,
+            baseline_proof=self._good_proof(new_tests_passed=None),
+            report_text=self._good_report(),
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["approve_allowed"])
+        self.assertTrue(any(
+            "new" in reason.lower() and "test" in reason.lower()
+            for reason in result["reasons"]
+        ))
+
+    def test_missing_unrelated_explanation_blocks_approval(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="approve",
+            full_suite_passed=False,
+            baseline_proof=self._good_proof(unrelated_explanation=""),
+            report_text=self._good_report(),
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["approve_allowed"])
+
+    def test_missing_suite_commands_block_approval(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="approve",
+            full_suite_passed=False,
+            baseline_proof=self._good_proof(
+                pr_suite_command="", baseline_suite_command=""),
+            report_text=self._good_report(),
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["approve_allowed"])
+        self.assertTrue(any(
+            "command" in reason.lower() for reason in result["reasons"]
+        ))
+
+    def test_unknown_suite_result_blocks_approval(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="approve",
+            full_suite_passed=None,
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["approve_allowed"])
+        self.assertTrue(result["block"])
+
+    def test_vague_same_as_master_claim_without_proof_blocks(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="approve",
+            full_suite_passed=False,
+            report_text="Validation: failures same as master; approving.",
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["approve_allowed"])
+        self.assertTrue(result["block"])
+
+    def test_non_approve_decision_never_blocks(self):
+        result = assess_full_suite_failure_approval_gate(
+            review_decision="comment",
+            full_suite_passed=None,
+            project_root=self.ROOT,
+        )
+        self.assertFalse(result["block"])
+
+
+if __name__ == "__main__":
+    unittest.main()
+
 if __name__ == "__main__":
     unittest.main()

tests/test_reviewer_merge_simulation.py

@@ -0,0 +1,135 @@
+"""Tests for merge-simulation worktree mutation verifier (#317)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_merge_simulation import (  # noqa: E402
+    assess_merge_simulation_report,
+    merge_simulation_commands,
+)
+
+
+def _good_simulation_report(**overrides) -> str:
+    lines = [
+        "Worktree/index mutations: merge simulation in branches/review-pr281",
+        "Worktree path: branches/review-pr281",
+        "Pre-simulation clean status: clean (git status --porcelain empty)",
+        "Merge result: conflicts in review_proofs.py",
+        "Abort command: git merge --abort",
+        "Post-abort clean status: clean after abort (git status --porcelain empty)",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        text = text.replace(f"{key}:", f"{key}: {value}")
+    return text
+
+
+class TestMergeSimulationDetection(unittest.TestCase):
+    def test_detects_no_commit_merge(self):
+        commands = merge_simulation_commands([
+            "git merge prgs/master --no-commit --no-ff",
+            "git status",
+        ])
+        self.assertEqual(commands, ["git merge prgs/master --no-commit --no-ff"])
+
+    def test_detects_merge_abort(self):
+        commands = merge_simulation_commands([
+            "git merge prgs/master --no-commit --no-ff",
+            "git merge --abort",
+        ])
+        self.assertIn("git merge --abort", commands)
+
+
+class TestMergeSimulationReport(unittest.TestCase):
+    def test_no_simulation_passes(self):
+        result = assess_merge_simulation_report(
+            "Review complete. Worktree/index mutations: none",
+            command_log=["git status --porcelain", "git diff prgs/master...HEAD"],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_documented_simulation_passes(self):
+        report = _good_simulation_report()
+        result = assess_merge_simulation_report(
+            report,
+            command_log=[
+                "git merge prgs/master --no-commit --no-ff",
+                "git merge --abort",
+            ],
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_readonly_classification_blocks(self):
+        report = "\n".join([
+            "Read-only diagnostics: git merge prgs/master --no-commit --no-ff, git status",
+            "Worktree/index mutations: none",
+        ])
+        result = assess_merge_simulation_report(
+            report,
+            command_log=["git merge prgs/master --no-commit --no-ff"],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(any("read-only" in r.lower() for r in result["reasons"]))
+
+    def test_missing_worktree_field_blocks(self):
+        report = "Merge simulation ran but only noted in prose."
+        result = assess_merge_simulation_report(
+            report,
+            command_log=["git merge prgs/master --no-commit --no-ff"],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("worktree/index mutations" in r.lower() for r in result["reasons"]))
+
+    def test_conflict_simulation_requires_merge_result(self):
+        report = _good_simulation_report().replace(
+            "Merge result: conflicts in review_proofs.py",
+            "",
+        )
+        result = assess_merge_simulation_report(
+            report,
+            command_log=["git merge prgs/master --no-commit --no-ff"],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("merge result" in r.lower() for r in result["reasons"]))
+
+    def test_aborted_simulation_requires_post_clean_proof(self):
+        report = _good_simulation_report().replace(
+            "Post-abort clean status: clean after abort (git status --porcelain empty)",
+            "",
+        )
+        result = assess_merge_simulation_report(
+            report,
+            command_log=[
+                "git merge prgs/master --no-commit --no-ff",
+                "git merge --abort",
+            ],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("post-abort" in r.lower() for r in result["reasons"]))
+
+    def test_successful_simulation_without_abort_omits_abort_fields(self):
+        report = "\n".join([
+            "Worktree/index mutations: merge simulation completed cleanly",
+            "Worktree path: branches/review-pr281",
+            "Pre-simulation clean status: clean before simulation",
+            "Merge result: clean fast-forward simulation",
+        ])
+        result = assess_merge_simulation_report(
+            report,
+            command_log=["git merge prgs/master --no-commit --no-ff"],
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_merge_simulation_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_mutation_categories.py

@@ -0,0 +1,88 @@
+"""Tests for precise mutation category verifier (#319)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_mutation_categories import assess_mutation_categories_report  # noqa: E402
+
+
+def _precise_handoff() -> str:
+    return "\n".join([
+        "Controller Handoff",
+        "File edits by reviewer: none",
+        "Worktree/index mutations: git worktree add branches/review-pr1",
+        "Git ref mutations: git fetch prgs master",
+        "MCP/Gitea mutations: gitea_view_pr",
+        "Review mutations: gitea_review_pr request_changes",
+        "Read-only diagnostics: git status, git diff",
+    ])
+
+
+class TestMutationCategories(unittest.TestCase):
+    def test_precise_categories_pass(self):
+        result = assess_mutation_categories_report(_precise_handoff())
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_legacy_workspace_mutations_blocks(self):
+        report = "\n".join([
+            "Controller Handoff",
+            "Workspace mutations: none (no local file changes)",
+            "File edits by reviewer: none",
+            "Worktree/index mutations: none",
+            "Git ref mutations: none",
+        ])
+        result = assess_mutation_categories_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
+
+    def test_workspace_none_with_worktree_command_blocks(self):
+        report = "\n".join([
+            "Controller Handoff",
+            "Workspace mutations: none (no local file changes)",
+            "File edits by reviewer: none",
+            "Worktree/index mutations: none",
+            "Git ref mutations: none",
+        ])
+        result = assess_mutation_categories_report(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["proven"])
+
+    def test_file_edits_none_with_worktree_mutation_passes(self):
+        report = "\n".join([
+            "Controller Handoff",
+            "File edits by reviewer: none",
+            "Worktree/index mutations: git reset --hard FETCH_HEAD",
+            "Git ref mutations: git fetch prgs",
+        ])
+        result = assess_mutation_categories_report(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD", "git fetch prgs"],
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_missing_required_categories_blocks(self):
+        report = "Controller Handoff\nFile edits by reviewer: none\n"
+        result = assess_mutation_categories_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("missing" in r.lower() for r in result["reasons"]))
+
+    def test_non_controller_handoff_skips(self):
+        result = assess_mutation_categories_report(
+            "Workspace mutations: none\n",
+        )
+        self.assertTrue(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_mutation_categories_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_validation_integrity.py

@@ -0,0 +1,130 @@
+"""Tests for PR-head vs diagnostic validation integrity verifier (#316)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_validation_integrity import assess_validation_integrity_report  # noqa: E402
+
+
+def _official_only_report() -> str:
+    return "\n".join([
+        "Official PR-head validation on unmodified worktree.",
+        "Official validation command: pytest tests/ -q",
+        "Official validation result: failed (3 failed)",
+        "Validation integrity status: failed",
+        "Worktree dirty after official validation: clean",
+        "Review decision: request_changes",
+    ])
+
+
+def _diagnostic_report() -> str:
+    return "\n".join([
+        _official_only_report(),
+        "Diagnostic local experiment — not PR-head validation",
+        "Diagnostic edit path: tests/test_example.py",
+        "File edits by reviewer: tests/test_example.py",
+        "Diagnostic command: pytest tests/test_example.py -q",
+        "Diagnostic result: passed after local fix",
+        "Diagnostic results used only for suggested fix; blocker remains PR-head failure.",
+    ])
+
+
+class TestValidationIntegrity(unittest.TestCase):
+    def test_official_only_passes(self):
+        result = assess_validation_integrity_report(
+            _official_only_report(),
+            validation_session={
+                "official_validation_ran": True,
+                "official_result": "fail",
+                "worktree_dirty_after_official": False,
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_failing_pr_head_with_diagnostic_pass_passes(self):
+        result = assess_validation_integrity_report(
+            _diagnostic_report(),
+            validation_session={
+                "official_validation_ran": True,
+                "official_result": "fail",
+                "diagnostic_validation_ran": True,
+                "diagnostic_result": "pass",
+                "diagnostic_edits": ["tests/test_example.py"],
+                "worktree_dirty_after_official": False,
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_post_edit_tests_as_official_blocks(self):
+        report = "\n".join([
+            "Official validation result: passed after diagnostic edit",
+            "pytest passed after local fix in validation worktree",
+        ])
+        result = assess_validation_integrity_report(
+            report,
+            validation_session={
+                "official_validation_ran": True,
+                "diagnostic_edits": ["tests/test_example.py"],
+                "diagnostic_validation_ran": True,
+            },
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_missing_dirty_after_validation_blocks(self):
+        report = _official_only_report().replace(
+            "Worktree dirty after official validation: clean",
+            "",
+        )
+        result = assess_validation_integrity_report(
+            report,
+            validation_session={"official_validation_ran": True, "official_result": "fail"},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("dirty" in r.lower() for r in result["reasons"]))
+
+    def test_diagnostic_without_label_blocks(self):
+        report = _diagnostic_report().replace(
+            "Diagnostic local experiment — not PR-head validation",
+            "",
+        )
+        result = assess_validation_integrity_report(
+            report,
+            validation_session={
+                "official_validation_ran": True,
+                "diagnostic_edits": ["tests/test_example.py"],
+                "diagnostic_validation_ran": True,
+            },
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"]))
+
+    def test_contaminated_requires_integrity_status(self):
+        report = "Validation worktree was edited before official validation completed."
+        result = assess_validation_integrity_report(
+            report,
+            validation_session={"contaminated": True, "official_validation_ran": True},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("contaminated" in r.lower() for r in result["reasons"]))
+
+    def test_action_log_edits_trigger_diagnostic_rules(self):
+        result = assess_validation_integrity_report(
+            _official_only_report(),
+            validation_session={"official_validation_ran": True, "official_result": "fail"},
+            action_log=[{"path": "tests/test_example.py", "kind": "file_edit", "performed": True}],
+        )
+        self.assertFalse(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_validation_integrity_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_validation_worktree_edits.py

@@ -0,0 +1,134 @@
+"""Tests for PR validation worktree no-edit verifier (#315)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_validation_worktree_edits import (  # noqa: E402
+    assess_validation_worktree_edit_report,
+)
+
+
+def _read_only_report() -> str:
+    return "\n".join([
+        "Validation worktree path: branches/review-pr280-feat-issue-275-claim",
+        "Official PR-head validation on unmodified worktree.",
+        "Official validation result: failed (1 failed)",
+        "File edits by reviewer: none",
+        "Review decision: request_changes",
+    ])
+
+
+def _diagnostic_report() -> str:
+    return "\n".join([
+        "Validation worktree path: branches/review-pr280-feat-issue-275-claim",
+        "Official PR-head validation on unmodified worktree.",
+        "Official validation result: failed (1 failed)",
+        "Review decision: request_changes",
+        "Diagnostic local experiment — not PR-head validation",
+        "Diagnostic scratch worktree path: branches/diagnostic-pr280-fix-test",
+        "File edits by reviewer: tests/test_agent_temp_artifacts.py (diagnostic only)",
+        "Diagnostic result: passed after local fix",
+    ])
+
+
+class TestValidationWorktreeEdits(unittest.TestCase):
+    def test_read_only_review_passes(self):
+        result = assess_validation_worktree_edit_report(
+            _read_only_report(),
+            validation_session={
+                "validation_worktree_path": (
+                    "branches/review-pr280-feat-issue-275-claim"
+                ),
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_validation_worktree_edit_blocks(self):
+        result = assess_validation_worktree_edit_report(
+            _read_only_report(),
+            validation_session={
+                "validation_worktree_path": (
+                    "branches/review-pr280-feat-issue-275-claim"
+                ),
+                "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"],
+            },
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(any("validation worktree" in r.lower() for r in result["reasons"]))
+
+    def test_file_edits_none_contradiction_blocks(self):
+        result = assess_validation_worktree_edit_report(
+            _read_only_report(),
+            validation_session={
+                "validation_worktree_path": (
+                    "branches/review-pr280-feat-issue-275-claim"
+                ),
+            },
+            action_log=[
+                {
+                    "path": "tests/test_agent_temp_artifacts.py",
+                    "kind": "file_edit",
+                    "performed": True,
+                    "worktree_path": "branches/review-pr280-feat-issue-275-claim",
+                }
+            ],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("file edits" in r.lower() for r in result["reasons"]))
+
+    def test_diagnostic_scratch_with_reporting_passes(self):
+        result = assess_validation_worktree_edit_report(
+            _diagnostic_report(),
+            validation_session={
+                "validation_worktree_path": (
+                    "branches/review-pr280-feat-issue-275-claim"
+                ),
+                "diagnostic_worktree_path": "branches/diagnostic-pr280-fix-test",
+                "diagnostic_edits": ["tests/test_agent_temp_artifacts.py"],
+                "diagnostic_experiment": True,
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_diagnostic_without_scratch_worktree_blocks(self):
+        result = assess_validation_worktree_edit_report(
+            "File edits by reviewer: tests/test_example.py",
+            validation_session={
+                "diagnostic_edits": ["tests/test_example.py"],
+                "diagnostic_experiment": True,
+            },
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"]))
+
+    def test_post_edit_official_validation_blocks(self):
+        report = "\n".join([
+            "Validation worktree path: branches/review-pr280-feat-issue-275-claim",
+            "Official validation result: passed after local edit",
+            "File edits by reviewer: tests/test_agent_temp_artifacts.py",
+        ])
+        result = assess_validation_worktree_edit_report(
+            report,
+            validation_session={
+                "validation_worktree_path": (
+                    "branches/review-pr280-feat-issue-275-claim"
+                ),
+                "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"],
+                "official_validation_after_edit": True,
+            },
+        )
+        self.assertFalse(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_validation_worktree_edit_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_worktree_ownership.py

@@ -0,0 +1,137 @@
+"""Tests for reviewer worktree ownership verifier (#312)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_worktree_ownership import assess_worktree_ownership_report  # noqa: E402
+
+
+def _fresh_review_report() -> str:
+    return "\n".join([
+        "Review worktree path: branches/review-pr280-feat-issue-275-claim",
+        "Worktree is inside branches/ and not the main checkout.",
+        "Tracked state: clean; untracked state: clean.",
+        "Official PR-head validation on session-owned worktree.",
+        "Worktree mutations: none",
+    ])
+
+
+def _safe_reuse_report() -> str:
+    return "\n".join([
+        "Safe-reuse proof for existing review worktree.",
+        "Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks",
+        "Worktree inside branches/; not the main checkout.",
+        "Not owned by another active task/session.",
+        "Clean tracked state; clean untracked state.",
+        "Branch/head before reset: feat/issue-260-example",
+        "Reset target SHA: abc123def4567890abcdef1234567890abcdef12",
+        "Project policy allows reuse/reset for this clean review worktree.",
+        "Worktree mutations: git reset --hard FETCH_HEAD",
+        "Destructive reset operations: git reset --hard FETCH_HEAD",
+    ])
+
+
+class TestWorktreeOwnership(unittest.TestCase):
+    def test_fresh_session_owned_passes(self):
+        result = assess_worktree_ownership_report(
+            _fresh_review_report(),
+            ownership_session={
+                "validation_ran": True,
+                "worktree_path": "branches/review-pr280-feat-issue-275-claim",
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["session_owned"])
+
+    def test_safe_reuse_with_reset_passes(self):
+        result = assess_worktree_ownership_report(
+            _safe_reuse_report(),
+            ownership_session={
+                "safe_reuse": True,
+                "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
+            },
+            command_log=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_destructive_reset_without_proof_blocks(self):
+        report = "\n".join([
+            "Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks",
+            "Worktree mutations: git reset --hard FETCH_HEAD",
+        ])
+        result = assess_worktree_ownership_report(
+            report,
+            ownership_session={
+                "validation_ran": True,
+                "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
+            },
+            command_log=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_reused_dirty_tracked_blocks(self):
+        result = assess_worktree_ownership_report(
+            _fresh_review_report(),
+            ownership_session={
+                "validation_ran": True,
+                "worktree_path": "branches/review-pr280-feat-issue-275-claim",
+                "dirty_tracked": True,
+            },
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("tracked" in r.lower() for r in result["reasons"]))
+
+    def test_safe_reuse_dirty_untracked_blocks(self):
+        result = assess_worktree_ownership_report(
+            _safe_reuse_report(),
+            ownership_session={
+                "safe_reuse": True,
+                "dirty_untracked": True,
+                "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
+            },
+            command_log=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("untracked" in r.lower() for r in result["reasons"]))
+
+    def test_workspace_none_with_reset_blocks(self):
+        report = "\n".join([
+            "Review worktree path: branches/review-pr280-feat-issue-275-claim",
+            "Workspace mutations: none",
+            "Worktree mutations: git reset --hard FETCH_HEAD",
+        ])
+        result = assess_worktree_ownership_report(
+            report,
+            ownership_session={
+                "session_owned": True,
+                "worktree_path": "branches/review-pr280-feat-issue-275-claim",
+            },
+            command_log=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
+
+    def test_main_checkout_blocks(self):
+        result = assess_worktree_ownership_report(
+            "Review worktree path: /Users/dev/Gitea-Tools",
+            ownership_session={
+                "validation_ran": True,
+                "main_checkout": True,
+                "worktree_path": "/Users/dev/Gitea-Tools",
+            },
+        )
+        self.assertFalse(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_worktree_ownership_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()