fix: resolve conflicts for PR #366 against current master

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)
@@ -3595,6 +3666,113 @@ def assess_reviewer_baseline_validation_proof(
     }
 
 
+# ---------------------------------------------------------------------------
+# 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)
 # ---------------------------------------------------------------------------
@@ -4073,6 +4251,13 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
     }
 
 
+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
+
+    return _assess(report_text, **kwargs)
+
+
 def assess_non_mergeable_skip_proof(report_text, **kwargs):
     """#322: require conflict proof when skipping non-mergeable PRs."""
     from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess
@@ -4085,3 +4270,10 @@ def assess_inventory_worktree_report(report_text, **kwargs):
     from reviewer_inventory_worktree import assess_inventory_worktree_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_blocker_skip.py

@@ -0,0 +1,164 @@
+"""Prior-blocker skip proof verifier for reviewer queue reports (#318)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_FULL_SHA = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
+_SHORT_SHA = re.compile(r"\b[0-9a-f]{7,39}\b", re.IGNORECASE)
+_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE)
+_BLOCKING_DECISION_RE = re.compile(
+    r"(?:blocking review decision|review decision|blocking decision)\s*:\s*request[_ ]changes|"
+    r"request[_ ]changes",
+    re.IGNORECASE,
+)
+_BLOCKING_HEAD_RE = re.compile(
+    r"(?:blocking review head(?:\s+sha)?|blocker head(?:\s+sha)?|review head at blocker)",
+    re.IGNORECASE,
+)
+_HEAD_CHANGED_RE = re.compile(
+    r"(?:head (?:changed|unchanged)|head sha (?:changed|unchanged)|"
+    r"head changed (?:after|since) (?:the )?blocker|head unchanged since blocker)",
+    re.IGNORECASE,
+)
+_BLOCKER_REASON_RE = re.compile(
+    r"(?:reason (?:it )?remains blocked|remains blocked because|blocking category)",
+    re.IGNORECASE,
+)
+_LIVE_BLOCKER_PROOF_RE = re.compile(
+    r"(?:blocker revalidated live|live proof|gitea_get_pr_review_feedback|"
+    r"gitea_view_pr|review feedback fetched|current review state)",
+    re.IGNORECASE,
+)
+_BLOCKER_UNVERIFIED_RE = re.compile(r"\bBLOCKER_STATUS_UNVERIFIED\b")
+
+
+def _pr_section(text: str, pr_number: int) -> str:
+    lines = (text or "").splitlines()
+    chunks: list[str] = []
+    capture = False
+    token = f"#{pr_number}"
+    for line in lines:
+        lower = line.lower()
+        if token in lower or f"pr {pr_number}" in lower or f"pr#{pr_number}" in lower.replace(" ", ""):
+            capture = True
+            chunks.append(line)
+            continue
+        if capture:
+            if _PR_NUMBER_RE.search(line) and token not in line.lower():
+                break
+            if line.strip() == "" and len(chunks) > 3:
+                break
+            chunks.append(line)
+    if chunks:
+        return "\n".join(chunks)
+    return text or ""
+
+
+def _has_head_sha(text: str, head_sha: str | None) -> bool:
+    if not head_sha:
+        return bool(_FULL_SHA.search(text) or _SHORT_SHA.search(text))
+    head = head_sha.strip().lower()
+    if head in text.lower():
+        return True
+    if len(head) >= 7 and head[:7] in text.lower():
+        return True
+    return bool(_FULL_SHA.search(text))
+
+
+def assess_prior_blocker_skip_proof(
+    report_text: str,
+    *,
+    skipped_prs: list[dict] | None = None,
+) -> dict[str, Any]:
+    """Validate live blocker proof for skipped earlier open PRs (#318)."""
+    text = report_text or ""
+    reasons: list[str] = []
+    assessments: list[dict[str, Any]] = []
+
+    for entry in skipped_prs or []:
+        pr_number = entry.get("pr_number")
+        if pr_number is None:
+            reasons.append("skipped PR entry missing pr_number")
+            continue
+        pr_number = int(pr_number)
+        section = _pr_section(text, pr_number)
+        verified = entry.get("blocker_verified")
+        head_changed = entry.get("head_changed_since_blocker")
+        blocking_head = (entry.get("blocking_review_head_sha") or "").strip() or None
+        current_head = (entry.get("head_sha") or "").strip() or None
+        skip_reason = (entry.get("skip_reason") or entry.get("blocking_decision") or "").lower()
+
+        item_reasons: list[str] = []
+
+        if head_changed is True:
+            item_reasons.append(
+                f"PR #{pr_number} head changed after blocker; it cannot be skipped "
+                "based on stale REQUEST_CHANGES"
+            )
+
+        if f"#{pr_number}" not in text.lower() and f"pr {pr_number}" not in text.lower():
+            item_reasons.append(f"skipped PR #{pr_number} not documented in final report")
+
+        if "request_changes" in skip_reason or entry.get("blocking_decision") == "request_changes":
+            if verified is False:
+                if not _BLOCKER_UNVERIFIED_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} blocker proof unavailable; report must classify "
+                        "BLOCKER_STATUS_UNVERIFIED"
+                    )
+            else:
+                if not _BLOCKING_DECISION_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing blocking review decision proof"
+                    )
+                if not _has_head_sha(section, current_head):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing current head SHA"
+                    )
+                if blocking_head and not _BLOCKING_HEAD_RE.search(section):
+                    if blocking_head.lower() not in section.lower():
+                        item_reasons.append(
+                            f"PR #{pr_number} skip missing blocking review head SHA"
+                        )
+                if head_changed is False and not _HEAD_CHANGED_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing head-changed-since-blocker proof"
+                    )
+                if not _BLOCKER_REASON_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing reason-it-remains-blocked"
+                    )
+                if not _LIVE_BLOCKER_PROOF_RE.search(section):
+                    item_reasons.append(
+                        f"PR #{pr_number} skip missing live blocker proof for this session"
+                    )
+
+        proven = not item_reasons
+        assessments.append({
+            "pr_number": pr_number,
+            "proven": proven,
+            "classification": (
+                "BLOCKER_STATUS_UNVERIFIED"
+                if verified is False
+                else "BLOCKED_SKIPPED"
+            ),
+            "reasons": item_reasons,
+        })
+        reasons.extend(item_reasons)
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "downgraded": False,
+        "assessments": assessments,
+        "reasons": reasons,
+        "safe_next_action": (
+            "fetch live review feedback and document blocker proof for each skipped PR, "
+            "or classify BLOCKER_STATUS_UNVERIFIED"
+            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"
+        ),
+    }

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_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,12 @@ def test_create_issue_final_report_verifier_exported():
     assert callable(assess_create_issue_final_report)
 
 
+def test_prior_blocker_skip_verifier_exported():
+    from review_proofs import assess_prior_blocker_skip_proof
+
+    assert callable(assess_prior_blocker_skip_proof)
+
+
 def test_non_mergeable_skip_verifier_exported():
     from review_proofs import assess_non_mergeable_skip_proof
 
@@ -146,3 +152,9 @@ def test_inventory_worktree_verifier_exported():
     from review_proofs import assess_inventory_worktree_report
 
     assert callable(assess_inventory_worktree_report)
+
+
+def test_worktree_ownership_verifier_exported():
+    from review_proofs import assess_worktree_ownership_report
+
+    assert callable(assess_worktree_ownership_report)

tests/test_review_proofs.py

@@ -24,6 +24,7 @@ from review_proofs import (  # noqa: E402
     ISSUE_SELECTION_CONTINUATION_EXPLICIT,
     ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
     assess_author_pr_report,
+    assess_full_suite_failure_approval_gate,
     assess_queue_ordering_proof,
     assess_reviewer_baseline_validation_proof,
     assess_capability_evidence,
@@ -2539,5 +2540,170 @@ class TestReviewerBaselineValidationProof(unittest.TestCase):
         self.assertFalse(report["baseline_validation_proven"])
 
 
+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()

tests/test_reviewer_blocker_skip.py

@@ -0,0 +1,139 @@
+"""Tests for prior-blocker skip proof verifier (#318)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_blocker_skip import assess_prior_blocker_skip_proof  # noqa: E402
+
+HEAD_CURRENT = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+HEAD_BLOCKER = "1111111111111111111111111111111111111111"
+
+
+def _good_blocker_skip_report(pr_number: int) -> str:
+    return "\n".join([
+        f"Skipped PR #{pr_number} due to prior REQUEST_CHANGES.",
+        f"- PR number: #{pr_number}",
+        f"- Current head SHA: {HEAD_CURRENT}",
+        "- Blocking review decision: request_changes",
+        f"- Blocking review head SHA: {HEAD_CURRENT}",
+        "- Head unchanged since blocker",
+        "- Reason remains blocked: unresolved review feedback at same head",
+        "- Blocker revalidated live in this session via gitea_get_pr_review_feedback",
+    ])
+
+
+class TestPriorBlockerSkipProof(unittest.TestCase):
+    def test_no_skips_passes(self):
+        result = assess_prior_blocker_skip_proof(
+            "Selected PR #281 for review.",
+            skipped_prs=[],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_blocked_unchanged_with_live_proof_passes(self):
+        report = _good_blocker_skip_report(280)
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_CURRENT,
+                "head_changed_since_blocker": False,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_head_changed_after_blocker_blocks_skip(self):
+        result = assess_prior_blocker_skip_proof(
+            _good_blocker_skip_report(280),
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_BLOCKER,
+                "head_changed_since_blocker": True,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("head changed" in r for r in result["reasons"]))
+
+    def test_missing_live_proof_blocks(self):
+        report = _good_blocker_skip_report(280).replace(
+            "- Blocker revalidated live in this session via gitea_get_pr_review_feedback",
+            "",
+        )
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_CURRENT,
+                "head_changed_since_blocker": False,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("live blocker proof" in r for r in result["reasons"]))
+
+    def test_blocker_unverified_classification_passes(self):
+        report = "\n".join([
+            "Skipped PR #280.",
+            "Classification: BLOCKER_STATUS_UNVERIFIED",
+            "Review feedback could not be fetched live in this session.",
+        ])
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "blocking_decision": "request_changes",
+                "blocker_verified": False,
+            }],
+        )
+        self.assertTrue(result["proven"])
+
+    def test_unverified_without_classification_blocks(self):
+        result = assess_prior_blocker_skip_proof(
+            "Skipped PR #280 based on memory from last session.",
+            skipped_prs=[{
+                "pr_number": 280,
+                "blocking_decision": "request_changes",
+                "blocker_verified": False,
+            }],
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("BLOCKER_STATUS_UNVERIFIED" in r for r in result["reasons"]))
+
+    def test_missing_blocking_head_sha_blocks(self):
+        report = _good_blocker_skip_report(280).replace(
+            f"- Blocking review head SHA: {HEAD_CURRENT}",
+            "",
+        )
+        result = assess_prior_blocker_skip_proof(
+            report,
+            skipped_prs=[{
+                "pr_number": 280,
+                "head_sha": HEAD_CURRENT,
+                "blocking_decision": "request_changes",
+                "blocking_review_head_sha": HEAD_CURRENT,
+                "head_changed_since_blocker": False,
+                "blocker_verified": True,
+            }],
+        )
+        self.assertFalse(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_prior_blocker_skip_proof 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()