diff --git a/.github/workflows/ci-guards.yml b/.github/workflows/ci-guards.yml index dbf6634f97d8..90fc6472fb26 100644 --- a/.github/workflows/ci-guards.yml +++ b/.github/workflows/ci-guards.yml @@ -715,6 +715,10 @@ jobs: if: ${{ matrix.group == 'ci' }} run: python3 tests/test_ci_admission_placement.py + - name: Validate that roll-up gates never read an absent signal as green + if: ${{ matrix.group == 'ci' }} + run: python3 tests/test_ci_gate_absent_signal.py + - name: Validate the Linux tests dispatch lane if: ${{ matrix.group == 'ci' }} run: python3 tests/test_linux_tests_lane.py diff --git a/.github/workflows/ci-macos.yml b/.github/workflows/ci-macos.yml index 78e75f98878c..24250041205b 100644 --- a/.github/workflows/ci-macos.yml +++ b/.github/workflows/ci-macos.yml @@ -4781,8 +4781,14 @@ jobs: macos-status: name: macOS status + # The placement jobs are here only so the gate verdict can read the `if:` + # of the jobs that need them. They are best effort (a 3-minute timeout or + # a lost runner can end them cancelled or failed), so the verdict step + # names them --context-only and never judges their own result. needs: + - admission-placement - macos-compile-admission + - late-placement - app-host-unit-tests - cli-product-tests - cli-regressions @@ -4796,6 +4802,35 @@ jobs: runs-on: ${{ github.repository_owner != 'manaflow-ai' && 'ubuntu-24.04' || github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository && !contains(fromJSON(inputs.owned_head_repos), github.event.pull_request.head.repo.full_name) && 'blacksmith-4vcpu-ubuntu-2404' || vars.LINUX_RUNNER || 'blacksmith-4vcpu-ubuntu-2404' }} timeout-minutes: 5 steps: + # An absent signal must never read green. The steps below allow a + # `skipped` need, which is right only when that job's routing turned it + # off. This asks each skipped job's own `if:` whether it was selected, + # with the same needs, variables, inputs and event fields that decided + # it, and fails when a selected job never ran (scripts/ci/gate_verdict.py; + # tests/test_ci_gate_absent_signal.py walks the needs graph). + - name: Check out the gate verdict + uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + persist-credentials: false + sparse-checkout: | + scripts/ci/gate_verdict.py + .github/workflows/ci-macos.yml + sparse-checkout-cone-mode: false + + - name: Every selected job reported + env: + GATE_NEEDS: ${{ toJSON(needs) }} + GATE_VARS: ${{ toJSON(vars) }} + GATE_INPUTS: ${{ toJSON(inputs) }} + GATE_GITHUB: >- + {"event_name": ${{ toJSON(github.event_name) }}, + "repository": ${{ toJSON(github.repository) }}, + "repository_owner": ${{ toJSON(github.repository_owner) }}, + "run_attempt": ${{ toJSON(github.run_attempt) }}, + "triggering_actor": ${{ toJSON(github.triggering_actor) }}, + "event": {"pull_request": {"head": {"repo": {"full_name": ${{ toJSON(github.event.pull_request.head.repo.full_name) }}}}}}} + run: python3 scripts/ci/gate_verdict.py --workflow .github/workflows/ci-macos.yml --gate macos-status --context-only admission-placement,late-placement + # Compile-avoidance shadow (RFC #15391, Work 3). The package lane # classifies the pull request's package edits; when every edited # package kept its interface and the package tests passed, the app diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e6bdb0bee157..43372c696ddf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1749,6 +1749,35 @@ jobs: # required check every pull request waits on. timeout-minutes: 5 steps: + # An absent signal must never read green. The steps below allow a + # `skipped` need, which is right only when that job's routing turned it + # off. This asks each skipped job's own `if:` whether it was selected, + # with the same needs, variables, inputs and event fields that decided + # it, and fails when a selected job never ran (scripts/ci/gate_verdict.py; + # tests/test_ci_gate_absent_signal.py walks the needs graph). + - name: Check out the gate verdict + uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 + with: + persist-credentials: false + sparse-checkout: | + scripts/ci/gate_verdict.py + .github/workflows/ci.yml + sparse-checkout-cone-mode: false + + - name: Every selected job reported + env: + GATE_NEEDS: ${{ toJSON(needs) }} + GATE_VARS: ${{ toJSON(vars) }} + GATE_INPUTS: ${{ toJSON(inputs) }} + GATE_GITHUB: >- + {"event_name": ${{ toJSON(github.event_name) }}, + "repository": ${{ toJSON(github.repository) }}, + "repository_owner": ${{ toJSON(github.repository_owner) }}, + "run_attempt": ${{ toJSON(github.run_attempt) }}, + "triggering_actor": ${{ toJSON(github.triggering_actor) }}, + "event": {"pull_request": {"head": {"repo": {"full_name": ${{ toJSON(github.event.pull_request.head.repo.full_name) }}}}}}} + run: python3 scripts/ci/gate_verdict.py --workflow .github/workflows/ci.yml --gate ci-status + - name: Check platform workflow routing env: PLATFORM_NEEDS: ${{ toJSON(needs) }} diff --git a/scripts/ci/gate_verdict.py b/scripts/ci/gate_verdict.py new file mode 100644 index 000000000000..74b32756f9fb --- /dev/null +++ b/scripts/ci/gate_verdict.py @@ -0,0 +1,581 @@ +#!/usr/bin/env python3 +"""A roll-up gate's verdict: every job it summarizes reported, or was routed off. + +A gate that reads `needs` and allows `success` or `skipped` cannot tell a job +its routing turned off from a job that was selected and never ran: an absent +signal reads green. This decides from the source that routed the job, its own +`if:` in the workflow file, so there is no second copy of any routing rule to +drift. For each job the gate needs: + +- `success` passes; +- `failure`, `cancelled` or anything else fails; +- `skipped` passes only when the job's `if:`, evaluated against the same + `needs` results and outputs, the repository variables, the workflow inputs + and the event fields that decided it, is false. A job whose `if:` says it + should have run was skipped for some other reason, and fails the gate. + +An `if:` this evaluator cannot resolve (an unknown function, a context field +the gate was not given, a need the gate does not see) counts as selected: it +fails closed. tests/test_ci_gate_absent_signal.py walks every gate's needs +graph and checks that each one resolves. + +Usage, from a gate job's step: + + env: + GATE_NEEDS: ${{ toJSON(needs) }} + GATE_VARS: ${{ toJSON(vars) }} + GATE_INPUTS: ${{ toJSON(inputs) }} + GATE_GITHUB: '{"event_name": ${{ toJSON(github.event_name) }}, ...}' + run: python3 scripts/ci/gate_verdict.py --workflow --gate +""" + +from __future__ import annotations + +import argparse +import json +import math +import os +import re +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +class Unresolved(Exception): + """The expression references something this evaluator cannot answer.""" + + +# -------------------------------------------------------------------------- +# Workflow reading. The runners that host gates have no PyYAML, so this reads +# only the two keys it needs, `if:` and `needs:`, from each job. The test +# compares the result with a full YAML parse of every gated workflow, so a +# layout this reader does not understand fails a pull request, not a gate. + +_JOB_HEADER = re.compile(r"^ ([A-Za-z_][A-Za-z0-9_-]*):\s*(#.*)?$") +_KEY = re.compile(r"^ ([A-Za-z_-]+):(.*)$") +_LIST_ITEM = re.compile(r"^ - ([A-Za-z0-9_-]+)\s*(#.*)?$") + + +@dataclass(frozen=True) +class Job: + name: str + needs: tuple[str, ...] + condition: str | None + + +def _scalar(raw: str) -> str: + value = raw.strip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in "'\"": + return value[1:-1] + return value + + +def read_jobs(text: str) -> dict[str, Job]: + lines = text.splitlines() + try: + start = lines.index("jobs:") + 1 + except ValueError as error: + raise ValueError("no top-level `jobs:` key") from error + jobs: dict[str, Job] = {} + current: str | None = None + needs: list[str] = [] + condition: str | None = None + in_needs = False + + def close() -> None: + if current is not None: + jobs[current] = Job(current, tuple(needs), condition) + + for line in lines[start:]: + if line and not line.startswith(" ") and not line.startswith("#"): + break # the next top-level key + header = _JOB_HEADER.match(line) + if header: + close() + current, needs, condition, in_needs = header.group(1), [], None, False + continue + if current is None: + continue + key = _KEY.match(line) + if key: + in_needs = False + name, rest = key.group(1), key.group(2).strip() + if name == "if": + if rest[:1] in {">", "|"}: + raise ValueError(f"{current}: a block-scalar `if:` is not supported") + condition = _scalar(rest) + elif name == "needs": + if rest.startswith("["): + needs = [_scalar(item) for item in rest.strip("[]").split(",") if item.strip()] + elif rest: + needs = [_scalar(rest)] + else: + in_needs = True + continue + if in_needs: + item = _LIST_ITEM.match(line) + if item: + needs.append(item.group(1)) + elif line.strip() and not line.strip().startswith("#"): + in_needs = False + close() + return jobs + + +# -------------------------------------------------------------------------- +# GitHub Actions expressions, the subset workflow `if:` conditions use. + +_TOKEN = re.compile( + r"\s*(?:" + r"(?P'(?:[^']|'')*')" + r"|(?P-?(?:0x[0-9a-fA-F]+|\d+(?:\.\d+)?(?:[eE][+-]?\d+)?))" + r"|(?P==|!=|<=|>=|&&|\|\||[<>!().,\[\]*])" + r"|(?P[A-Za-z_][A-Za-z0-9_-]*)" + r")" +) + + +def _tokens(source: str) -> list[tuple[str, str]]: + out: list[tuple[str, str]] = [] + position = 0 + source = source.rstrip() + while position < len(source): + match = _TOKEN.match(source, position) + if not match or match.end() == position: + raise Unresolved(f"cannot read the expression at {source[position:]!r}") + position = match.end() + kind = match.lastgroup + assert kind is not None + out.append((kind, match.group(kind))) + return out + + +class _Filtered(list): + """The result of `.*`: property access maps over the elements.""" + + +def _truthy(value: Any) -> bool: + if value is None or value is False: + return False + if isinstance(value, (int, float)) and not isinstance(value, bool): + return not (value == 0 or (isinstance(value, float) and math.isnan(value))) + if isinstance(value, str): + return value != "" + return True + + +def _number(value: Any) -> float: + if value is None: + return 0.0 + if isinstance(value, bool): + return 1.0 if value else 0.0 + if isinstance(value, (int, float)): + return float(value) + if isinstance(value, str): + text = value.strip() + if text == "": + return 0.0 + try: + return float(int(text, 16)) if text.lower().startswith("0x") else float(text) + except ValueError: + return math.nan + return math.nan + + +def _equal(left: Any, right: Any) -> bool: + if isinstance(left, str) and isinstance(right, str): + return left.casefold() == right.casefold() + if left is None and right is None: + return True + if isinstance(left, (dict, list)) or isinstance(right, (dict, list)): + return left is right + if type(left) is type(right) and not isinstance(left, str): + return left == right + return _number(left) == _number(right) + + +def _compare(left: Any, right: Any) -> float: + if isinstance(left, str) and isinstance(right, str): + a, b = left.casefold(), right.casefold() + return (a > b) - (a < b) + a, b = _number(left), _number(right) + if math.isnan(a) or math.isnan(b): + return math.nan + return (a > b) - (a < b) + + +def _string(value: Any) -> str: + if value is None: + return "" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, float) and value.is_integer(): + return str(int(value)) + if isinstance(value, (dict, list)): + return json.dumps(value) + return str(value) + + +class _Strict(dict): + """A context object whose missing keys are a gap in what the gate was given. + + `deep` objects (the event fields) are strict all the way down; `needs` is + strict only about which jobs it holds, since a job's outputs may be unset. + """ + + def __init__(self, data: dict[str, Any], *, deep: bool): + super().__init__(data) + self.deep = deep + + def lookup(self, name: str) -> Any: + for key, item in self.items(): + if key.casefold() == name.casefold(): + if self.deep and isinstance(item, dict): + return _Strict(item, deep=True) + return item + raise Unresolved(f"context field {name!r} was not passed to the gate") + + +class _Vars(dict): + """Repository variables: an unset one reads as the empty string, as in GitHub.""" + + +def _member(value: Any, name: str) -> Any: + if isinstance(value, _Filtered): + return _Filtered(_member(item, name) for item in value if isinstance(item, dict)) + if isinstance(value, _Strict): + return value.lookup(name) + if isinstance(value, dict): + for key, item in value.items(): + if key.casefold() == name.casefold(): + return item + return "" if isinstance(value, _Vars) else None + return None + + +class _Parser: + """Builds a small tree, so `&&` and `||` can short-circuit as GitHub does.""" + + def __init__(self, tokens: list[tuple[str, str]]): + self.tokens = tokens + self.index = 0 + + def peek(self) -> tuple[str, str] | None: + return self.tokens[self.index] if self.index < len(self.tokens) else None + + def take(self, value: str | None = None) -> tuple[str, str]: + token = self.peek() + if token is None or (value is not None and token[1] != value): + raise Unresolved(f"expected {value!r}, found {token!r}") + self.index += 1 + return token + + def parse(self) -> tuple: + node = self.binary(0) + if self.peek() is not None: + raise Unresolved(f"unexpected {self.peek()!r}") + return node + + LEVELS = (("||",), ("&&",), ("==", "!="), ("<", "<=", ">", ">=")) + + def binary(self, level: int) -> tuple: + if level == len(self.LEVELS): + return self.unary() + node = self.binary(level + 1) + while self.peek() is not None and self.peek()[0] == "op" and self.peek()[1] in self.LEVELS[level]: + op = self.take()[1] + node = ("bin", op, node, self.binary(level + 1)) + return node + + def unary(self) -> tuple: + if self.peek() == ("op", "!"): + self.take() + return ("not", self.unary()) + return self.postfix() + + def postfix(self) -> tuple: + kind, text = self.take() + if kind == "str": + node: tuple = ("lit", text[1:-1].replace("''", "'")) + elif kind == "num": + node = ("lit", _number(text)) + elif text == "(": + node = self.binary(0) + self.take(")") + elif kind == "ident" and self.peek() == ("op", "("): + self.take("(") + args: list[tuple] = [] + if self.peek() != ("op", ")"): + args.append(self.binary(0)) + while self.peek() == ("op", ","): + self.take() + args.append(self.binary(0)) + self.take(")") + node = ("call", text.lower(), args) + elif kind == "ident" and text in {"true", "false", "null"}: + node = ("lit", {"true": True, "false": False, "null": None}[text]) + elif kind == "ident": + node = ("ctx", text) + else: + raise Unresolved(f"unexpected {text!r}") + while self.peek() in {("op", "."), ("op", "[")}: + if self.take()[1] == ".": + kind, name = self.take() + if name == "*": + node = ("star", node) + elif kind == "ident": + node = ("get", node, ("lit", name)) + else: + raise Unresolved(f"bad property {name!r}") + else: + key = self.binary(0) + self.take("]") + node = ("get", node, key) + return node + + +def _run(node: tuple, context: dict[str, Any], status: dict[str, bool]) -> Any: + kind = node[0] + if kind == "lit": + return node[1] + if kind == "ctx": + if node[1] not in context: + raise Unresolved(f"unknown context {node[1]!r}") + return context[node[1]] + if kind == "not": + return not _truthy(_run(node[1], context, status)) + if kind == "star": + value = _run(node[1], context, status) + return _Filtered(value.values() if isinstance(value, dict) + else value if isinstance(value, list) else []) + if kind == "get": + value = _run(node[1], context, status) + key = _run(node[2], context, status) + if isinstance(value, list) and not isinstance(value, _Filtered): + index = _number(key) + return value[int(index)] if index.is_integer() and 0 <= index < len(value) else None + return _member(value, _string(key)) + if kind == "bin": + op = node[1] + left = _run(node[2], context, status) + if op == "&&": + return _run(node[3], context, status) if _truthy(left) else left + if op == "||": + return left if _truthy(left) else _run(node[3], context, status) + right = _run(node[3], context, status) + if op == "==": + return _equal(left, right) + if op == "!=": + return not _equal(left, right) + order = _compare(left, right) + if math.isnan(order): + return False + return {"<": order < 0, "<=": order <= 0, ">": order > 0, ">=": order >= 0}[op] + if kind == "call": + name, arg_nodes = node[1], node[2] + if name in status: + if arg_nodes: + raise Unresolved(f"{name}() takes no arguments here") + return status[name] + args = [_run(arg, context, status) for arg in arg_nodes] + if name == "contains" and len(args) == 2: + haystack, needle = args + if isinstance(haystack, list): + return any(_equal(item, needle) for item in haystack) + return _string(needle).casefold() in _string(haystack).casefold() + if name == "startswith" and len(args) == 2: + return _string(args[0]).casefold().startswith(_string(args[1]).casefold()) + if name == "endswith" and len(args) == 2: + return _string(args[0]).casefold().endswith(_string(args[1]).casefold()) + if name == "fromjson" and len(args) == 1: + try: + return json.loads(_string(args[0])) + except ValueError as error: + raise Unresolved(f"fromJSON of {args[0]!r}") from error + if name == "tojson" and len(args) == 1: + return json.dumps(args[0], indent=2) + if name == "format" and args: + template = _string(args[0]).replace("{{", "\0").replace("}}", "\1") + filled = re.sub(r"\{(\d+)\}", lambda m: _string(args[1 + int(m.group(1))]), template) + return filled.replace("\0", "{").replace("\1", "}") + if name == "join" and args: + separator = _string(args[1]) if len(args) > 1 else "," + items = args[0] if isinstance(args[0], list) else [args[0]] + return separator.join(_string(item) for item in items) + raise Unresolved(f"unsupported function {name}()") + raise Unresolved(f"unknown node {kind}") + + +_STATUS = re.compile(r"\b(success|failure|always|cancelled)\s*\(", re.IGNORECASE) + + +def strip_expression(condition: str) -> str: + text = condition.strip() + if text.startswith("${{") and text.endswith("}}"): + return text[3:-2].strip() + return text + + +def evaluate(expression: str, context: dict[str, Any], status: dict[str, bool]) -> Any: + return _run(_Parser(_tokens(expression)).parse(), context, status) + + +# -------------------------------------------------------------------------- +# The verdict. + +PASSING = "success" + + +def _ancestors(job: Job, jobs: dict[str, Job] | None) -> set[str]: + """Every job upstream of this one. GitHub's status functions see the whole + chain: a skipped or failed ancestor skips an implicit-success dependent + even when the job between them ran under !cancelled().""" + seen: set[str] = set() + pending = list(job.needs) + while pending: + name = pending.pop() + if name in seen: + continue + seen.add(name) + if jobs is not None: + if name not in jobs: + raise Unresolved(f"{job.name} has ancestor {name}, which is not in the workflow") + pending.extend(jobs[name].needs) + return seen + + +def selected(job: Job, needs: dict[str, Any], context: dict[str, Any], *, + jobs: dict[str, Job] | None = None, tolerate_variable_flips: bool = True) -> bool: + """Whether the job's own `if:` says it should have run, given what the gate sees. + + With `tolerate_variable_flips` off this is plain routing: the `if:` as + GitHub evaluates it with these variables. Pass `jobs`: without it the + status functions see only direct needs, and a skipped grandparent can make + a job look selected when GitHub skipped it.""" + results = {} + for name in _ancestors(job, jobs): + if name not in needs: + raise Unresolved(f"{job.name} depends on {name}, which the gate does not see") + results[name] = needs[name].get("result") + status = { + # The gate itself runs under !cancelled(), so the run is not cancelled. + "cancelled": False, + "always": True, + "success": all(result == "success" for result in results.values()), + "failure": any(result == "failure" for result in results.values()), + } + expression = strip_expression(job.condition) if job.condition else "" + if not expression: + return status["success"] + scope = dict(context) + scope["needs"] = _Strict({name: {"result": needs[name].get("result"), + "outputs": needs[name].get("outputs") or {}} + for name in job.needs}, deep=False) + implicit = not _STATUS.search(expression) + if implicit and not status["success"]: + # A condition without a status function runs only when its needs passed. + return False + # Repository variables are read again here, after the job was routed, and + # an operator may have flipped one in between (or between re-run + # attempts). A skip that some value of the variables this `if:` reads + # explains is therefore accepted: a variable is a deliberate switch, not + # a routing signal, and reading it late must not red every PR in flight. + current = context.get("vars") or _Vars() + choices = _variable_choices(expression, current) if tolerate_variable_flips else [_Vars(current)] + for variables in choices: + scope["vars"] = variables + if not _truthy(evaluate(expression, scope, status)): + return False + return True + + +_VAR_REF = re.compile(r"\bvars\.([A-Za-z_][A-Za-z0-9_]*)", re.IGNORECASE) + + +def _variable_choices(expression: str, current: dict[str, Any]) -> list[dict[str, Any]]: + """The current variables, then each referenced one set to every value the + expression compares it with, or to unset.""" + choices = [_Vars(current)] + literals = [match.group(0)[1:-1].replace("''", "'") + for match in re.finditer(r"'(?:[^']|'')*'", expression)] + for name in sorted(set(_VAR_REF.findall(expression))): + for value in ["", *literals, "x"]: + # Replace the variable however its key is cased: lookups ignore case. + others = {key: item for key, item in current.items() if key.casefold() != name.casefold()} + choices.append(_Vars({**others, name: value})) + return choices + + +def problems(jobs: dict[str, Job], gate: str, needs: dict[str, Any], context: dict[str, Any], + context_only: frozenset[str] = frozenset()) -> list[str]: + """What keeps the gate from passing. `context_only` names best-effort needs + the gate reads only to evaluate other needs' `if:`; their own result, + whatever it is, is not a verdict.""" + found: list[str] = [] + for name in jobs[gate].needs: + entry = needs.get(name) + if entry is None: + found.append(f"{name}: no result reached the gate") + continue + if name in context_only: + continue + result = entry.get("result") + if result == PASSING: + continue + if result != "skipped": + found.append(f"{name}: {result}") + continue + try: + wanted = selected(jobs[name], needs, context, jobs=jobs) + except Unresolved as error: + found.append(f"{name}: skipped, and its routing could not be read ({error}); " + "a skip nobody can explain is not a pass") + continue + if wanted: + found.append(f"{name}: skipped, though its own `if:` selected it; " + "it never ran, so nothing stands behind this gate") + return found + + +def context_from_environment(environ: dict[str, str]) -> dict[str, Any]: + def load(name: str) -> Any: + return json.loads(environ.get(name) or "{}") or {} + + return { + "vars": _Vars(load("GATE_VARS")), + "inputs": load("GATE_INPUTS"), + "github": _Strict(load("GATE_GITHUB"), deep=True), + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--workflow", required=True, type=Path) + parser.add_argument("--gate", required=True) + parser.add_argument("--context-only", default="", + help="comma-separated best-effort needs read only for other needs' `if:`") + args = parser.parse_args(argv) + context_only = frozenset(name for name in args.context_only.split(",") if name) + + jobs = read_jobs(args.workflow.read_text()) + if args.gate not in jobs: + print(f"no job {args.gate!r} in {args.workflow}", file=sys.stderr) + return 1 + needs = json.loads(os.environ.get("GATE_NEEDS") or "{}") + unknown = context_only - set(jobs[args.gate].needs) + if unknown: + print(f"--context-only names jobs {args.gate} does not need: {sorted(unknown)}", file=sys.stderr) + return 1 + found = problems(jobs, args.gate, needs, context_from_environment(dict(os.environ)), context_only) + if found: + for line in found: + print(f"::error::{line}") + return 1 + for name in jobs[args.gate].needs: + print(f"{name}: {needs[name]['result']}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/ci/workflow_guard_groups.py b/scripts/ci/workflow_guard_groups.py index 16e27aa4ee5b..b947971fb383 100644 --- a/scripts/ci/workflow_guard_groups.py +++ b/scripts/ci/workflow_guard_groups.py @@ -53,6 +53,8 @@ # derived from ci-guards.yml by direct_path_owners() and need no entry here. PATH_OWNERS = { ".github/workflows/ci-main-full-suite.yml": frozenset(("ci",)), + # tests/test_ci_gate_absent_signal.py imports the gate verdict. + "scripts/ci/gate_verdict.py": frozenset(("ci",)), # test_ci_runner_capability_resolver.py reads the capability map, the # resolver it imports, the reusable workflow that publishes the map, and diff --git a/tests/test-execution.toml b/tests/test-execution.toml index e3b653338b5d..1f25f2de1400 100644 --- a/tests/test-execution.toml +++ b/tests/test-execution.toml @@ -1677,6 +1677,10 @@ lane = "linux-guard" path = "tests/test_ci_admission_placement.py" lane = "linux-guard" +[[test]] +path = "tests/test_ci_gate_absent_signal.py" +lane = "linux-guard" + [[test]] path = "tests/test_ci_run_guards.py" lane = "linux-guard" diff --git a/tests/test_ci_gate_absent_signal.py b/tests/test_ci_gate_absent_signal.py new file mode 100644 index 000000000000..14a50bf6665c --- /dev/null +++ b/tests/test_ci_gate_absent_signal.py @@ -0,0 +1,436 @@ +#!/usr/bin/env python3 +"""Roll-up gates never read an absent signal as green. + +ci-status and macOS status summarize the jobs they need. A `skipped` need is +a pass only when that job's routing turned it off; a job its routing selected +that never ran (a cancelled, missing or skipped upstream) must fail the gate. +scripts/ci/gate_verdict.py decides that from each job's own `if:`. This test: + +- checks the gate's workflow reader against a full YAML parse; +- checks each gate runs the verdict, on everything it needs, and that the + needs graph is closed so every `if:` can be evaluated; +- checks every job of the workflow is summarized by its gate unless it cannot + fail the run; +- walks the needs graph over seeded routing scenarios: the routed results must + pass, and each selected need turned skipped, cancelled or failed must fail; +- pins the GitHub expression semantics the verdict relies on. +""" +from __future__ import annotations + +import functools +import importlib.util +import json +import random +import re +import sys +import unittest +from pathlib import Path + +import yaml + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "scripts/ci/gate_verdict.py" +GATES = { + ".github/workflows/ci.yml": "ci-status", + ".github/workflows/ci-macos.yml": "macos-status", +} +# Jobs a gate does not summarize, other than ones that cannot fail the run +# (job-level continue-on-error), each with the reason it does not block. +NOT_SUMMARIZED = { + (".github/workflows/ci.yml", "dogfood-build"): + "label-requested delivery of a dogfood ZIP; it tests nothing", +} +# Gates that decide admission rather than report a verdict: everything they +# read must also reach the verdict gate, so letting a skip through admits work +# but cannot pass a pull request. +PRE_GATES = { + ".github/workflows/ci.yml": ("macos-admission-gate",), +} +VERDICT_STEP = "Every selected job reported" +SCENARIOS = 400 + + +def load(): + spec = importlib.util.spec_from_file_location("gate_verdict", SCRIPT) + module = importlib.util.module_from_spec(spec) + sys.modules["gate_verdict"] = module + spec.loader.exec_module(module) + return module + + +verdict = load() + + +@functools.lru_cache(maxsize=None) +def parsed(path: str) -> dict: + return yaml.safe_load((ROOT / path).read_text()) + + +def condition_of(job: dict) -> str | None: + value = job.get("if") + if value is None: + return None + return value if isinstance(value, str) else str(value).lower() + + +def needs_of(job: dict) -> tuple[str, ...]: + value = job.get("needs") or () + return (value,) if isinstance(value, str) else tuple(value) + + +def verdict_step(path: str) -> dict: + gate = parsed(path)["jobs"][GATES[path]] + steps = [step for step in gate["steps"] if step.get("name") == VERDICT_STEP] + if len(steps) != 1: + raise AssertionError(f"{path} {GATES[path]}: expected one {VERDICT_STEP!r} step") + return steps[0] + + +def context_only(path: str) -> frozenset[str]: + match = re.search(r"--context-only (\S+)", verdict_step(path)["run"]) + return frozenset(match.group(1).split(",")) if match else frozenset() + + +def reads_variables(job) -> bool: + return "vars." in (job.condition or "") + + +REFERENCE = re.compile(r"\b(needs|inputs|vars|github)((?:\.[A-Za-z0-9_*-]+)+)") +LITERAL = re.compile(r"'((?:[^']|'')*)'|\b(\d+)\b") +_REF = r"(?:needs|inputs|vars|github)(?:\.[A-Za-z0-9_*-]+)+" +COMPARED = re.compile( + rf"(?P{_REF})\s*(?:==|!=|<=|>=|<|>)\s*'(?P(?:[^']|'')*)'" + rf"|(?PstartsWith|endsWith|contains)\(\s*(?P{_REF})\s*,\s*'(?P(?:[^']|'')*)'\s*\)", + re.IGNORECASE) +EXPRESSION = re.compile(r"\$\{\{\s*toJSON\((github(?:\.[A-Za-z0-9_]+)+)\)\s*\}\}") + + +@functools.lru_cache(maxsize=None) +def jobs_of(path: str) -> dict: + return verdict.read_jobs((ROOT / path).read_text()) + + +class Scenario: + """One seeded routing: event fields, variables, inputs and job outputs.""" + + def __init__(self, path: str, rng: random.Random): + self.path = path + self.jobs = jobs_of(path) + self.gate = GATES[path] + conditions = " ".join(job.condition or "" for job in self.jobs.values()) + literals = {match.group(1) if match.group(1) is not None else match.group(2) + for match in LITERAL.finditer(conditions)} + domain = sorted(literals | {"", "x"}) + refs = {(m.group(1), m.group(2)) for m in REFERENCE.finditer(conditions)} + + # Each reference mostly takes a value its own conditions compare it + # with, so long conjunctions are reached as often as their negations. + compared: dict[str, set[str]] = {} + for match in COMPARED.finditer(conditions): + if match.group("fn"): + ref, literal = match.group("fref"), match.group("flit") + else: + ref, literal = match.group("ref"), match.group("lit") + literal = literal.replace("''", "'") + if match.group("fn"): + literal = {"startswith": literal + "std", "endswith": "std" + literal}.get( + match.group("fn").lower(), "a" + literal + "b") + compared.setdefault(ref, set()).add(literal) + + def pick(ref: str = ""): + near = sorted(compared.get(ref, ())) + return rng.choice(near) if near and rng.random() < 0.8 else rng.choice(domain) + + self.outputs: dict[str, dict[str, str]] = {} + self.inputs: dict[str, str] = {} + variables: dict[str, str] = {} + for root, rest in sorted(refs): + parts = rest.strip(".").split(".") + if root == "needs" and len(parts) == 3 and parts[1] == "outputs": + self.outputs.setdefault(parts[0], {})[parts[2]] = pick(root + rest) + elif root == "inputs": + self.inputs[parts[0]] = pick(root + rest) + elif root == "vars" and rng.random() < 0.8: + variables[parts[0]] = pick(root + rest) # sometimes unset + template = verdict_step(path)["env"]["GATE_GITHUB"] + events = [value for value in domain if value in {"pull_request", "merge_group", "workflow_dispatch", "push"}] + + def fill(match: re.Match) -> str: + field = match.group(1) + if field == "github.event_name": + return json.dumps(rng.choice(events or ["pull_request"])) + if field == "github.run_attempt": + return json.dumps(str(rng.randint(1, 4))) + return json.dumps(pick(field) if rng.random() < 0.9 else None) + + self.context = verdict.context_from_environment({ + "GATE_VARS": json.dumps(variables), + "GATE_INPUTS": json.dumps(self.inputs), + "GATE_GITHUB": EXPRESSION.sub(fill, template), + }) + + def routed(self) -> dict[str, dict]: + """Each job's result as GitHub would route it: ran and passed, or skipped.""" + needs: dict[str, dict] = {} + pending = list(self.jobs[self.gate].needs) + while pending: + progressed = False + for name in list(pending): + job = self.jobs[name] + if any(need not in needs for need in job.needs): + continue + ran = verdict.selected(job, needs, self.context, jobs=self.jobs, + tolerate_variable_flips=False) + needs[name] = {"result": "success" if ran else "skipped", + "outputs": dict(self.outputs.get(name, {})) if ran else {}} + pending.remove(name) + progressed = True + if not progressed: + raise AssertionError(f"{self.path}: needs cycle among {pending}") + return {name: needs[name] for name in self.jobs[self.gate].needs} + + +class ReaderTests(unittest.TestCase): + def test_reader_matches_yaml_for_every_job(self): + for path in GATES: + with self.subTest(path=path): + jobs = verdict.read_jobs((ROOT / path).read_text()) + full = parsed(path)["jobs"] + self.assertEqual(sorted(jobs), sorted(full)) + for name, job in full.items(): + self.assertEqual(jobs[name].needs, needs_of(job), name) + self.assertEqual(jobs[name].condition, condition_of(job), name) + + +class WiringTests(unittest.TestCase): + def test_gate_runs_the_verdict_on_everything_it_needs(self): + for path, gate in GATES.items(): + with self.subTest(path=path): + step = verdict_step(path) + self.assertNotIn("continue-on-error", step) + expected = f"python3 scripts/ci/gate_verdict.py --workflow {path} --gate {gate}" + if context_only(path): + expected += " --context-only " + ",".join(sorted(context_only(path))) + self.assertEqual(step["run"].strip(), expected) + self.assertEqual(step["env"]["GATE_NEEDS"], "${{ toJSON(needs) }}") + self.assertEqual(step["env"]["GATE_VARS"], "${{ toJSON(vars) }}") + self.assertEqual(step["env"]["GATE_INPUTS"], "${{ toJSON(inputs) }}") + job = parsed(path)["jobs"][gate] + steps = job["steps"] + checkout = steps[steps.index(step) - 1] + self.assertTrue(checkout.get("uses", "").startswith("actions/checkout@")) + sparse = checkout["with"]["sparse-checkout"].split() + self.assertIn("scripts/ci/gate_verdict.py", sparse) + self.assertIn(path, sparse) + self.assertEqual(job.get("if"), "${{ !cancelled() }}") + + def test_needs_graph_is_closed_under_the_gate(self): + for path, gate in GATES.items(): + jobs = verdict.read_jobs((ROOT / path).read_text()) + seen = set(jobs[gate].needs) + for name in jobs[gate].needs: + for need in verdict._ancestors(jobs[name], jobs): + with self.subTest(path=path, job=name, need=need): + self.assertIn(need, seen, f"{gate} must need {need} to read {name}'s `if:`") + + def test_every_job_that_can_fail_is_summarized(self): + for path, gate in GATES.items(): + full = parsed(path)["jobs"] + summarized = set(needs_of(full[gate])) + for name, job in full.items(): + if name == gate or name in summarized or job.get("continue-on-error") is True: + continue + with self.subTest(path=path, job=name): + self.assertIn((path, name), NOT_SUMMARIZED, + f"{name} can fail but {gate} does not summarize it") + + def test_pre_gates_feed_the_verdict_gate(self): + for path, names in PRE_GATES.items(): + full = parsed(path)["jobs"] + summarized = set(needs_of(full[GATES[path]])) + for name in names: + with self.subTest(path=path, job=name): + self.assertIn(name, summarized) + self.assertLessEqual(set(needs_of(full[name])), summarized) + + def test_every_event_field_an_if_reads_is_passed(self): + for path, gate in GATES.items(): + jobs = verdict.read_jobs((ROOT / path).read_text()) + passed = set(EXPRESSION.findall(verdict_step(path)["env"]["GATE_GITHUB"])) + for name in jobs[gate].needs: + for match in REFERENCE.finditer(jobs[name].condition or ""): + if match.group(1) == "github": + with self.subTest(path=path, job=name): + self.assertIn("github" + match.group(2), passed) + + +class NeedsGraphWalk(unittest.TestCase): + def test_absent_signals_fail_and_routed_skips_pass(self): + for path, gate in GATES.items(): + rng = random.Random(f"{path}:20261020") + best_effort = context_only(path) + selected_seen: set[str] = set() + skipped_seen: set[str] = set() + for index in range(SCENARIOS): + scenario = Scenario(path, rng) + needs = scenario.routed() + + def check(broken): + return verdict.problems(scenario.jobs, gate, broken, scenario.context, best_effort) + + self.assertEqual(check(needs), [], f"{path} scenario {index}: routed results must pass") + for name, entry in needs.items(): + if entry["result"] == "skipped": + skipped_seen.add(name) + mutations = ("failure", "cancelled") + else: + selected_seen.add(name) + mutations = ("cancelled", "failure") + # A skip a repository-variable flip explains is accepted. + if not reads_variables(scenario.jobs[name]): + mutations = ("skipped", *mutations) + for result in mutations: + found = check({**needs, name: {"result": result, "outputs": {}}}) + mentions = any(line.startswith(f"{name}:") for line in found) + if name in best_effort: + self.assertFalse(mentions, f"{path} scenario {index}: best-effort {name} " + f"{result} must not fail {gate}") + else: + self.assertTrue(mentions, f"{path} scenario {index}: {gate} passed with {name} {result}") + missing = {key: value for key, value in needs.items() if key != name} + self.assertTrue(check(missing)) + # The walk reached every need both ways, so no job escaped it. + every = set(scenario.jobs[gate].needs) + self.assertEqual(selected_seen, every, f"{path}: needs never selected") + self.assertTrue(skipped_seen, f"{path}: no routed skip was exercised") + + def test_best_effort_needs_cannot_hold_back_what_needs_them(self): + # A context-only need is never judged, so it must be one whose result + # cannot stop a dependent: every step continue-on-error, and every job + # that needs it runs whatever its result (a status function, and no + # read of its result). + for path, gate in GATES.items(): + full = parsed(path)["jobs"] + jobs = jobs_of(path) + for name in context_only(path): + with self.subTest(path=path, job=name): + self.assertIn(name, jobs[gate].needs) + self.assertTrue(all(step.get("continue-on-error") is True + for step in full[name]["steps"])) + for other, job in jobs.items(): + if other == gate or name not in job.needs: + continue + condition = verdict.strip_expression(job.condition or "") + self.assertRegex(condition, verdict._STATUS, other) + self.assertNotIn(f"needs.{name}.result", condition, other) + + def test_variable_flip_between_routing_and_gate_is_not_red(self): + path = ".github/workflows/ci-macos.yml" + jobs = jobs_of(path) + job = jobs["late-placement"] + needs = {"macos-compile-admission": {"result": "success", "outputs": {}}} + github = {"event_name": "pull_request", "repository": "manaflow-ai/cmux", + "repository_owner": "manaflow-ai", "run_attempt": "1", "triggering_actor": "a", + "event": {"pull_request": {"head": {"repo": {"full_name": "manaflow-ai/cmux"}}}}} + context = verdict.context_from_environment({ + "GATE_VARS": json.dumps({"CI_PR_POOL_OWNED": "1", "GLAEDA_ROUTE_APP_ID": "7"}), + "GATE_GITHUB": json.dumps(github)}) + # Routed with these variables, it runs ... + self.assertTrue(verdict.selected(job, needs, context, tolerate_variable_flips=False)) + # ... but had CI_PR_POOL_OWNED been off when it was routed, its skip is explained. + self.assertFalse(verdict.selected(job, needs, context)) + + def test_pr_21287_compile_admission_never_ran(self): + # macOS compile admission sat queued until merge; had it ended without + # running, the gate must not have read its skip as a routed one. + path = ".github/workflows/ci-macos.yml" + jobs = verdict.read_jobs((ROOT / path).read_text()) + inputs = {"macos": "true", "cli": "false", "full_suite": "false", "compile_admitted": "false", + "unit_suite": "false", "swift_packages": "false", "release_build": "true", + "pr_root_runner": "glaeda-root-std-xcode-26.6", "pr_owned_jobs": " admission "} + context = verdict.context_from_environment({ + "GATE_INPUTS": json.dumps(inputs), + "GATE_VARS": json.dumps({"CI_OWNED_SPREAD": "0"}), + "GATE_GITHUB": json.dumps({ + "event_name": "pull_request", "repository": "manaflow-ai/cmux", + "repository_owner": "manaflow-ai", "run_attempt": "1", "triggering_actor": "a", + "event": {"pull_request": {"head": {"repo": {"full_name": "manaflow-ai/cmux"}}}}}), + }) + needs = {name: {"result": "skipped", "outputs": {}} for name in jobs["macos-status"].needs} + found = verdict.problems(jobs, "macos-status", needs, context, context_only(path)) + self.assertEqual([line.split(":")[0] for line in found], ["macos-compile-admission"]) + + +class ExpressionSemantics(unittest.TestCase): + STATUS = {"cancelled": False, "always": True, "success": True, "failure": False} + + def run_expression(self, text: str, **context): + return verdict.evaluate(text, context, self.STATUS) + + def test_string_comparison_ignores_case(self): + self.assertTrue(self.run_expression("a == 'TRUE'", a="true")) + + def test_missing_output_is_null_and_differs_from_false(self): + needs = verdict._Strict({"changes": {"result": "success", "outputs": {}}}, deep=False) + self.assertTrue(self.run_expression("needs.changes.outputs.web != 'false'", needs=needs)) + + def test_unknown_need_or_event_field_is_unresolved(self): + needs = verdict._Strict({}, deep=False) + with self.assertRaises(verdict.Unresolved): + self.run_expression("needs.other.result == 'success'", needs=needs) + github = verdict._Strict({"event_name": "pull_request"}, deep=True) + with self.assertRaises(verdict.Unresolved): + self.run_expression("github.ref == 'x'", github=github) + + def test_unset_variable_is_empty_string(self): + self.assertTrue(self.run_expression("vars.X == ''", vars=verdict._Vars())) + self.assertTrue(self.run_expression("vars.X != '1'", vars=verdict._Vars())) + + def test_and_or_short_circuit_and_return_operands(self): + needs = verdict._Strict({}, deep=False) + self.assertFalse(self.run_expression("false && needs.other.result", needs=needs)) + self.assertEqual(self.run_expression("'' || 'b'"), "b") + self.assertEqual(self.run_expression("'a' && 'b'"), "b") + + def test_number_coercion(self): + self.assertTrue(self.run_expression("a <= 2", a="2")) + self.assertFalse(self.run_expression("a <= 2", a="3")) + + def test_functions(self): + self.assertTrue(self.run_expression("startsWith(a, 'glaeda-root-')", a="glaeda-root-std")) + self.assertTrue(self.run_expression("contains(a, ' admission ')", a="x admission y")) + self.assertTrue(self.run_expression("contains(fromJSON(a), 'o/r')", a='["o/r"]')) + self.assertTrue(self.run_expression("contains(e.labels.*.name, 'full-ci')", + e={"labels": [{"name": "full-ci"}]})) + with self.assertRaises(verdict.Unresolved): + self.run_expression("hashFiles('x')") + + def test_status_functions_see_every_ancestor(self): + # A skipped, B ran under !cancelled(), C has an implicit success(): + # GitHub skips C, so its skip is a routed one, not an absent signal. + jobs = { + "a": verdict.Job("a", (), "${{ inputs.x == 'true' }}"), + "b": verdict.Job("b", ("a",), "${{ !cancelled() }}"), + "c": verdict.Job("c", ("b",), "${{ inputs.y == 'true' }}"), + "gate": verdict.Job("gate", ("a", "b", "c"), "${{ !cancelled() }}"), + } + context = {"inputs": {"x": "false", "y": "true"}} + needs = {"a": {"result": "skipped"}, "b": {"result": "success"}, "c": {"result": "skipped"}} + self.assertFalse(verdict.selected(jobs["c"], needs, context, jobs=jobs)) + self.assertEqual(verdict.problems(jobs, "gate", needs, context), []) + ran = {**needs, "a": {"result": "success"}} + found = verdict.problems(jobs, "gate", ran, {"inputs": {"x": "true", "y": "true"}}) + self.assertEqual(len(found), 1) + self.assertTrue(found[0].startswith("c: skipped, though its own `if:` selected it"), found) + + def test_implicit_success_without_a_status_function(self): + job = verdict.Job("j", ("a",), "${{ inputs.x == 'true' }}") + context = {"inputs": {"x": "true"}} + self.assertFalse(verdict.selected(job, {"a": {"result": "skipped"}}, context)) + self.assertTrue(verdict.selected(job, {"a": {"result": "success"}}, context)) + explicit = verdict.Job("j", ("a",), "${{ !cancelled() && inputs.x == 'true' }}") + self.assertTrue(verdict.selected(explicit, {"a": {"result": "skipped"}}, context)) + + +if __name__ == "__main__": + unittest.main()