From 46d3b81192487e46e4996cd371ffa76659e10c89 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Mon, 3 Aug 2026 15:58:53 -0500 Subject: [PATCH 1/2] fix(ci): surface collection errors behind invalid_payload truth-gate verdicts The agent-completion truth gate blocks ~47 of the 69 open PRs with a bare `invalid_payload` and no remediation path. Root cause: `agentTaskApplicable()` in pr-checks.yml classifies any branch matching /^(?:agent|claude|codex|copilot|jules)[\/-]/ as agent work, so human-authored Claude Code worktree branches are held to the full AgentTask provenance contract. With no linked AgentTask issue, the collector emits `policy.agent_login` and `policy.run_id` as null and records the real reasons in `collection_errors` (missing_linked_issue, missing_agent_login, missing_agent_run_id). `evaluate()` then returned at the schema check and discarded `collection_errors` entirely -- they are only read further down, after the early return. Authors saw `invalid_payload` and nothing else. This keeps the gate fail-closed and byte-identical in `verdict` and `reasons`, and only adds `details.collection_errors` so the gate says what to fix. Verified: 112 passed against the reproduced PR #1270 payload. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- scripts/ci/agent_completion_gate.py | 29 +++++++--- tests/unit/test_agent_completion_gate.py | 68 ++++++++++++++++++++++++ 2 files changed, 90 insertions(+), 7 deletions(-) diff --git a/scripts/ci/agent_completion_gate.py b/scripts/ci/agent_completion_gate.py index d342c1323..2a6a4d742 100644 --- a/scripts/ci/agent_completion_gate.py +++ b/scripts/ci/agent_completion_gate.py @@ -5,7 +5,18 @@ 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 evaluate(payload: Any) -> Dict[str, Any]: @@ -244,10 +255,18 @@ def evaluate(payload: Any) -> Dict[str, Any]: if type(evidence.get(field)) is not bool: invalid_fields.append("evidence." + field) if invalid_fields: + # A malformed payload is still fail-closed, but the collector already + # knows *why* the fields are missing. Surfacing those errors here keeps + # the verdict identical while telling the author what to fix, instead of + # stranding them on a bare "invalid_payload". + 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": {"invalid_fields": sorted(set(invalid_fields))}, + "details": details, } reasons = [] @@ -257,11 +276,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 diff --git a/tests/unit/test_agent_completion_gate.py b/tests/unit/test_agent_completion_gate.py index 301263cd1..0c14d8f6f 100644 --- a/tests/unit/test_agent_completion_gate.py +++ b/tests/unit/test_agent_completion_gate.py @@ -1196,6 +1196,74 @@ 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"]) + class CompletionGateWorkflowTests(unittest.TestCase): def _workflow(self): From df6a5db03b9fa4642e85605bbcd1dd775bfa6d9f Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 3 Aug 2026 21:05:03 +0000 Subject: [PATCH 2/2] fix(ci): route every invalid_payload verdict through a shared builder MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The collection_errors diagnostic was only attached to the late field-validation return. The three early invalid_payload short-circuits (payload not a dict, policy not a dict, missing/invalid policy.applicable) returned bare verdicts, so a malformed payload that never reaches field validation stayed just as opaque despite the collector having already recorded why — exactly the case the review thread raised (evaluate({"policy": {}, "collection_errors": [...]})). Extract _invalid_payload(payload, invalid_fields) and route all four invalid_payload returns through it so the diagnostic is applied consistently. verdict and reasons stay byte-identical for every input; only details is enriched, and only when the collector recorded errors — the gate remains fail-closed. Add regression coverage for both early paths and confirm a non-dict payload still returns an empty details. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_011L6jqdhrKTYLinTnJYKEg9 --- scripts/ci/agent_completion_gate.py | 55 ++++++++++++------------ tests/unit/test_agent_completion_gate.py | 48 +++++++++++++++++++++ 2 files changed, 75 insertions(+), 28 deletions(-) diff --git a/scripts/ci/agent_completion_gate.py b/scripts/ci/agent_completion_gate.py index 2a6a4d742..80c3ee41b 100644 --- a/scripts/ci/agent_completion_gate.py +++ b/scripts/ci/agent_completion_gate.py @@ -19,29 +19,40 @@ def _collection_errors(payload: Any) -> List[str]: 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": {}} @@ -255,19 +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: - # A malformed payload is still fail-closed, but the collector already - # knows *why* the fields are missing. Surfacing those errors here keeps - # the verdict identical while telling the author what to fix, instead of - # stranding them on a bare "invalid_payload". - 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, - } + return _invalid_payload(payload, invalid_fields) reasons = [] identity_projection = { diff --git a/tests/unit/test_agent_completion_gate.py b/tests/unit/test_agent_completion_gate.py index 0c14d8f6f..0cd300f63 100644 --- a/tests/unit/test_agent_completion_gate.py +++ b/tests/unit/test_agent_completion_gate.py @@ -1264,6 +1264,54 @@ def test_invalid_payload_drops_blank_collection_errors(self): 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):