fix(ci): include ordinary PR Dylint in the local gate

ci/README.md

@@ -3,13 +3,19 @@
 Python scripts for CI, packaging, and development tooling. All invoked via `uv run`.
 
 `local_gate.py` runs both required Ubuntu jobs through bosn's pinned act2
-engine and checks clean source identity, native Linux x64 execution and both
-job verdicts. Install bosn 0.1.11 or newer for the stock runner tools.
+engine, then replays the entire existing `dylint.yml` workflow with the ordinary
+unlabelled PR event. It checks clean source identity, native Linux x64 execution,
+all three Dylint job verdicts and the completed policy, library and workspace
+steps. Install bosn 0.1.11 or newer for the stock runner tools.
 `local-gate.toml` enforces PR attestations. The local gate uses the existing
-workflow-dispatch event to run these same jobs before stamping the tree;
+workflow-dispatch event for the Ubuntu pair and PR/minimal for Dylint before
+stamping the tree. Dispatch Dylint adds cross-target work, so it is not used as
+a substitute for the ordinary PR selection. Library UI fixtures retain their
+existing source-change condition. Neither workflow is modified or filtered;
 PR verification is confined to PR events. Board builds, extended/full
 mode and other native hosts retain their remote coverage. The migration is
-tracked in [#1635](https://github.com/FastLED/fbuild/issues/1635).
+tracked in [#1635](https://github.com/FastLED/fbuild/issues/1635), with Dylint
+parity and complete cold/warm measurements in [#1643](https://github.com/FastLED/fbuild/issues/1643).
 
 ## Contents
 

ci/local_dylint_gate.py

@@ -0,0 +1,136 @@
+"""Replay the existing ordinary-PR Dylint workflow without selecting a job."""
+
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+from ci.local_gate import (
+    JsonValue,
+    RunProof,
+    RunSelection,
+    document,
+    verify_run,
+    wire_string,
+)
+
+SELECTION = RunSelection(
+    ".github/workflows/dylint.yml",
+    None,
+    "pull_request",
+    frozenset({"policy", "dylint", "gate"}),
+)
+
+
+@dataclass(frozen=True)
+class RequiredSteps:
+    job: str
+    names: tuple[str, ...]
+
+
+REQUIRED_STEPS = (
+    RequiredSteps("policy", ("Validate Dylint policy files",)),
+    RequiredSteps(
+        "dylint",
+        (
+            "Validate Dylint allowlist paths",
+            "Enforce shrink-only .fbuild allowlist",
+            "Validate platform-boundary ledgers",
+            "Prepare prebuilt Dylint tools and driver",
+            "Install rustfmt for the Dylint toolchain",
+            "Check Dylint library formatting",
+            "Test Dylint libraries",
+            "Run dylint over workspace",
+        ),
+    ),
+)
+
+
+@dataclass(frozen=True)
+class StepProof:
+    job: str
+    name: str
+    stage: str
+    status: str
+    conclusion: str | None
+
+
+def read_steps(raw: dict[str, JsonValue]) -> tuple[StepProof, ...]:
+    tree = raw.get("tree")
+    if not isinstance(tree, dict) or not isinstance(tree.get("groups"), list):
+        raise ValueError("Dylint proof lacks job groups")
+    steps: list[StepProof] = []
+    for group in tree["groups"]:
+        if not isinstance(group, dict) or not isinstance(group.get("jobs"), list):
+            raise ValueError("invalid Dylint job group")
+        for job in group["jobs"]:
+            if not isinstance(job, dict) or not isinstance(job.get("sections"), list):
+                raise ValueError("Dylint proof lacks step evidence")
+            job_id = wire_string(job, "job_id")
+            for section in job["sections"]:
+                if not isinstance(section, dict):
+                    raise ValueError("invalid Dylint step evidence")
+                conclusion = section.get("conclusion")
+                if conclusion is not None and not isinstance(conclusion, str):
+                    raise ValueError("invalid Dylint step conclusion")
+                steps.append(
+                    StepProof(
+                        job_id,
+                        *(
+                            wire_string(section, key)
+                            for key in ("name", "stage", "status")
+                        ),
+                        conclusion,
+                    )
+                )
+    return tuple(steps)
+
+
+def verify_dylint(raw: dict[str, JsonValue], workspace: Path, sha: str) -> None:
+    verify_run(RunProof.from_json(raw), workspace, sha, SELECTION)
+    steps = read_steps(raw)
+    for required in REQUIRED_STEPS:
+        for name in required.names:
+            matching = [
+                step
+                for step in steps
+                if step.job == required.job
+                and step.name == name
+                and step.stage == "Main"
+            ]
+            if (
+                len(matching) != 1
+                or matching[0].status != "completed"
+                or matching[0].conclusion != "success"
+            ):
+                raise ValueError(
+                    f"required Dylint step did not pass: {required.job}/{name}"
+                )
+
+
+def run_dylint(workspace: Path, sha: str) -> None:
+    submitted = document(
+        [
+            "bosn",
+            "ci",
+            "run",
+            "--workspace",
+            str(workspace),
+            "--workflow",
+            SELECTION.workflow,
+            "--trigger",
+            "pr",
+            "--mode",
+            "minimal",
+            "--sha",
+            sha,
+            "--timeout-secs",
+            "7200",
+            "--json",
+        ]
+    )
+    run_id = wire_string(submitted, "run")
+    print(f"bosn ordinary-PR Dylint run: {run_id}", flush=True)
+    subprocess.run(["bosn", "ci", "wait", run_id], cwd=workspace, check=True)
+    verify_dylint(document(["bosn", "ci", "show", run_id, "--json"]), workspace, sha)

ci/local_gate.py

@@ -16,7 +16,18 @@
 JsonValue = str | int | float | bool | None | list["JsonValue"] | dict[str, "JsonValue"]
 ROOT = Path(__file__).resolve().parent.parent
 WORKFLOW = ".github/workflows/ci-minimal.yml"
-REQUIRED_JOBS = frozenset({"check", "python-facade-tests"})
+REQUIRED_JOBS = frozenset({"verify", "check", "python-facade-tests"})
+
+
+@dataclass(frozen=True)
+class RunSelection:
+    workflow: str
+    job: str | None
+    event: str
+    required_jobs: frozenset[str]
+
+
+UBUNTU_SELECTION = RunSelection(WORKFLOW, "linux", "workflow_dispatch", REQUIRED_JOBS)
 
 
 def wire_string(raw: dict[str, JsonValue], key: str) -> str:
@@ -48,7 +59,7 @@ class RunProof:
     engine: str
     act_version: str
     workflow: str
-    job: str
+    job: str | None
     mode: str
     event: str
     state: str
@@ -89,53 +100,56 @@ def from_json(cls, raw: dict[str, JsonValue]) -> RunProof:
                         )
                     )
                 )
+        selected_job = raw.get("job")
+        if "job" not in raw or (
+            selected_job is not None and not isinstance(selected_job, str)
+        ):
+            raise ValueError("bosn record lacks a valid job selection")
         return cls(
             Path(wire_string(raw, "workspace")).resolve(),
             wire_string(raw, "sha"),
             raw["dirty"],
+            *(wire_string(raw, key) for key in ("engine", "act_version", "workflow")),
+            selected_job,
             *(
                 wire_string(raw, key)
-                for key in (
-                    "engine",
-                    "act_version",
-                    "workflow",
-                    "job",
-                    "mode",
-                    "event",
-                    "state",
-                    "conclusion",
-                )
+                for key in ("mode", "event", "state", "conclusion")
             ),
             wire_count(raw, "exit_code"),
             *(wire_count(counts, key) for key in ("total", "completed", "failed")),
             tuple(jobs),
         )
 
 
-def verify_run(proof: RunProof, workspace: Path, sha: str) -> None:
+def verify_run(
+    proof: RunProof,
+    workspace: Path,
+    sha: str,
+    selection: RunSelection = UBUNTU_SELECTION,
+) -> None:
     if (
         proof.workspace != workspace.resolve()
         or proof.sha != sha
         or proof.dirty is not None
         or proof.engine != "act"
         or "-act2." not in proof.act_version
-        or proof.workflow != WORKFLOW
-        or proof.job != "linux"
+        or proof.workflow != selection.workflow
+        or proof.job != selection.job
         or proof.mode != "minimal"
-        or proof.event != "workflow_dispatch"
+        or proof.event != selection.event
     ):
         raise ValueError("run does not prove this clean minimal Linux source")
     if (
         proof.state != "done"
         or proof.conclusion != "success"
         or proof.exit_code != 0
-        or proof.total < 2
+        or proof.total != len(selection.required_jobs)
         or proof.completed != proof.total
         or proof.failed != 0
         or len(proof.jobs) != proof.total
     ):
         raise ValueError("selected jobs did not all complete successfully")
-    for job_id in REQUIRED_JOBS:
+    for job_id in selection.required_jobs:
         matching = [job for job in proof.jobs if job.job_id == job_id]
         if (
             len(matching) != 1
@@ -182,6 +196,7 @@ def main() -> None:
             "-m",
             "unittest",
             "ci.test_local_gate",
+            "ci.test_local_dylint_gate",
             "ci.test_fractional_workflows",
             "ci.test_test_cache_status",
         ],
@@ -240,12 +255,18 @@ def main() -> None:
         ROOT,
         sha,
     )
+    from ci.local_dylint_gate import run_dylint
+
+    run_dylint(ROOT, sha)
     if (
         output(["git", "rev-parse", "HEAD"]).strip() != sha
         or output(["git", "status", "--porcelain", "--untracked-files=normal"]).strip()
     ):
         raise ValueError("worktree changed while the local gate ran")
-    print(f"Passed both required Ubuntu jobs: {sha}", flush=True)
+    print(
+        f"Passed both required Ubuntu jobs and the ordinary-PR Dylint workflow: {sha}",
+        flush=True,
+    )
 
 
 if __name__ == "__main__":

ci/test_local_dylint_gate.py

@@ -0,0 +1,106 @@
+"""A successful Ubuntu proof cannot substitute for ordinary-PR Dylint."""
+
+import copy
+import unittest
+from pathlib import Path
+
+from ci.local_dylint_gate import REQUIRED_STEPS, SELECTION, verify_dylint
+from ci.local_gate import JsonValue
+
+
+def completed_record() -> dict[str, JsonValue]:
+    jobs: list[JsonValue] = []
+    for job_id in sorted(SELECTION.required_jobs):
+        names = next(
+            (required.names for required in REQUIRED_STEPS if required.job == job_id),
+            (),
+        )
+        jobs.append(
+            {
+                "job_id": job_id,
+                "status": "completed",
+                "conclusion": "success",
+                "sections": [
+                    {
+                        "name": name,
+                        "stage": "Main",
+                        "status": "completed",
+                        "conclusion": "success",
+                    }
+                    for name in names
+                ],
+            }
+        )
+    return {
+        "workspace": "/repo",
+        "sha": "a" * 40,
+        "dirty": None,
+        "engine": "act",
+        "act_version": "0.2.89-act2.7",
+        "workflow": SELECTION.workflow,
+        "job": None,
+        "mode": "minimal",
+        "event": "pull_request",
+        "state": "done",
+        "conclusion": "success",
+        "exit_code": 0,
+        "jobs": {"total": 3, "completed": 3, "failed": 0},
+        "tree": {"malformed_lines": 0, "groups": [{"jobs": jobs}]},
+    }
+
+
+class DylintProofTests(unittest.TestCase):
+    def test_complete_workflow_and_executed_steps_pass(self) -> None:
+        verify_dylint(completed_record(), Path("/repo"), "a" * 40)
+
+    def test_dispatch_full_or_selected_job_cannot_prove_ordinary_pr(self) -> None:
+        for change in (
+            {"event": "workflow_dispatch"},
+            {"mode": "full"},
+            {"job": "dylint"},
+            {"workflow": ".github/workflows/ci-minimal.yml"},
+        ):
+            with self.subTest(change=change), self.assertRaises(ValueError):
+                verify_dylint({**completed_record(), **change}, Path("/repo"), "a" * 40)
+
+    def test_green_job_with_missing_skipped_or_unfinished_workspace_step_fails(
+        self,
+    ) -> None:
+        for verdict in ("skipped", "failure", None):
+            raw = completed_record()
+            tree = raw["tree"]
+            assert isinstance(tree, dict)
+            groups = tree["groups"]
+            assert isinstance(groups, list) and isinstance(groups[0], dict)
+            jobs = groups[0]["jobs"]
+            assert isinstance(jobs, list)
+            dylint = next(
+                job
+                for job in jobs
+                if isinstance(job, dict) and job["job_id"] == "dylint"
+            )
+            assert isinstance(dylint, dict) and isinstance(dylint["sections"], list)
+            workspace = dylint["sections"][-1]
+            assert isinstance(workspace, dict)
+            workspace["conclusion"] = verdict
+            with self.subTest(verdict=verdict), self.assertRaises(ValueError):
+                verify_dylint(raw, Path("/repo"), "a" * 40)
+            dylint["sections"].pop()
+            with self.assertRaises(ValueError):
+                verify_dylint(raw, Path("/repo"), "a" * 40)
+
+    def test_duplicate_successful_step_is_not_unambiguous_evidence(self) -> None:
+        raw = completed_record()
+        tree = raw["tree"]
+        assert isinstance(tree, dict) and isinstance(tree["groups"], list)
+        group = tree["groups"][0]
+        assert isinstance(group, dict) and isinstance(group["jobs"], list)
+        job = group["jobs"][0]
+        assert isinstance(job, dict) and isinstance(job["sections"], list)
+        job["sections"].append(copy.deepcopy(job["sections"][0]))
+        with self.assertRaises(ValueError):
+            verify_dylint(raw, Path("/repo"), "a" * 40)
+
+
+if __name__ == "__main__":
+    unittest.main()