feat: require live linked issue proof in reconciliation handoffs (#300)

Add assess_reconcile_linked_issue_report verifier to block open/closed/resolved
linked issue status claims without gitea_view_issue proof in the current session,
require not-verified wording when the issue is missing, and gate recommended
issue close/reopen/comment actions on exact capability proof. Wire into
build_final_report and export from review_proofs.

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

review_proofs.py

@@ -1405,6 +1405,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         if report_text
         else {"proven": True, "block": False, "reasons": [], "violations": []}
     )
+    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": []}
+    )
     if baseline_validation is None and report_text:
         baseline_validation = assess_reviewer_baseline_validation_proof(
             report_text=report_text,
@@ -1561,6 +1566,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
             "queue-status report violates loaded workflow proof gates (#339)"
         )
         downgrade_reasons.extend(queue_status_report.get("reasons", []))
+    if not reconcile_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 baseline_validation.get("proven"):
         downgrade_reasons.append(
             "reviewer baseline validation proof missing or failed (#325)"
@@ -1646,6 +1656,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
         "queue_status_violations": list(
             queue_status_report.get("violations") or []
         ),
+        "reconcile_linked_issue_proven": bool(reconcile_linked_issue.get("proven")),
+        "reconcile_linked_issue_violations": list(
+            reconcile_linked_issue.get("reasons") or []
+        ),
         "baseline_validation_proven": bool(baseline_validation.get("proven")),
         "baseline_validation_violations": list(
             baseline_validation.get("violations") or []
@@ -3682,6 +3696,11 @@ _QUEUE_STATUS_REPORT_HINT = re.compile(
     re.I,
 )
 
+_RECONCILE_LINKED_ISSUE_HINT = re.compile(
+    r"already[- ]landed|reconcil|linked issue(?:\s+live)?\s+status",
+    re.I,
+)
+
 _GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
 
 _NOT_APPLICABLE_VALUE = re.compile(
@@ -4071,3 +4090,10 @@ 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_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)

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_llm_workflow_split.py

@@ -146,3 +146,9 @@ 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_reconcile_linked_issue_verifier_exported():
+    from review_proofs import assess_reconcile_linked_issue_report
+
+    assert callable(assess_reconcile_linked_issue_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()