fix(ci): route every invalid_payload verdict through a shared builder

scripts/ci/agent_completion_gate.py

@@ -5,32 +5,54 @@
 import re
 import sys
 from pathlib import Path
-from typing import Any, Dict
+from typing import Any, Dict, List
+
+
+def _collection_errors(payload: Any) -> List[str]:
+    """Return the non-empty collection errors recorded by evidence collection."""
+
+    if not isinstance(payload, dict):
+        return []
+    raw = payload.get("collection_errors")
+    if not isinstance(raw, list):
+        return []
+    return [str(error) for error in raw if str(error).strip()]
+
+
+def _invalid_payload(payload: Any, invalid_fields: List[str]) -> Dict[str, Any]:
+    """Build a fail-closed ``invalid_payload`` verdict with collector diagnostics.
+
+    Every ``invalid_payload`` return routes through here so the collector's own
+    ``collection_errors`` are surfaced consistently — not only on the late
+    field-validation path. ``verdict`` and ``reasons`` stay byte-identical for
+    every input; the helper only enriches ``details``, keeping the gate
+    fail-closed while telling the author what to fix.
+    """
+
+    details: Dict[str, Any] = {}
+    if invalid_fields:
+        details["invalid_fields"] = sorted(set(invalid_fields))
+    surfaced_errors = _collection_errors(payload)
+    if surfaced_errors:
+        details["collection_errors"] = surfaced_errors
+    return {
+        "verdict": "blocked",
+        "reasons": ["invalid_payload"],
+        "details": details,
+    }
 
 
 def evaluate(payload: Any) -> Dict[str, Any]:
     """Evaluate agent execution evidence and return a fail-closed verdict."""
 
     if not isinstance(payload, dict):
-        return {
-            "verdict": "blocked",
-            "reasons": ["invalid_payload"],
-            "details": {},
-        }
+        return _invalid_payload(payload, [])
 
     policy = payload.get("policy")
     if not isinstance(policy, dict):
-        return {
-            "verdict": "blocked",
-            "reasons": ["invalid_payload"],
-            "details": {"invalid_fields": ["policy"]},
-        }
+        return _invalid_payload(payload, ["policy"])
     if "applicable" not in policy or type(policy["applicable"]) is not bool:
-        return {
-            "verdict": "blocked",
-            "reasons": ["invalid_payload"],
-            "details": {"invalid_fields": ["policy.applicable"]},
-        }
+        return _invalid_payload(payload, ["policy.applicable"])
     if policy.get("applicable") is False:
         return {"verdict": "not_applicable", "reasons": [], "details": {}}
 
@@ -244,11 +266,7 @@ def evaluate(payload: Any) -> Dict[str, Any]:
             if type(evidence.get(field)) is not bool:
                 invalid_fields.append("evidence." + field)
     if invalid_fields:
-        return {
-            "verdict": "blocked",
-            "reasons": ["invalid_payload"],
-            "details": {"invalid_fields": sorted(set(invalid_fields))},
-        }
+        return _invalid_payload(payload, invalid_fields)
 
     reasons = []
     identity_projection = {
@@ -257,11 +275,7 @@ def evaluate(payload: Any) -> Dict[str, Any]:
         "run_id": str(policy.get("run_id") or "").strip() or None,
     }
     details = {"identity_projection": identity_projection}
-    collection_errors = [
-        str(error)
-        for error in (payload.get("collection_errors") or [])
-        if str(error).strip()
-    ]
+    collection_errors = _collection_errors(payload)
     if collection_errors:
         reasons.append("evidence_collection_failed")
         details["collection_errors"] = collection_errors

tests/unit/test_agent_completion_gate.py

@@ -1196,6 +1196,122 @@ def test_cli_returns_zero_for_ready_input(self):
         self.assertEqual(exit_code, 0)
         self.assertEqual(json.loads(output.getvalue())["verdict"], "ready")
 
+    def test_invalid_payload_surfaces_the_underlying_collection_errors(self):
+        """A malformed payload must still explain *why* the fields are missing.
+
+        Reproduces the production failure that blocked ~47 open PRs: branches
+        matching the agent heuristic (``claude/*``, ``codex/*``, ...) are marked
+        applicable, but with no AgentTask issue the collector emits
+        ``agent_login``/``run_id`` as null. The gate correctly blocks, yet
+        previously reported a bare ``invalid_payload`` and discarded the
+        ``collection_errors`` that name the actual remediation.
+        """
+
+        payload = _valid_payload()
+        payload["policy"]["agent_login"] = None
+        payload["policy"]["run_id"] = None
+        payload["collection_errors"] = [
+            "missing_linked_issue",
+            "missing_agent_run_id",
+            "missing_agent_login",
+        ]
+
+        result = _evaluate(payload)
+
+        self.assertEqual(result["verdict"], "blocked")
+        self.assertEqual(result["reasons"], ["invalid_payload"])
+        self.assertIn("policy.agent_login", result["details"]["invalid_fields"])
+        self.assertIn("policy.run_id", result["details"]["invalid_fields"])
+        self.assertEqual(
+            result["details"]["collection_errors"],
+            [
+                "missing_linked_issue",
+                "missing_agent_run_id",
+                "missing_agent_login",
+            ],
+        )
+
+    def test_invalid_payload_omits_collection_errors_when_there_are_none(self):
+        payload = _valid_payload()
+        payload["policy"]["agent_login"] = None
+        payload["collection_errors"] = []
+
+        result = _evaluate(payload)
+
+        self.assertEqual(result["verdict"], "blocked")
+        self.assertEqual(result["reasons"], ["invalid_payload"])
+        self.assertNotIn("collection_errors", result["details"])
+
+    def test_invalid_payload_tolerates_unusable_collection_errors(self):
+        for unusable in (None, "missing_agent_login", {"a": 1}, 7):
+            with self.subTest(collection_errors=unusable):
+                payload = _valid_payload()
+                payload["policy"]["agent_login"] = None
+                payload["collection_errors"] = unusable
+
+                result = _evaluate(payload)
+
+                self.assertEqual(result["verdict"], "blocked")
+                self.assertEqual(result["reasons"], ["invalid_payload"])
+                self.assertNotIn("collection_errors", result["details"])
+
+    def test_invalid_payload_drops_blank_collection_errors(self):
+        payload = _valid_payload()
+        payload["policy"]["run_id"] = None
+        payload["collection_errors"] = ["", "   ", "stale_head"]
+
+        result = _evaluate(payload)
+
+        self.assertEqual(result["details"]["collection_errors"], ["stale_head"])
+
+    def test_early_invalid_payload_paths_still_surface_collection_errors(self):
+        """Collector diagnostics must survive the *early* ``invalid_payload``
+        returns, not only the late field-validation path.
+
+        Regression for the reviewer's example: a payload whose ``policy`` is
+        malformed short-circuits before field validation, so it previously
+        returned a bare ``invalid_payload`` and discarded the
+        ``collection_errors`` the collector had already recorded. Every
+        invalid-payload response now routes through ``_invalid_payload`` and
+        carries those diagnostics consistently.
+        """
+
+        cases = (
+            # policy is not a dict -> earliest invalid_fields return
+            ({"policy": "nope"}, "policy"),
+            # policy is a dict but missing `applicable` -> the exact example
+            # from the review thread
+            ({"policy": {}}, "policy.applicable"),
+        )
+        for base, expected_field in cases:
+            with self.subTest(invalid_field=expected_field):
+                payload = dict(base)
+                payload["collection_errors"] = ["missing_linked_issue"]
+
+                result = _evaluate(payload)
+
+                self.assertEqual(result["verdict"], "blocked")
+                self.assertEqual(result["reasons"], ["invalid_payload"])
+                self.assertIn(
+                    expected_field, result["details"]["invalid_fields"]
+                )
+                self.assertEqual(
+                    result["details"]["collection_errors"],
+                    ["missing_linked_issue"],
+                )
+
+    def test_non_dict_payload_reports_no_collection_errors(self):
+        """A payload that is not even a dict has no diagnostics to surface and
+        must keep returning an empty ``details`` (unchanged behaviour)."""
+
+        for payload in ([], "nope", 7, None):
+            with self.subTest(payload=payload):
+                result = _evaluate(payload)
+
+                self.assertEqual(result["verdict"], "blocked")
+                self.assertEqual(result["reasons"], ["invalid_payload"])
+                self.assertEqual(result["details"], {})
+
 
 class CompletionGateWorkflowTests(unittest.TestCase):
     def _workflow(self):