Context collectors

docs/design/harness-adapter.md

@@ -132,7 +132,7 @@ A collector is a named command registered for one of two points, `before` or `af
 {"blas": "openblas-0.3.27", "simd_baseline": ["SSE", "SSE2", "SSE3"]}
 ```
 
-The driving half captures that output as a provenance artifact and adds the facts to the context document's `environment.system` under the collector's name, here `environment.system.numpy-build.blas`, with the collector's name as their source. Translation therefore stays repeatable from artifacts, and a collector cannot collide with the snapshot or with another collector. Facts are copied to every result of the invocation they were collected around. A collector that fails, times out, or prints anything else contributes no facts, and those results carry the warning `collector-failed`; nothing is substituted and the run continues.
+The driving half captures that output as a provenance artifact and adds the facts to the context document's `environment.system` under the collector's name, here `environment.system.numpy-build.blas`, with the collector's name as their source. Translation therefore stays repeatable from artifacts, and a collector cannot collide with the snapshot or with another collector. Facts are copied to every result of the invocation they were collected around. A collector may also be held to declared facts (`verify`, `expect`) and refuse the order when they differ; the prototype implements this beyond the design's "nothing checks one against the other" (`prototype-design.md` §3). A collector that fails, times out, or prints anything else contributes no facts, and those results carry the warning `collector-failed`; nothing is substituted and the run continues.
 
 **Setup file.** The setup file is a channel for declared facts (`benchmark-environments.md` §3.2), next to the work order and project configuration, for the case where the party that prepares an environment is not the party that runs benchmarks in it. That party writes what it knows into `.benchx/setup.json` at the root of the benchmark suite, as one JSON object of facts in the same shape:
 

docs/design/prototype-design.md

@@ -80,7 +80,7 @@ a TARGET ending in `.py` selects it.
 walking up from the working directory; its directory is the root, and with no file the
 root is the working directory (`schemas/config/0.1.0`). It holds `project`, `source.uri`,
 `provider.command`, an `adapter` block (`name`, `suite`, `quantities`, `precision`),
-`environment_variables`, `timeouts`, `paths.store` and `paths.results` (default
+`environment_variables`, `collectors`, `timeouts`, `paths.store` and `paths.results` (default
 `store.parquet` and `results/` in the root, and relative to the root), and `compare`
 (`profile`, `rounds`, `k`, `min_rounds`). A command-line argument wins; if it is missing the
 value is read from the file; if that is missing too, a built-in default applies, where one
@@ -321,6 +321,24 @@ the same way report the same configuration. Nobody declared these, so the detect
   (`harness-adapter.md` §5.1), for what the cache cannot tell, such as how a
   dependency was installed. The prototype has no placement rules, so they
   land in `environment.system.setup`, which the default identity policy matches.
+- **Context collectors** (`harness-adapter.md` §5.1), named under `collectors` in
+  `benchx.conf.json` and copied into the work order. The runner runs each command once,
+  before the cases, in the harness's working directory and environment (`{root}` in an
+  argument is the project root), under `timeout_s` (default 60). It prints one JSON object.
+  The runner saves it whole as an artifact, and puts its `summary` object and its
+  sha256 into every result's `environment.system.<name>`. A collector that fails adds no
+  facts and gives the results `collector-failed`. Only `before`, once per order, is
+  implemented. The commands are executed, so `bx run` refuses an order that has
+  collectors unless given `--allow-collectors`; `bx bench` and `bx compare` build the
+  order from the user's own configuration and allow them.
+- **Verification** of a collector's facts: `verify: "how_built"` requires every key of the
+  target's `how_built`, when that is an object (a provider may declare facts there; as a
+  string it is text and never checked), to be among the collected facts with exactly the
+  same value, at any depth; extra collected keys are fine. `expect` requires facts by
+  value. A failed check, or a collector that cannot run, refuses the order before any case
+  (`on_fail: refuse`, the default), or with `on_fail: warn` gives every result the
+  warning `collector-mismatch`. `prototype/examples/numpy-bench` uses this to check that
+  the numpy a provider described is the numpy the workers import.
 
 `benchmark-environments.md` §3.4 asks that every fact carry its source
 (`detected`, `declared`, or a plugin name). The prototype does not record it

docs/design/prototype-scope.md

@@ -40,7 +40,7 @@ one run:
 
 | Component | In prototype | Deferred | Out of scope |
 |---|---|---|---|
-| Runner | execute a `build` or `working_tree` target per a work order; apply the order's `environment_variables` to the launched process; capture the zero-configuration snapshot, the build configuration, and source identity; account for planned cases; refuse an order it cannot apply exactly, including one that names a deferred field | `python_env` targets, context collectors beyond the built-in ones, environment policy (enforcing launch settings, verifying requested hardware, refusing an unsatisfiable run) | building revisions, tuning the machine (`benchmark-environments.md` §2.3) |
+| Runner | execute a `build` or `working_tree` target per a work order; apply the order's `environment_variables` to the launched process; capture the zero-configuration snapshot, the build configuration, and source identity; account for planned cases; refuse an order it cannot apply exactly, including one that names a deferred field | `python_env` targets, context collectors that run after the harness or per case (the prototype runs them once per order, before the cases), environment policy (enforcing launch settings, verifying requested hardware, refusing an unsatisfiable run) | building revisions, tuning the machine (`benchmark-environments.md` §2.3) |
 | Adapter | Google Benchmark, both halves: drive filter and repetitions, translate output; pyperf, both halves, for Python functions in a working tree, through `bx bench` and `bx compare` (the default adapter of both) | all other harnesses; pyperf's `tags`, `date`, per-run load and frequency, and `pyperf check` | — |
 | Store | one Parquet file; ingest per schema §5.4 (validated and durable, auto-create, unit-conflict rejection, idempotency, attempt integrity); one default identity policy; queries: list series, series history, get result | continuity events, quarantine, artifact storage, identity-policy authoring | — |
 | Comparator | run mode with the `revisions` and `environments` profiles; eligibility guard; comparison document | history mode; `single-run`, `variants`, and `cross-machine` profiles; batch requests; provisional noise rules | — |

prototype/benchx/cli.py

@@ -55,7 +55,7 @@ def cmd_run(args):
     out = Path(args.out) if args.out else cfg.results_dir
     snapshot.ignore(cfg.store_path, cfg.results_dir, out)
     try:
-        summary = runner.run(args.order, out)
+        summary = runner.run(args.order, out, allow_collectors=args.allow_collectors)
     except runner.Refused as e:
         return _refused(e)
     print(f"{summary['results']} result(s) for {summary['cases']} case(s) -> {summary['out']}"
@@ -127,6 +127,7 @@ def _common(cfg, args, *, forced_adapter=None, forced_suite=None) -> dict:
         "source_uri": args.source_uri or cfg.section("source").get("uri"),
         "build": not args.no_build,
         "provider": cfg.provider,
+        "collectors": cfg.collectors,
         "results_dir": cfg.results_dir,
         # An asv case is every round of its benchmark, each bounded by the benchmark's own timeout.
         "case_s": cfg.pick(getattr(args, "case_timeout", None), timeouts.get("case_s"),
@@ -465,6 +466,8 @@ def main(argv=None):
     p.add_argument("--out", help="directory for result files (default ./results)")
     p.add_argument("--no-ingest", action="store_true",
                    help="only write the result files; do not deliver them to the local store")
+    p.add_argument("--allow-collectors", action="store_true",
+                   help="run the order's context collectors: commands the order names (otherwise refused)")
     p.set_defaults(fn=cmd_run)
 
     p = sub.add_parser("bench", help="benchmark one target: the single-target form of bx compare",

prototype/benchx/collectors.py

@@ -0,0 +1,113 @@
+"""Context collectors (harness-adapter.md §5.1), and the check against declared facts.
+
+A collector is a command the order names. The runner runs it once, before the
+cases, in the harness's working directory and environment, outside any timed
+region and under a time limit. It prints one JSON object of facts. The facts go
+to the artifacts in full, and to every result's `environment.system.<name>` as their
+`summary` object (if they have one) and their sha256.
+
+A collector that fails, times out, or prints anything else adds no facts and
+gives the results the warning `collector-failed`; nothing is substituted.
+
+A collector may also be held to what was declared: `verify: how_built` requires
+every key of the target's `how_built` object to be among the collected facts
+with exactly the same value, and `expect` requires facts by value. A failed
+check refuses the order, or with `on_fail: warn` gives the results the warning
+`collector-mismatch`. A collector that cannot run cannot verify, so under
+`refuse` that refuses too.
+"""
+
+import json
+import subprocess
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from . import core
+from .core import Refusal, loader
+
+DEFAULT_TIMEOUT_S = 60.0
+MAX_DIFFS = 5
+
+
+@dataclass
+class Collected:
+    system: dict = field(default_factory=dict)  # name -> what goes into environment.system
+    artifacts: list = field(default_factory=list)  # (kind, media_type, path)
+    warnings: list = field(default_factory=list)
+    refusals: list = field(default_factory=list)
+
+
+def _differences(wanted, found, path=""):
+    """The paths at which `found` does not contain `wanted` exactly. A mapping is
+    compared key by key (extra keys in `found` are fine); anything else by its
+    canonical JSON, so `true` is not `1` as Python's `==` would have it."""
+    if isinstance(wanted, dict):
+        if not isinstance(found, dict):
+            yield f"{path or '.'}: expected an object, found {_short(found)}"
+            return
+        for key, value in wanted.items():
+            if key not in found:
+                yield f"{path}/{key}: missing (declared {_short(value)})"
+            else:
+                yield from _differences(value, found[key], f"{path}/{key}")
+    elif core.canonical(wanted) != core.canonical(found):
+        yield f"{path or '.'}: declared {_short(wanted)}, observed {_short(found)}"
+
+
+def _short(value, limit=80) -> str:
+    text = json.dumps(value, sort_keys=True)
+    return text if len(text) <= limit else text[:limit] + "..."
+
+
+def _collect(spec, cwd, env) -> tuple[dict | None, str | None]:
+    """The facts a collector prints, or (None, why not)."""
+    try:
+        done = subprocess.run(spec["command"], cwd=cwd, env=env, capture_output=True, text=True,
+                              timeout=spec.get("timeout_s", DEFAULT_TIMEOUT_S))
+    except OSError as e:
+        return None, f"cannot run {spec['command'][0]!r}: {e}"
+    except subprocess.TimeoutExpired:
+        return None, f"timed out after {spec.get('timeout_s', DEFAULT_TIMEOUT_S):g} s"
+    if done.returncode != 0:
+        last = (done.stderr.strip().splitlines() or ["no message"])[-1]
+        return None, f"exit {done.returncode}: {last}"
+    facts, errors = loader.parse(done.stdout)
+    if facts is None:
+        return None, "; ".join(str(error) for error in errors)
+    return facts, None
+
+
+def run_all(specs, *, declared, cwd, env, artifacts_dir, taken=()) -> Collected:
+    """Run each collector. `declared` is the target's how_built, `taken` the
+    environment.system names already in use."""
+    out = Collected()
+    for spec in specs:
+        name = spec["name"]
+        if name in taken or name in out.system:
+            out.refusals.append(Refusal(f"collector {name!r}: that name is already in environment.system"))
+            continue
+        checks = bool(spec.get("verify") or spec.get("expect"))
+        refuse = spec.get("on_fail", "refuse") == "refuse"
+        facts, why = _collect(spec, cwd, env)
+        problems = []
+        if facts is None:
+            out.warnings.append("collector-failed")
+            problems.append(f"collector {name!r} failed: {why}")
+        else:
+            digest = core.sha256_hex(core.canonical(facts))
+            path = Path(artifacts_dir) / f"collector-{name}.json"
+            path.write_bytes(core.pretty(facts))
+            out.artifacts.append(("collector-output", "application/json", path))
+            summary = facts.get("summary") if isinstance(facts.get("summary"), dict) else {}
+            out.system[name] = {**summary, "sha256": digest}
+            if spec.get("verify") == "how_built":
+                if isinstance(declared, dict):
+                    problems += [f"collector {name!r}: how_built {d}" for d in _differences(declared, facts)]
+                else:
+                    problems.append(f"collector {name!r}: verify: how_built, but the target's how_built is not an object")
+            problems += [f"collector {name!r}: expect {d}" for d in _differences(spec.get("expect", {}), facts)]
+            if problems:
+                out.warnings.append("collector-mismatch")
+        if problems and checks and refuse:  # an unchecked collector that fails only warns
+            out.refusals += [Refusal(problem) for problem in problems[:MAX_DIFFS]]
+    return out

prototype/benchx/compare.py

@@ -239,6 +239,8 @@ def _declared(document) -> dict:
     runner saw: the declarations are recorded as given and never judged."""
     info = document["provenance"].get("info", {})
     side = {k: info[k] for k in ("how_built", "activation", "shell", "target_provider") if k in info}
+    if isinstance(side.get("how_built"), dict):  # facts, shown as the JSON text they are
+        side["how_built"] = json.dumps(side["how_built"], sort_keys=True, separators=(",", ":"))
     system = document["coordinates"]["environment"].get("system") or {}
     if env := (system.get("process") or {}).get("environment"):
         side["observed_env"] = env

prototype/benchx/config.py

@@ -92,6 +92,17 @@ def store_path(self) -> Path:
     def results_dir(self) -> Path:
         return self._path(self.section("paths").get("results", DEFAULT_RESULTS))
 
+    @property
+    def collectors(self) -> list[dict]:
+        """The context collectors, as a work order lists them. `{root}` in a command
+        argument is the project root, because a collector runs in the harness's
+        working directory, not here."""
+        out = []
+        for name, spec in self.section("collectors").items():
+            command = [argument.replace("{root}", str(self.root)) for argument in spec["command"]]
+            out.append({"name": name, **spec, "command": command})
+        return out
+
     @property
     def provider(self) -> tuple[list[str], Path] | None:
         """The provider command, and the directory it runs in (the root)."""

prototype/benchx/orders.py

@@ -27,7 +27,8 @@ def default_run_key(prefix: str) -> str:
 
 def make_order(*, suite: dict, target: dict, precision: dict, run_key: str, quantities=("wall-time",),
                case_s: float = CASE_TIMEOUT_S, order_s: float = ORDER_TIMEOUT_S, project=None, labels=None,
-               environment_variables=None, round_=None, slot=None, benchmark=None) -> dict:
+               environment_variables=None, round_=None, slot=None, benchmark=None,
+               collectors=None) -> dict:
     """One resolved work order. Optional fields that are None or empty are left out.
 
     Raises OrderError when the result is not a valid work order.
@@ -59,6 +60,8 @@ def make_order(*, suite: dict, target: dict, precision: dict, run_key: str, quan
         order["project"] = project
     if benchmark:
         order["benchmark"] = benchmark
+    if collectors:
+        order["collectors"] = list(collectors)
     errors = core.check(order, "work-order")
     if errors:
         raise OrderError("; ".join(str(error) for error in errors))

prototype/benchx/runner.py

@@ -12,7 +12,7 @@
 from datetime import datetime, timezone
 from pathlib import Path
 
-from . import __version__, adapters, core, snapshot
+from . import __version__, adapters, collectors, core, snapshot
 from . import target as target_mod
 from .adapters.base import Unsupported
 from .core import Refusal, Rejected
@@ -118,15 +118,20 @@ def _benchmark_mismatches(stated: dict | None, found: dict | None):
                       f"but the suite is in {found['uri']}{'@' + found['revision'] if 'revision' in found else ''}")
 
 
-def run(order_path, out_dir) -> dict:
-    """Execute one order; write results and the order into out_dir."""
+def run(order_path, out_dir, allow_collectors=False) -> dict:
+    """Execute one order; write results and the order into out_dir. The order's
+    context collectors are commands the runner would execute, so they run only
+    if `allow_collectors`."""
     inspection = core.inspect(Path(order_path), "work-order")
     if inspection.data is None:
         raise Refused(inspection.errors)
     order = inspection.data
     refusals = list(check(order))
     if refusals:
         raise Refused(refusals)
+    if order.get("collectors") and not allow_collectors:
+        raise Refused([Refusal("the order names context collectors, which are commands to run; "
+                               "pass --allow-collectors to run it")])
     adapter = adapters.get(order["suites"][0]["adapter"])
     applied_protocol, invocation = adapter.protocol(order["precision"])
 
@@ -177,6 +182,11 @@ def run(order_path, out_dir) -> dict:
     slot, round_ = order.get("slot"), order.get("round")
     artifacts_dir = out_dir / "artifacts" / (f"slot-{slot}" if slot is not None else ref[7:19])
     artifacts_dir.mkdir(parents=True, exist_ok=True)
+    collected = collectors.run_all(order.get("collectors", []), declared=order["target"].get("how_built"),
+                                   cwd=target_path, env=child_env, artifacts_dir=artifacts_dir, taken=system)
+    if collected.refusals:
+        raise Refused(collected.refusals)
+    system.update(collected.system)
     order_file = out_dir / f"workorder-{ref[7:]}.json"
     order_file.write_bytes(core.pretty(order))
 
@@ -209,6 +219,7 @@ def run(order_path, out_dir) -> dict:
         if native_path.exists():
             artifacts.insert(0, _artifact("native-output", "application/json", native_path))
         artifacts += [_artifact(kind, media_type, path) for kind, media_type, path in run_.get("artifacts", [])]
+        artifacts += [_artifact(kind, media_type, path) for kind, media_type, path in collected.artifacts]
         native_system, native_info = adapter.context_facts(run_["native"])
         case_system, case_warnings = snapshot.merge(system, native_system), []
         if hasattr(adapter, "observed_environment"):
@@ -290,7 +301,7 @@ def run(order_path, out_dir) -> dict:
                 "procedure": procedure,
                 "provenance": provenance,
             }
-            warnings = [] if condition else ([setup_warning] if setup_warning else []) + case_warnings
+            warnings = [] if condition else ([setup_warning] if setup_warning else []) + collected.warnings + case_warnings
             if warnings:
                 document["quality"] = {"warnings": warnings}
             errors = core.check(document, "measurement-result")