fix: preserve duplicate required check identities

Aggregate same-name check runs and commit statuses fail-closed after reducing each identity to its latest state so DeployBot does not mark a required context passed while another matching entry is failed or pending.

Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>

Author: 2 people

src/agent_merge_queue/cli.py

@@ -148,8 +148,15 @@ def normalize_check_state(check: dict[str, Any]) -> str:
     return "UNKNOWN"
 
 
+def check_identity(check: dict[str, Any], name: str) -> tuple[str, str]:
+    typename = str(check.get("__typename") or "")
+    if not typename:
+        typename = "StatusContext" if check.get("context") else "CheckRun"
+    return (typename, name)
+
+
 def check_states(checks: Iterable[dict[str, Any]]) -> dict[str, str]:
-    grouped: dict[str, tuple[str, int, str]] = {}
+    grouped: dict[tuple[str, tuple[str, str]], tuple[str, int, str]] = {}
     for index, check in enumerate(checks):
         name = str(check.get("name") or check.get("context") or "")
         if not name:
@@ -168,17 +175,23 @@ def check_states(checks: Iterable[dict[str, Any]]) -> dict[str, str]:
         state = normalize_check_state(check)
         # A newly queued run may not have a timestamp yet. Fail closed instead
         # of letting an older success hide that pending rerun.
-        order = "\uffff" if not timestamp and state not in PASSED_CHECK_STATES else timestamp
+        order = (
+            "\uffff" if not timestamp and state not in PASSED_CHECK_STATES else timestamp
+        )
         candidate = (order, index, state)
-        if name not in grouped or candidate[:2] > grouped[name][:2]:
-            grouped[name] = candidate
+        key = (name, check_identity(check, name))
+        if key not in grouped or candidate[:2] > grouped[key][:2]:
+            grouped[key] = candidate
+
+    states_by_name: dict[str, list[str]] = {}
+    for (name, _identity), value in grouped.items():
+        states_by_name.setdefault(name, []).append(value[2])
 
     result: dict[str, str] = {}
-    for name, value in grouped.items():
-        state = value[2]
-        if state in FAILED_CHECK_STATES:
+    for name, states in states_by_name.items():
+        if any(state in FAILED_CHECK_STATES for state in states):
             result[name] = "failed"
-        elif state in PASSED_CHECK_STATES:
+        elif all(state in PASSED_CHECK_STATES for state in states):
             result[name] = "passed"
         else:
             result[name] = "pending"

tests/test_cli.py

@@ -1655,6 +1655,59 @@ def test_undated_queued_rerun_hides_older_success(self) -> None:
 
         self.assertEqual(states["CI"], "pending")
 
+    def test_same_name_check_run_and_failed_status_blocks_required_check(self) -> None:
+        states = check_states(
+            [
+                {"__typename": "CheckRun", "name": "CI", "conclusion": "SUCCESS"},
+                {"__typename": "StatusContext", "context": "CI", "state": "FAILURE"},
+            ]
+        )
+        self.assertEqual(states["CI"], "failed")
+
+        value = entry(1)
+        value.checks = states
+        value.classify(CONFIG)
+
+        self.assertEqual(value.state, "blocked")
+        self.assertIn("CI failed", value.reasons)
+
+    def test_same_name_check_run_and_pending_status_waits_required_check(self) -> None:
+        states = check_states(
+            [
+                {"__typename": "CheckRun", "name": "CI", "conclusion": "SUCCESS"},
+                {"__typename": "StatusContext", "context": "CI", "state": "PENDING"},
+            ]
+        )
+        self.assertEqual(states["CI"], "pending")
+
+        value = entry(1)
+        value.checks = states
+        value.classify(CONFIG)
+
+        self.assertEqual(value.state, "waiting")
+        self.assertIn("CI is not complete", value.reasons)
+
+    def test_exact_check_run_replaces_stale_rollup_check_run_identity(self) -> None:
+        states = check_states(
+            [
+                {
+                    "__typename": "CheckRun",
+                    "name": "CI",
+                    "status": "QUEUED",
+                    "workflowName": "Build",
+                    "startedAt": "2026-06-20T00:00:00Z",
+                },
+                {
+                    "name": "CI",
+                    "conclusion": "SUCCESS",
+                    "app": {"slug": "github-actions"},
+                    "started_at": "2026-06-20T00:01:00Z",
+                },
+            ]
+        )
+
+        self.assertEqual(states["CI"], "passed")
+
     def test_review_verdicts_are_classified_generically(self) -> None:
         value = entry(1)
         value.review_verdicts = (ReviewVerdict("Any bot", "blocked", ("one finding",)),)