Procedure-compliance metric + pure scorer module (#85)

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/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"],