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: coygeek

src/agent_merge_queue/cli.py

@@ -129,8 +129,23 @@ 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"
+    app = check.get("app")
+    app_slug = app.get("slug") if isinstance(app, dict) else ""
+    source = str(
+        app_slug
+        or check.get("workflowName")
+        or check.get("context")
+        or name
+    )
+    return (typename, source)
+
+
 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:
@@ -148,15 +163,19 @@ def check_states(checks: Iterable[dict[str, Any]]) -> dict[str, str]:
         # of letting an older success hide that pending rerun.
         order = "\uffff" if not timestamp and state != "SUCCESS" 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 == "SUCCESS":
+        elif all(state == "SUCCESS" for state in states):
             result[name] = "passed"
         else:
             result[name] = "pending"

tests/test_cli.py

@@ -197,6 +197,38 @@ 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_review_verdicts_are_classified_generically(self) -> None:
         value = entry(1)
         value.review_verdicts = (ReviewVerdict("Any bot", "blocked", ("one finding",)),)