fix: validate runner candidates using full commit statuses

.github/scripts/runner-image.py

@@ -68,6 +68,22 @@ def changed_requirements(pr, manifest_path=MANIFEST):
     return False
 
 
+def latest_status(head, context):
+    # The combined /status endpoint omits creator. Full statuses are newest
+    # first: select before validating, never fall back to an older success.
+    page = 1
+    while True:
+        statuses = api(f"commits/{head}/statuses?per_page=100&page={page}")
+        # GitHub contexts are case-insensitive; a case variant must shadow
+        # older canonical statuses even though it cannot be trusted below.
+        candidate = next((s for s in statuses if s["context"].casefold() == context.casefold()), None)
+        if candidate is not None:
+            return candidate
+        if len(statuses) < 100:
+            return None
+        page += 1
+
+
 def export_environment(manifest, output):
     lock = manifest["requirements"]
     versions = lock["versions"]
@@ -140,11 +156,13 @@ def select(manifest, kind, output, wait_seconds, arch=None, validation=False):
             require(fingerprint(expected) == fingerprint(manifest),
                     "Merge-tree requirements differ from PR head; rebase before building a candidate")
             deadline = time.monotonic() + wait_seconds
+            context = f"Runner image candidate / PR {pr['number']}"
             while True:
-                statuses = api(f"commits/{head}/status")["statuses"]
-                candidate = next((s for s in statuses if s["context"] == f"Runner image candidate / PR {pr['number']}"), None)
+                candidate = latest_status(head, context)
                 if candidate and candidate["state"] == "success":
-                    require(candidate.get("creator", {}).get("login") == "github-actions[bot]",
+                    require(candidate["context"] == context, "Candidate status must use the exact publisher context")
+                    creator = candidate.get("creator")
+                    require(isinstance(creator, dict) and creator.get("login") == "github-actions[bot]",
                             "Candidate status must come from the trusted publisher")
                     require(re.fullmatch(r"sha256:[0-9a-f]{64}", candidate.get("description", "")),
                             "Publisher did not record an immutable digest")

.github/scripts/tests/fixtures/candidate-status-pr5151.json

@@ -0,0 +1,40 @@
+{
+  "combined": {
+    "state": "pending",
+    "sha": "a02b1460736e18b6345bb4722c622e55e787371d",
+    "total_count": 4,
+    "statuses": [
+      {
+        "id": 55116170875,
+        "context": "Runner image candidate / PR 5151",
+        "state": "success",
+        "description": "sha256:e5ebd957d28d15976320023b76cffa8b982e1d131c572d92eb9c91814dbf807b",
+        "target_url": "https://github.com/dashpay/platform/actions/runs/36474975257",
+        "created_at": "2026-09-28T20:13:10Z",
+        "updated_at": "2026-09-28T20:13:10Z"
+      }
+    ]
+  },
+  "statuses": [
+    {
+      "id": 55116170875,
+      "context": "Runner image candidate / PR 5151",
+      "state": "success",
+      "description": "sha256:e5ebd957d28d15976320023b76cffa8b982e1d131c572d92eb9c91814dbf807b",
+      "target_url": "https://github.com/dashpay/platform/actions/runs/36474975257",
+      "created_at": "2026-09-28T20:13:10Z",
+      "updated_at": "2026-09-28T20:13:10Z",
+      "creator": {
+        "login": "github-actions[bot]",
+        "id": 41898282
+      }
+    }
+  ],
+  "publisher_run": {
+    "id": 36474975257,
+    "path": ".github/workflows/runner-image-candidate.yml",
+    "event": "pull_request_target",
+    "status": "completed",
+    "conclusion": "success"
+  }
+}

.github/scripts/tests/test_runner_image.py

@@ -8,14 +8,20 @@
 from pathlib import Path
 import tempfile
 import unittest
+from urllib.error import URLError
 from unittest.mock import patch
 
 ROOT = Path(__file__).resolve().parents[3]
 spec = importlib.util.spec_from_file_location("runner_image", ROOT / ".github/scripts/runner-image.py")
 runner = importlib.util.module_from_spec(spec)
 spec.loader.exec_module(runner)
-HEAD = "a" * 40
-DIGEST = "sha256:" + "d" * 64
+# Minimal projections of public REST responses captured 2026-09-28 for PR 5151.
+# /status has Simple Commit Status objects (no creator); /statuses has creator.
+FIXTURE = json.loads((Path(__file__).parent / "fixtures/candidate-status-pr5151.json").read_text())
+HEAD = FIXTURE["combined"]["sha"]
+DIGEST = FIXTURE["statuses"][0]["description"]
+STATUS_PATH = f"commits/{HEAD}/statuses?per_page=100&page=1"
+RUN_PATH = f"actions/runs/{FIXTURE['publisher_run']['id']}"
 
 
 class SelectorTests(unittest.TestCase):
@@ -25,43 +31,52 @@ def setUp(self):
         self.addCleanup(self.temp.cleanup)
         self.event = Path(self.temp.name) / "event.json"
         self.output = Path(self.temp.name) / "output"
-        self.pr = {"number": 4702, "state": "open", "changed_files": 1,
+        self.pr = {"number": 5151, "state": "open", "changed_files": 1,
                    "head": {"sha": HEAD}}
         self.responses = {
-            "pulls/4702": self.pr,
-            "pulls/4702/files?per_page=100&page=1": [{"filename": runner.MANIFEST}],
+            "pulls/5151": self.pr,
+            "pulls/5151/files?per_page=100&page=1": [{"filename": runner.MANIFEST}],
             f"contents/{runner.MANIFEST}?ref={HEAD}": {
                 "content": base64.b64encode(json.dumps(self.manifest).encode()).decode()},
-            f"commits/{HEAD}/status": {"statuses": [{
-                "context": "Runner image candidate / PR 4702", "state": "success",
-                "creator": {"login": "github-actions[bot]"}, "description": DIGEST,
-                "target_url": "https://github.com/dashpay/platform/actions/runs/7",
-            }]},
-            "actions/runs/7": {"path": ".github/workflows/runner-image-candidate.yml",
-                              "event": "pull_request_target", "conclusion": "success"},
+            f"commits/{HEAD}/status": copy.deepcopy(FIXTURE["combined"]),
+            STATUS_PATH: copy.deepcopy(FIXTURE["statuses"]),
+            RUN_PATH: copy.deepcopy(FIXTURE["publisher_run"]),
         }
 
-    def select(self, event=None, kind="rust", arch=None, validation=False):
+    def api_response(self, path):
+        response = self.responses[path]
+        if isinstance(response, Exception):
+            raise response
+        return response
+
+    def status_calls(self):
+        return [call.args[0] for call in self.api.call_args_list
+                if call.args[0].startswith("commits/")]
+
+    def select(self, event=None, kind="rust", arch=None, validation=False, wait_seconds=0):
         self.event.write_text(json.dumps(event if event is not None else {"pull_request": self.pr}))
         with patch.dict(os.environ, {"GITHUB_EVENT_PATH": str(self.event)}), \
-             patch.object(runner, "api", side_effect=lambda path: self.responses[path]):
-            runner.select(self.manifest, kind, self.output, 0, arch, validation)
+             patch.object(runner, "api", side_effect=self.api_response) as api:
+            self.api = api
+            runner.select(self.manifest, kind, self.output, wait_seconds, arch, validation)
         return dict(line.split("=", 1) for line in self.output.read_text().splitlines())
 
     def test_non_pr_and_unchanged_pr_use_existing_pool(self):
         self.assertEqual(json.loads(self.select({})["labels"]), ["self-hosted", "Linux", "rust-ci"])
-        self.responses["pulls/4702/files?per_page=100&page=1"] = [{"filename": "Cargo.lock"}]
+        self.responses["pulls/5151/files?per_page=100&page=1"] = [{"filename": "Cargo.lock"}]
         self.assertEqual(self.select()["image_changed"], "false")
 
     def test_exact_candidate_includes_head_digest_and_kind(self):
-        output = self.select()
-        labels = json.loads(output["labels"])
-        self.assertEqual(labels[-1], f"platform-image-pr-4702-{HEAD}-{DIGEST[7:]}-rust")
-        self.assertEqual(output["image_changed"], "true")
+        self.assertNotIn("creator", FIXTURE["combined"]["statuses"][0])
+        for kind in ("kotlin", "rust", "npm"):
+            with self.subTest(kind=kind):
+                output = self.select(kind=kind)
+                self.assertEqual(json.loads(output["labels"]), ["self-hosted", "Linux", "X64",
+                                 f"platform-image-pr-5151-{HEAD}-{DIGEST[7:]}-{kind}"])
+                self.assertEqual(output["image_changed"], "true")
+                self.assertEqual(self.status_calls(), [STATUS_PATH])
 
-    def test_npm_candidates_and_ordinary_pool_have_distinct_labels(self):
-        labels = json.loads(self.select(kind="npm")["labels"])
-        self.assertEqual(labels[-1], f"platform-image-pr-4702-{HEAD}-{DIGEST[7:]}-npm")
+    def test_should_keep_npm_ordinary_pool_labels(self):
         self.assertEqual(json.loads(self.select({}, kind="npm")["labels"]), ["self-hosted", "npm-pr"])
 
     def test_new_head_or_closed_pr_rejects_stale_run(self):
@@ -80,17 +95,145 @@ def test_merge_tree_cannot_mix_requirements_from_both_branches(self):
             self.select()
 
     def test_missing_or_incomplete_publisher_cannot_select_image(self):
-        self.responses["actions/runs/7"]["conclusion"] = None
-        with self.assertRaisesRegex(ValueError, "not published"):
-            self.select()
-        self.responses[f"commits/{HEAD}/status"]["statuses"] = []
+        for conclusion in (None, "failure", "cancelled"):
+            with self.subTest(conclusion=conclusion):
+                self.responses[RUN_PATH]["conclusion"] = conclusion
+                with self.assertRaisesRegex(ValueError, "not published"):
+                    self.select()
+                self.assertFalse(self.output.exists())
+        self.responses[STATUS_PATH] = []
         with self.assertRaisesRegex(ValueError, "not published"):
             self.select()
 
     def test_other_workflow_cannot_supply_candidate_status(self):
-        self.responses["actions/runs/7"]["path"] = ".github/workflows/tests.yml"
-        with self.assertRaisesRegex(ValueError, "Unexpected candidate"):
-            self.select()
+        for field, value in (("path", ".github/workflows/tests.yml"), ("event", "pull_request")):
+            with self.subTest(field=field):
+                self.responses[RUN_PATH] = dict(FIXTURE["publisher_run"], **{field: value})
+                with self.assertRaisesRegex(ValueError, "Unexpected candidate"):
+                    self.select()
+                self.assertFalse(self.output.exists())
+
+    def test_should_reject_missing_null_malformed_or_wrong_creator_without_fallback(self):
+        # The real combined response is also a regression case: no creator.
+        missing = FIXTURE["combined"]["statuses"][0]
+        good = FIXTURE["statuses"][0]
+        candidates = [missing] + [dict(good, creator=creator) for creator in (
+            None, {}, "github-actions[bot]", [], 42,
+            {"login": None}, {"login": "untrusted-user"},
+        )]
+        for candidate in candidates:
+            with self.subTest(creator=candidate.get("creator", "absent")):
+                self.responses[STATUS_PATH] = [candidate, good]
+                with self.assertRaisesRegex(ValueError, "trusted publisher"):
+                    self.select()
+                self.assertEqual(self.status_calls(), [STATUS_PATH])
+                self.assertNotIn(RUN_PATH, [call.args[0] for call in self.api.call_args_list])
+                self.assertFalse(self.output.exists())
+
+    def test_should_reject_invalid_digest_or_publisher_url_without_fallback(self):
+        good = FIXTURE["statuses"][0]
+        for field, value, error in (
+            ("description", "sha256:abc", "immutable digest"),
+            ("description", "latest", "immutable digest"),
+            ("target_url", "https://github.com/other/platform/actions/runs/7", "publishing workflow"),
+            ("target_url", good["target_url"] + "/jobs/1", "publishing workflow"),
+        ):
+            with self.subTest(field=field, value=value):
+                self.responses[STATUS_PATH] = [dict(good, **{field: value}), good]
+                with self.assertRaisesRegex(ValueError, error):
+                    self.select()
+                self.assertFalse(self.output.exists())
+
+    def test_should_block_older_success_when_newest_is_pending_failure_or_error(self):
+        good = FIXTURE["statuses"][0]
+        for state in ("pending", "failure", "error"):
+            with self.subTest(state=state):
+                self.responses[STATUS_PATH] = [dict(good, state=state), good]
+                with self.assertRaisesRegex(ValueError, "not published"):
+                    self.select()
+                self.assertEqual(self.status_calls(), [STATUS_PATH])
+                self.assertNotIn(RUN_PATH, [call.args[0] for call in self.api.call_args_list])
+                self.assertFalse(self.output.exists())
+
+    def test_should_use_first_success_without_unnecessary_pagination(self):
+        good = FIXTURE["statuses"][0]
+        older = dict(good, description="sha256:" + "e" * 64)
+        self.responses[STATUS_PATH] = [good] + [older] * 99
+        labels = json.loads(self.select()["labels"])
+        self.assertEqual(labels[-1], f"platform-image-pr-5151-{HEAD}-{DIGEST[7:]}-rust")
+        self.assertEqual(self.status_calls(), [STATUS_PATH])
+
+    def test_should_shadow_canonical_success_with_newer_case_variant(self):
+        good = FIXTURE["statuses"][0]
+        for state in ("success", "pending", "failure", "error"):
+            with self.subTest(state=state):
+                newer = dict(good, context=good["context"].lower(), state=state)
+                self.responses[STATUS_PATH] = [newer] + [good] * 99
+                error = "exact publisher context" if state == "success" else "not published"
+                with self.assertRaisesRegex(ValueError, error):
+                    self.select()
+                self.assertEqual(self.status_calls(), [STATUS_PATH])
+                self.assertNotIn(RUN_PATH, [call.args[0] for call in self.api.call_args_list])
+                self.assertFalse(self.output.exists())
+
+    def test_should_select_first_match_on_later_page_before_validation(self):
+        good = FIXTURE["statuses"][0]
+        self.responses[STATUS_PATH] = [dict(good, context="unrelated")] * 100
+        page2 = STATUS_PATH.replace("&page=1", "&page=2")
+        for state in ("pending", "success"):
+            with self.subTest(state=state):
+                self.responses[page2] = [dict(good, state=state)] + [good] * 99
+                if state == "pending":
+                    with self.assertRaisesRegex(ValueError, "not published"):
+                        self.select()
+                    self.assertFalse(self.output.exists())
+                else:
+                    self.assertEqual(self.select()["image_changed"], "true")
+                self.assertEqual(self.status_calls(), [STATUS_PATH, page2])
+
+    def test_should_require_exact_context_and_stop_at_page_exhaustion(self):
+        good = FIXTURE["statuses"][0]
+        unrelated = [dict(good, context=context) for context in (
+            "Runner image candidate / PR 51510", "Runner image candidate / PR 5151 suffix",
+            "Runner image candidate / PR 5151 ", "Runner image candidate / PR 515",
+        )]
+        page2 = STATUS_PATH.replace("&page=1", "&page=2")
+        for last_page in ([], unrelated):
+            with self.subTest(last_page_size=len(last_page)):
+                self.responses[STATUS_PATH] = unrelated * 25
+                self.responses[page2] = last_page
+                with self.assertRaisesRegex(ValueError, "not published"):
+                    self.select()
+                self.assertEqual(self.status_calls(), [STATUS_PATH, page2])
+                self.assertFalse(self.output.exists())
+
+    def test_should_fail_closed_on_status_api_failure(self):
+        page2 = STATUS_PATH.replace("&page=1", "&page=2")
+        for failed_path in (STATUS_PATH, page2):
+            with self.subTest(failed_path=failed_path):
+                self.responses[STATUS_PATH] = [dict(FIXTURE["statuses"][0], context="other")] * 100
+                self.responses[failed_path] = URLError("status API unavailable")
+                with self.assertRaisesRegex(URLError, "status API unavailable"):
+                    self.select()
+                self.assertFalse(self.output.exists())
+
+    def test_should_retry_pending_status_and_publisher_run(self):
+        good = FIXTURE["statuses"][0]
+        for pending in ("status", "run"):
+            with self.subTest(pending=pending):
+                self.responses[STATUS_PATH] = [dict(good, state="pending"), good] if pending == "status" else [good]
+                self.responses[RUN_PATH]["conclusion"] = None if pending == "run" else "success"
+
+                def publish(_seconds):
+                    self.responses[STATUS_PATH] = [good]
+                    self.responses[RUN_PATH]["conclusion"] = "success"
+
+                with patch.object(runner.time, "monotonic", return_value=0), \
+                     patch.object(runner.time, "sleep", side_effect=publish) as sleep:
+                    self.assertEqual(self.select(wait_seconds=60)["image_changed"], "true")
+                sleep.assert_called_once_with(20)
+                self.assertEqual(self.status_calls(), [STATUS_PATH, STATUS_PATH])
+                self.assertEqual([call.args[0] for call in self.api.call_args_list].count("pulls/5151"), 2)
 
     def test_environment_export_rejects_multiline_values_before_writing(self):
         self.manifest["requirements"]["versions"]["protoc"] = "32.0\nINJECTED=yes"
@@ -99,30 +242,30 @@ def test_environment_export_rejects_multiline_values_before_writing(self):
         self.assertFalse(self.output.exists())
 
     def test_arm64_validation_never_consumes_amd64_candidate_status(self):
-        self.responses[f"commits/{HEAD}/status"]["statuses"] = []
+        self.responses[STATUS_PATH] = []
         output = self.select(arch="ARM64")
         self.assertEqual(json.loads(output["labels"]), ["self-hosted", "Linux", "ARM64", "rust-ci"])
 
     def test_arm64_manifest_change_does_not_use_stale_ordinary_arm64_runners(self):
-        self.responses["pulls/4702/files?per_page=100&page=1"] = [{"filename": runner.ARM64_MANIFEST}]
+        self.responses["pulls/5151/files?per_page=100&page=1"] = [{"filename": runner.ARM64_MANIFEST}]
         output = self.select()
         self.assertEqual(json.loads(output["labels"]), ["self-hosted", "Linux", "X64", "rust-ci"])
         self.assertEqual(output["image_changed"], "false")
         # Mac-backed ARM64 capacity remains available to ordinary, unrelated PRs.
-        self.responses["pulls/4702/files?per_page=100&page=1"] = [{"filename": "Cargo.lock"}]
+        self.responses["pulls/5151/files?per_page=100&page=1"] = [{"filename": "Cargo.lock"}]
         self.assertEqual(json.loads(self.select()["labels"]), ["self-hosted", "Linux", "rust-ci"])
 
     def test_both_manifest_changes_still_require_exact_amd64_candidate(self):
-        self.responses["pulls/4702/files?per_page=100&page=1"] = [
+        self.responses["pulls/5151/files?per_page=100&page=1"] = [
             {"filename": runner.MANIFEST}, {"filename": runner.ARM64_MANIFEST}]
         output = self.select()
         self.assertEqual(json.loads(output["labels"]), ["self-hosted", "Linux", "X64",
-                         f"platform-image-pr-4702-{HEAD}-{DIGEST[7:]}-rust"])
+                         f"platform-image-pr-5151-{HEAD}-{DIGEST[7:]}-rust"])
         self.assertEqual(output["image_changed"], "true")
 
     def test_arm64_validation_rejects_merge_tree_drift(self):
         arm = runner.read_manifest(ROOT / runner.ARM64_MANIFEST)
-        self.responses["pulls/4702/files?per_page=100&page=1"] = [{"filename": runner.ARM64_MANIFEST}]
+        self.responses["pulls/5151/files?per_page=100&page=1"] = [{"filename": runner.ARM64_MANIFEST}]
         remote = copy.deepcopy(arm)
         remote["recipe_revision"] = "e" * 40
         self.responses[f"contents/{runner.ARM64_MANIFEST}?ref={HEAD}"] = {