fix(pr-review): select latest check attempts

loopx/capabilities/pr_review_queue/README.md

@@ -390,8 +390,11 @@ reports a typed verdict and invalid-reason codes.
 re-reads the named PR instead of trusting a saved review packet. In particular,
 GitHub may retain or reassociate an approval after an update-from-base commit;
 the gate still requires the public review body to name the observed exact head.
-It also rejects missing, pending, failed, or unknown checks and incomplete or
-unresolved review threads. An admin bypass may satisfy GitHub's author-owned
+For check-runs with a reliable workflow/job identity and start time, it evaluates
+only the latest attempt and reports raw and superseded counts; ambiguous rows are
+retained so the gate fails closed. It also rejects missing, pending, failed, or
+unknown effective checks and incomplete or unresolved review threads. An admin
+bypass may satisfy GitHub's author-owned
 self-review limitation, but it never overrides this capability gate or supplies
 user merge authority.
 

loopx/capabilities/pr_review_queue/check_attempts.py

@@ -0,0 +1,67 @@
+"""Fail-closed normalization for repeated GitHub status-check attempts."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from datetime import datetime
+from typing import Any
+
+
+def _attempt_timestamp(item: Mapping[str, Any]) -> float:
+    for field in ("startedAt", "createdAt"):
+        value = str(item.get(field) or "").strip()
+        if not value:
+            continue
+        try:
+            return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
+        except ValueError:
+            continue
+    return 0.0
+
+
+def _attempt_identity(item: Mapping[str, Any]) -> tuple[str, ...] | None:
+    name = str(
+        item.get("name")
+        or item.get("context")
+        or item.get("workflowName")
+        or ""
+    ).strip().casefold()
+    workflow = str(item.get("workflowName") or "").strip().casefold()
+    typename = str(item.get("__typename") or "check").strip().casefold()
+    if name and workflow:
+        return (typename, workflow, name)
+
+    context = str(item.get("context") or "").strip().casefold()
+    app = item.get("app")
+    app = app if isinstance(app, Mapping) else {}
+    app_identity = str(app.get("slug") or app.get("name") or "").strip().casefold()
+    if context and app_identity:
+        return (typename, app_identity, context)
+    return None
+
+
+def latest_check_attempts(
+    items: Sequence[dict[str, Any]],
+) -> tuple[list[dict[str, Any]], int]:
+    """Drop provably superseded attempts while retaining ambiguous rows."""
+
+    latest_by_identity: dict[tuple[str, ...], float] = {}
+    for item in items:
+        identity = _attempt_identity(item)
+        timestamp = _attempt_timestamp(item)
+        if identity is not None and timestamp:
+            latest_by_identity[identity] = max(
+                latest_by_identity.get(identity, 0.0), timestamp
+            )
+
+    effective: list[dict[str, Any]] = []
+    for item in items:
+        identity = _attempt_identity(item)
+        timestamp = _attempt_timestamp(item)
+        if (
+            identity is None
+            or not timestamp
+            or timestamp >= latest_by_identity[identity]
+        ):
+            effective.append(item)
+    return effective, len(items) - len(effective)

loopx/pr_review.py

@@ -29,6 +29,7 @@
 from .capabilities.pr_review_queue.github_source import (
     attach_pr_review_details_concurrently as _attach_pr_review_details_concurrently,
 )
+from .capabilities.pr_review_queue.check_attempts import latest_check_attempts
 from .control_plane.runtime.time import now_utc_iso
 from .presentation.markdown import as_dict as _as_dict
 from .presentation.markdown import as_list as _as_list
@@ -674,7 +675,12 @@ def _check_state(item: dict[str, Any]) -> str:
 
 
 def _checks(pr: dict[str, Any]) -> dict[str, Any]:
-    items = [item for item in _as_list(pr.get("statusCheckRollup")) if isinstance(item, dict)]
+    raw_items = [
+        item
+        for item in _as_list(pr.get("statusCheckRollup"))
+        if isinstance(item, dict)
+    ]
+    items, superseded = latest_check_attempts(raw_items)
     counts: dict[str, int] = {}
     failures: list[str] = []
     pending: list[str] = []
@@ -688,6 +694,8 @@ def _checks(pr: dict[str, Any]) -> dict[str, Any]:
     if not items:
         return {
             "total": 0,
+            "raw_total": len(raw_items),
+            "superseded": superseded,
             "counts": {},
             "summary": "No status-check rollup was available from the source.",
             "failures": [],
@@ -701,6 +709,8 @@ def _checks(pr: dict[str, Any]) -> dict[str, Any]:
         summary = f"{counts.get('success', 0)} successful check(s)."
     return {
         "total": len(items),
+        "raw_total": len(raw_items),
+        "superseded": superseded,
         "counts": counts,
         "summary": summary,
         "failures": failures[:5],

tests/test_pr_review_github_scan.py

@@ -746,6 +746,108 @@ def test_merge_readiness_rejects_red_pending_and_unresolved_remote_gates() -> No
     }.issubset(blocked["blocking_reasons"]), blocked
 
 
+def test_merge_readiness_uses_latest_check_attempt_per_workflow_job() -> None:
+    pr = _merge_ready_pr()
+    pr["statusCheckRollup"] = [
+        {
+            "__typename": "CheckRun",
+            "workflowName": "Python Tests",
+            "name": "merge-gate",
+            "status": "COMPLETED",
+            "conclusion": "CANCELLED",
+            "startedAt": "2026-09-09T11:00:00Z",
+        },
+        {
+            "__typename": "CheckRun",
+            "workflowName": "Python Tests",
+            "name": "merge-gate",
+            "status": "COMPLETED",
+            "conclusion": "SUCCESS",
+            "startedAt": "2026-09-09T11:05:00Z",
+        },
+        {
+            "__typename": "CheckRun",
+            "workflowName": "Security",
+            "name": "merge-gate",
+            "status": "COMPLETED",
+            "conclusion": "SUCCESS",
+            "startedAt": "2026-09-09T11:01:00Z",
+        },
+    ]
+    ready = merge_readiness_module.build_pr_merge_readiness_packet(
+        pull_request=pr,
+        repository="owner/repo",
+        expected_exact_head=f"4110@{HEAD_1}",
+        reviewer_login="maintainer",
+        review_threads=_complete_review_threads(),
+        source="fixture",
+    )
+
+    assert ready["ready"] is True, ready
+    assert ready["checks"] == {
+        "total": 2,
+        "raw_total": 3,
+        "superseded": 1,
+        "counts": {"success": 2},
+        "summary": "2 successful check(s).",
+        "failures": [],
+        "pending": [],
+    }
+
+
+def test_merge_readiness_keeps_latest_pending_and_ambiguous_attempts() -> None:
+    pr = _merge_ready_pr()
+    pr["statusCheckRollup"] = [
+        {
+            "__typename": "CheckRun",
+            "workflowName": "Python Tests",
+            "name": "pytest",
+            "status": "COMPLETED",
+            "conclusion": "SUCCESS",
+            "startedAt": "2026-09-09T11:00:00Z",
+        },
+        {
+            "__typename": "CheckRun",
+            "workflowName": "Python Tests",
+            "name": "pytest",
+            "status": "IN_PROGRESS",
+            "conclusion": "",
+            "startedAt": "2026-09-09T11:05:00Z",
+        },
+        {
+            "name": "legacy-context",
+            "status": "COMPLETED",
+            "conclusion": "FAILURE",
+        },
+        {
+            "name": "legacy-context",
+            "status": "COMPLETED",
+            "conclusion": "SUCCESS",
+        },
+    ]
+    blocked = merge_readiness_module.build_pr_merge_readiness_packet(
+        pull_request=pr,
+        repository="owner/repo",
+        expected_exact_head=f"4110@{HEAD_1}",
+        reviewer_login="maintainer",
+        review_threads=_complete_review_threads(),
+        source="fixture",
+    )
+
+    assert blocked["ready"] is False, blocked
+    assert blocked["checks"]["raw_total"] == 4
+    assert blocked["checks"]["total"] == 3
+    assert blocked["checks"]["superseded"] == 1
+    assert blocked["checks"]["counts"] == {
+        "pending": 1,
+        "failure": 1,
+        "success": 1,
+    }
+    assert {"status_checks_failed", "status_checks_pending"}.issubset(
+        blocked["blocking_reasons"]
+    )
+
+
 def test_merge_readiness_accepts_titled_author_owned_approval_only_with_bypass() -> (
     None
 ):