final_report_validator.py

@@ -87,14 +87,34 @@ _ALREADY_LANDED_RE = re.compile(
     r"already[- ]landed|ancestor proof\s*:\s*passed|eligibility class\s*:\s*already_landed",
     re.IGNORECASE,
 )
+_ALREADY_LANDED_GATE_FIRED_RE = re.compile(
+    r"\b(?:fired|passed|true|yes|already_landed_reconcile_required)\b",
+    re.IGNORECASE,
+)
 _ELIGIBLE_REVIEW_RE = re.compile(
     r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr",
     re.IGNORECASE,
 )
+_APPROVED_STATE_RE = re.compile(r"\bapproved?\b", re.IGNORECASE)
+_MERGED_STATE_RE = re.compile(r"\bmerged\b", re.IGNORECASE)
+_READY_TO_MERGE_STATE_RE = re.compile(r"\bready_to_merge\b|\bready to merge\b", re.IGNORECASE)
+_NEGATED_STATE_RE = re.compile(
+    r"\b(?:not|no|none|n/a|never)\b|request_changes|not attempted",
+    re.IGNORECASE,
+)
 _REVIEWED_LANDED_RE = re.compile(
     r"(?:reviewed|approved).{0,40}already[- ]landed|already[- ]landed.{0,40}(?:reviewed|eligible)",
     re.IGNORECASE | re.DOTALL,
 )
+_TARGET_BRANCH_UP_TO_DATE_RE = re.compile(
+    r"\btarget branch (?:is )?up[- ]to[- ]date\b",
+    re.IGNORECASE,
+)
+_TARGET_BRANCH_SHA_RE = re.compile(
+    r"target branch sha\s*:\s*[0-9a-f]{40}",
+    re.IGNORECASE,
+)
+_FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
 _RECONCILE_STALE_FIELDS = (
     "pr number opened",
     "pinned reviewed head",
@@ -190,6 +210,54 @@ def _handoff_fields(report_text: str) -> dict[str, str]:
     return fields
 
 
+def _already_landed_gate_fired(report_text: str) -> bool:
+    text = report_text or ""
+    fields = _handoff_fields(text)
+
+    gate_value = fields.get("already-landed gate", "")
+    if gate_value and _ALREADY_LANDED_GATE_FIRED_RE.search(gate_value):
+        return True
+
+    ancestor_value = fields.get("ancestor proof", "")
+    if re.search(r"\bpassed\b", ancestor_value, re.IGNORECASE):
+        return True
+
+    eligibility_value = fields.get("eligibility class", "")
+    if "already_landed" in eligibility_value.lower():
+        return True
+
+    if re.search(
+        r"\bpr\b.{0,60}\balready[- ]landed\b|\balready[- ]landed\b.{0,60}\bpr\b",
+        text,
+        re.IGNORECASE | re.DOTALL,
+    ):
+        return True
+
+    return bool(_ALREADY_LANDED_RE.search(text))
+
+
+def _positive_review_state_terms(fields: dict[str, str]) -> list[str]:
+    terms: list[str] = []
+    for key, value in fields.items():
+        lowered_key = key.lower()
+        lowered_value = value.lower()
+        if _NEGATED_STATE_RE.search(value):
+            continue
+        if lowered_key == "review decision" and _APPROVED_STATE_RE.search(value):
+            terms.append("APPROVED")
+        elif lowered_key == "merge result" and _MERGED_STATE_RE.search(value):
+            terms.append("MERGED")
+        elif _READY_TO_MERGE_STATE_RE.search(value):
+            terms.append("READY_TO_MERGE")
+        elif lowered_value in {"approved", "approve"}:
+            terms.append("APPROVED")
+        elif lowered_value == "merged":
+            terms.append("MERGED")
+        elif lowered_value == "ready_to_merge":
+            terms.append("READY_TO_MERGE")
+    return sorted(set(terms))
+
+
 def _rule_shared_controller_handoff(
     report_text: str,
     task_kind: str,
@@ -492,7 +560,7 @@ def _rule_reviewer_main_checkout_path(report_text: str) -> list[dict[str, str]]:
 
 def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, str]]:
     text = report_text or ""
-    if not _ALREADY_LANDED_RE.search(text):
+    if not _already_landed_gate_fired(text):
         return []
     if not _ELIGIBLE_REVIEW_RE.search(text):
         return []
@@ -507,6 +575,65 @@ def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, s
     ]
 
 
+def _rule_reviewer_already_landed_state(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _already_landed_gate_fired(text):
+        return []
+
+    terms = _positive_review_state_terms(_handoff_fields(text))
+    if not terms:
+        return []
+
+    return [
+        validator_finding(
+            "reviewer.already_landed_review_state",
+            "block",
+            "Review decision",
+            "already-landed gate fired but final report claims "
+            f"{', '.join(terms)} state",
+            "stop normal review/merge; classify as ALREADY_LANDED_RECONCILE_REQUIRED",
+        )
+    ]
+
+
+def _rule_reviewer_target_branch_freshness(
+    report_text: str,
+    *,
+    action_log: list[dict] | None = None,
+) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _TARGET_BRANCH_UP_TO_DATE_RE.search(text):
+        return []
+
+    fields = _handoff_fields(text)
+    fetch_reported = bool(_GIT_FETCH_RE.search(text)) or any(
+        _GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
+        for e in (action_log or [])
+    )
+    target_sha_reported = bool(_TARGET_BRANCH_SHA_RE.search(text)) or any(
+        "target branch" in key and "sha" in key and _FULL_SHA_RE.search(value)
+        for key, value in fields.items()
+    )
+    if fetch_reported and target_sha_reported:
+        return []
+
+    missing = []
+    if not fetch_reported:
+        missing.append("fresh git fetch")
+    if not target_sha_reported:
+        missing.append("target branch SHA")
+    return [
+        validator_finding(
+            "reviewer.target_branch_freshness_proof",
+            "block",
+            "Target branch SHA",
+            "target branch up-to-date claim lacks " + " and ".join(missing),
+            "fetch the target branch and report the exact target branch SHA before "
+            "claiming it is up to date",
+        )
+    ]
+
+
 def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]:
     from review_proofs import _handoff_section_lines
 
@@ -706,6 +833,8 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_main_checkout_baseline,
         _rule_reviewer_main_checkout_path,
         _rule_reviewer_already_landed_eligible,
+        _rule_reviewer_already_landed_state,
+        _rule_reviewer_target_branch_freshness,
         _rule_reviewer_mutation_ledger,
         _rule_reviewer_review_mutation,
     ],
@@ -875,4 +1004,4 @@ def assess_final_report_validator(
         "safe_next_action": _primary_safe_next_action(findings),
         "complete": grade == "A",
         "handoff_heading": HANDOFF_HEADING,
-    }
+    }

review_proofs.py

@@ -1715,6 +1715,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         if report_text and _RECONCILE_INVENTORY_HINT.search(report_text)
         else {"proven": True, "block": False, "reasons": []}
     )
+    reconcile_linked_issue = (
+        assess_reconcile_linked_issue_report(report_text)
+        if report_text and _RECONCILE_LINKED_ISSUE_HINT.search(report_text)
+        else {"proven": True, "block": False, "reasons": []}
+    )
     already_landed_handoff = (
         assess_already_landed_handoff_report(report_text)
         if report_text
@@ -1891,6 +1896,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "reconciliation PR inventory lacks pagination proof (#308)"
         )
         downgrade_reasons.extend(reconcile_inventory.get("reasons", []))
+    if not reconcile_linked_issue.get("proven"):
+        downgrade_reasons.append(
+            "reconciliation linked issue status lacks live fetch proof (#300)"
+        )
+        downgrade_reasons.extend(reconcile_linked_issue.get("reasons", []))
     if not pr_queue_cleanup_report.get("proven"):
         downgrade_reasons.append(
             "PR-only cleanup report violates cleanup-mode proof gates (#390)"
@@ -2003,6 +2013,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "reconcile_inventory_violations": list(
             reconcile_inventory.get("reasons") or []
         ),
+        "reconcile_linked_issue_proven": bool(reconcile_linked_issue.get("proven")),
+        "reconcile_linked_issue_violations": list(
+            reconcile_linked_issue.get("reasons") or []
+        ),
         "pr_queue_cleanup_report_proven": bool(
             pr_queue_cleanup_report.get("proven")
         ),
@@ -4907,6 +4921,11 @@ _RECONCILE_INVENTORY_HINT = re.compile(
     re.I,
 )
 
+_RECONCILE_LINKED_ISSUE_HINT = re.compile(
+    r"already[- ]landed|reconcil|linked issue(?:\s+live)?\s+status",
+    re.I,
+)
+
 _PR_QUEUE_CLEANUP_REPORT_HINT = re.compile(
     r"workflows/pr-queue-cleanup\.md|task:\s*pr-queue-cleanup|"
     r"pr[- ]only queue cleanup",
@@ -4920,6 +4939,7 @@ def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
 
     return _assess(report_text or "")
 
+
 _GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
 
 _NOT_APPLICABLE_VALUE = re.compile(
@@ -5341,6 +5361,13 @@ def assess_reconcile_inventory_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
+def assess_reconcile_linked_issue_report(report_text, **kwargs):
+    """#300: prove linked issue status was fetched live in reconciliation handoffs."""
+    from reviewer_reconcile_linked_issue import assess_reconcile_linked_issue_report as _assess
+
+    return _assess(report_text, **kwargs)
+
+
 def assess_already_landed_handoff_report(report_text, **kwargs):
     """#299: reject stale fields after the already-landed gate fires."""
     from reviewer_already_landed_handoff import assess_already_landed_handoff_report as _assess

reviewer_reconcile_linked_issue.py

@@ -0,0 +1,191 @@
+"""Reconciliation linked-issue live proof verifier (#300)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_HANDOFF_SECTION_RE = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M)
+
+_LINKED_ISSUE_FIELD_RE = re.compile(
+    r"linked issue\s*:\s*(.+)$",
+    re.I | re.M,
+)
+_LINKED_ISSUE_STATUS_RE = re.compile(
+    r"linked issue(?:\s+live)?\s+status\s*:\s*(.+)$",
+    re.I | re.M,
+)
+
+_STATUS_CLAIM_RE = re.compile(
+    r"\b(?:issue\s+#?\d+\s*\()?(open|closed|resolved)\b|"
+    r"issue\s+(?:is\s+)?(open|closed|resolved)\b|"
+    r"linked issue.*\b(open|closed|resolved)\b",
+    re.I,
+)
+_NOT_VERIFIED_RE = re.compile(r"not verified in this session", re.I)
+
+_LIVE_FETCH_PROOF_RE = re.compile(
+    r"gitea_view_issue|issue fetched live|live issue fetch|"
+    r"fetched linked issue.*(?:current|this) session|"
+    r"linked issue (?:fetch|proof).*(?:current|this) session|"
+    r"live linked issue proof",
+    re.I,
+)
+
+_ISSUE_ACTION_RECOMMEND_RE = re.compile(
+    r"(?:recommend(?:ed)?|should|must)\s+(?:close|reopen|comment on)\s+"
+    r"(?:the\s+)?(?:linked\s+)?issue|"
+    r"(?:close|reopen|comment on)\s+linked issue\s+#?\d+",
+    re.I,
+)
+_CAPABILITY_PROOF_RE = re.compile(
+    r"capability proof|exact capability|gitea_close_issue|"
+    r"gitea_mark_issue|gitea_create_issue_comment|gitea_edit_issue",
+    re.I,
+)
+
+_NO_LINKED_ISSUE_VALUES = frozenset({
+    "",
+    "none",
+    "n/a",
+    "not applicable",
+    "no linked issue",
+    "unknown",
+})
+
+
+def _handoff_field_map(report_text: str) -> dict[str, str]:
+    text = report_text or ""
+    match = _HANDOFF_SECTION_RE.search(text)
+    if not match:
+        return {}
+    fields: dict[str, str] = {}
+    for line in text[match.end() :].splitlines():
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" not in stripped:
+            continue
+        key, value = stripped.split(":", 1)
+        fields[key.strip().lower()] = value.strip()
+    return fields
+
+
+def _extract_linked_issue_number(text: str, fields: dict[str, str]) -> int | None:
+    linked = fields.get("linked issue", "")
+    if linked.lower() in _NO_LINKED_ISSUE_VALUES:
+        return None
+    match = re.search(r"#?(\d+)", linked)
+    if match:
+        return int(match.group(1))
+    field_match = _LINKED_ISSUE_FIELD_RE.search(text)
+    if field_match:
+        num = re.search(r"#?(\d+)", field_match.group(1))
+        if num:
+            return int(num.group(1))
+    return None
+
+
+def _status_value(text: str, fields: dict[str, str]) -> str:
+    for key in ("linked issue live status", "linked issue status"):
+        if fields.get(key):
+            return fields[key]
+    match = _LINKED_ISSUE_STATUS_RE.search(text)
+    return match.group(1).strip() if match else ""
+
+
+def _live_proof_present(text: str, session: dict, issue_number: int | None) -> bool:
+    if session.get("live_fetched") or session.get("verified_live"):
+        return True
+    if session.get("issue_fetch_proof"):
+        return True
+    if _LIVE_FETCH_PROOF_RE.search(text):
+        return True
+    if issue_number is not None:
+        pattern = re.compile(
+            rf"gitea_view_issue.*#?{issue_number}|"
+            rf"issue\s+#?{issue_number}.*(?:fetched|viewed).*(?:current|this) session",
+            re.I,
+        )
+        if pattern.search(text):
+            return True
+    return False
+
+
+def assess_reconcile_linked_issue_report(
+    report_text: str,
+    *,
+    reconcile_session: dict | None = None,
+) -> dict[str, Any]:
+    """Validate linked issue status claims in reconciliation handoffs (#300)."""
+    text = report_text or ""
+    session = dict(reconcile_session or {})
+    fields = _handoff_field_map(text)
+    reasons: list[str] = []
+
+    issue_number = session.get("linked_issue_number")
+    if issue_number is None:
+        issue_number = _extract_linked_issue_number(text, fields)
+
+    if issue_number is None:
+        status = _status_value(text, fields)
+        if status and status.lower() not in _NO_LINKED_ISSUE_VALUES:
+            if _STATUS_CLAIM_RE.search(status):
+                reasons.append(
+                    "linked issue status reported without identifying the linked issue (#300)"
+                )
+        if not reasons:
+            return {
+                "proven": True,
+                "block": False,
+                "reasons": [],
+                "linked_issue_number": None,
+                "safe_next_action": "proceed",
+            }
+
+    status = _status_value(text, fields)
+    status_lower = status.lower()
+
+    if session.get("issue_missing"):
+        if status_lower and "not verified" not in status_lower:
+            reasons.append(
+                "missing linked issue must be reported as "
+                "'not verified in this session' (#300)"
+            )
+    elif status:
+        if _NOT_VERIFIED_RE.search(status):
+            pass
+        elif _STATUS_CLAIM_RE.search(status):
+            if not _live_proof_present(text, session, issue_number):
+                reasons.append(
+                    "linked issue open/closed/resolved status requires live "
+                    "gitea_view_issue proof in the current session (#300)"
+                )
+        elif status_lower not in _NO_LINKED_ISSUE_VALUES:
+            if not _live_proof_present(text, session, issue_number):
+                reasons.append(
+                    "linked issue status claim requires live fetch proof or "
+                    "'not verified in this session' (#300)"
+                )
+
+    if _ISSUE_ACTION_RECOMMEND_RE.search(text):
+        if not _CAPABILITY_PROOF_RE.search(text) and not session.get(
+            "issue_action_capability_proven"
+        ):
+            reasons.append(
+                "recommended linked issue close/reopen/comment requires "
+                "exact capability proof (#300)"
+            )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": list(dict.fromkeys(reasons)),
+        "linked_issue_number": issue_number,
+        "live_proof_present": _live_proof_present(text, session, issue_number),
+        "safe_next_action": (
+            "fetch linked issue with gitea_view_issue in this session or "
+            "report 'Linked issue status: not verified in this session'"
+            if reasons
+            else "proceed"
+        ),
+    }

tests/test_final_report_validator.py

@@ -25,11 +25,15 @@ def _review_handoff(**overrides):
         "- Files changed: review_proofs.py",
         "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
         "- Mutations: review only",
-        "- Workspace mutations: none",
         "- File edits by reviewer: none",
         "- Worktree/index mutations: none",
         "- Git ref mutations: git fetch prgs master",
         "- MCP/Gitea mutations: review submitted",
+        "- Review mutations: submitted request_changes review",
+        "- Merge mutations: none",
+        "- Cleanup mutations: none",
+        "- External-state mutations: none",
+        "- Read-only diagnostics: issue and PR metadata inspected",
         "- Current status: PR open",
         "- Blockers: none",
         "- Next: await author",
@@ -190,11 +194,63 @@ class TestReviewPrRules(unittest.TestCase):
             )
         )
 
+    def test_already_landed_gate_blocks_review_terminal_states(self):
+        cases = {
+            "APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"),
+            "MERGED": ("- Merge result: not attempted", "- Merge result: MERGED"),
+            "READY_TO_MERGE": ("- Current status: PR open", "- Current status: READY_TO_MERGE"),
+        }
+        for state, (old, new) in cases.items():
+            with self.subTest(state=state):
+                report = (
+                    _review_handoff()
+                    .replace("- Reviewer eligibility: eligible", "- Reviewer eligibility: blocked")
+                    .replace(old, new)
+                    + "\n- Already-landed gate: fired"
+                    + "\n- Ancestor proof: passed"
+                    + "\n- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED"
+                )
+                result = assess_final_report_validator(report, "review_pr")
+                self.assertTrue(result["blocked"])
+                self.assertTrue(
+                    any(
+                        f["rule_id"] == "reviewer.already_landed_review_state"
+                        for f in result["findings"]
+                    )
+                )
+
+    def test_target_branch_up_to_date_requires_fetch_and_sha(self):
+        report = (
+            _review_handoff()
+            .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
+            + "\nTarget branch up to date."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.target_branch_freshness_proof"
+                for f in result["findings"]
+            )
+        )
+
+    def test_target_branch_up_to_date_with_fetch_and_sha_passes(self):
+        report = (
+            _review_handoff()
+            + "\n- Target branch SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+            + "\nTarget branch up to date."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertEqual(result["grade"], "A")
+
     def test_git_fetch_must_not_be_readonly_only(self):
         report = (
             _review_handoff()
             .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
-            + "\n- Read-only diagnostics: git fetch prgs master"
+            .replace(
+                "- Read-only diagnostics: issue and PR metadata inspected",
+                "- Read-only diagnostics: git fetch prgs master",
+            )
         )
         result = assess_final_report_validator(
             report,
@@ -270,4 +326,4 @@ class TestEntryPoint(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()

tests/test_llm_workflow_split.py

@@ -207,6 +207,12 @@ def test_reconcile_inventory_verifier_exported():
     assert callable(assess_reconcile_inventory_report)
 
 
+def test_reconcile_linked_issue_verifier_exported():
+    from review_proofs import assess_reconcile_linked_issue_report
+
+    assert callable(assess_reconcile_linked_issue_report)
+
+
 def test_already_landed_handoff_verifier_exported():
     from review_proofs import assess_already_landed_handoff_report
 

tests/test_reviewer_reconcile_linked_issue.py

@@ -0,0 +1,117 @@
+"""Tests for reconciliation linked-issue live proof verifier (#300)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_reconcile_linked_issue import assess_reconcile_linked_issue_report  # noqa: E402
+
+
+def _good_handoff() -> str:
+    return "\n".join([
+        "## PR reconciliation summary",
+        "Fetched linked issue #263 live with gitea_view_issue in this session.",
+        "",
+        "## Controller Handoff",
+        "- Selected PR: #278",
+        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+        "- Linked issue: #263",
+        "- Linked issue live status: Issue #263 (open)",
+        "- Safe next action: reconcile open PR",
+    ])
+
+
+class TestReconcileLinkedIssueLiveProof(unittest.TestCase):
+    def test_live_fetch_with_open_status_passes(self):
+        result = assess_reconcile_linked_issue_report(
+            _good_handoff(),
+            reconcile_session={"live_fetched": True, "linked_issue_number": 263},
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertEqual(result["linked_issue_number"], 263)
+
+    def test_no_linked_issue_passes(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Selected PR: #99",
+            "- Linked issue: none",
+            "- Linked issue live status: not applicable",
+        ])
+        result = assess_reconcile_linked_issue_report(report)
+        self.assertTrue(result["proven"])
+        self.assertIsNone(result["linked_issue_number"])
+
+    def test_closed_status_without_live_proof_blocks(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Linked issue: #263",
+            "- Linked issue live status: Issue #263 (closed)",
+        ])
+        result = assess_reconcile_linked_issue_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+        self.assertTrue(any("live" in r.lower() for r in result["reasons"]))
+
+    def test_not_verified_wording_passes_without_live_proof(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Linked issue: #263",
+            "- Linked issue live status: not verified in this session",
+        ])
+        result = assess_reconcile_linked_issue_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_missing_issue_requires_not_verified(self):
+        report = "\n".join([
+            "## Controller Handoff",
+            "- Linked issue: #999",
+            "- Linked issue live status: Issue #999 (open)",
+        ])
+        result = assess_reconcile_linked_issue_report(
+            report,
+            reconcile_session={"issue_missing": True, "linked_issue_number": 999},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("not verified" in r.lower() for r in result["reasons"]))
+
+    def test_text_live_proof_without_session_lock_passes(self):
+        report = _good_handoff()
+        result = assess_reconcile_linked_issue_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["live_proof_present"])
+
+    def test_issue_action_recommendation_requires_capability_proof(self):
+        report = "\n".join([
+            _good_handoff(),
+            "Recommended: close linked issue #263 after PR reconciliation.",
+        ])
+        result = assess_reconcile_linked_issue_report(
+            report,
+            reconcile_session={"live_fetched": True},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("capability" in r.lower() for r in result["reasons"]))
+
+    def test_issue_action_with_capability_proof_passes(self):
+        report = "\n".join([
+            _good_handoff(),
+            "Recommended: close linked issue #263.",
+            "Capability proof: gitea_close_issue allowed for prgs-author.",
+        ])
+        result = assess_reconcile_linked_issue_report(
+            report,
+            reconcile_session={"live_fetched": True},
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_reconcile_linked_issue_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()