fix: distinguish completed contributor risk assessments from execution failures

.github/actions/contributor-check/tests/test_contributor_check_action.py

@@ -103,7 +103,11 @@ def test_genuine_unknown_is_still_returned(tmp_path):
 
 @pytest.mark.parametrize("risk,expected", [("LOW", "LOW"), ("MEDIUM", "MEDIUM"), ("HIGH", "HIGH")])
 def test_computed_risk_passes_through(tmp_path, risk, expected):
-    script = _stub(tmp_path, "profile.py", f'print(\'{{"risk": "{risk}"}}\')\n')
+    exit_code = {"LOW": 0, "MEDIUM": 1, "HIGH": 2}[risk]
+    script = _stub(
+        tmp_path, "profile.py",
+        f'import sys\nprint(\'{{"risk": "{risk}"}}\')\nsys.exit({exit_code})\n',
+    )
 
     assert action._run_check(script, [], str(tmp_path / "out.json")) == expected
 

.github/actions/contributor-check/tests/test_risk_exit_contract.py

@@ -0,0 +1,74 @@
+"""Contract tests for contributor assessment process exit codes."""
+import importlib.util
+import json
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+MODULE = Path(__file__).resolve().parents[1] / "vendor" / "agt" / "contributor_check_action.py"
+spec = importlib.util.spec_from_file_location("contributor_check_action_under_test", MODULE)
+action = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(action)
+
+
+class RiskExitContractTests(unittest.TestCase):
+    def run_case(self, code, value, expected=None):
+        with tempfile.TemporaryDirectory() as directory:
+            result = subprocess.CompletedProcess(["python"], code, json.dumps(value), "")
+            with patch.object(action.subprocess, "run", return_value=result):
+                if expected is None:
+                    with self.assertRaises(action.CheckExecutionError):
+                        action._run_check("dummy.py", [], str(Path(directory) / "out"))
+                else:
+                    self.assertEqual(
+                        action._run_check("dummy.py", [], str(Path(directory) / "out")),
+                        expected,
+                    )
+
+    def test_valid_assessments(self):
+        for risk, code, normalized in [
+            ("NONE", 0, "LOW"), ("LOW", 0, "LOW"),
+            ("UNKNOWN", 0, "UNKNOWN"),
+            ("MEDIUM", 1, "MEDIUM"), ("HIGH", 2, "HIGH"),
+        ]:
+            with self.subTest(risk=risk):
+                self.run_case(code, {"risk": risk}, normalized)
+
+    def test_contradictions_and_bad_output_fail_closed(self):
+        for code, value in [
+            (0, {"risk": "HIGH"}), (1, {"risk": "LOW"}),
+            (2, {"risk": "MEDIUM"}), (0, {"risk": "INVALID"}),
+            (0, {}), (0, []), (0, {"risk": []}),
+            (0, {"risk": {}}), (0, {"risk": 3}), (3, {"risk": "HIGH"}),
+            (127, {"risk": "LOW"}),
+        ]:
+            with self.subTest(code=code, value=value):
+                self.run_case(code, value)
+
+    def test_malformed_and_empty_json_fail_closed(self):
+        for content in ["", "{", "null", "[1]", "42"]:
+            with self.subTest(content=content):
+                with tempfile.TemporaryDirectory() as directory:
+                    result = subprocess.CompletedProcess(["python"], 0, content, "")
+                    with patch.object(action.subprocess, "run", return_value=result):
+                        with self.assertRaises(action.CheckExecutionError):
+                            action._run_check("dummy.py", [], str(Path(directory) / "out"))
+
+    def test_operational_failures_are_not_risk_assessments(self):
+        for error in [subprocess.TimeoutExpired("test", 120), OSError("missing binary")]:
+            with self.subTest(error=type(error).__name__):
+                with tempfile.TemporaryDirectory() as directory:
+                    with patch.object(action.subprocess, "run", side_effect=error):
+                        with self.assertRaises(action.CheckExecutionError):
+                            action._run_check("dummy.py", [], str(Path(directory) / "out"))
+
+    def test_aggregation_preserves_high_and_unknown(self):
+        self.assertEqual(action._aggregate_risk("LOW", "HIGH"), "HIGH")
+        self.assertEqual(action._aggregate_risk("LOW", "UNKNOWN"), "UNKNOWN")
+        self.assertEqual(action._aggregate_risk("MEDIUM", "HIGH"), "HIGH")
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/actions/contributor-check/vendor/agt/contributor_check_action.py

@@ -99,10 +99,10 @@ def _run_check(script: str, args: list[str], out_path: str) -> str:
 
     Path(out_path).write_text(result.stdout)
 
-    # A non-zero exit means the check did not run to completion. An import
-    # error, a missing file or an unhandled exception all land here, and all
-    # of them used to reach the bare `except` below as UNKNOWN.
-    if result.returncode != 0:
+    # Scanners use 0/1/2 for completed LOW-or-NONE/MEDIUM/HIGH
+    # assessments. A nonzero assessment is not a process failure.
+    expected_exit = {"NONE": 0, "LOW": 0, "UNKNOWN": 0, "MEDIUM": 1, "HIGH": 2}
+    if result.returncode not in (0, 1, 2):
         detail = (result.stderr or "").strip().splitlines()
         tail = detail[-1] if detail else "no stderr"
         raise CheckExecutionError(
@@ -113,14 +113,19 @@ def _run_check(script: str, args: list[str], out_path: str) -> str:
         data = json.loads(result.stdout)
     except ValueError as exc:
         raise CheckExecutionError(
-            f"{name} exited 0 but produced no parseable JSON: {exc}"
+            f"{name} exited {result.returncode} but produced invalid JSON: {exc}"
         ) from exc
 
-    risk = data.get("risk", "UNKNOWN")
-    # credential_audit returns "NONE" when a user has no merged PRs in the
-    # target repo (nothing to launder = not applicable). Normalize to "LOW"
-    # so it aggregates correctly rather than being treated as UNKNOWN (which
-    # is reserved for checks that errored or could not be determined).
+    if not isinstance(data, dict):
+        raise CheckExecutionError(f"{name} produced non-object JSON")
+    risk = data.get("risk")
+    if not isinstance(risk, str) or risk not in expected_exit:
+        raise CheckExecutionError(f"{name} returned missing or unknown risk: {risk!r}")
+    if result.returncode != expected_exit[risk]:
+        raise CheckExecutionError(
+            f"{name} reported risk {risk} but exited {result.returncode}"
+        )
+    # NONE means not applicable, not an unknown or failed scan.
     return "LOW" if risk == "NONE" else risk