ci: add selectable external CI runners (#3451)

.github/scripts/backend-tests-workflow-contract_test.py

@@ -25,7 +25,7 @@ class BackendTestsWorkflowContractTest(unittest.TestCase):
         generate = step_block(workflow, "Generate test report")
         publish = step_block(workflow, "Publish bounded test report summary")
 
-        temporary_summary = "$RUNNER_TEMP/backend-test-summary-${{ matrix.shard }}.md"
+        temporary_summary = "$RUNNER_TEMP/backend-test-summary-${BACKEND_TEST_SHARD}.md"
         self.assertIn(
             "uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10",
             checkout,
@@ -37,12 +37,13 @@ class BackendTestsWorkflowContractTest(unittest.TestCase):
         )
         self.assertIn("path: .github/vendor/go-test-action", checkout)
         self.assertIn(f'report_summary="{temporary_summary}"', generate)
+        self.assertIn("BACKEND_TEST_SHARD: ${{ matrix.shard }}", generate)
         self.assertNotIn("${{ runner.temp }}", generate)
         self.assertIn(': > "$report_summary"', generate)
         self.assertIn('GITHUB_STEP_SUMMARY="$report_summary"', generate)
         self.assertIn(
             'INPUT_FROMJSONFILE="$GITHUB_WORKSPACE/apps/backend/'
-            'test-results-${{ matrix.shard }}.json"',
+            'test-results-${BACKEND_TEST_SHARD}.json"',
             generate,
         )
         self.assertIn(
@@ -60,8 +61,8 @@ class BackendTestsWorkflowContractTest(unittest.TestCase):
         self.assertIn(f'--input "{temporary_summary}"', publish)
         self.assertIn('--output "$GITHUB_STEP_SUMMARY"', publish)
         self.assertIn("--diagnostics-label", publish)
-        self.assertIn("backend-test-results-${{ matrix.shard }}", publish)
-        self.assertIn("${{ github.run_id }}#artifacts", publish)
+        self.assertIn("backend-test-results-${BACKEND_TEST_SHARD}", publish)
+        self.assertIn("$GITHUB_RUN_ID#artifacts", publish)
 
     def test_full_json_diagnostics_remain_in_the_named_artifact(self) -> None:
         workflow = BACKEND_WORKFLOW.read_text(encoding="utf-8")

.github/scripts/e2e-tests-workflow-contract_test.py

@@ -193,6 +193,58 @@ cat "${FAKE_DOCKER_MANIFEST}"
         self.assertIn("timeout-minutes: 35", normal_job)
         self.assertNotIn("timeout-minutes: 25", normal_job)
 
+    # @covers AC-PLATFORM-EXTERNAL-E2E-RUNNER-CAPACITY-001.1
+    # @covers AC-PLATFORM-EXTERNAL-E2E-RUNNER-CAPACITY-001.2
+    # @covers AC-PLATFORM-EXTERNAL-E2E-RUNNER-CAPACITY-001.3
+    # @covers AC-PLATFORM-EXTERNAL-E2E-RUNNER-CAPACITY-001.4
+    # @covers AC-PLATFORM-EXTERNAL-E2E-RUNNER-CAPACITY-001.5
+    # @covers AC-PLATFORM-EXTERNAL-E2E-RUNNER-CAPACITY-001.6
+    # @covers AC-PLATFORM-EXTERNAL-E2E-RUNNER-CAPACITY-002.1
+    # @covers AC-PLATFORM-EXTERNAL-E2E-RUNNER-CAPACITY-002.2
+    def test_external_runner_tiers_are_toggleable_for_eligible_jobs(self) -> None:
+        workflow = E2E_WORKFLOW.read_text(encoding="utf-8")
+
+        light_jobs = {
+            "changes": ("build", "changes_runner"),
+            "e2e-gate": (None, "e2e_gate_runner"),
+        }
+
+        for job, (next_job, output_name) in light_jobs.items():
+            if next_job is None:
+                job_text = workflow.partition("  e2e-gate:\n")[2]
+            else:
+                job_text = job_block(workflow, job, next_job)
+            expected = (
+                "runs-on: ${{ fromJSON(needs.runner_plan.outputs.plan)."
+                f"{output_name} }}}}"
+            )
+            self.assertIn(
+                expected,
+                job_text,
+                f"{job} must select its configured tier",
+            )
+        e2e_job = job_block(workflow, "e2e", "playwright_image")
+        self.assertIn("runs-on: ${{ matrix.runner }}", e2e_job)
+        self.assertIn(
+            "matrix: ${{ fromJSON(needs.runner_plan.outputs.plan).e2e_matrix }}",
+            e2e_job,
+        )
+
+        protected_jobs = {
+            "build": "e2e",
+            "e2e-report": "e2e-gate",
+            "playwright_image": "e2e-containers",
+            "e2e-containers": "e2e-kubernetes-compatibility",
+            "e2e-kubernetes-compatibility": "desktop-e2e",
+            "desktop-e2e": "e2e-report",
+        }
+        for job, next_job in protected_jobs.items():
+            protected_text = job_block(workflow, job, next_job)
+            self.assertIn("runs-on: ubuntu-latest", protected_text)
+            self.assertNotIn("runner_plan", protected_text)
+            self.assertNotIn("KANDEV_CI_EXTERNAL", protected_text)
+            self.assertNotIn("KANDEV_CI_RUNNER_", protected_text)
+
     def test_contract_runs_in_the_unfiltered_required_workflow(self) -> None:
         workflow = LINT_WORKFLOW.read_text(encoding="utf-8")
 

.github/scripts/external-runner-workflow-contract_test.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+"""Contract tests for external runner planner wiring and protected jobs."""
+
+from pathlib import Path
+import unittest
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+WORKFLOW_ROOT = REPO_ROOT / ".github" / "workflows"
+PLAN_SCRIPT = REPO_ROOT / ".github" / "scripts" / "runner-plan.py"
+
+
+def job_block(workflow: str, job: str, next_job: str | None) -> str:
+    marker = f"  {job}:\n"
+    _, separator, remainder = workflow.partition(marker)
+    if not separator:
+        raise AssertionError(f"Workflow has no {job} job")
+    if next_job is None:
+        return remainder
+    return remainder.partition(f"\n  {next_job}:\n")[0]
+
+
+class ExternalRunnerWorkflowContractTest(unittest.TestCase):
+    def assert_planner(
+        self, workflow_name: str, workflow_key: str, families: tuple[str, ...]
+    ) -> str:
+        workflow = (WORKFLOW_ROOT / workflow_name).read_text(encoding="utf-8")
+        planner = job_block(workflow, "runner_plan", "changes")
+        self.assertIn("runs-on: ubuntu-latest", planner)
+        self.assertIn("plan: ${{ steps.plan.outputs.plan }}", planner)
+        self.assertIn("uses: ./.github/actions/plan-external-runners", planner)
+        self.assertIn(f"workflow: {workflow_key}", planner)
+        self.assertIn("families: >-", planner)
+        self.assertIn("run-id: ${{ github.run_id }}", planner)
+        self.assertIn("burst: ${{ vars.KANDEV_CI_EXTERNAL_ENABLED }}", planner)
+        self.assertIn("percent: ${{ vars.KANDEV_CI_EXTERNAL_PERCENT }}", planner)
+        self.assertIn("light-label: ${{ vars.KANDEV_CI_RUNNER_LIGHT }}", planner)
+        self.assertIn("standard-label: ${{ vars.KANDEV_CI_RUNNER_STANDARD }}", planner)
+        for family in families:
+            self.assertIn(f'"name":"{family}"', planner)
+        return workflow
+
+    def test_e2e_uses_planner_for_eligible_jobs_and_keeps_protected_jobs_hosted(self) -> None:
+        workflow = self.assert_planner(
+            "e2e-tests.yml",
+            "e2e",
+            ("changes", "e2e", "e2e_gate"),
+        )
+        self.assertIn(
+            "runs-on: ${{ fromJSON(needs.runner_plan.outputs.plan).changes_runner }}",
+            job_block(workflow, "changes", "build"),
+        )
+        build = job_block(workflow, "build", "e2e")
+        self.assertIn("runs-on: ubuntu-latest", build)
+        self.assertNotIn("runner_plan", build)
+        e2e = job_block(workflow, "e2e", "playwright_image")
+        self.assertIn(
+            "matrix: ${{ fromJSON(needs.runner_plan.outputs.plan).e2e_matrix }}",
+            e2e,
+        )
+        self.assertIn("runs-on: ${{ matrix.runner }}", e2e)
+        report = job_block(workflow, "e2e-report", "e2e-gate")
+        self.assertIn("runs-on: ubuntu-latest", report)
+        self.assertNotIn("runner_plan", report)
+        self.assertIn(
+            "runs-on: ${{ fromJSON(needs.runner_plan.outputs.plan).e2e_gate_runner }}",
+            job_block(workflow, "e2e-gate", None),
+        )
+        for job, next_job in (
+            ("playwright_image", "e2e-containers"),
+            ("e2e-containers", "e2e-kubernetes-compatibility"),
+            ("e2e-kubernetes-compatibility", "desktop-e2e"),
+            ("desktop-e2e", "e2e-report"),
+        ):
+            protected = job_block(workflow, job, next_job)
+            self.assertIn("runs-on: ubuntu-latest", protected)
+            self.assertNotIn("runner_plan", protected)
+            self.assertNotIn("KANDEV_CI_EXTERNAL", protected)
+
+    def test_backend_uses_planner_and_keeps_services_and_windows_hosted(self) -> None:
+        workflow = self.assert_planner(
+            "backend-tests.yml",
+            "backend",
+            ("test",),
+        )
+        for job, next_job in (
+            ("changes", "static_checks"),
+            ("static_checks", "test_shards"),
+            ("test_ambient_env", "test"),
+        ):
+            protected = job_block(workflow, job, next_job)
+            self.assertIn("runs-on: ubuntu-latest", protected)
+            self.assertNotIn("runner_plan", protected)
+        shards = job_block(workflow, "test_shards", "test_ambient_env")
+        self.assertIn("- shard: 1", shards)
+        self.assertIn("- shard: 2", shards)
+        self.assertIn("runs-on: ubuntu-latest", shards)
+        self.assertNotIn("runner_plan", shards)
+        self.assertNotIn("matrix.runner", shards)
+        test_gate = job_block(workflow, "test", "postgres-boot")
+        self.assertIn(
+            "runs-on: ${{ fromJSON(needs.runner_plan.outputs.plan).test_runner }}",
+            test_gate,
+        )
+        for job, next_job in (("postgres-boot", "test-windows"), ("test-windows", None)):
+            protected = job_block(workflow, job, next_job)
+            self.assertIn("runs-on: ubuntu-latest" if job == "postgres-boot" else "runs-on: windows-latest", protected)
+            self.assertNotIn("runner_plan", protected)
+
+    def test_frontend_uses_planner_for_test_and_gate(self) -> None:
+        workflow = self.assert_planner(
+            "frontend-tests.yml",
+            "frontend",
+            ("changes", "frontend", "frontend_gate"),
+        )
+        self.assertIn(
+            "runs-on: ${{ fromJSON(needs.runner_plan.outputs.plan).changes_runner }}",
+            job_block(workflow, "changes", "frontend"),
+        )
+        self.assertIn(
+            "runs-on: ${{ fromJSON(needs.runner_plan.outputs.plan).frontend_runner }}",
+            job_block(workflow, "frontend", "frontend-gate"),
+        )
+        self.assertIn(
+            "runs-on: ${{ fromJSON(needs.runner_plan.outputs.plan).frontend_gate_runner }}",
+            job_block(workflow, "frontend-gate", None),
+        )
+
+    def test_protected_lint_workflows_do_not_run_the_planner(self) -> None:
+        for workflow_name in (
+            "architecture-lint.yml",
+            "lint-action-pinning.yml",
+            "lint-harness-files.yml",
+        ):
+            workflow = (WORKFLOW_ROOT / workflow_name).read_text(encoding="utf-8")
+            lint = job_block(workflow, "lint", None)
+            self.assertNotIn("runner_plan:", workflow)
+            self.assertNotIn("name: Require runner plan", lint)
+            self.assertIn("runs-on: ubuntu-latest", lint)
+
+    def test_plan_script_is_checked_by_required_action_pinning_workflow(self) -> None:
+        workflow = (WORKFLOW_ROOT / "lint-action-pinning.yml").read_text(encoding="utf-8")
+        self.assertTrue(PLAN_SCRIPT.is_file())
+        action = (WORKFLOW_ROOT / ".." / "actions" / "plan-external-runners" / "action.yml").resolve()
+        self.assertTrue(action.is_file())
+        action_text = action.read_text(encoding="utf-8")
+        self.assertIn("using: composite", action_text)
+        self.assertIn("families:", action_text)
+        self.assertIn("plan:", action_text)
+        self.assertIn("runner-plan.py", action_text)
+        self.assertNotIn("changes_runner:", action_text)
+        self.assertIn("python3 .github/scripts/runner-plan_test.py", workflow)
+        self.assertIn("python3 .github/scripts/external-runner-workflow-contract_test.py", workflow)
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/scripts/runner-plan.py

@@ -0,0 +1,226 @@
+#!/usr/bin/env python3
+"""Build deterministic runner assignments from workflow-owned family data."""
+
+import argparse
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class RunnerPlan:
+    plan: dict[str, object]
+    warnings: list[str]
+
+
+@dataclass(frozen=True)
+class Family:
+    name: str
+    tier: str
+    instances: tuple[dict[str, object], ...]
+    is_matrix: bool
+
+
+GITHUB_RUNNER = "ubuntu-latest"
+VALID_TIERS = frozenset({"light", "standard"})
+FAMILY_NAME = re.compile(r"[a-z][a-z0-9_]*\Z")
+
+
+def _parse_percentage(raw: str | None) -> tuple[int, list[str]]:
+    if raw is None or raw.strip() == "":
+        return 0, []
+    value = raw.strip()
+    warning = (
+        "KANDEV_CI_EXTERNAL_PERCENT must be an integer from 0 to 100, "
+        f"got {raw!r}"
+    )
+    if not value.isascii() or not value.isdigit():
+        return 0, [warning]
+    percentage = int(value)
+    if percentage > 100:
+        return 0, [warning]
+    return percentage, []
+
+
+def _validate_families(raw: object) -> tuple[Family, ...]:
+    if not isinstance(raw, list) or not raw:
+        raise ValueError("families must be a non-empty JSON array")
+
+    families: list[Family] = []
+    names: set[str] = set()
+    for entry in raw:
+        if not isinstance(entry, dict):
+            raise ValueError("each family must be a JSON object")
+        name = entry.get("name")
+        tier = entry.get("tier")
+        if not isinstance(name, str) or FAMILY_NAME.fullmatch(name) is None:
+            raise ValueError(f"invalid family name: {name!r}")
+        if name in names:
+            raise ValueError(f"duplicate family name: {name}")
+        if tier not in VALID_TIERS:
+            raise ValueError(f"family {name!r} has invalid tier: {tier!r}")
+
+        is_matrix = "instances" in entry
+        instances_value = entry.get("instances", [{}])
+        if not isinstance(instances_value, list) or not instances_value:
+            raise ValueError(f"family {name!r} instances must be a non-empty array")
+        instances: list[dict[str, object]] = []
+        for instance in instances_value:
+            if not isinstance(instance, dict):
+                raise ValueError(f"family {name!r} instances must be JSON objects")
+            if "runner" in instance:
+                raise ValueError(f"family {name!r} instance cannot define runner")
+            instances.append(instance)
+
+        names.add(name)
+        families.append(
+            Family(
+                name=name,
+                tier=tier,
+                instances=tuple(instances),
+                is_matrix=is_matrix,
+            )
+        )
+    return tuple(families)
+
+
+def _external_for_singleton(
+    *, workflow: str, run_id: str, family: str, percentage: int
+) -> bool:
+    if percentage <= 0:
+        return False
+    if percentage >= 100:
+        return True
+    key = f"{workflow}\0{run_id}\0{family}".encode("utf-8")
+    bucket = int.from_bytes(hashlib.sha256(key).digest()[:8], "big") % 100
+    return bucket < percentage
+
+
+def _matrix_assignments(
+    *,
+    workflow: str,
+    run_id: str,
+    family: str,
+    instances: tuple[dict[str, object], ...],
+    percentage: int,
+) -> set[int]:
+    external_count = (len(instances) * percentage) // 100
+    ranked: list[tuple[bytes, int]] = []
+    for index, instance in enumerate(instances):
+        matrix_key = json.dumps(instance, sort_keys=True, separators=(",", ":"))
+        digest = hashlib.sha256(
+            f"{workflow}\0{run_id}\0{family}\0{matrix_key}".encode("utf-8")
+        ).digest()
+        ranked.append((digest, index))
+    return {index for _, index in sorted(ranked)[:external_count]}
+
+
+def _plan_family(
+    *,
+    workflow: str,
+    run_id: str,
+    family: Family,
+    burst: str,
+    percentage: int,
+    external_label: str,
+) -> list[dict[str, object]]:
+    if burst != "true" or not external_label or percentage == 0:
+        external_indices: set[int] = set()
+    elif not family.is_matrix:
+        external_indices = (
+            {0}
+            if _external_for_singleton(
+                workflow=workflow,
+                run_id=run_id,
+                family=family.name,
+                percentage=percentage,
+            )
+            else set()
+        )
+    else:
+        external_indices = _matrix_assignments(
+            workflow=workflow,
+            run_id=run_id,
+            family=family.name,
+            instances=family.instances,
+            percentage=percentage,
+        )
+
+    return [
+        {
+            **instance,
+            "runner": external_label if index in external_indices else GITHUB_RUNNER,
+        }
+        for index, instance in enumerate(family.instances)
+    ]
+
+
+def build_plan(
+    *,
+    workflow: str,
+    run_id: str,
+    families: object,
+    burst: str,
+    percent: str | None,
+    light_label: str,
+    standard_label: str,
+) -> RunnerPlan:
+    percentage, warnings = _parse_percentage(percent)
+    labels = {"light": light_label, "standard": standard_label}
+    plan: dict[str, object] = {}
+    for family in _validate_families(families):
+        planned = _plan_family(
+            workflow=workflow,
+            run_id=run_id,
+            family=family,
+            burst=burst,
+            percentage=percentage,
+            external_label=labels[family.tier],
+        )
+        if family.is_matrix:
+            plan[f"{family.name}_matrix"] = {"include": planned}
+        else:
+            plan[f"{family.name}_runner"] = planned[0]["runner"]
+    return RunnerPlan(plan=plan, warnings=warnings)
+
+
+def _write_output(plan: RunnerPlan, output_path: Path) -> None:
+    value = json.dumps(plan.plan, separators=(",", ":"), sort_keys=True)
+    with output_path.open("a", encoding="utf-8") as output:
+        output.write(f"plan={value}\n")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--workflow", required=True)
+    parser.add_argument("--run-id", required=True)
+    parser.add_argument("--families", required=True)
+    parser.add_argument("--burst", default="")
+    parser.add_argument("--percent")
+    parser.add_argument("--light-label", default="")
+    parser.add_argument("--standard-label", default="")
+    parser.add_argument("--output", type=Path, default=Path("/dev/stdout"))
+    args = parser.parse_args()
+    try:
+        families = json.loads(args.families)
+        plan = build_plan(
+            workflow=args.workflow,
+            run_id=args.run_id,
+            families=families,
+            burst=args.burst,
+            percent=args.percent,
+            light_label=args.light_label,
+            standard_label=args.standard_label,
+        )
+    except (json.JSONDecodeError, ValueError) as error:
+        parser.error(str(error))
+    for warning in plan.warnings:
+        print(f"::warning::{warning}")
+    _write_output(plan, args.output)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/scripts/runner-plan_test.py

@@ -0,0 +1,223 @@
+#!/usr/bin/env python3
+"""Tests for deterministic external CI runner allocation."""
+
+import importlib.util
+from pathlib import Path
+import unittest
+
+
+SCRIPT = Path(__file__).with_name("runner-plan.py")
+SPEC = importlib.util.spec_from_file_location("runner_plan", SCRIPT)
+assert SPEC is not None and SPEC.loader is not None
+runner_plan = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(runner_plan)
+
+E2E_FAMILIES = [
+    {"name": "changes", "tier": "light"},
+    {
+        "name": "e2e",
+        "tier": "standard",
+        "instances": [{"shard": shard} for shard in range(1, 15)],
+    },
+    {"name": "e2e_gate", "tier": "light"},
+]
+FRONTEND_FAMILIES = [
+    {"name": "frontend", "tier": "standard"},
+    {"name": "frontend_gate", "tier": "light"},
+]
+BACKEND_MATRIX_FAMILY = [
+    {
+        "name": "backend_test",
+        "tier": "standard",
+        "instances": [{"shard": 1, "index": 0}, {"shard": 2, "index": 1}],
+    }
+]
+
+
+class RunnerPlanTest(unittest.TestCase):
+    def test_missing_percentage_defaults_to_github(self) -> None:
+        plan = runner_plan.build_plan(
+            workflow="e2e",
+            run_id="100",
+            families=E2E_FAMILIES,
+            burst="true",
+            percent=None,
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+
+        self.assertEqual(plan.plan["changes_runner"], "ubuntu-latest")
+        self.assertFalse(plan.warnings)
+
+    def test_burst_off_ignores_percentage(self) -> None:
+        plan = runner_plan.build_plan(
+            workflow="frontend",
+            run_id="101",
+            families=FRONTEND_FAMILIES,
+            burst="false",
+            percent="100",
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+
+        self.assertEqual(plan.plan["frontend_runner"], "ubuntu-latest")
+        self.assertEqual(plan.plan["frontend_gate_runner"], "ubuntu-latest")
+
+    def test_matrix_percentage_uses_floor_and_is_stable(self) -> None:
+        first = runner_plan.build_plan(
+            workflow="e2e",
+            run_id="102",
+            families=E2E_FAMILIES,
+            burst="true",
+            percent="50",
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+        second = runner_plan.build_plan(
+            workflow="e2e",
+            run_id="102",
+            families=E2E_FAMILIES,
+            burst="true",
+            percent="50",
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+
+        first_matrix = first.plan["e2e_matrix"]["include"]
+        second_matrix = second.plan["e2e_matrix"]["include"]
+        self.assertEqual(first_matrix, second_matrix)
+        self.assertEqual(
+            sum(item["runner"] == "standard-runner" for item in first_matrix), 7
+        )
+
+    def test_backend_matrix_percentage_uses_floor(self) -> None:
+        plan = runner_plan.build_plan(
+            workflow="backend",
+            run_id="103",
+            families=BACKEND_MATRIX_FAMILY,
+            burst="true",
+            percent="50",
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+
+        matrix = plan.plan["backend_test_matrix"]["include"]
+        self.assertEqual(
+            sum(item["runner"] == "standard-runner" for item in matrix), 1
+        )
+        self.assertEqual(
+            [(item["shard"], item["index"]) for item in matrix],
+            [(1, 0), (2, 1)],
+        )
+
+    def test_workflow_owned_family_names_need_no_python_change(self) -> None:
+        plan = runner_plan.build_plan(
+            workflow="new-workflow",
+            run_id="109",
+            families=[{"name": "new_job", "tier": "standard"}],
+            burst="true",
+            percent="100",
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+
+        self.assertEqual(plan.plan, {"new_job_runner": "standard-runner"})
+
+    def test_invalid_family_definition_is_rejected(self) -> None:
+        with self.assertRaisesRegex(ValueError, "duplicate family name"):
+            runner_plan.build_plan(
+                workflow="frontend",
+                run_id="110",
+                families=[
+                    {"name": "same", "tier": "light"},
+                    {"name": "same", "tier": "standard"},
+                ],
+                burst="true",
+                percent="50",
+                light_label="light-runner",
+                standard_label="standard-runner",
+            )
+
+    def test_singleton_allocation_is_stable_for_reruns(self) -> None:
+        first = runner_plan.build_plan(
+            workflow="frontend",
+            run_id="104",
+            families=FRONTEND_FAMILIES,
+            burst="true",
+            percent="50",
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+        second = runner_plan.build_plan(
+            workflow="frontend",
+            run_id="104",
+            families=FRONTEND_FAMILIES,
+            burst="true",
+            percent="50",
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+        self.assertEqual(first.plan, second.plan)
+
+    def test_empty_tier_label_falls_back_to_github(self) -> None:
+        plan = runner_plan.build_plan(
+            workflow="frontend",
+            run_id="105",
+            families=FRONTEND_FAMILIES,
+            burst="true",
+            percent="100",
+            light_label="",
+            standard_label="",
+        )
+
+        self.assertEqual(plan.plan["frontend_runner"], "ubuntu-latest")
+
+    def test_invalid_percentage_fails_closed_with_warning(self) -> None:
+        plan = runner_plan.build_plan(
+            workflow="e2e",
+            run_id="106",
+            families=E2E_FAMILIES,
+            burst="true",
+            percent="101",
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+
+        self.assertTrue(plan.warnings)
+        self.assertEqual(plan.plan["changes_runner"], "ubuntu-latest")
+        matrix = plan.plan["e2e_matrix"]["include"]
+        self.assertTrue(all(item["runner"] == "ubuntu-latest" for item in matrix))
+
+    def test_unicode_digits_fail_closed_with_warning(self) -> None:
+        plan = runner_plan.build_plan(
+            workflow="e2e",
+            run_id="108",
+            families=E2E_FAMILIES,
+            burst="true",
+            percent="²",
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+
+        self.assertTrue(plan.warnings)
+        self.assertEqual(plan.plan["changes_runner"], "ubuntu-latest")
+        matrix = plan.plan["e2e_matrix"]["include"]
+        self.assertTrue(all(item["runner"] == "ubuntu-latest" for item in matrix))
+
+    def test_full_percentage_uses_external_for_non_empty_tier(self) -> None:
+        plan = runner_plan.build_plan(
+            workflow="frontend",
+            run_id="107",
+            families=FRONTEND_FAMILIES,
+            burst="true",
+            percent="100",
+            light_label="light-runner",
+            standard_label="standard-runner",
+        )
+
+        self.assertEqual(plan.plan["frontend_runner"], "standard-runner")
+        self.assertEqual(plan.plan["frontend_gate_runner"], "light-runner")
+
+
+if __name__ == "__main__":
+    unittest.main()