v0.6.0: procedure-compliance metric, scorer, HYD coverage, experiment runbook

bench/README.md

@@ -43,6 +43,68 @@ pip install -e bench/
   the client's transcript (`--output-format stream-json`), and the client's
   system prompt is a confound for model-vs-model baselines — those go
   through `a320-bench run`.
+- `a320_bench/scoring.py` + `a320_bench/cli_score.py` — `a320-bench score
+  PATH... [--json] [--detail] [--include-errors]`: turns recorded
+  trajectories (files or run directories) into procedure-compliance
+  ScoreCards. Pure and offline — no Sim, no network, no compiled binding — so
+  a reviewer scores results with `pip install -e bench/` alone. The metric is
+  a reported **vector** (coverage, order, end_state, safety, extraneous) plus
+  a **derived scalar** for rankings; full spec in
+  [`docs/fase5-metrica.md`](../docs/fase5-metrica.md).
+
+## Score a batch
+
+```powershell
+a320-bench score runs/                       # human table + per (scenario,model) aggregates
+a320-bench score runs/ --json > scores.json  # ScoreCards + aggregates for plots
+a320-bench score runs/elec-apu-gen-fault/one.jsonl --json --detail
+```
+
+## Experiment runbook (needs provider API keys)
+
+The baselines and ablations of #20 are a loop over `run` (records) and `score`
+(measures) — no bespoke code. The protocol (models, N per cell, ablation axes,
+statistical power) is in [`docs/fase5-metrica.md`](../docs/fase5-metrica.md);
+here are the exact commands the matrix executes.
+
+Keys live in the environment, never in a file or a trajectory (the adapter
+records the model and sampling, never the credential):
+
+```powershell
+$env:ANTHROPIC_API_KEY = "..."          # Claude via litellm
+$env:GEMINI_API_KEY   = "AQ..."          # Gemini via Vertex express (AQ.* keys)
+```
+
+One cell — a model against every scenario, N runs each, into a run tree:
+
+```powershell
+$SCEN = Get-ChildItem scenarios -Recurse -Filter *.yaml |
+        Where-Object { $_.Directory.Name -ne "schema" }
+foreach ($s in $SCEN) {
+  a320-bench run --scenario $s.FullName `
+    --model anthropic/claude-opus-4-8 --runs 10 --out runs/baseline
+}
+# Gemini needs its Vertex express api_base passed through --sampling:
+foreach ($s in $SCEN) {
+  a320-bench run --scenario $s.FullName --model gemini/gemini-2.5-flash --runs 10 `
+    --out runs/baseline `
+    --sampling '{"api_base": "https://aiplatform.googleapis.com/v1beta1/publishers/google"}'
+}
+```
+
+Score the whole tree — the table separates by `(scenario, model)`, the JSON
+feeds the plots:
+
+```powershell
+a320-bench score runs/baseline                 # human table + aggregates
+a320-bench score runs/baseline --json > baseline.json
+```
+
+An ablation is the same loop with one lever changed (a scenario variant with a
+different `instructions_profile`, a `--sampling` temperature, a tool-surface
+profile) into a separate `--out runs/ablation-<name>`, scored the same way.
+`score` needs no binding and no network, so the whole measurement half runs on
+a reviewer's machine with `pip install -e bench/` alone.
 
 ## Tests
 

bench/a320_bench/__init__.py

@@ -21,6 +21,11 @@
     "ScenarioError": "a320_bench.scenario",
     "evaluate_predicate": "a320_bench.scenario",
     "load_scenario": "a320_bench.scenario",
+    "ScoreCard": "a320_bench.scoring",
+    "ScoringError": "a320_bench.scoring",
+    "score_trajectory": "a320_bench.scoring",
+    "score_file": "a320_bench.scoring",
+    "aggregate": "a320_bench.scoring",
 }
 
 __all__ = sorted(_EXPORTS)

bench/a320_bench/cli.py

@@ -15,8 +15,11 @@
 import sys
 from typing import Any
 
-from a320_bench.episode import run_episode
-from a320_bench.scenario import ScenarioError, load_scenario
+# NB: a320_bench.episode / .scenario are imported lazily inside main(), not at
+# module top. Both pull the compiled a320_sim binding (episode) or jsonschema
+# catalog checks (scenario); importing them here would make `a320-bench score`
+# — which needs neither — fail to even start on a reviewer's binding-free
+# machine (pip install -e bench/ alone). See the score dispatch below.
 
 
 def _positive_int(text: str) -> int:
@@ -75,12 +78,37 @@ def build_parser() -> argparse.ArgumentParser:
         default=None,
         help="path to write the harness's success evaluation JSON on shutdown",
     )
+
+    score = sub.add_parser(
+        "score",
+        help="score recorded trajectories (files or run directories) into "
+        "compliance ScoreCards; needs no binding, no network",
+    )
+    score.add_argument("paths", nargs="+", help="trajectory .jsonl files or directories")
+    score.add_argument("--json", action="store_true", help="emit ScoreCards + aggregates as JSON")
+    score.add_argument(
+        "--detail", action="store_true", help="include per-command classification (with --json)"
+    )
+    score.add_argument(
+        "--include-errors",
+        action="store_true",
+        help="fold provider_error runs into the aggregate means (excluded by default)",
+    )
     return parser
 
 
 def main(argv: "list[str] | None" = None) -> int:
     args = build_parser().parse_args(argv)
 
+    if args.command == "score":
+        from a320_bench.cli_score import cmd_score
+
+        return cmd_score(args.paths, as_json=args.json, detail=args.detail,
+                         include_errors=args.include_errors)
+
+    # Every path below this point needs a scenario (and, for run, the binding).
+    from a320_bench.scenario import ScenarioError, load_scenario
+
     try:
         scenario = load_scenario(args.scenario)
     except ScenarioError as exc:
@@ -97,7 +125,10 @@ def main(argv: "list[str] | None" = None) -> int:
         )
 
     # Imported here, not at module top: `run` is the only piece that needs
-    # litellm, and the error message tells the user exactly what to install.
+    # litellm (and the a320_sim binding via episode), and the error message
+    # tells the user exactly what to install.
+    from a320_bench.episode import run_episode
+
     try:
         from a320_bench.providers.litellm_adapter import LiteLLMAdapter
     except ImportError as exc:

bench/a320_bench/cli_score.py

@@ -0,0 +1,119 @@
+"""``a320-bench score``: trajectories in, ScoreCards + aggregates out.
+
+Kept apart from cli.py and binding-free on purpose: the whole point of the
+scorer is that a paper reviewer analyses trajectories with ``pip install -e
+bench/`` and nothing else — no compiled ``a320_sim``, no litellm. This module
+imports only the pure scoring layer and the stdlib.
+"""
+
+import dataclasses
+import json
+import sys
+from pathlib import Path
+
+from a320_bench.scoring import (
+    AggregateRow,
+    ScoreCard,
+    ScoringError,
+    aggregate,
+    score_file,
+)
+
+
+def _collect(paths: list[str]) -> list[Path]:
+    """Expand files and directories to a list of .jsonl trajectories.
+
+    Directories are searched recursively for ``*.jsonl`` (sorted); an explicit
+    file path is taken as-is, whatever its extension. The same trajectory
+    reached through two arguments — e.g. a directory and a file inside it, or a
+    path repeated — is collapsed by resolved path, so a run is never
+    double-counted into the aggregate means.
+    """
+    out: list[Path] = []
+    seen: set[Path] = set()
+    for raw in paths:
+        p = Path(raw)
+        candidates = sorted(p.rglob("*.jsonl")) if p.is_dir() else [p]
+        for c in candidates:
+            key = c.resolve()
+            if key in seen:
+                continue
+            seen.add(key)
+            out.append(c)
+    return out
+
+
+def _fmt(x: "float | None", width: int = 5) -> str:
+    return "  -  ".rjust(width) if x is None else f"{x:.2f}".rjust(width)
+
+
+def _print_table(cards: list[ScoreCard], aggregates: list[AggregateRow]) -> None:
+    # per-run rows
+    print(f"{'scenario':<26} {'model':<26} {'reason':<20} "
+          f"{'score':>5} {'cov':>5} {'ord':>5} {'end':>5} {'D':>2} {'A':>2} {'ext':>3}")
+    print("-" * 120)
+    for c in cards:
+        v = c.vector
+        print(
+            f"{c.scenario_id:<26.26} {c.model:<26.26} {c.reason:<20.20} "
+            f"{_fmt(c.score)} {_fmt(v.coverage)} {_fmt(v.order)} {_fmt(v.end_state)} "
+            f"{v.safety_dangerous:>2} {v.safety_anti_procedure:>2} {v.extraneous:>3}"
+        )
+    # aggregate block
+    print()
+    print(f"{'AGGREGATE  scenario':<26} {'model':<26} "
+          f"{'n':>3} {'scored':>6} {'score':>13} {'cov':>5} {'ord':>5} "
+          f"{'pass':>5} {'dngr':>5} {'perr':>5} {'inval':>5}")
+    print("-" * 120)
+    for a in aggregates:
+        # ASCII '+/-', not U+00B1: the table is printed to consoles whose
+        # encoding (cp1252 on Windows) would mojibake the headline number.
+        score_cell = (
+            f"{a.score_mean:.2f}+/-{a.score_std:.2f}"
+            if a.score_mean is not None and a.score_std is not None
+            else "   -   "
+        )
+        print(
+            f"{a.scenario_id:<26.26} {a.model:<26.26} "
+            f"{a.n:>3} {a.n_scored:>6} {score_cell:>13} "
+            f"{_fmt(a.coverage_mean)} {_fmt(a.order_mean)} "
+            f"{_fmt(a.pass_rate)} {_fmt(a.dangerous_rate)} "
+            f"{_fmt(a.provider_error_rate)} {_fmt(a.invalid_rate)}"
+        )
+
+
+def cmd_score(
+    paths: list[str], *, as_json: bool, detail: bool, include_errors: bool
+) -> int:
+    files = _collect(paths)
+    if not files:
+        print("a320-bench: no .jsonl trajectories found in the given paths", file=sys.stderr)
+        return 2
+
+    cards = []
+    for f in files:
+        try:
+            cards.append(score_file(f))
+        except (ScoringError, OSError, json.JSONDecodeError) as exc:
+            print(f"a320-bench: cannot score {f}: {exc}", file=sys.stderr)
+            return 2
+
+    aggregates = aggregate(cards, include_errors=include_errors)
+
+    if as_json:
+        payload = {
+            "cards": [_card_dict(c, detail=detail) for c in cards],
+            "aggregates": [dataclasses.asdict(a) for a in aggregates],
+        }
+        json.dump(payload, sys.stdout, indent=2, default=str)
+        sys.stdout.write("\n")
+    else:
+        _print_table(cards, aggregates)
+    return 0
+
+
+def _card_dict(card: ScoreCard, *, detail: bool) -> dict:
+    d = dataclasses.asdict(card)
+    if not detail:
+        d.pop("detail", None)
+    return d

bench/a320_bench/scenario.py

@@ -209,6 +209,50 @@ def _action(data: dict[str, Any]) -> Action:
     return Action(control=data["control"], value=data["value"], rationale=data.get("rationale", ""))
 
 
+def parse_ground_truth(raw: dict[str, Any]) -> GroundTruth:
+    """Build a GroundTruth from a scenario's ``ground_truth`` block.
+
+    Module-level and reusable so the scorer can parse the ground truth
+    embedded in a trajectory's ``meta.scenario`` with the same code path that
+    validated it at load time — one parser, no drift.
+    """
+    return GroundTruth(
+        source=SourceRef(
+            document=raw["source"]["document"],
+            revision=raw["source"]["revision"],
+            accessed=raw["source"]["accessed"],
+            url=raw["source"].get("url", ""),
+            notes=raw["source"].get("notes", ""),
+        ),
+        procedure=tuple(
+            ProcedureBlock(
+                block=b["block"],
+                ordered=b.get("ordered", False),
+                actions=tuple(_action(a) for a in b["actions"]),
+            )
+            for b in raw["procedure"]
+        ),
+        optional_actions=tuple(_action(a) for a in raw.get("optional_actions", [])),
+        forbidden_actions=tuple(
+            ForbiddenAction(
+                control=a["control"],
+                value=a["value"],
+                severity=a["severity"],
+                rationale=a.get("rationale", ""),
+            )
+            for a in raw.get("forbidden_actions", [])
+        ),
+    )
+
+
+def parse_success(raw: dict[str, Any]) -> Success:
+    """Build a Success from a scenario's ``success`` block (reused by the scorer)."""
+    return Success(
+        final_state=tuple(_predicate(p) for p in raw["final_state"]),
+        ecam_clear_of=tuple(raw.get("ecam_clear_of", [])),
+    )
+
+
 def load_scenario(path: "str | Path", *, check_catalogs: bool = True) -> Scenario:
     """Load and validate one scenario YAML.
 
@@ -254,39 +298,8 @@ def load_scenario(path: "str | Path", *, check_catalogs: bool = True) -> Scenari
             must_not_appear=tuple(data["expected_ecam"].get("must_not_appear", [])),
         ),
         task_prompt=data["task_prompt"],
-        ground_truth=GroundTruth(
-            source=SourceRef(
-                document=data["ground_truth"]["source"]["document"],
-                revision=data["ground_truth"]["source"]["revision"],
-                accessed=data["ground_truth"]["source"]["accessed"],
-                url=data["ground_truth"]["source"].get("url", ""),
-                notes=data["ground_truth"]["source"].get("notes", ""),
-            ),
-            procedure=tuple(
-                ProcedureBlock(
-                    block=b["block"],
-                    ordered=b.get("ordered", False),
-                    actions=tuple(_action(a) for a in b["actions"]),
-                )
-                for b in data["ground_truth"]["procedure"]
-            ),
-            optional_actions=tuple(
-                _action(a) for a in data["ground_truth"].get("optional_actions", [])
-            ),
-            forbidden_actions=tuple(
-                ForbiddenAction(
-                    control=a["control"],
-                    value=a["value"],
-                    severity=a["severity"],
-                    rationale=a.get("rationale", ""),
-                )
-                for a in data["ground_truth"].get("forbidden_actions", [])
-            ),
-        ),
-        success=Success(
-            final_state=tuple(_predicate(p) for p in data["success"]["final_state"]),
-            ecam_clear_of=tuple(data["success"].get("ecam_clear_of", [])),
-        ),
+        ground_truth=parse_ground_truth(data["ground_truth"]),
+        success=parse_success(data["success"]),
         budget=Budget(
             max_tool_calls=data["budget"]["max_tool_calls"],
             max_sim_time_s=data["budget"]["max_sim_time_s"],