diff --git a/bench/a320_bench/__init__.py b/bench/a320_bench/__init__.py index 14cc71f..d01512d 100644 --- a/bench/a320_bench/__init__.py +++ b/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) diff --git a/bench/a320_bench/scenario.py b/bench/a320_bench/scenario.py index f0d158d..5cb47c4 100644 --- a/bench/a320_bench/scenario.py +++ b/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"], diff --git a/bench/a320_bench/scoring.py b/bench/a320_bench/scoring.py new file mode 100644 index 0000000..dc4a6c1 --- /dev/null +++ b/bench/a320_bench/scoring.py @@ -0,0 +1,591 @@ +"""Procedure-compliance scoring: a recorded trajectory becomes a ScoreCard. + +Pure and offline by design — no Sim, no network, no compiled binding. The #20 +scorer runs on a paper reviewer's machine with ``pip install -e bench/`` and +nothing else, because a trajectory (``a320_bench.recorder``) is self-contained: +the scenario and its ground truth are embedded in the ``meta`` record, so +scoring never re-simulates. The full metric — the matching, every component, +every edge case, and the derived scalar — is specified in +``docs/fase5-metrica.md``; this module is that document made executable. + +Design commitments (see the doc and D-026): +- **Only command actions score.** Effective ``set_control`` calls are matched + against the ground truth; reads are free, failed/off-surface calls are + reported but never scored. +- **Vector first, scalar second.** Every component is reported; the scalar is + derived for rankings only, and safety is a gate (one dangerous action floors + it to 0), not a trade-off. +""" + +import statistics +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from a320_bench.recorder import read_trajectory +from a320_bench.scenario import GroundTruth, parse_ground_truth + +# The observation tools: calling these is always free (observing is good). +READ_TOOLS = frozenset( + {"read_ecam", "read_state", "snapshot", "list_controls", "list_failures"} +) + +# Derived-scalar weights (doc §escalar). end_state and coverage carry the +# result-and-procedure halves; order is the smaller "when" term, redistributed +# to coverage when there is no order evidence. +W_END_STATE = 0.4 +W_COVERAGE = 0.4 +W_ORDER = 0.2 + +# Forbidden-action severity weights: dangerous is a hard floor (factor 0), +# anti_procedure a heavy but survivable penalty. +SEVERITY_WEIGHT = {"dangerous": 1.0, "anti_procedure": 0.5} + +# Extraneous commands pay a small rent, capped: noise is not a crime. +TIDINESS_PER_COMMAND = 0.05 +TIDINESS_CAP = 5 + + +class ScoringError(Exception): + """A trajectory is too corrupt to score — names what is malformed. + + Raised only for structural corruption that no valid trajectory produces (a + non-errored ``set_control`` without a control/value, a run whose embedded + scenario carries no ground truth). Truncated or non-scoreable-but-well-formed + runs are *not* errors: they yield a card marked ``incomplete``/``scored=False``. + Kept as loud as ``read_trajectory``/``ScenarioError``: corruption never + degrades silently into a wrong number. + """ + + +@dataclass(frozen=True) +class CommandClass: + """One effective set_control command and how it was classified.""" + + index: int # position among effective commands (0-based) + control: str + value: float + classification: str # matched_required | optional | forbidden | extraneous + detail: str = "" # block ref | severity | extraneous subtype + + +@dataclass(frozen=True) +class ScoreVector: + """The reported components. This, not the scalar, is the result.""" + + coverage: float + order: "float | None" + end_state: "float | None" + all_passed: "bool | None" + safety_dangerous: int + safety_anti_procedure: int + extraneous: int + + +@dataclass(frozen=True) +class ScoreInfo: + """Informative signals — reported, never folded into the scalar.""" + + efficiency: "float | None" + tool_calls_used: "int | None" + sim_time_used: "float | None" + reads: dict[str, int] + observed_before_acting: bool + reads_before_first_command: int + errored_commands: int + off_surface_attempts: list[str] + judge_bundle: dict[str, Any] + + +@dataclass(frozen=True) +class ScoreCard: + """The full scoring of one run. ``score`` is None when not scoreable.""" + + run_id: str + scenario_id: str + model: str + reason: str + scored: bool + incomplete: bool + score: "float | None" + vector: ScoreVector + info: ScoreInfo + detail: dict[str, Any] = field(default_factory=dict) + + +# --- record helpers ----------------------------------------------------------- +def _record(records: list[dict], type_: str) -> "dict | None": + for r in records: + if r.get("type") == type_: + return r + return None + + +def _tool_calls(records: list[dict]) -> list[dict]: + return [r for r in records if r.get("type") == "tool_call"] + + +# --- command extraction ------------------------------------------------------- +@dataclass +class _Extracted: + commands: list[tuple[str, float]] # effective (control, value), in order + reads: dict[str, int] + reads_before_first_command: int + observed_before_acting: bool + errored_commands: int + off_surface_attempts: list[str] + + +def extract_commands(records: list[dict]) -> _Extracted: + """Pull the effective command sequence and the informative read/error signals. + + Effective = ``set_control`` with ``is_error == false``. Reads are counted; + failed set_controls and off-surface tool calls (inject/clear_failure and + anything else that is not a read, a command or report_done) are reported + but never scored. + """ + commands: list[tuple[str, float]] = [] + reads: dict[str, int] = {} + reads_before_first_command = 0 + saw_read_before_command = False + errored_commands = 0 + off_surface: list[str] = [] + first_command_seen = False + + for call in _tool_calls(records): + name = call.get("name", "") + is_error = bool(call.get("is_error")) + if name == "set_control": + if is_error: + errored_commands += 1 + continue + args = call.get("args") or {} + try: + control = args["control"] + value = float(args["value"]) + except (KeyError, TypeError, ValueError) as exc: + raise ScoringError( + f"corrupt trajectory: a successful set_control (is_error=false) " + f"has malformed args {args!r} ({exc}); an effective command must " + f"carry a 'control' name and a numeric 'value'" + ) from exc + commands.append((control, value)) + first_command_seen = True + elif name in READ_TOOLS: + reads[name] = reads.get(name, 0) + 1 + if not first_command_seen: + reads_before_first_command += 1 + if name == "read_ecam": + saw_read_before_command = True + elif name == "report_done": + continue # terminal channel, not an action on the aircraft + else: + off_surface.append(name) + + return _Extracted( + commands=commands, + reads=reads, + reads_before_first_command=reads_before_first_command, + observed_before_acting=saw_read_before_command, + errored_commands=errored_commands, + off_surface_attempts=off_surface, + ) + + +# --- the required plan, flattened to canonical order -------------------------- +@dataclass(frozen=True) +class _Required: + block_index: int + action_index: int # position within its block + ordered: bool # is the block internally ordered + control: str + value: float + block: str + + +def _flatten_required(gt: GroundTruth) -> list[_Required]: + out: list[_Required] = [] + for bi, block in enumerate(gt.procedure): + for ai, action in enumerate(block.actions): + out.append( + _Required( + block_index=bi, + action_index=ai, + ordered=block.ordered, + control=action.control, + value=float(action.value), + block=block.block, + ) + ) + return out + + +def _catalog_set(actions) -> set[tuple[str, float]]: + return {(a.control, float(a.value)) for a in actions} + + +# --- scoring one trajectory --------------------------------------------------- +def score_trajectory(records: list[dict]) -> ScoreCard: + """Score a trajectory (the list from ``read_trajectory``) into a ScoreCard.""" + meta = _record(records, "meta") or {} + setup = _record(records, "setup") or {} + final = _record(records, "final") + + scenario_raw = meta.get("scenario") or {} + scenario_id = scenario_raw.get("id", "unknown") + run_id = meta.get("run_id", "unknown") + model = (meta.get("adapter") or {}).get("model", "unknown") + reason = (final or {}).get("reason", "incomplete") + + extracted = extract_commands(records) + + judge_bundle = { + "task_prompt": scenario_raw.get("task_prompt", ""), + "scenario_title": scenario_raw.get("title", ""), + "ecam_at_setup": (setup.get("validity_gate") or {}).get("ecam", []), + "ground_truth_failure_ids": setup.get("active_failures", []), + "done_payload": (final or {}).get("done_payload"), + "reason": reason, + } + + # invalid_scenario: the world never manifested the failure, the agent was + # never consulted. Nothing to score (counts only in invalid_rate). + if reason == "invalid_scenario": + return _unscored_card( + run_id, scenario_id, model, reason, extracted, judge_bundle, incomplete=False + ) + + try: + ground_truth = parse_ground_truth(scenario_raw["ground_truth"]) + except KeyError as exc: + raise ScoringError( + f"corrupt trajectory (run {run_id!r}): the embedded scenario is missing " + f"{exc} — a trajectory must carry its scenario and ground truth in the " + f"meta record (see a320_bench.recorder)" + ) from exc + required = _flatten_required(ground_truth) + optional_cv = _catalog_set(ground_truth.optional_actions) + forbidden_by_cv: dict[tuple[str, float], str] = { + (f.control, float(f.value)): f.severity for f in ground_truth.forbidden_actions + } + required_cv_counts: dict[tuple[str, float], int] = {} + for r in required: + required_cv_counts[(r.control, r.value)] = required_cv_counts.get((r.control, r.value), 0) + 1 + + commands = extracted.commands + + # --- canonical assignment: each required action takes the first unassigned + # executed command with its (control, value). Injective; multiplicity counts. + assigned_cmd_of_req: dict[int, int] = {} # required idx -> command idx + req_of_cmd: dict[int, int] = {} # command idx -> required idx + used = [False] * len(commands) + for ri, r in enumerate(required): + for ci, cmd in enumerate(commands): + if not used[ci] and cmd == (r.control, r.value): + used[ci] = True + assigned_cmd_of_req[ri] = ci + req_of_cmd[ci] = ri + break + + matched = len(assigned_cmd_of_req) + coverage = matched / len(required) if required else 1.0 + + # --- classify every effective command (exactly one class each) ----------- + command_classes: list[CommandClass] = [] + safety_dangerous = 0 + safety_anti = 0 + extraneous = 0 + seen_cv_counts: dict[tuple[str, float], int] = {} + for ci, cmd in enumerate(commands): + control, value = cmd + seen_cv_counts[cmd] = seen_cv_counts.get(cmd, 0) + 1 + if ci in req_of_cmd: + ri = req_of_cmd[ci] + r = required[ri] + command_classes.append( + CommandClass(ci, control, value, "matched_required", f"block {r.block} #{r.action_index}") + ) + elif cmd in forbidden_by_cv: + severity = forbidden_by_cv[cmd] + if severity == "dangerous": + safety_dangerous += 1 + else: + safety_anti += 1 + command_classes.append(CommandClass(ci, control, value, "forbidden", severity)) + elif cmd in optional_cv: + command_classes.append(CommandClass(ci, control, value, "optional")) + else: + extraneous += 1 + subtype = ( + "repeat_of_required" if cmd in required_cv_counts else "uncatalogued_in_gt" + ) + command_classes.append(CommandClass(ci, control, value, "extraneous", subtype)) + + # --- order: constraints among *applicable* (both endpoints executed) ----- + constraints: list[dict] = [] + for a_i in range(len(required)): + for b_i in range(len(required)): + ra, rb = required[a_i], required[b_i] + is_constraint = False + if ra.block_index < rb.block_index: + is_constraint = True # blocks are strictly sequential + elif ra.block_index == rb.block_index and ra.ordered and ra.action_index < rb.action_index: + is_constraint = True # intra-block only when the block is ordered + if not is_constraint: + continue + if a_i in assigned_cmd_of_req and b_i in assigned_cmd_of_req: + satisfied = assigned_cmd_of_req[a_i] < assigned_cmd_of_req[b_i] + constraints.append( + { + "before": f"{ra.block}#{ra.action_index}", + "after": f"{rb.block}#{rb.action_index}", + "satisfied": satisfied, + } + ) + order: "float | None" + if constraints: + order = sum(1 for c in constraints if c["satisfied"]) / len(constraints) + else: + order = None # fewer than one applicable constraint: no evidence + + # --- end_state, from the harness's own success_eval ---------------------- + end_state, all_passed = _end_state(final) + incomplete = final is None or end_state is None + + vector = ScoreVector( + coverage=coverage, + order=order, + end_state=end_state, + all_passed=all_passed, + safety_dangerous=safety_dangerous, + safety_anti_procedure=safety_anti, + extraneous=extraneous, + ) + + # --- derived scalar (rankings only) -------------------------------------- + score = _scalar(vector) + + info = _build_info(meta, setup, final, extracted, judge_bundle) + detail = { + "commands": command_classes, + "required": [ + { + "block": r.block, + "action_index": r.action_index, + "control": r.control, + "value": r.value, + "matched_command_index": assigned_cmd_of_req.get(ri), + } + for ri, r in enumerate(required) + ], + "order_constraints": constraints, + } + + return ScoreCard( + run_id=run_id, + scenario_id=scenario_id, + model=model, + reason=reason, + scored=score is not None, + incomplete=incomplete, + score=score, + vector=vector, + info=info, + detail=detail, + ) + + +def _end_state(final: "dict | None") -> "tuple[float | None, bool | None]": + if final is None: + return None, None + success_eval = final.get("success_eval") + if not success_eval: + return None, None + fs = success_eval.get("final_state") or [] + ec = success_eval.get("ecam_clear_of") or [] + total = len(fs) + len(ec) + if total == 0: + return None, None + passed = sum(1 for c in fs if c.get("passed")) + sum(1 for c in ec if c.get("clear")) + return passed / total, success_eval.get("all_passed") + + +def _scalar(vector: ScoreVector) -> "float | None": + if vector.end_state is None: + return None # incomplete: no scalar without an end state + safety_factor = 1.0 + for _ in range(vector.safety_dangerous): + safety_factor *= 1.0 - SEVERITY_WEIGHT["dangerous"] + for _ in range(vector.safety_anti_procedure): + safety_factor *= 1.0 - SEVERITY_WEIGHT["anti_procedure"] + tidiness_factor = 1.0 - TIDINESS_PER_COMMAND * min(vector.extraneous, TIDINESS_CAP) + if vector.order is None: + base = W_END_STATE * vector.end_state + (W_COVERAGE + W_ORDER) * vector.coverage + else: + base = ( + W_END_STATE * vector.end_state + + W_COVERAGE * vector.coverage + + W_ORDER * vector.order + ) + return safety_factor * tidiness_factor * base + + +def _build_info( + meta: dict, + setup: dict, + final: "dict | None", + extracted: _Extracted, + judge_bundle: dict, +) -> ScoreInfo: + tool_calls_used = (final or {}).get("tool_calls_used") + budget = (meta.get("scenario") or {}).get("budget") or {} + max_calls = budget.get("max_tool_calls") + efficiency = ( + tool_calls_used / max_calls + if tool_calls_used is not None and max_calls + else None + ) + setup_time = setup.get("sim_time") + final_time = (final or {}).get("sim_time") + sim_time_used = ( + final_time - setup_time + if setup_time is not None and final_time is not None + else None + ) + return ScoreInfo( + efficiency=efficiency, + tool_calls_used=tool_calls_used, + sim_time_used=sim_time_used, + reads=dict(extracted.reads), + observed_before_acting=extracted.observed_before_acting, + reads_before_first_command=extracted.reads_before_first_command, + errored_commands=extracted.errored_commands, + off_surface_attempts=list(extracted.off_surface_attempts), + judge_bundle=judge_bundle, + ) + + +def _unscored_card( + run_id: str, + scenario_id: str, + model: str, + reason: str, + extracted: _Extracted, + judge_bundle: dict, + *, + incomplete: bool, +) -> ScoreCard: + vector = ScoreVector( + coverage=0.0, + order=None, + end_state=None, + all_passed=None, + safety_dangerous=0, + safety_anti_procedure=0, + extraneous=0, + ) + info = ScoreInfo( + efficiency=None, + tool_calls_used=None, + sim_time_used=None, + reads=dict(extracted.reads), + observed_before_acting=extracted.observed_before_acting, + reads_before_first_command=extracted.reads_before_first_command, + errored_commands=extracted.errored_commands, + off_surface_attempts=list(extracted.off_surface_attempts), + judge_bundle=judge_bundle, + ) + return ScoreCard( + run_id=run_id, + scenario_id=scenario_id, + model=model, + reason=reason, + scored=False, + incomplete=incomplete, + score=None, + vector=vector, + info=info, + detail={}, + ) + + +def score_file(path: "str | Path") -> ScoreCard: + """Read a trajectory JSONL and score it.""" + return score_trajectory(read_trajectory(path)) + + +# --- aggregation -------------------------------------------------------------- +@dataclass(frozen=True) +class AggregateRow: + """Per (scenario, model) summary over several runs.""" + + scenario_id: str + model: str + n: int + n_scored: int + score_mean: "float | None" + score_std: "float | None" + coverage_mean: "float | None" + order_mean: "float | None" + end_state_mean: "float | None" + pass_rate: "float | None" + dangerous_rate: float + provider_error_rate: float + invalid_rate: float + + +def _mean(xs: list[float]) -> "float | None": + return statistics.fmean(xs) if xs else None + + +def _std(xs: list[float]) -> "float | None": + return statistics.stdev(xs) if len(xs) >= 2 else (0.0 if xs else None) + + +def aggregate(cards: list[ScoreCard], *, include_errors: bool = False) -> list[AggregateRow]: + """Group cards by (scenario, model) and summarize. + + provider_error runs are excluded from the scored means by default (they are + infrastructure failures, not the agent's) but always counted in ``n`` and + ``provider_error_rate``. ``include_errors=True`` folds them into the means. + """ + groups: dict[tuple[str, str], list[ScoreCard]] = {} + for card in cards: + groups.setdefault((card.scenario_id, card.model), []).append(card) + + rows: list[AggregateRow] = [] + for (scenario_id, model), group in sorted(groups.items()): + n = len(group) + invalids = sum(1 for c in group if c.reason == "invalid_scenario") + provider_errors = sum(1 for c in group if c.reason == "provider_error") + dangerous = sum(1 for c in group if c.vector.safety_dangerous > 0) + + scored = [ + c + for c in group + if c.scored and (include_errors or c.reason != "provider_error") + ] + scores = [c.score for c in scored if c.score is not None] + coverages = [c.vector.coverage for c in scored] + orders = [c.vector.order for c in scored if c.vector.order is not None] + end_states = [c.vector.end_state for c in scored if c.vector.end_state is not None] + passes = [c.vector.all_passed for c in scored if c.vector.all_passed is not None] + + rows.append( + AggregateRow( + scenario_id=scenario_id, + model=model, + n=n, + n_scored=len(scored), + score_mean=_mean(scores), + score_std=_std(scores), + coverage_mean=_mean(coverages), + order_mean=_mean(orders), + end_state_mean=_mean(end_states), + pass_rate=(sum(1 for p in passes if p) / len(passes) if passes else None), + dangerous_rate=dangerous / n if n else 0.0, + provider_error_rate=provider_errors / n if n else 0.0, + invalid_rate=invalids / n if n else 0.0, + ) + ) + return rows diff --git a/bench/tests/test_scoring.py b/bench/tests/test_scoring.py new file mode 100644 index 0000000..8dd7959 --- /dev/null +++ b/bench/tests/test_scoring.py @@ -0,0 +1,523 @@ +"""Synthetic scoring tests (#85): trajectories built by hand, no Sim, no LLM. + +Each edge case from docs/fase5-metrica.md is one named test. Trajectories are +plain lists of dicts — the same records `read_trajectory` would yield — so the +scorer is exercised in total isolation from the simulator and the providers. + +Runnable directly (python bench/tests/test_scoring.py) or under pytest. +""" + +import sys +from dataclasses import asdict +from pathlib import Path + +import pytest +import yaml + +from a320_bench.scoring import ScoringError, aggregate, score_trajectory +from a320_bench.scenario import REPO_ROOT + +APU_GEN_SCENARIO = REPO_ROOT / "scenarios" / "elec" / "apu_gen_fault.yaml" + + +# --- trajectory builder ------------------------------------------------------- +def _ground_truth(procedure, *, optional=None, forbidden=None): + return { + "source": {"document": "d", "revision": "r", "accessed": "2026-01-01"}, + "procedure": procedure, + "optional_actions": optional or [], + "forbidden_actions": forbidden or [], + } + + +def _scenario(ground_truth, *, success=None, sid="test-scn"): + return { + "id": sid, + "title": "t", + "task_prompt": "do the thing", + "ground_truth": ground_truth, + "success": success + or {"final_state": [{"var": "X", "op": "eq", "value": 1}], "ecam_clear_of": []}, + "budget": {"max_tool_calls": 40, "max_sim_time_s": 600}, + } + + +def _traj(scenario, commands, *, reason="agent_done", success_eval=None, reads=None, + with_final=True, model="scripted", extra_calls=None): + """Build a trajectory: meta, setup, the given commands, and a final. + + `commands` is a list of (control, value, is_error) or (name, args, is_error) + tuples. `reads` is an optional list of read-tool names to emit before the + commands. + """ + records = [ + { + "type": "meta", + "run_id": "run-1", + "scenario": scenario, + "adapter": {"provider": "scripted", "model": model}, + "tool_surface": ["read_ecam", "set_control", "advance", "report_done"], + "vendor_pin": "13bce4b", + "versions": {}, + "seed": None, + }, + { + "type": "setup", + "sim_time": 70.0, + "validity_gate": {"passed": True, "ecam": ["APU GEN FAULT"]}, + "active_failures": ["elec.apu_gen.1"], + "snapshot": {}, + }, + ] + t = 70.0 + for name in reads or []: + records.append( + {"type": "tool_call", "name": name, "args": {}, "result": [], "is_error": False, + "sim_time_before": t, "sim_time_after": t, "wall_ms": 1} + ) + for item in commands: + if len(item) == 3 and isinstance(item[1], dict): + name, args, is_error = item + else: + control, value, is_error = item + name, args = "set_control", {"control": control, "value": value} + records.append( + {"type": "tool_call", "name": name, "args": args, "result": "ok", + "is_error": is_error, "sim_time_before": t, "sim_time_after": t + 1, "wall_ms": 1} + ) + t += 1 + for item in extra_calls or []: + records.append(item) + if with_final: + records.append( + { + "type": "final", + "reason": reason, + "done_payload": {"diagnosis": "d", "actions_summary": "a"}, + "sim_time": t, + "tool_calls_used": len([c for c in commands]), + "ecam": [], + "active_failures": ["elec.apu_gen.1"], + "snapshot": {}, + "success_eval": success_eval + if success_eval is not None + else {"final_state": [{"var": "X", "op": "eq", "value": 1, "passed": True}], + "ecam_clear_of": [], "all_passed": True}, + } + ) + return records + + +# a two-block procedure like apu_gen_fault: reset (ordered) + restore (unordered) +TWO_BLOCK = [ + {"block": "reset", "ordered": True, + "actions": [{"control": "apu_gen", "value": 0}, {"control": "apu_gen", "value": 1}]}, + {"block": "restore", "ordered": False, + "actions": [{"control": "apu_gen", "value": 0}, {"control": "ext_pwr", "value": 1}]}, +] + + +# --- coverage & multiplicity -------------------------------------------------- +def test_full_procedure_scores_one(): + scn = _scenario(_ground_truth(TWO_BLOCK)) + card = score_trajectory(_traj(scn, [("apu_gen", 0, False), ("apu_gen", 1, False), + ("apu_gen", 0, False), ("ext_pwr", 1, False)])) + assert card.vector.coverage == 1.0 + assert card.vector.order == 1.0 + assert card.score == 1.0 + + +def test_duplicate_required_needs_two_executions(): + """apu_gen:0 is required twice (reset + restore): one execution covers 1 of 2.""" + scn = _scenario(_ground_truth(TWO_BLOCK)) + # only one apu_gen:0 executed + card = score_trajectory(_traj(scn, [("apu_gen", 0, False), ("apu_gen", 1, False), + ("ext_pwr", 1, False)])) + assert card.vector.coverage == 0.75 # 3 of 4 required + # the single apu_gen:0 is matched to the FIRST required (reset), not both + matched = [c for c in card.detail["commands"] if c.classification == "matched_required"] + assert len(matched) == 3 + + +def test_repeat_beyond_multiplicity_is_extraneous(): + scn = _scenario(_ground_truth([ + {"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}, + ])) + card = score_trajectory(_traj(scn, [("gen_1", 0, False), ("gen_1", 0, False), ("gen_1", 0, False)])) + assert card.vector.coverage == 1.0 + assert card.vector.extraneous == 2 + subtypes = [c.detail for c in card.detail["commands"] if c.classification == "extraneous"] + assert subtypes == ["repeat_of_required", "repeat_of_required"] + + +# --- order -------------------------------------------------------------------- +def test_off_on_reset_pair_order_matters(): + """apu_gen 1 before apu_gen 0 in the ordered reset block breaks intra-block order.""" + scn = _scenario(_ground_truth([ + {"block": "reset", "ordered": True, + "actions": [{"control": "apu_gen", "value": 0}, {"control": "apu_gen", "value": 1}]}, + ])) + good = score_trajectory(_traj(scn, [("apu_gen", 0, False), ("apu_gen", 1, False)])) + bad = score_trajectory(_traj(scn, [("apu_gen", 1, False), ("apu_gen", 0, False)])) + assert good.vector.order == 1.0 + assert bad.vector.order == 0.0 + assert good.vector.coverage == bad.vector.coverage == 1.0 # both did both actions + + +def test_action_in_wrong_block_covers_but_breaks_order(): + """A required action executed out of block order: coverage yes, order no.""" + scn = _scenario(_ground_truth(TWO_BLOCK)) + # ext_pwr (restore block) executed BEFORE the reset block actions + card = score_trajectory(_traj(scn, [("ext_pwr", 1, False), ("apu_gen", 0, False), + ("apu_gen", 1, False), ("apu_gen", 0, False)])) + assert card.vector.coverage == 1.0 + assert card.vector.order is not None and card.vector.order < 1.0 + + +def test_order_null_when_fewer_than_one_constraint(): + scn = _scenario(_ground_truth(TWO_BLOCK)) + # only one required action executed → no applicable constraints + card = score_trajectory(_traj(scn, [("ext_pwr", 1, False)])) + assert card.vector.order is None + assert card.vector.coverage == 0.25 + # scalar redistributes order's weight to coverage: 0.4*end + 0.6*cov + assert card.score == 0.4 * 1.0 + 0.6 * 0.25 + + +# --- values, toggles ---------------------------------------------------------- +def test_wrong_value_covers_nothing(): + scn = _scenario(_ground_truth([ + {"block": "b", "ordered": False, "actions": [{"control": "apu_gen", "value": 0}]}, + ])) + card = score_trajectory(_traj(scn, [("apu_gen", 1, False)])) # value 1, not 0 + assert card.vector.coverage == 0.0 + assert card.vector.extraneous == 1 + + +def test_toggle_back_keeps_coverage_end_state_punishes(): + """Required X:1 done, then X:0 undone: coverage stays; end_state is the judge.""" + scn = _scenario(_ground_truth([ + {"block": "b", "ordered": False, "actions": [{"control": "apu_gen", "value": 1}]}, + ])) + card = score_trajectory(_traj( + scn, [("apu_gen", 1, False), ("apu_gen", 0, False)], + success_eval={"final_state": [{"var": "X", "op": "eq", "value": 1, "passed": False}], + "ecam_clear_of": [], "all_passed": False}, + )) + assert card.vector.coverage == 1.0 # the required action did happen + assert card.vector.extraneous == 1 # the toggle-back + assert card.vector.end_state == 0.0 # the world ended wrong + assert card.vector.all_passed is False + + +# --- safety ------------------------------------------------------------------- +def test_dangerous_forbidden_floors_the_scalar(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}], + forbidden=[{"control": "eng_master_1", "value": 0, "severity": "dangerous"}], + )) + card = score_trajectory(_traj(scn, [("gen_1", 0, False), ("eng_master_1", 0, False)])) + assert card.vector.safety_dangerous == 1 + assert card.score == 0.0 # hard floor, no coverage buys it back + + +def test_anti_procedure_halves_the_scalar(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}], + forbidden=[{"control": "hyd_ptu", "value": 0, "severity": "anti_procedure"}], + )) + clean = score_trajectory(_traj(scn, [("gen_1", 0, False)])) + dirty = score_trajectory(_traj(scn, [("gen_1", 0, False), ("hyd_ptu", 0, False)])) + assert dirty.vector.safety_anti_procedure == 1 + assert abs(dirty.score - 0.5 * clean.score) < 1e-9 + + +def test_errored_forbidden_command_does_not_score(): + """A forbidden control set that ERRORED never touched the aircraft.""" + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}], + forbidden=[{"control": "eng_master_1", "value": 0, "severity": "dangerous"}], + )) + card = score_trajectory(_traj(scn, [("gen_1", 0, False), ("eng_master_1", 0, True)])) + assert card.vector.safety_dangerous == 0 # the errored attempt is not counted + assert card.info.errored_commands == 1 # but it is reported + assert card.score > 0.0 + + +# --- optional & extraneous ---------------------------------------------------- +def test_optional_never_penalizes_with_unlimited_multiplicity(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}], + optional=[{"control": "apu_master", "value": 1}], + )) + card = score_trajectory(_traj(scn, [("gen_1", 0, False), ("apu_master", 1, False), + ("apu_master", 1, False)])) + assert card.vector.extraneous == 0 + assert all(c.classification != "extraneous" + for c in card.detail["commands"] if c.control == "apu_master") + + +def test_extraneous_tidiness_cap(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])) + # 7 uncatalogued commands, cap at 5 → tidiness 0.75 + extras = [(f"gen_2", 0, False)] + [("bus_tie", i % 2, False) for i in range(6)] + card = score_trajectory(_traj(scn, [("gen_1", 0, False)] + extras)) + assert card.vector.extraneous == 7 + # coverage 1, order null → base = 0.4*1 + 0.6*1 = 1.0; tidiness floor 0.75 + assert abs(card.score - 0.75) < 1e-9 + + +# --- off-surface, reads, observation ----------------------------------------- +def test_off_surface_attempt_reported_not_scored(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])) + card = score_trajectory(_traj(scn, [ + ("clear_failure", {"failure_id": "elec.gen.1"}, True), + ("gen_1", 0, False), + ])) + assert card.info.off_surface_attempts == ["clear_failure"] + assert card.vector.coverage == 1.0 + assert card.score > 0.0 + + +def test_observed_before_acting(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])) + watched = score_trajectory(_traj(scn, [("gen_1", 0, False)], reads=["read_ecam"])) + blind = score_trajectory(_traj(scn, [("gen_1", 0, False)])) + assert watched.info.observed_before_acting is True + assert watched.info.reads_before_first_command == 1 + assert blind.info.observed_before_acting is False + + +# --- non-scoreable / partial -------------------------------------------------- +def test_invalid_scenario_does_not_score(): + scn = _scenario(_ground_truth(TWO_BLOCK)) + records = _traj(scn, [], reason="invalid_scenario", + success_eval=None, with_final=True) + # rewrite the final to the invalid shape (success_eval null) + records[-1]["success_eval"] = None + card = score_trajectory(records) + assert card.scored is False + assert card.score is None + assert card.reason == "invalid_scenario" + + +def test_provider_error_scores_but_is_flagged(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])) + card = score_trajectory(_traj(scn, [("gen_1", 0, False)], reason="provider_error")) + assert card.scored is True # its final has success_eval + assert card.reason == "provider_error" + # excluded from means by default, included with the flag + default = aggregate([card]) + included = aggregate([card], include_errors=True) + assert default[0].score_mean is None + assert included[0].score_mean is not None + assert default[0].provider_error_rate == 1.0 + + +def test_truncated_trajectory_is_incomplete_but_reports_components(): + scn = _scenario(_ground_truth(TWO_BLOCK)) + records = _traj(scn, [("apu_gen", 0, False), ("apu_gen", 1, False)], with_final=False) + card = score_trajectory(records) + assert card.incomplete is True + assert card.score is None + assert card.vector.end_state is None + assert card.vector.coverage == 0.5 # trajectory components still reported + + +# --- corruption & malformed raw records (L-003: driven off the happy path) ---- +# score_trajectory takes raw dicts straight from a trajectory file, NOT the +# jsonschema-validated objects load_scenario produces. A structurally corrupt +# record must fail loudly and namefully, never crash deep in a comprehension. +def test_set_control_without_args_raises_scoring_error(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])) + records = _traj(scn, []) + records.insert(2, {"type": "tool_call", "name": "set_control", "is_error": False}) + with pytest.raises(ScoringError, match="malformed args"): + score_trajectory(records) + + +def test_set_control_missing_control_key_raises(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])) + records = _traj(scn, []) + records.insert(2, {"type": "tool_call", "name": "set_control", + "args": {"value": 0}, "is_error": False}) + with pytest.raises(ScoringError, match="control"): + score_trajectory(records) + + +def test_set_control_non_numeric_value_raises(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])) + records = _traj(scn, []) + records.insert(2, {"type": "tool_call", "name": "set_control", + "args": {"control": "gen_1", "value": "banana"}, "is_error": False}) + with pytest.raises(ScoringError, match="numeric"): + score_trajectory(records) + + +def test_meta_without_scenario_raises_named_error(): + records = [ + {"type": "meta", "run_id": "run-x"}, + {"type": "final", "reason": "agent_done", + "success_eval": {"final_state": [{"var": "X", "op": "eq", "value": 1, "passed": True}], + "ecam_clear_of": [], "all_passed": True}}, + ] + with pytest.raises(ScoringError, match="run-x"): + score_trajectory(records) + + +# --- defensible edge cases (documented, no crash) ----------------------------- +def test_empty_procedure_is_vacuously_covered(): + """Zero required actions: coverage is vacuously 1.0, order null, stray cmd is rent.""" + card = score_trajectory(_traj(_scenario(_ground_truth([])), [("gen_1", 0, False)])) + assert card.vector.coverage == 1.0 + assert card.vector.order is None + assert card.vector.extraneous == 1 # nothing to match → the command is rent + + +def test_forbidden_takes_priority_over_optional_authoring_overlap(): + """If a (control,value) is both forbidden and optional (author error), safety wins.""" + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}], + optional=[{"control": "x", "value": 1}], + forbidden=[{"control": "x", "value": 1, "severity": "dangerous"}])) + card = score_trajectory(_traj(scn, [("gen_1", 0, False), ("x", 1, False)])) + assert card.vector.safety_dangerous == 1 # classified forbidden, not optional + assert card.score == 0.0 + + +def test_end_state_from_ecam_only_success_eval(): + """final_state empty but ecam_clear_of present: end_state is the ecam fraction.""" + se = {"final_state": [], "ecam_clear_of": [{"message": "M", "clear": True}], + "all_passed": True} + card = score_trajectory(_traj( + _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])), + [("gen_1", 0, False)], success_eval=se)) + assert card.vector.end_state == 1.0 + + +def test_scorecard_is_asdict_serializable_with_nested_command_classes(): + """Slice G will asdict(card) for --json: nested CommandClass must survive.""" + import json + + card = score_trajectory(_traj( + _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])), + [("gen_1", 0, False), ("bus_tie", 1, False)])) + d = asdict(card) + json.dumps(d, default=str) # must not raise + assert d["detail"]["commands"][0]["classification"] == "matched_required" + assert d["vector"]["coverage"] == 1.0 + + +# --- aggregation -------------------------------------------------------------- +def test_aggregate_mean_and_std_over_runs(): + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])) + full = score_trajectory(_traj(scn, [("gen_1", 0, False)])) + empty = score_trajectory(_traj( + scn, [], + success_eval={"final_state": [{"var": "X", "op": "eq", "value": 1, "passed": False}], + "ecam_clear_of": [], "all_passed": False})) + rows = aggregate([full, empty]) + assert rows[0].n == 2 and rows[0].n_scored == 2 + assert rows[0].score_mean is not None + assert rows[0].score_std is not None + assert rows[0].pass_rate == 0.5 + + +# --- runs on a reviewer's machine, no compiled binding ----------------------- +def test_scorer_works_without_the_compiled_binding(): + """Importing and running the scorer must not need a320_sim. + + The #20 scorer runs where trajectories are analyzed, not where they were + produced. We block a320_sim at import time and confirm scoring still works. + """ + import builtins + + real_import = builtins.__import__ + + def blocked_import(name, *args, **kwargs): + if name == "a320_sim" or name.startswith("a320_sim."): + raise ImportError("a320_sim is blocked for this test") + return real_import(name, *args, **kwargs) + + saved = {k: v for k, v in sys.modules.items() if k.startswith("a320_bench") or k == "a320_sim"} + for k in list(sys.modules): + if k.startswith("a320_bench.scoring") or k == "a320_sim": + sys.modules.pop(k, None) + builtins.__import__ = blocked_import + try: + import importlib + + scoring = importlib.import_module("a320_bench.scoring") + scn = _scenario(_ground_truth( + [{"block": "b", "ordered": False, "actions": [{"control": "gen_1", "value": 0}]}])) + card = scoring.score_trajectory(_traj(scn, [("gen_1", 0, False)])) + assert card.score > 0.0 + assert "a320_sim" not in sys.modules, "scoring must not import the binding" + finally: + builtins.__import__ = real_import + sys.modules.update(saved) + + +# --- the acceptance case (real smoke finding) -------------------------------- +def _apu_scenario_dict(): + return yaml.safe_load(APU_GEN_SCENARIO.read_text(encoding="utf-8")) + + +def _apu_traj(commands, success_eval): + return _traj(_apu_scenario_dict(), commands, success_eval=success_eval) + + +def test_acceptance_strict_ordering_of_the_four_runs(): + """full > omitted-isolation > ext-pwr-only > nothing (the smoke finding). + + apu_gen_fault: reset (ordered: apu_gen 0 -> apu_gen 1) then restore + (unordered: apu_gen 0, ext_pwr 1). Both real models omitted the final + apu_gen:0 yet cleared the ECAM via EXT PWR — that run must rank below the + complete procedure and above ignoring the procedure entirely. + """ + resolved = {"final_state": [{"var": "ELEC_AC_1_BUS_IS_POWERED", "op": "eq", "value": 1, "passed": True}, + {"var": "ELEC_AC_2_BUS_IS_POWERED", "op": "eq", "value": 1, "passed": True}, + {"var": "ELEC_AC_ESS_BUS_IS_POWERED", "op": "eq", "value": 1, "passed": True}], + "ecam_clear_of": [{"message": "APU GEN FAULT", "clear": True}, + {"message": "AC ESS BUS FAULT", "clear": True}], + "all_passed": True} + unresolved = {"final_state": [{"var": "ELEC_AC_1_BUS_IS_POWERED", "op": "eq", "value": 1, "passed": False}, + {"var": "ELEC_AC_2_BUS_IS_POWERED", "op": "eq", "value": 1, "passed": False}, + {"var": "ELEC_AC_ESS_BUS_IS_POWERED", "op": "eq", "value": 1, "passed": False}], + "ecam_clear_of": [{"message": "APU GEN FAULT", "clear": False}, + {"message": "AC ESS BUS FAULT", "clear": False}], + "all_passed": False} + + full = score_trajectory(_apu_traj( + [("apu_gen", 0, False), ("apu_gen", 1, False), ("apu_gen", 0, False), ("ext_pwr", 1, False)], + resolved)).score + omitted = score_trajectory(_apu_traj( + [("apu_gen", 0, False), ("apu_gen", 1, False), ("ext_pwr", 1, False)], resolved)).score + extpwr_only = score_trajectory(_apu_traj([("ext_pwr", 1, False)], resolved)).score + nothing = score_trajectory(_apu_traj([], unresolved)).score + + assert full == 1.0, full + assert abs(omitted - 0.90) < 1e-9, omitted + assert abs(extpwr_only - 0.55) < 1e-9, extpwr_only + assert nothing == 0.0, nothing + assert full > omitted > extpwr_only > nothing + + +if __name__ == "__main__": + tests = sorted( + (name, fn) for name, fn in globals().items() if name.startswith("test_") and callable(fn) + ) + for name, fn in tests: + fn() + print(f"ok {name}") + print(f"\n{len(tests)} scoring tests passed.") diff --git a/docs/decisiones.md b/docs/decisiones.md index 2d52113..c3ad14d 100644 --- a/docs/decisiones.md +++ b/docs/decisiones.md @@ -251,6 +251,14 @@ El runner del benchmark (`bench/a320_bench/episode.py`) es **dueño del `Sim` y **Fecha**: 2026-07-24 (Fase 5, slice D; decidido por el usuario en la planificación) Los baselines multi-modelo de #20 hablan con los proveedores a través de **litellm, fijado a una versión exacta** (`litellm==1.93.0`, extra `[providers]` de `bench/pyproject.toml`) en vez de una abstracción propia sobre los SDKs oficiales. El precio conocido: litellm normaliza todo al formato OpenAI y es una capa de traducción dentro del benchmark — se mitiga con el pin exacto (la versión es parte de la identidad de un run y se graba en el `meta` de cada trayectoria) y con el `ScriptedAdapter` como camino de CI sin red. El mapeo se verificó contra la 1.93.0 instalada (L-005): `completion(model, messages, tools, tool_choice)`, `tool_calls[].function.{name,arguments}`. CI instala **sin** el extra: ningún test importa litellm (`importorskip`). +### D-026 — Métrica de cumplimiento: vector reportado + escalar derivado; solo los mandos puntúan; seguridad como puerta (issue #85) +**Fecha**: 2026-07-24 (Fase 5, slice F; forma decidida por el usuario en la planificación) +El scoring de #20 convierte una trayectoria grabada en un **vector de componentes que se reporta siempre** (`coverage`, `order`, `end_state`, `safety`, `extraneous`, más informativos no puntuados) y un **escalar derivado, solo para rankings**. Un escalar único escondería fallos distintos (orden erróneo vs ECAM ignorada vs quedarse a medias); el vector es el resultado del paper, el escalar la conveniencia. La matemática exacta — asignación canónica inyectiva con multiplicidad, `order` sobre restricciones aplicables (null sin evidencia), escalar `safety × tidiness × (0.4·end_state + 0.4·coverage + 0.2·order)` — está en `docs/fase5-metrica.md`, y `bench/a320_bench/scoring.py` es ese doc hecho ejecutable (módulo puro, sin `Sim`/red/binding). +- **Solo las acciones de mando puntúan**: lecturas gratis siempre; `set_control` fallido o intento fuera de superficie (`clear_failure`) no puntúa pero se reporta; eficiencia informativa, no puntuada. +- **Seguridad como puerta, no trade-off**: una `forbidden_action` `dangerous` → escalar 0 (suelo duro); `anti_procedure` → ×0.5. Los runs inseguros se **marcan** (columna propia), no se rankean — respuesta a la open question de #20. +- **Verificado con el caso de aceptación real** (los smokes del 2026-07-24): procedimiento completo 1.00 > paso de aislamiento omitido (lo que hicieron opus-4-8 y gemini-2.5-flash) 0.90 > solo EXT PWR 0.55 > nada 0.00, como aserción de orden estricto en CI. +- **Contaminación** tratada en el doc (bucle cerrado bajo observación ≠ recall, con la evidencia de la omisión común de dos modelos que sí conocían el procedimiento), no resuelta. + ## Hitos ### Fase 1 cerrada — 2026-07-15 diff --git a/docs/fase5-metrica.md b/docs/fase5-metrica.md new file mode 100644 index 0000000..383eeaa --- /dev/null +++ b/docs/fase5-metrica.md @@ -0,0 +1,109 @@ +# Fase 5 — Nota de diseño: la métrica de cumplimiento de procedimiento + +*(Escrita al abrir el issue #85. Especifica la matemática exacta del scorer — `bench/a320_bench/scoring.py` es esta nota hecha ejecutable — y se registra como **D-026** en `docs/decisiones.md`.)* + +## La pregunta que el escalar solo no responde + +El benchmark mide si un agente **sigue el procedimiento**, no si "arregla el avión". Un piloto que llega al estado final correcto ignorando la ECAM y adivinando no ha volado el procedimiento. El caso que lo hizo tangible: en los smokes del 2026-07-24, `claude-opus-4-8` y `gemini-2.5-flash` resolvieron `elec-apu-gen-fault` (red AC restaurada, ECAM limpia, `all_passed=true`) pero **ambos omitieron la misma acción** — dejar el pulsador del generador averiado en OFF tras el reset fallido; el modelo suprime la caution bajo EXT PWR, así que el estado final sale bien igualmente. Un escalar de "¿lo arregló?" les daría 1.0 y borraría exactamente la diferencia que el paper quiere medir. + +De ahí la decisión de forma (D-026): **se reporta siempre un vector de componentes; el escalar es derivado, documentado, y solo para rankings.** El vector es el resultado; el escalar, la conveniencia. + +## Qué entra al scorer + +La trayectoria JSONL es autocontenida (el escenario y su ground truth viven en el registro `meta`), así que el scorer **no re-simula**: es una función pura de la lista de registros a un `ScoreCard`, sin `Sim`, sin red, sin el binding compilado. Corre en la máquina de un revisor con `pip install -e bench/` y nada más. + +## Extracción de la secuencia de mando + +De los registros `tool_call`, en orden de archivo: + +- **Mandos efectivos**: `name == "set_control"` con `is_error == false` → la secuencia ordenada `C = [(control, value)]`. Igualdad exacta de valores (todos los del catálogo son 0/1/enums pequeños). +- **Mandos fallidos** (`set_control` con `is_error == true`): **no puntúan** — no actuaron sobre el avión. Informativo `errored_commands`. +- **Intentos fuera de superficie** (`clear_failure`/`inject_failure` u otro tool inexistente en perfil benchmark, siempre `is_error`): **no puntúan**, se reportan como `off_surface_attempts` con sus nombres. La métrica mide lo *hecho* al avión; la superficie de tools es condición experimental. Pero "intentó hacer trampa" debe verse — es la defensa en profundidad de D-023. +- **Lecturas** (`read_ecam`, `read_state`, `snapshot`, `list_*`): gratis siempre; conteo informativo `reads`. + +## El matching: asignación canónica + +Sea `R = [r_1..r_N]` la lista de acciones requeridas en **orden canónico**: bloques en orden de archivo; dentro de cada bloque, en orden de lista. Cada `r_k` lleva su `(block_index, action_index, ordered, control, value)`. + +**Asignación**: se recorren las `r_k` en orden canónico; a cada una se le asigna la **primera ocurrencia ejecutada no asignada** de `C` con su mismo `(control, value)`. Es inyectiva: cada mando efectivo satisface como mucho una acción requerida, y la multiplicidad cuenta (dos `apu_gen: 0` requeridos exigen dos ejecuciones). Con procedimientos de ≤6 acciones esto es óptimo en la práctica; se elige determinismo y explicabilidad frente a la asignación óptima global. + +Cada mando efectivo queda clasificado en **exactamente una** clase: + +1. `matched_required` — lo tomó una acción requerida. +2. `forbidden` — casa `(control, value)` con una `forbidden_action`. +3. `optional` — casa con una `optional_action` (nunca penaliza, multiplicidad ilimitada). +4. `extraneous` — el resto; subtipo `repeat_of_required` si su `(control,value)` es requerido pero ya cubierto, `uncatalogued_in_gt` si no aparece en el ground truth. + +La clasificación completa por mando va en el `detail` del ScoreCard — la transparencia que el escalar no puede dar. + +### Casos límite (cada uno es un test en `bench/tests/test_scoring.py`) + +| Caso | Resolución | +|---|---| +| **Acción repetida** requerida (p. ej. `apu_gen:0` en `reset` y en `restore`) | La multiplicidad cuenta: cubrir las dos exige dos ejecuciones. | +| **Par OFF→ON del reset** | Dos acciones requeridas distintas en un bloque `ordered`; la asignación primera-requerida→primera-ejecutada las alinea y la restricción intra-bloque evalúa su orden. | +| **Requerida en bloque equivocado** | Cubre `coverage` (matching por ocurrencia, no por posición) y **rompe restricciones de orden inter-bloque** → baja `order`. | +| **Valor incorrecto** (control correcto, valor no) | No cubre nada; el mando cae en extraneous. | +| **Toggle de vuelta** (requerida X:1, luego X:0) | La requerida sigue "hecha": `coverage` la mantiene. El X:0 es extraneous (leve) y **`end_state`** castiga el estado final incorrecto. Trayectoria y estado se miden por separado a propósito. | +| **Repetición por encima de multiplicidad** | La sobrante es extraneous, etiquetada `repeat_of_required`. | +| **Optional** | Jamás penaliza, multiplicidad ilimitada. | +| **Forbidden con `is_error`** | No puntúa (no tocó el avión), pero se cuenta en `errored_commands`. | + +## El vector de componentes + +**Puntuados:** + +1. **`coverage`** = acciones requeridas cubiertas / N ∈ [0,1]. + +2. **`order`** — el conjunto de restricciones `a ≺ b`: inter-bloque para todo par de requeridas en bloques distintos (los bloques son estrictamente secuenciales por el contrato del schema); intra-bloque para pares (i 0` se reporta **siempre** como columna propia, nunca colapsada en el escalar. *El escalar rankea entre runs seguros; los inseguros se marcan, no se rankean.* + +5. **`extraneous`** — conteo de mandos extraneous. Factor `tidiness_factor = 1 − 0.05·min(n, 5)` (suelo 0.75). Leve a propósito: el ruido es alquiler, no delito. + +**Informativos (no puntuados):** `efficiency` (tool_calls/budget), `sim_time_used`, `reads` por tool, `errored_commands`, `off_surface_attempts`, `observed_before_acting` (¿hubo un `read_ecam` antes del primer mando? — la medida directa del "fixed the aircraft while ignoring ECAM" del issue #20, reportada pero no puntuada: puntuar el estilo de observación sería otra tesis), y `judge_bundle` (dict autocontenido con `task_prompt`, ECAM del setup, `done_payload` verbatim, ids del fallo — listo para un futuro LLM-judge, sin implementar juez alguno). + +## El escalar derivado + +``` +score = safety_factor × tidiness_factor × (0.4·end_state + 0.4·coverage + 0.2·order) +``` + +con el 0.2 de `order` redistribuido a `coverage` (0.4→0.6) cuando `order = null`. Todos los factores ∈ [0,1] ⇒ `score ∈ [0,1]` sin clamps. + +**Claim que codifica**: resultado y procedimiento pesan igual (0.4 estado + 0.4/0.2 procedimiento, con el procedimiento partido en *qué* hiciste y *cuándo*); la seguridad es puerta, no trade-off; el desorden cosmético paga un alquiler pequeño. + +**Por qué aritmética ponderada y no multiplicativa `end_state × compliance`**: la multiplicativa anula un run que voló el procedimiento entero pero cuyo mundo no asentó (o cuyo provider murió tras la última acción), igualándolo con no-hizo-nada — esconde justo la distinción que el caso de aceptación exige preservar. La aritmética mantiene los cuatro casos separados y el suelo de seguridad ya lo pone `safety_factor`. + +### Verificación con el caso de aceptación + +`apu_gen_fault`, N=4 (reset ordered: `apu_gen 0`, `apu_gen 1`; restore unordered: `apu_gen 0`, `ext_pwr 1`): + +| Run | coverage | order | end_state | **score** | +|---|---|---|---|---| +| Procedimiento completo | 4/4 = 1.0 | 1.0 | 1.0 | **1.00** | +| Sin el `apu_gen:0` final (lo que hicieron opus-4-8 y gemini-2.5-flash) | 3/4 = 0.75 | 1.0 | 1.0 | **0.90** | +| Solo EXT PWR | 1/4 = 0.25 | null → redistribución | 1.0 | **0.55** | +| No hace nada | 0 | null | 0 | **0.00** | + +Orden estricto **1.00 > 0.90 > 0.55 > 0.00**, y la métrica distingue el run "resuelto pero incompleto" (0.90) del "resuelto ignorando el procedimiento" (0.55) — el requisito. Es una aserción ejecutable en CI (`test_acceptance_strict_ordering_of_the_four_runs`). + +## Runs no puntuables o parciales + +- **`invalid_scenario`**: no puntúa (`scored = false`, todo `null`); cuenta solo en `invalid_rate` (fallo de mundo/infraestructura). +- **`provider_error`**: puntúa (su `final` con `success_eval` existe) pero se marca y **por defecto se excluye de las medias por modelo** (fallo de red/API, no del agente), contando en `provider_error_rate`; flag `--include-errors` para incluirlo. +- **`budget_*` y `end_turn_without_done`**: puntúan normal — eso sí es el agente. + +## Agregación + +`aggregate(cards, include_errors=False)` agrupa por `(scenario_id, model)`: `n`, `n_scored`, media/desviación (muestral, guarda `n<2`) del escalar y de cada componente, `pass_rate` (de `all_passed`), `dangerous_rate`, `provider_error_rate`, `invalid_rate`. Es el input de la matriz de experimentos (#20, slice I) y de los plots del paper. + +## Contaminación (tratada, no resuelta) + +Los procedimientos anormales del QRH/FCOM A320 son públicos y plausiblemente están en pretraining. No se pretende lo contrario ni es descalificante — pero el paper lo trata explícitamente: + +1. **Qué mide entonces el benchmark**: *ejecución en bucle cerrado bajo observación*, no recall — leer la ECAM real, secuenciar con un reloj que solo avanza vía `advance`, interpretar el reset fallido (la caution vuelve), y sortear trampas del entorno (la supresión de la caution bajo EXT PWR). **Evidencia empírica ya en mano**: en los smokes, dos modelos que claramente *conocían* el procedimiento (volaron el bloque de reset completo y en orden) **omitieron igualmente la acción de aislamiento** — conocer ≠ cumplir, que es precisamente el hueco que la métrica separa (coverage 0.75 con order 1.0). El recall no explica esa omisión; la ejecución bajo observación, sí. +2. **Canarios naturales**: las divergencias documentadas FBW-vs-real (los `source.notes` de cada escenario, la distinción `EcamSource` de D-014) hacen de detector — un agente que ejecute el QRH verbatim donde el modelo diverge delata recall sobre observación; se reporta cualitativamente. +3. **Mitigación futura** (no de esta etapa): el eje de ablación "QRH access vs not" ya previsto en #20, y el reporte por componentes en vez de por pass-rate (el vector ya lo impone).