open-standards/scripts/standards-touched.py

#!/usr/bin/env python3
"""standards-touched.py: which standards workspaces does a pull request touch?

The standards build tests every Rust workspace of the tree, one matrix leg each, on
three kinds of runner. Building all of them for every pull request filled the queue:
a change to one standard rebuilt and retested all the others, though none of their
inputs had changed. This decides, for each workspace, whether a change can have
changed what its build and tests see, and the workflow builds only those.

A workspace counts as touched when the change touches any of:
  1. its own area: the first component of its directory (for `jcp/jcp-rs`, all of
     `jcp/`), which holds its conformance data, documents and examples too;
  2. the area of every crate it depends on by path, followed through those crates'
     own path dependencies;
  3. any file outside those areas that one of its Rust sources names with a relative
     path (`include_str!`, `include_bytes!`, `#[path]`, or a string literal of a
     path a test reads);
  4. anything every leg shares: the workflow file, this script and the scripts the
     workflow runs, the toolchain pin, and the cargo configuration.

    standards-touched.py <base-sha> <head-sha> [--workflow FILE] [--github-output]

Prints one JSON object: {"run": [names], "skip": [names], "all": bool, "why": {...}}.
With --github-output it also writes, to $GITHUB_OUTPUT, `include` (the matrix entries
to build, as JSON), `any` and `energy` (whether the macOS energy job runs).
With no base or head (a push, not a pull request), or anything it cannot determine,
every workspace runs: a wrong "skip" silently drops a test, a wrong "run" only costs
runner time, so it guesses toward running.
"""
import json
import os
import re
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
WORKFLOW = ".github/workflows/standards-build.yml"

# Every leg reads these; a change to any of them runs every workspace.
SHARED = [
    WORKFLOW,
    "scripts/standards-touched.py",
    "rust-toolchain.toml",
    "rust-toolchain",
    ".cargo/",
]


def matrix(workflow: Path) -> list[dict]:
    """The `include:` entries of the build job's matrix, read without a YAML library
    (the runner image has none guaranteed): each entry is a `- { key: value, ... }`
    flow mapping on one line, as the workflow writes them."""
    out = []
    for line in workflow.read_text().splitlines():
        m = re.match(r"\s*-\s*\{(.*)\}\s*$", line)
        if not m:
            continue
        entry = {}
        for part in re.findall(r'(\w+)\s*:\s*("(?:[^"\\]|\\.)*"|[^,]+)', m.group(1)):
            k, v = part
            v = v.strip()
            if v.startswith('"') and v.endswith('"'):
                v = v[1:-1]
            elif v in ("true", "false"):
                v = v == "true"
            entry[k] = v
        if "name" in entry and "dir" in entry:
            out.append(entry)
    return out


def scripts_used(workflow: Path) -> list[str]:
    """Paths under scripts/ that the workflow names; each is shared by every leg."""
    return sorted(set(re.findall(r"\b(scripts/[\w./-]+)", workflow.read_text())))


def area(path: str) -> str:
    """The first component of a repository path, as a prefix: `jcp/jcp-rs` → `jcp/`."""
    first = path.strip("/").split("/", 1)[0]
    return first + "/"


def path_deps(crate_dir: Path) -> set[Path]:
    """The directories this crate or workspace names with `path = "..."` in any of its
    manifests, resolved."""
    found = set()
    for manifest in crate_dir.rglob("Cargo.toml"):
        if "target" in manifest.relative_to(crate_dir).parts:
            continue
        for rel in re.findall(r'path\s*=\s*"([^"]+)"', manifest.read_text(errors="replace")):
            p = (manifest.parent / rel).resolve()
            if p.exists():
                found.add(p if p.is_dir() else p.parent)
    return found


REL = re.compile(r'"((?:\.\./)+[^"\s]+)"')


def outside_reads(crate_dir: Path, inside: list[str]) -> set[str]:
    """Repository paths outside `inside` that the crate's Rust sources name by a
    relative path climbing out of their own directory."""
    found = set()
    for src in crate_dir.rglob("*.rs"):
        parts = src.relative_to(crate_dir).parts
        if "target" in parts:
            continue
        for rel in REL.findall(src.read_text(errors="replace")):
            for base in (src.parent, crate_dir):
                p = (base / rel).resolve()
                if not p.exists():
                    continue
                try:
                    r = p.relative_to(ROOT).as_posix()
                except ValueError:
                    continue
                if not any(r.startswith(a) for a in inside):
                    found.add(r)
    return found


def inputs_of(entry: dict) -> tuple[list[str], list[str]]:
    """(areas, exact paths) whose change can change what this workspace's leg sees."""
    ws = ROOT / entry["dir"]
    areas = {area(entry["dir"])}
    seen, todo = set(), [ws]
    while todo:
        d = todo.pop()
        if d in seen:
            continue
        seen.add(d)
        for dep in path_deps(d):
            try:
                areas.add(area(dep.relative_to(ROOT).as_posix()))
            except ValueError:
                continue
            todo.append(dep)
    exact = outside_reads(ws, sorted(areas))
    return sorted(areas), sorted(exact)


def changed(base: str, head: str) -> list[str]:
    out = subprocess.run(
        ["git", "-C", str(ROOT), "diff", "--name-only", f"{base}...{head}"],
        capture_output=True, text=True, check=True,
    ).stdout
    return [l for l in out.splitlines() if l]


def decide(entries: list[dict], files: list[str], shared: list[str]) -> dict:
    run, skip, why = [], [], {}
    hit_shared = [f for f in files if any(f == s or (s.endswith("/") and f.startswith(s)) for s in shared)]
    for e in entries:
        if hit_shared:
            run.append(e["name"]); why[e["name"]] = f"shared: {hit_shared[0]}"
            continue
        areas, exact = inputs_of(e)
        hit = next((f for f in files if any(f.startswith(a) for a in areas)
                    or any(f == x or f.startswith(x.rstrip("/") + "/") for x in exact)), None)
        if hit:
            run.append(e["name"]); why[e["name"]] = hit
        else:
            skip.append(e["name"])
    return {"run": run, "skip": skip, "all": not skip, "why": why}


# The workspaces whose energy backends the macOS energy job tests; it runs when any of
# them does.
ENERGY = {"jcp-rs", "proof-rs", "eoc-rs", "jouleclaw-rs", "sandbox-rs", "joule-code-rs"}


def main(argv: list[str]) -> int:
    workflow = WORKFLOW
    argv = [a for a in argv if a != "--github-output"]
    if "--workflow" in argv:
        i = argv.index("--workflow")
        workflow = argv[i + 1]
        argv = argv[:i] + argv[i + 2:]
    base, head = (argv + ["", ""])[:2]
    try:
        entries = matrix(ROOT / workflow)
        if not entries:
            raise ValueError("no matrix entries read")
    except Exception as exc:  # cannot even list the workspaces: say so and fail
        print(f"standards-touched: cannot read the matrix: {exc}", file=sys.stderr)
        return 2
    names = [e["name"] for e in entries]
    try:
        if not base or not head or set(base) == {"0"}:
            raise ValueError("not a pull request")
        files = changed(base, head)
        result = decide(entries, files, SHARED + scripts_used(ROOT / workflow))
    except Exception as exc:  # fail open: every workspace runs
        result = {"run": names, "skip": [], "all": True, "why": {"*": f"running all: {exc}"}}
    print(json.dumps(result, sort_keys=True))
    out = os.environ.get("GITHUB_OUTPUT")
    if "--github-output" in sys.argv[1:] and out:
        keep = set(result["run"])
        include = [e for e in entries if e["name"] in keep]
        with open(out, "a") as fh:
            fh.write("include=" + json.dumps(include, separators=(",", ":")) + "\n")
            fh.write("any=" + ("true" if include else "false") + "\n")
            fh.write("energy=" + ("true" if keep & ENERGY else "false") + "\n")
        for name in result["run"]:
            print(f"  runs:  {name} ({result['why'].get(name, result['why'].get('*', ''))})")
        for name in result["skip"]:
            print(f"  skips: {name}")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))