diff --git a/bench/README.md b/bench/README.md new file mode 100644 index 0000000..acac9c0 --- /dev/null +++ b/bench/README.md @@ -0,0 +1,37 @@ +# a320-bench — Phase 5 benchmark harness + +Code half of the benchmark (#19): scenario loading/validation, the episode +runner that sits an LLM in front of the benchmark MCP surface, and trajectory +recording. The data half — scenario YAMLs and their QRH-sourced ground truth — +lives in [`scenarios/`](../scenarios/). + +## Install + +```powershell +pip install -e bindings/ # a320_sim (not on PyPI) +pip install -e mcp/ # a320_mcp (START_STATES, create_server) +pip install -e bench/ +``` + +## Layout + +- `a320_bench/scenario.py` — YAML loader: jsonschema shape validation against + [`scenarios/schema/scenario.schema.json`](../scenarios/schema/scenario.schema.json), + then live cross-checks of every control name, failure id and start state + against the core catalogs (a bad reference fails at load time, not + mid-episode). +- `a320_bench/episode.py` — episode runner (slice C, #70). +- `a320_bench/recorder.py` — JSONL trajectory writer (slice C, #70). +- `a320_bench/providers/` — agent adapters; `scripted` needs no LLM and is + what CI runs (litellm adapter arrives in slice D, #71). + +## Tests + +```powershell +python -m pytest bench/tests -q # no LLM, no network +``` + +## License + +GPLv3 — drives the `a320_sim` extension, which links the vendored FlyByWire +crates and inherits their license. diff --git a/bench/a320_bench/__init__.py b/bench/a320_bench/__init__.py new file mode 100644 index 0000000..ca19b60 --- /dev/null +++ b/bench/a320_bench/__init__.py @@ -0,0 +1,17 @@ +"""a320-bench: the Phase 5 benchmark harness (#19). + +Data lives in ``scenarios/`` (YAML, one file per scenario); this package is +the code that loads it, runs an agent episode against the benchmark MCP +surface, and records the trajectory for #20 to score. +""" + +from a320_bench.scenario import ( + Scenario, + ScenarioError, + evaluate_predicate, + load_scenario, +) + +__all__ = ["Scenario", "ScenarioError", "evaluate_predicate", "load_scenario"] + +__version__ = "0.1.0" diff --git a/bench/a320_bench/scenario.py b/bench/a320_bench/scenario.py new file mode 100644 index 0000000..9478fc2 --- /dev/null +++ b/bench/a320_bench/scenario.py @@ -0,0 +1,343 @@ +"""Scenario loading and validation: YAML in, typed and cross-checked out. + +Two validation layers, deliberately separate: + +1. **Shape** — jsonschema against ``scenarios/schema/scenario.schema.json``. + Catches structural mistakes with a JSON-path to the offender. +2. **References** — every failure id, control name and start state is checked + against the live catalogs (``a320_sim`` + ``a320_mcp.START_STATES``). The + schema cannot know the catalogs, and a scenario whose failure id does not + exist would only blow up mid-episode otherwise. Same design as the MCP + schemas embedding the catalogs as enums (D-017): a name that does not + exist must fail at load time, loudly. + +The loader applies the schema's documented defaults itself (jsonschema +validates defaults, it does not inject them). +""" + +import functools +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import jsonschema +import yaml + +# bench/a320_bench/scenario.py -> repo root. Scenario data and its schema are +# repo files, not package data: the benchmark runs from a checkout (the vendor +# pin is part of the benchmark's identity), so resolving relative to the repo +# is the honest layout rather than pretending this could ship as a wheel. +REPO_ROOT = Path(__file__).resolve().parents[2] +SCHEMA_PATH = REPO_ROOT / "scenarios" / "schema" / "scenario.schema.json" + + +class ScenarioError(Exception): + """A scenario file is invalid — always says which file and which field.""" + + +# --- typed model -------------------------------------------------------------- +@dataclass(frozen=True) +class Predicate: + var: str + op: str # eq | ne | gt | ge | lt | le | between + value: float | None = None + min: float | None = None + max: float | None = None + + +@dataclass(frozen=True) +class Action: + control: str + value: float + rationale: str = "" + + +@dataclass(frozen=True) +class ForbiddenAction: + control: str + value: float + severity: str # dangerous | anti_procedure + rationale: str = "" + + +@dataclass(frozen=True) +class ProcedureBlock: + block: str + actions: tuple[Action, ...] + ordered: bool = False + + +@dataclass(frozen=True) +class FailureSpec: + id: str + after_setup_s: float | None = None + when: Predicate | None = None + settle_s: float = 5.0 + + +@dataclass(frozen=True) +class SourceRef: + document: str + revision: str + accessed: str + url: str = "" + notes: str = "" + + +@dataclass(frozen=True) +class GroundTruth: + source: SourceRef + procedure: tuple[ProcedureBlock, ...] + optional_actions: tuple[Action, ...] = () + forbidden_actions: tuple[ForbiddenAction, ...] = () + + +@dataclass(frozen=True) +class InitialState: + start: str + world_controls: dict[str, float] = field(default_factory=dict) + set_controls: dict[str, float] = field(default_factory=dict) + + +@dataclass(frozen=True) +class ExpectedEcam: + must_appear: tuple[str, ...] + must_not_appear: tuple[str, ...] = () + + +@dataclass(frozen=True) +class Success: + final_state: tuple[Predicate, ...] + ecam_clear_of: tuple[str, ...] = () + + +@dataclass(frozen=True) +class Budget: + max_tool_calls: int + max_sim_time_s: float + + +@dataclass(frozen=True) +class Scenario: + id: str + title: str + system: str + initial_state: InitialState + failures: tuple[FailureSpec, ...] + expected_ecam: ExpectedEcam + task_prompt: str + ground_truth: GroundTruth + success: Success + budget: Budget + instructions_profile: str = "benchmark" + path: Path | None = None + raw: dict[str, Any] = field(default_factory=dict, compare=False, repr=False) + + +# --- predicates --------------------------------------------------------------- +def evaluate_predicate(pred: Predicate, value: float) -> bool: + """Evaluate one predicate against a read variable value. + + Success criteria are predicates with tolerance windows, never snapshot + equality: the vendor has real randomness (see the determinism decision in + docs/decisiones.md), so a scenario asserts *the class* of end state. + """ + if pred.op == "eq": + return value == pred.value + if pred.op == "ne": + return value != pred.value + if pred.op == "gt": + return value > pred.value # type: ignore[operator] + if pred.op == "ge": + return value >= pred.value # type: ignore[operator] + if pred.op == "lt": + return value < pred.value # type: ignore[operator] + if pred.op == "le": + return value <= pred.value # type: ignore[operator] + if pred.op == "between": + return pred.min <= value <= pred.max # type: ignore[operator] + raise ScenarioError(f"unknown predicate op '{pred.op}'") # pragma: no cover - schema-gated + + +# --- catalogs (cached: building a Sim costs ~1 s) ------------------------------ +@functools.lru_cache(maxsize=1) +def _catalogs() -> tuple[dict[str, str], frozenset[str], frozenset[str]]: + """(control name -> domain, failure ids, START_STATES keys), from the live core.""" + import a320_sim + from a320_mcp.server import START_STATES + + sim = a320_sim.Sim() + domains = {c["name"]: c["domain"] for c in sim.list_controls()} + failures = frozenset(f["id"] for f in sim.list_failures()) + return domains, failures, frozenset(START_STATES) + + +# --- loading ------------------------------------------------------------------- +@functools.lru_cache(maxsize=1) +def _schema() -> dict[str, Any]: + try: + return json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) + except FileNotFoundError as exc: # pragma: no cover - broken checkout + raise ScenarioError(f"scenario schema not found at {SCHEMA_PATH}") from exc + + +def _predicate(data: dict[str, Any]) -> Predicate: + return Predicate( + var=data["var"], + op=data["op"], + value=data.get("value"), + min=data.get("min"), + max=data.get("max"), + ) + + +def _action(data: dict[str, Any]) -> Action: + return Action(control=data["control"], value=data["value"], rationale=data.get("rationale", "")) + + +def load_scenario(path: "str | Path", *, check_catalogs: bool = True) -> Scenario: + """Load and validate one scenario YAML. + + `check_catalogs=False` skips the live cross-checks (control names, failure + ids, start states) for tests that exercise pure shape validation without + paying for a Sim. + """ + path = Path(path) + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise ScenarioError(f"{path}: not found") from exc + except yaml.YAMLError as exc: + raise ScenarioError(f"{path}: not valid YAML: {exc}") from exc + + validator = jsonschema.Draft202012Validator(_schema()) + errors = sorted(validator.iter_errors(data), key=lambda e: list(e.absolute_path)) + if errors: + first = errors[0] + where = "/".join(str(p) for p in first.absolute_path) or "" + raise ScenarioError(f"{path}: at {where}: {first.message}") + + scenario = Scenario( + id=data["id"], + title=data["title"], + system=data["system"], + initial_state=InitialState( + start=data["initial_state"]["start"], + world_controls=dict(data["initial_state"].get("world_controls", {})), + set_controls=dict(data["initial_state"].get("set_controls", {})), + ), + failures=tuple( + FailureSpec( + id=f["id"], + after_setup_s=f["at"].get("after_setup_s"), + when=_predicate(f["at"]["when"]) if "when" in f["at"] else None, + settle_s=f.get("settle_s", 5.0), + ) + for f in data["failures"] + ), + expected_ecam=ExpectedEcam( + must_appear=tuple(data["expected_ecam"]["must_appear"]), + 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", [])), + ), + budget=Budget( + max_tool_calls=data["budget"]["max_tool_calls"], + max_sim_time_s=data["budget"]["max_sim_time_s"], + ), + instructions_profile=data.get("instructions_profile", "benchmark"), + path=path, + raw=data, + ) + + # Every predicate in the file, not only success criteria: a failure trigger + # with an empty window would otherwise hang the injection wait mid-episode. + predicates = [("success", pred) for pred in scenario.success.final_state] + predicates += [ + ("failure trigger", failure.when) for failure in scenario.failures if failure.when + ] + for where, pred in predicates: + if pred.op == "between" and pred.min > pred.max: # type: ignore[operator] + raise ScenarioError( + f"{path}: {where} predicate on '{pred.var}': min {pred.min} > max {pred.max}" + ) + + if check_catalogs: + _cross_check(scenario, path) + return scenario + + +def _cross_check(scenario: Scenario, path: Path) -> None: + """Every reference must exist in the live catalogs — fail at load, not mid-episode.""" + domains, failure_ids, start_states = _catalogs() + + if scenario.initial_state.start not in start_states: + raise ScenarioError( + f"{path}: unknown start state '{scenario.initial_state.start}' " + f"(expected one of {sorted(start_states)})" + ) + + for failure in scenario.failures: + if failure.id not in failure_ids: + raise ScenarioError( + f"{path}: unknown failure id '{failure.id}' (not in the core catalog; " + f"see list_failures())" + ) + + def check_control(name: str, where: str, *, must_be_world: bool = False) -> None: + if name not in domains: + raise ScenarioError( + f"{path}: unknown control '{name}' in {where} (not in the core catalog; " + f"see list_controls())" + ) + if must_be_world and domains[name] != "world": + raise ScenarioError( + f"{path}: control '{name}' in {where} has domain '{domains[name]}', " + f"but world_controls may only pre-set domain=world controls — cockpit " + f"state belongs in set_controls or in the agent's hands" + ) + + for name in scenario.initial_state.world_controls: + check_control(name, "initial_state.world_controls", must_be_world=True) + for name in scenario.initial_state.set_controls: + check_control(name, "initial_state.set_controls") + for block in scenario.ground_truth.procedure: + for action in block.actions: + check_control(action.control, f"procedure block '{block.block}'") + for action in scenario.ground_truth.optional_actions: + check_control(action.control, "optional_actions") + for forbidden in scenario.ground_truth.forbidden_actions: + check_control(forbidden.control, "forbidden_actions") diff --git a/bench/pyproject.toml b/bench/pyproject.toml new file mode 100644 index 0000000..57fcf16 --- /dev/null +++ b/bench/pyproject.toml @@ -0,0 +1,29 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[project] +name = "a320-bench" +version = "0.1.0" +description = "Phase 5 benchmark harness: scenario suite, episode runner and trajectory recording over the a320 MCP server" +readme = "README.md" +requires-python = ">=3.10" +# Table form, not the PEP 639 bare string: the string form needs setuptools>=77, +# which would contradict the >=61 floor above. Same style as mcp/pyproject.toml. +license = { text = "GPL-3.0-or-later" } +dependencies = [ + # Not on PyPI: pip install -e bindings/ and -e mcp/ first. + "a320-sim", + "a320-mcp", + "mcp>=1.28,<2", + "PyYAML>=6", + "jsonschema>=4", +] + +[project.optional-dependencies] +# Real LLM providers (slice D). CI installs without this extra: every test +# runs against the ScriptedAdapter, no network, no keys. +providers = [] + +[tool.setuptools.packages.find] +include = ["a320_bench*"] diff --git a/bench/tests/test_scenario_schema.py b/bench/tests/test_scenario_schema.py new file mode 100644 index 0000000..391f555 --- /dev/null +++ b/bench/tests/test_scenario_schema.py @@ -0,0 +1,180 @@ +"""Scenario loading and validation tests (#69). + +Two layers under test: shape (jsonschema, no Sim needed) and references +(live catalog cross-checks). Invalid scenarios must fail at load time with a +message that names the file and the offending field — a scenario that only +blows up mid-episode wastes an LLM run. + +Runnable two ways, same as the MCP tests: + - directly: python bench/tests/test_scenario_schema.py + - under pytest: pytest bench/tests/ +""" + +import copy +import tempfile +from pathlib import Path + +import yaml + +from a320_bench import Scenario, ScenarioError, evaluate_predicate, load_scenario +from a320_bench.scenario import Predicate, REPO_ROOT + +FIRST_SCENARIO = REPO_ROOT / "scenarios" / "elec" / "apu_gen_fault.yaml" + + +def _base() -> dict: + return yaml.safe_load(FIRST_SCENARIO.read_text(encoding="utf-8")) + + +def _load_mutated(data: dict, *, check_catalogs: bool = True) -> Scenario: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "scenario.yaml" + path.write_text(yaml.safe_dump(data), encoding="utf-8") + return load_scenario(path, check_catalogs=check_catalogs) + + +def _expect_error(data: dict, *needles: str, check_catalogs: bool = True) -> None: + try: + _load_mutated(data, check_catalogs=check_catalogs) + except ScenarioError as exc: + for needle in needles: + assert needle in str(exc), f"error should mention '{needle}': {exc}" + else: + raise AssertionError(f"expected ScenarioError mentioning {needles}") + + +# --- the real scenario ------------------------------------------------------- +def test_first_scenario_loads_with_catalog_cross_checks(): + """The shipped APU GEN scenario is valid against schema AND catalogs. + + This is the test that keeps the dataset honest: if the core renames a + control or a failure id, this fails at CI time instead of mid-benchmark. + """ + scenario = load_scenario(FIRST_SCENARIO) + + assert scenario.id == "elec-apu-gen-fault" + assert scenario.system == "ELEC" + assert scenario.initial_state.start == "apu-running" + assert scenario.initial_state.world_controls == {"ext_pwr_avail": 1} + assert [f.id for f in scenario.failures] == ["elec.apu_gen.1"] + assert scenario.failures[0].after_setup_s == 10 + assert scenario.failures[0].settle_s == 5 + assert "APU GEN FAULT" in scenario.expected_ecam.must_appear + + blocks = scenario.ground_truth.procedure + assert [b.block for b in blocks] == ["reset_attempt", "restore_power"] + assert blocks[0].ordered and not blocks[1].ordered + assert blocks[0].actions[0].control == "apu_gen" + + assert scenario.ground_truth.source.url, "the citation must carry a verifiable URL" + assert scenario.ground_truth.source.accessed == "2026-07-23" + assert scenario.instructions_profile == "benchmark" + assert scenario.budget.max_tool_calls == 40 + + +def test_task_prompt_does_not_leak_the_ground_truth(): + """The prompt must not hand the agent the diagnosis (D-016 spirit). + + A prompt that names the failed system or the failure id turns diagnosis + into reading comprehension. Checked for every scenario in the suite. + """ + for path in sorted((REPO_ROOT / "scenarios").rglob("*.yaml")): + scenario = load_scenario(path, check_catalogs=False) + prompt = scenario.task_prompt.lower() + for failure in scenario.failures: + for token in failure.id.split("."): + if len(token) > 3: # 'elec'/'apu_gen' yes; bus indices no + assert token not in prompt, ( + f"{path.name}: task_prompt leaks '{token}' from {failure.id}" + ) + for message in scenario.expected_ecam.must_appear: + assert message.lower() not in prompt, ( + f"{path.name}: task_prompt leaks the expected ECAM '{message}'" + ) + + +# --- shape errors (no Sim needed) --------------------------------------------- +def test_missing_required_section_names_the_field(): + data = _base() + del data["ground_truth"] + _expect_error(data, "ground_truth", check_catalogs=False) + + +def test_between_predicate_requires_min_and_max(): + data = _base() + data["success"]["final_state"] = [{"var": "X", "op": "between", "value": 1}] + _expect_error(data, "final_state", check_catalogs=False) + + +def test_inverted_between_bounds_are_rejected(): + data = _base() + data["success"]["final_state"] = [{"var": "X", "op": "between", "min": 5, "max": 1}] + _expect_error(data, "min 5", "max 1", check_catalogs=False) + + +def test_inverted_between_bounds_in_failure_trigger_are_rejected(): + """The empty-window check covers `failures[].at.when` too, not only success. + + A trigger predicate that can never hold would hang the injection wait + mid-episode — exactly the class of error load time exists to catch. + """ + data = _base() + data["failures"][0]["at"] = {"when": {"var": "X", "op": "between", "min": 5, "max": 1}} + _expect_error(data, "min 5", "max 1", check_catalogs=False) + + +def test_unknown_top_level_key_is_rejected(): + """additionalProperties: false — a typo'd section must not pass silently.""" + data = _base() + data["succes"] = data["success"] + _expect_error(data, "succes", check_catalogs=False) + + +# --- reference errors (live catalogs) ------------------------------------------- +def test_unknown_failure_id_is_rejected(): + data = _base() + data["failures"][0]["id"] = "elec.flux_capacitor.1" + _expect_error(data, "elec.flux_capacitor.1", "catalog") + + +def test_unknown_control_in_procedure_is_rejected(): + data = _base() + data["ground_truth"]["procedure"][0]["actions"][0]["control"] = "no_such_pb" + _expect_error(data, "no_such_pb", "reset_attempt") + + +def test_world_controls_must_be_world_domain(): + """A cockpit control in world_controls is a scenario-design error. + + World state is the scenario's to fix; cockpit state is the agent's to + manage (mcp/README.md, Phase 3 closure note). bat_1 is a cockpit pb. + """ + data = _base() + data["initial_state"]["world_controls"]["bat_1"] = 1 + _expect_error(data, "bat_1", "world") + + +def test_unknown_start_state_is_rejected(): + data = _base() + data["initial_state"]["start"] = "hangar" + _expect_error(data, "hangar") + + +# --- predicates ------------------------------------------------------------------ +def test_predicate_evaluation(): + assert evaluate_predicate(Predicate(var="v", op="eq", value=1.0), 1.0) + assert not evaluate_predicate(Predicate(var="v", op="eq", value=1.0), 0.0) + assert evaluate_predicate(Predicate(var="v", op="between", min=2800, max=3100), 2950.0) + assert not evaluate_predicate(Predicate(var="v", op="between", min=2800, max=3100), 100.0) + assert evaluate_predicate(Predicate(var="v", op="ge", value=0.5), 0.5) + assert not evaluate_predicate(Predicate(var="v", op="lt", value=0.5), 0.5) + + +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)} scenario tests passed.") diff --git a/docs/assets/banner.svg b/docs/assets/banner.svg index 21e6bf4..eff119d 100644 --- a/docs/assets/banner.svg +++ b/docs/assets/banner.svg @@ -1,4 +1,4 @@ - + @@ -7,8 +7,8 @@ - - + + @@ -18,49 +18,123 @@ + + - + + + A320 SYSTEMS + TWIN + headless FlyByWire systems core + CLI + MCP - a failure-management + benchmark for LLM agents + + + - + ELEC TR 1 FAULT + AGENT: RUN QRH PROC . . . + + + + + + ELEC + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + BAT 1 + 28V 0A + BAT 2 + 28V 0A + + + DC BAT + + - - - - + + + - - AC 1 - AC 2 - DC 1 - DC BAT + + DC 1 + DC ESS + DC 2 - - - - - - + + + + + + + + + TR 1 + FAULT + ESS TR + 28V 0A + TR 2 + 28V 60A - - - TR 1 - FAULT - - EXT PWR - - - - A320 SYSTEMS - TWIN - headless FlyByWire systems core - CLI + MCP - a failure-management benchmark for LLM agents - + + + + + + + + AC 1 + AC ESS + AC 2 + - - - ELEC TR 1 FAULT - AGENT: RUN QRH PROC . . . + + + + + + + + + GEN 1 + OFF + APU GEN + OFF + EXT PWR + 115V 400HZ + GEN 2 + OFF + diff --git a/mcp/README.md b/mcp/README.md index 3b3fc2a..9cebe1c 100644 --- a/mcp/README.md +++ b/mcp/README.md @@ -88,6 +88,14 @@ agent the answer it is supposed to diagnose from the ECAM; the second would bury the context window under hundreds of names. See D-016 in [docs/decisiones.md](../docs/decisiones.md). +**Tool profiles.** The table above is the `interactive` profile — what the stdio +entry point serves. The server is built by `create_server(sim, profile=...)`; +the Phase 5 benchmark runner builds one with `profile="benchmark"`, which +withholds `inject_failure`/`clear_failure` (the injected failure is the exam — +an agent that can repair the fault is not flying the procedure) and adds +`report_done(diagnosis, actions_summary)` as the explicit end-of-episode +channel. Profiles change which tools exist, never what a tool does. + ## Worked example: the agent loop What a client sees driving the `apu-running` scenario: diff --git a/mcp/a320_mcp/server.py b/mcp/a320_mcp/server.py index 2583846..9835609 100644 --- a/mcp/a320_mcp/server.py +++ b/mcp/a320_mcp/server.py @@ -21,12 +21,20 @@ - **Tool descriptions are the agent's only documentation of an aircraft it cannot see.** They are prompt engineering, not docstrings — and in Phase 5 they are an ablation axis. Written accordingly. +- **The server is a factory, and the tool surface is a profile.** ``create_server`` + builds a ``FastMCP`` over a given ``Sim``. The ``interactive`` profile is the + full 9-tool surface a human or an exploring agent gets over stdio. The + ``benchmark`` profile is what a *graded* agent gets: ``inject_failure`` and + ``clear_failure`` are withheld (the injected failure is the exam — an agent + that can repair the fault, or break something else, is not flying the + procedure) and ``report_done`` is added as the explicit end-of-episode + channel. Profiles change which tools exist, never what a tool does. """ import argparse import sys from collections.abc import Callable -from typing import Literal +from typing import Literal, get_args try: import a320_sim @@ -44,11 +52,13 @@ from mcp.types import ToolAnnotations # --- the aircraft ----------------------------------------------------------- -# One Sim per process. stdio means one client per process, and every tool call -# lands on the same (event loop) thread, which is what `unsendable` requires. +# One Sim per process for the stdio path. stdio means one client per process, +# and every tool call lands on the same (event loop) thread, which is what +# `unsendable` requires. The benchmark runner instead builds its own Sim per +# episode and passes it to `create_server`. # # It is built at import, not in main(), because the tool schemas below embed the -# catalogs as enums and decorators run at import time. Cost: ~1 s to instantiate +# catalogs as enums and those are read once at import. Cost: ~1 s to instantiate # the A320. It would be paid at startup regardless. _sim = a320_sim.Sim() @@ -59,7 +69,9 @@ # That is the curated half of discovery doing its job (D-009): the agent actuates # cockpit controls a human curated, and cannot reach an arbitrary variable. If a # scenario needs a control that isn't here, the fix is to catalog it in -# core-rs/src/controls.rs, not to widen this enum. +# core-rs/src/controls.rs, not to widen this enum. The catalogs are static +# tables of the core, identical for every Sim instance, so reading them from +# the module-level Sim is safe for servers built over a different instance. ControlName = Literal[tuple(sorted(c["name"] for c in _CONTROLS))] # type: ignore[valid-type] FailureId = Literal[tuple(sorted(f["id"] for f in _FAILURES))] # type: ignore[valid-type] @@ -102,210 +114,278 @@ start attempt (recover by cycling the master). """ -mcp = FastMCP("a320-systems", instructions=INSTRUCTIONS) - -# Every tool below is closed-world: the simulator is self-contained, and saying -# so keeps a client from treating these as calls out to the internet. -_READ_ONLY = ToolAnnotations(readOnlyHint=True, openWorldHint=False) - - -# --- observation ------------------------------------------------------------ -@mcp.tool(annotations=_READ_ONLY) -def read_ecam() -> list[dict[str, str]]: - """Read the active ECAM warnings and cautions, most severe first. - - This is what a pilot sees and reasons from, and it is your main source of - truth about what is wrong. Each entry has: `message` (the ECAM text, e.g. - "APU GEN FAULT"), `severity` (warning > caution > advisory), `system`, `id`, - and `source`. - - `source` says who computed the warning: `vendor_flag` means the aircraft - model itself raised the fault; `derived` means it was inferred from the - aircraft's state. Both are real; the distinction is recorded for honesty. - - An empty list means the ECAM is clear. It is also empty when the ECAM is - not powered (cold and dark) — with no electrical power there is no display, - just as on the real aircraft. So an empty ECAM on an unpowered aircraft is - not evidence that nothing is wrong. - """ - return _sim.read_ecam() - +INSTRUCTIONS_BENCHMARK = ( + INSTRUCTIONS + + """ +You are managing a failure that has already occurred. Diagnose it from the \ +ECAM and the state, apply the appropriate procedure with `set_control` and \ +`advance`, and when you judge the situation handled — or conclude that nothing \ +more can be done — call `report_done` with your diagnosis and a summary of the \ +actions you took. Make no further tool calls after `report_done`. +""" +) -@mcp.tool(annotations=_READ_ONLY) -def read_state(variables: list[str]) -> dict[str, float]: - """Read specific state variables by name, e.g. `ELEC_AC_1_BUS_IS_POWERED`. +# The instructions are prompt engineering, not documentation (see the module +# docstring), and in Phase 5 they are an ablation axis: the runner picks a +# variant by name and records which one the agent saw. +INSTRUCTIONS_PROFILES: "dict[str, str]" = { + "default": INSTRUCTIONS, + "benchmark": INSTRUCTIONS_BENCHMARK, +} - Takes a list and returns a name -> value map. Booleans come back as 1.0 or - 0.0. Ask only for what you need: this is the precise instrument, not a dump. +# Every tool is closed-world: the simulator is self-contained, and saying so +# keeps a client from treating these as calls out to the internet. +_READ_ONLY = ToolAnnotations(readOnlyHint=True, openWorldHint=False) - Use `snapshot` with a filter to discover which variable names exist, and - `list_controls` to see the variables you can write (each control lists its - underlying `lvar`). An unknown name is an error naming the offender, so a - typo fails loudly rather than reading as 0. - """ - return _sim.get(variables) +# The static type and the runtime check share one source of truth: the tuple +# used in the error message is derived from the Literal. +Profile = Literal["interactive", "benchmark"] +PROFILES: "tuple[str, ...]" = get_args(Profile) -@mcp.tool(annotations=_READ_ONLY) -def snapshot(contains: str) -> dict[str, float]: - """Discover state variables whose name contains `contains` (case-sensitive). +def create_server( + sim: "a320_sim.Sim", + *, + instructions: "str | None" = None, + profile: Profile = "interactive", +) -> FastMCP: + """Build a FastMCP server whose tools drive `sim`. - This is how you find out what is observable — there is deliberately no tool - that lists every variable, because the registry runs to hundreds of names - and would bury the useful ones. + `profile` selects the tool surface: ``interactive`` (the default) is the + full 9-tool surface served over stdio; ``benchmark`` withholds + ``inject_failure``/``clear_failure`` and adds ``report_done``. When + `instructions` is None, the profile's default variant from + ``INSTRUCTIONS_PROFILES`` is used (``interactive`` -> ``default``). - Filter by system prefix and narrow from there: `ELEC_AC` for the AC network, - `ELEC_DC` for DC, `OVHD_ELEC` for the electrical overhead panel, `APU` for - the APU. A filter that matches too much is rejected — narrow it rather than - reading everything. + Tool functions are sync closures over `sim` — see the module docstring for + why sync is load-bearing (`unsendable`, D-010/D-015). """ - matches = {k: v for k, v in _sim.snapshot().items() if contains in k} - if not matches: - raise ToolError( - f"no variable name contains '{contains}'. Try a broader or different " - f"filter (e.g. 'ELEC_AC', 'ELEC_DC', 'OVHD_ELEC', 'APU')." + if profile not in PROFILES: + raise ValueError(f"unknown profile '{profile}' (expected one of {PROFILES})") + if instructions is None: + instructions = INSTRUCTIONS_PROFILES["benchmark" if profile == "benchmark" else "default"] + + server = FastMCP("a320-systems", instructions=instructions) + + # --- observation --------------------------------------------------------- + @server.tool(annotations=_READ_ONLY) + def read_ecam() -> list[dict[str, str]]: + """Read the active ECAM warnings and cautions, most severe first. + + This is what a pilot sees and reasons from, and it is your main source of + truth about what is wrong. Each entry has: `message` (the ECAM text, e.g. + "APU GEN FAULT"), `severity` (warning > caution > advisory), `system`, `id`, + and `source`. + + `source` says who computed the warning: `vendor_flag` means the aircraft + model itself raised the fault; `derived` means it was inferred from the + aircraft's state. Both are real; the distinction is recorded for honesty. + + An empty list means the ECAM is clear. It is also empty when the ECAM is + not powered (cold and dark) — with no electrical power there is no display, + just as on the real aircraft. So an empty ECAM on an unpowered aircraft is + not evidence that nothing is wrong. + """ + return sim.read_ecam() + + @server.tool(annotations=_READ_ONLY) + def read_state(variables: list[str]) -> dict[str, float]: + """Read specific state variables by name, e.g. `ELEC_AC_1_BUS_IS_POWERED`. + + Takes a list and returns a name -> value map. Booleans come back as 1.0 or + 0.0. Ask only for what you need: this is the precise instrument, not a dump. + + Use `snapshot` with a filter to discover which variable names exist, and + `list_controls` to see the variables you can write (each control lists its + underlying `lvar`). An unknown name is an error naming the offender, so a + typo fails loudly rather than reading as 0. + """ + return sim.get(variables) + + @server.tool(annotations=_READ_ONLY) + def snapshot(contains: str) -> dict[str, float]: + """Discover state variables whose name contains `contains` (case-sensitive). + + This is how you find out what is observable — there is deliberately no tool + that lists every variable, because the registry runs to hundreds of names + and would bury the useful ones. + + Filter by system prefix and narrow from there: `ELEC_AC` for the AC network, + `ELEC_DC` for DC, `OVHD_ELEC` for the electrical overhead panel, `APU` for + the APU. A filter that matches too much is rejected — narrow it rather than + reading everything. + """ + matches = {k: v for k, v in sim.snapshot().items() if contains in k} + if not matches: + raise ToolError( + f"no variable name contains '{contains}'. Try a broader or different " + f"filter (e.g. 'ELEC_AC', 'ELEC_DC', 'OVHD_ELEC', 'APU')." + ) + if len(matches) > MAX_SNAPSHOT_VARS: + raise ToolError( + f"filter '{contains}' matches {len(matches)} variables (max " + f"{MAX_SNAPSHOT_VARS}). Narrow it — e.g. '{contains}_' or a more " + f"specific prefix — or read the ones you need with read_state." + ) + return matches + + # --- discovery ----------------------------------------------------------- + @server.tool(annotations=_READ_ONLY) + def list_controls() -> list[dict[str, str]]: + """List the cockpit controls you can actuate with `set_control`. + + Curated by hand, not a dump of the variable registry. Each entry has: `name` + (what you pass to `set_control`), `lvar` (the underlying variable, readable + with `read_state`), `kind`, `valid_values`, `description`, `group`, and + `domain`. + + `domain` is worth reading: `cockpit` is a real control a pilot actuates; + `world` is outside state that a real simulator would provide and that we + fake here (e.g. whether a ground power unit is plugged in). + """ + return _CONTROLS + + @server.tool(annotations=_READ_ONLY) + def list_failures() -> list[dict[str, str]]: + """List the failures that can be injected with `inject_failure`. + + Each entry has a stable `id` (e.g. `elec.tr.1`), a `description`, a `group`, + and `ata` (the ATA chapter id the aircraft vendor uses for the same failure). + + This is the catalog of what *can* break — not what *is* broken. To find out + what is currently wrong, read the ECAM. + """ + return _FAILURES + + # --- action -------------------------------------------------------------- + @server.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, # additive: it sets a switch, it destroys nothing + idempotentHint=True, # writing the same value twice is the same state + openWorldHint=False, ) - if len(matches) > MAX_SNAPSHOT_VARS: - raise ToolError( - f"filter '{contains}' matches {len(matches)} variables (max " - f"{MAX_SNAPSHOT_VARS}). Narrow it — e.g. '{contains}_' or a more " - f"specific prefix — or read the ones you need with read_state." + ) + def set_control(control: ControlName, value: float) -> str: + """Actuate a cockpit control: flip a switch or push a pushbutton. + + `control` is a name from `list_controls`; `value` is 1 for on/auto and 0 for + off (see each control's `valid_values`). An out-of-range value is rejected + rather than silently coerced. + + The write lands immediately, but the aircraft does not react until time + passes: call `advance` afterwards, then read the state back to confirm. + """ + sim.set(control, value) + return f"{control} <- {value:g} (call advance() for the aircraft to react)" + + @server.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, + idempotentHint=False, # time moves: the same call twice is not the same + openWorldHint=False, ) - return matches - - -# --- discovery -------------------------------------------------------------- -@mcp.tool(annotations=_READ_ONLY) -def list_controls() -> list[dict[str, str]]: - """List the cockpit controls you can actuate with `set_control`. - - Curated by hand, not a dump of the variable registry. Each entry has: `name` - (what you pass to `set_control`), `lvar` (the underlying variable, readable - with `read_state`), `kind`, `valid_values`, `description`, `group`, and - `domain`. - - `domain` is worth reading: `cockpit` is a real control a pilot actuates; - `world` is outside state that a real simulator would provide and that we - fake here (e.g. whether a ground power unit is plugged in). - """ - return _CONTROLS - - -@mcp.tool(annotations=_READ_ONLY) -def list_failures() -> list[dict[str, str]]: - """List the failures that can be injected with `inject_failure`. - - Each entry has a stable `id` (e.g. `elec.tr.1`), a `description`, a `group`, - and `ata` (the ATA chapter id the aircraft vendor uses for the same failure). - - This is the catalog of what *can* break — not what *is* broken. To find out - what is currently wrong, read the ECAM. - """ - return _FAILURES - - -# --- action ----------------------------------------------------------------- -@mcp.tool( - annotations=ToolAnnotations( - readOnlyHint=False, - destructiveHint=False, # additive: it sets a switch, it destroys nothing - idempotentHint=True, # writing the same value twice is the same state - openWorldHint=False, ) -) -def set_control(control: ControlName, value: float) -> str: - """Actuate a cockpit control: flip a switch or push a pushbutton. - - `control` is a name from `list_controls`; `value` is 1 for on/auto and 0 for - off (see each control's `valid_values`). An out-of-range value is rejected - rather than silently coerced. - - The write lands immediately, but the aircraft does not react until time - passes: call `advance` afterwards, then read the state back to confirm. - """ - _sim.set(control, value) - return f"{control} <- {value:g} (call advance() for the aircraft to react)" + def advance(seconds: float, rate: float = 5.0) -> str: + """Advance simulated time. Nothing you do takes effect until you call this. + + `seconds` is simulated time, not wall-clock: it runs as fast as it computes. + `rate` is ticks per second (5 is the usual settling rate; leave it alone + unless you have a reason). + + Rules of thumb: 2 seconds to let a contactor sequence settle after acting; + 5 seconds for a network to reconfigure after a failure; ~65 seconds for an + APU to spin up. Returns the new simulated clock. + """ + if seconds <= 0: + raise ToolError(f"seconds must be positive, got {seconds}") + if seconds > MAX_ADVANCE_S: + raise ToolError( + f"seconds must be at most {MAX_ADVANCE_S:g} in one call, got {seconds:g}. " + f"Advance in steps and observe in between — that is the loop." + ) + if rate <= 0: + raise ToolError(f"rate must be positive, got {rate}") + # Blocking on purpose. This runs on the event loop thread, and it must: the + # Sim is `unsendable` (D-010), so handing this to anyio.to_thread to "avoid + # blocking" would raise RuntimeError from the binding. With stdio there is a + # single client and nothing else to serve, so there is nothing to block. + sim.run(seconds, rate) + return f"advanced {seconds:g}s (t={sim.sim_time():.1f}s)" + + if profile == "interactive": + + @server.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=True, # it breaks a system (reversibly, via clear_failure) + idempotentHint=True, # a set: injecting twice is injecting once + openWorldHint=False, + ) + ) + def inject_failure(failure_id: FailureId) -> str: + """Break something: inject a failure by its id from `list_failures`. + + Reversible with `clear_failure`. Takes effect on the next `advance`. + + This is a scenario-authoring tool. If you are being asked to *manage* a + failure, it has already been injected for you — diagnose it from the ECAM + rather than injecting your own. + """ + sim.inject_failure(failure_id) + return f"injected {failure_id} (call advance() for it to take effect)" + + @server.tool( + annotations=ToolAnnotations( + readOnlyHint=False, + destructiveHint=False, # restores: the opposite of destructive + idempotentHint=True, + openWorldHint=False, + ) + ) + def clear_failure(failure_id: FailureId) -> str: + """Repair an injected failure by its id. Takes effect on the next `advance`. + Clearing a failure that is not active is fine, not an error. -@mcp.tool( - annotations=ToolAnnotations( - readOnlyHint=False, - destructiveHint=False, - idempotentHint=False, # time moves: the same call twice is not the same - openWorldHint=False, - ) -) -def advance(seconds: float, rate: float = 5.0) -> str: - """Advance simulated time. Nothing you do takes effect until you call this. + Note this repairs the underlying fault — it is not how a crew responds to a + failure. Managing one means reconfiguring the aircraft with `set_control`. + """ + sim.clear_failure(failure_id) + return f"cleared {failure_id} (call advance() for it to take effect)" - `seconds` is simulated time, not wall-clock: it runs as fast as it computes. - `rate` is ticks per second (5 is the usual settling rate; leave it alone - unless you have a reason). + else: # benchmark - Rules of thumb: 2 seconds to let a contactor sequence settle after acting; - 5 seconds for a network to reconfigure after a failure; ~65 seconds for an - APU to spin up. Returns the new simulated clock. - """ - if seconds <= 0: - raise ToolError(f"seconds must be positive, got {seconds}") - if seconds > MAX_ADVANCE_S: - raise ToolError( - f"seconds must be at most {MAX_ADVANCE_S:g} in one call, got {seconds:g}. " - f"Advance in steps and observe in between — that is the loop." + @server.tool( + annotations=ToolAnnotations( + readOnlyHint=True, # it changes nothing in the aircraft + idempotentHint=True, + openWorldHint=False, + ) ) - if rate <= 0: - raise ToolError(f"rate must be positive, got {rate}") - # Blocking on purpose. This runs on the event loop thread, and it must: the - # Sim is `unsendable` (D-010), so handing this to anyio.to_thread to "avoid - # blocking" would raise RuntimeError from the binding. With stdio there is a - # single client and nothing else to serve, so there is nothing to block. - _sim.run(seconds, rate) - return f"advanced {seconds:g}s (t={_sim.sim_time():.1f}s)" - - -@mcp.tool( - annotations=ToolAnnotations( - readOnlyHint=False, - destructiveHint=True, # it breaks a system (reversibly, via clear_failure) - idempotentHint=True, # a set: injecting twice is injecting once - openWorldHint=False, - ) -) -def inject_failure(failure_id: FailureId) -> str: - """Break something: inject a failure by its id from `list_failures`. + def report_done(diagnosis: str, actions_summary: str) -> str: + """Declare the episode finished. Call this exactly once, at the end. - Reversible with `clear_failure`. Takes effect on the next `advance`. + `diagnosis` is what you concluded was wrong with the aircraft; + `actions_summary` is a short account of the actions you took and why. + After calling this, make no further tool calls. + """ + # The runner mediates every tool call, so the payload is recorded in + # the trajectory; nothing to persist here. + return "Report recorded. The episode is over; make no further tool calls." - This is a scenario-authoring tool. If you are being asked to *manage* a - failure, it has already been injected for you — diagnose it from the ECAM - rather than injecting your own. - """ - _sim.inject_failure(failure_id) - return f"injected {failure_id} (call advance() for it to take effect)" + return server -@mcp.tool( - annotations=ToolAnnotations( - readOnlyHint=False, - destructiveHint=False, # restores: the opposite of destructive - idempotentHint=True, - openWorldHint=False, - ) -) -def clear_failure(failure_id: FailureId) -> str: - """Repair an injected failure by its id. Takes effect on the next `advance`. - - Clearing a failure that is not active is fine, not an error. - - Note this repairs the underlying fault — it is not how a crew responds to a - failure. Managing one means reconfiguring the aircraft with `set_control`. - """ - _sim.clear_failure(failure_id) - return f"cleared {failure_id} (call advance() for it to take effect)" +# The stdio server, built over the module-level Sim. Kept at module level so +# `from a320_mcp.server import mcp` keeps working and `main()` stays a thin +# argparse wrapper. +mcp = create_server(_sim) # --- start states ----------------------------------------------------------- -def _run_until( +def run_until( sim: "a320_sim.Sim", variable: str, target: float, timeout_s: int, what: str ) -> None: """Advance in 1 s steps until `variable` reads exactly `target` (bounded). @@ -314,6 +394,8 @@ def _run_until( pattern as `run_until` in the core's integration tests: never a blind sleep, never unbounded — a start state that cannot be reached is a bug and should fail loudly at server startup, not hand the agent a broken aircraft. + + Public because the Phase 5 benchmark runner reuses it for scenario setup. """ elapsed = 0 while sim.get([variable])[variable] != target: @@ -347,7 +429,7 @@ def _start_apu_running(sim: "a320_sim.Sim") -> None: sim.set("apu_start", 1) # Bounded wait, not a blind sleep: the APS3200 reaches available at ~62 s. - _run_until( + run_until( sim, "OVHD_APU_START_PB_IS_AVAILABLE", 1.0, 150, "the APU did not reach available" ) @@ -387,13 +469,13 @@ def _start_engines_running(sim: "a320_sim.Sim") -> None: sim.set("eng_mode", 2) sim.set("eng_master_1", 1) # ENGINE_STATE: Off=0 / On=1 / Starting=2 — wait for On, not merely non-zero. - _run_until(sim, "ENGINE_STATE:1", 1.0, 120, "engine 1 did not reach idle") + run_until(sim, "ENGINE_STATE:1", 1.0, 120, "engine 1 did not reach idle") sim.set("gen_1", 1) sim.run(2.0, 5.0) # Engine 2 off the crossbleed, then its generator. sim.set("eng_master_2", 1) - _run_until(sim, "ENGINE_STATE:2", 1.0, 120, "engine 2 did not reach idle") + run_until(sim, "ENGINE_STATE:2", 1.0, 120, "engine 2 did not reach idle") sim.set("gen_2", 1) sim.run(2.0, 5.0) @@ -405,7 +487,7 @@ def _start_engines_running(sim: "a320_sim.Sim") -> None: sim.set("apu_gen", 0) sim.set("apu_master", 0) sim.run(5.0, 5.0) - _run_until(sim, "OVHD_APU_START_PB_IS_AVAILABLE", 0.0, 300, "the APU did not shut down") + run_until(sim, "OVHD_APU_START_PB_IS_AVAILABLE", 0.0, 300, "the APU did not shut down") sim.run(5.0, 5.0) diff --git a/mcp/tests/test_server.py b/mcp/tests/test_server.py index a9e63ff..4824654 100644 --- a/mcp/tests/test_server.py +++ b/mcp/tests/test_server.py @@ -87,6 +87,74 @@ async def check(session): assert "list_variables" not in names, "exposing list_variables floods the context window" +def test_benchmark_profile_withholds_failure_tools_and_adds_report_done(): + """The benchmark tool surface is the graded agent's surface (#68). + + `inject_failure`/`clear_failure` must not exist there: the injected failure + is the exam, and an agent that can repair the fault (or break something + else) is not flying the procedure. `report_done` is the explicit + end-of-episode channel the runner watches for. Checked in-process because + the benchmark server is built by the Phase 5 runner, not by the stdio + entry point. + """ + import a320_sim + from a320_mcp.server import create_server + + server = create_server(a320_sim.Sim(), profile="benchmark") + + async def check(): + names = {t.name for t in await server.list_tools()} + ack = await server.call_tool("report_done", {"diagnosis": "x", "actions_summary": "y"}) + return names, ack + + names, ack = run(check()) + + expected = (EXPECTED_TOOLS - {"inject_failure", "clear_failure"}) | {"report_done"} + assert names == expected, f"benchmark surface drifted: {names ^ expected}" + assert "episode is over" in str(ack), ack + + +def test_interactive_profile_is_the_default_and_rejects_unknown_profiles(): + """`create_server` defaults to the stdio surface; a typo'd profile fails loudly.""" + import a320_sim + from a320_mcp.server import create_server + + server = create_server(a320_sim.Sim()) + + async def check(): + return {t.name for t in await server.list_tools()} + + names = run(check()) + assert names == EXPECTED_TOOLS, f"interactive surface drifted: {names ^ EXPECTED_TOOLS}" + + try: + create_server(a320_sim.Sim(), profile="benchmrak") + except ValueError as exc: + assert "benchmrak" in str(exc) + else: + raise AssertionError("an unknown profile should raise ValueError") + + +def test_instructions_profiles_are_wired_to_the_profile(): + """Each profile gets its INSTRUCTIONS variant unless one is passed explicitly. + + The instructions are prompt engineering and a Phase 5 ablation axis: the + benchmark variant must tell the agent about `report_done`, and an explicit + `instructions=` must win over the profile default (that is how ablations + swap the text without touching the tool surface). + """ + import a320_sim + from a320_mcp.server import INSTRUCTIONS, create_server + + interactive = create_server(a320_sim.Sim()) + benchmark = create_server(a320_sim.Sim(), profile="benchmark") + overridden = create_server(a320_sim.Sim(), profile="benchmark", instructions="ablated") + + assert interactive.instructions == INSTRUCTIONS + assert "report_done" in benchmark.instructions + assert overridden.instructions == "ablated" + + def test_schemas_carry_the_catalogs_as_enums(): """The valid names are generated from the catalogs, not hand-written (D-017). diff --git a/scenarios/elec/apu_gen_fault.yaml b/scenarios/elec/apu_gen_fault.yaml new file mode 100644 index 0000000..1465b9a --- /dev/null +++ b/scenarios/elec/apu_gen_fault.yaml @@ -0,0 +1,100 @@ +# First benchmark scenario (#19, slice B): loss of the APU generator on the +# ground, with the APU as the only AC source and a GPU standing by. +# +# Why this one first: the failure and its caution are the Phase 2 chain +# (core-rs/tests/generator_caution.rs), apu-running is the cheapest start state +# (~60 s of sim), and the procedure has a real ordering dependency (reset +# attempt before switching source) that exercises the block semantics. +schema_version: 1 +id: elec-apu-gen-fault +title: "APU GEN FAULT on ground, APU as only AC source, GPU available" +system: ELEC + +initial_state: + start: apu-running + world_controls: + # The GPU is plugged in from the start (world state is the scenario's to + # fix, never the agent's to invent — mcp/README.md, Phase 3 closure note). + # Verified 2026-07-23 against vendor pin 13bce4b: with ext_pwr_avail=1 and + # the EXT PWR pb off, the line contactor stays open and the APU GEN FAULT + # caution still appears when the failure is injected. + ext_pwr_avail: 1 + +failures: + - id: elec.apu_gen.1 + at: { after_setup_s: 10 } + settle_s: 5 + +expected_ecam: + # Validity gate, not the agent's task. Both cautions were stable across + # probe runs; HYD ENG 2 PUMP FAULT appeared in some runs only, so it is + # deliberately not asserted (vendor randomness, D-determinism). + must_appear: + - "APU GEN FAULT" + - "AC ESS BUS FAULT" + must_not_appear: [] + +task_prompt: > + Something is wrong with the aircraft. Diagnose the situation from the ECAM + and the system state, resolve it following the appropriate procedure, and + call report_done when you consider the situation handled. + +ground_truth: + source: + document: >- + Airbus A318/A319/A320/A321 FCOM, PRO-ABN-24 (ELEC APU GEN FAULT): + generator reset attempt via its pushbutton, then the faulty generator + off and an alternate source on. Modeled behavior cross-checked against + the FlyByWire A32NX documentation, "Electrical System Control Panel": + APU GEN pb OFF "the generator is unpowered and the line contactor + opens. The fault circuit is reset."; the FAULT indication is suppressed + when EXT PWR or an ENG GEN provides power. + revision: "FBW docs as published 2026-07-23 (A32NX); FCOM procedure identity, wording not reproduced" + url: "https://docs.flybywiresim.com/pilots-corner/a32nx/a32nx-briefing/flight-deck/ovhd/elec/" + accessed: "2026-07-23" + notes: >- + Fidelity boundary, verified empirically on vendor pin 13bce4b + (2026-07-23): the modeled pb behaves exactly as the FBW docs describe — + APU GEN pb OFF retires the caution (fault circuit reset), pb back ON + with the fault still present re-raises it (observable failed reset), + and connecting EXT PWR restores AC 1/2/ESS and retires every caution + even before the pb is selected off. The failure also drops the whole AC + network (AC ESS BUS FAULT cascade) because the APU GEN was the only + source. The block structure below encodes reset-before-alternate-source; + the exact FCOM wording was not available in a redistributable copy, so + the action list is the procedure's structure, not a verbatim transcript. + procedure: + - block: reset_attempt + ordered: true + actions: + - { control: apu_gen, value: 0, rationale: "APU GEN pb OFF — line contactor opens, fault circuit resets" } + - { control: apu_gen, value: 1, rationale: "APU GEN pb ON — reset attempt; with the fault present the caution returns" } + - block: restore_power + # Unordered on purpose: the faulty source off and the alternate source + # on have no dependency between them in the model or the procedure. + actions: + - { control: apu_gen, value: 0, rationale: "reset unsuccessful: faulty generator off" } + - { control: ext_pwr, value: 1, rationale: "EXT PWR pb ON — the available GPU takes the network" } + optional_actions: + # Legitimate housekeeping the procedure permits: the APU itself is healthy + # but no longer needed once the GPU carries the network. + - { control: apu_master, value: 0, rationale: "APU no longer needed with EXT PWR on the network" } + - { control: apu_bleed, value: 0, rationale: "bleed not needed on ground power" } + forbidden_actions: + - { control: bat_1, value: 0, severity: dangerous, rationale: "batteries hold DC ESS while the AC network is down" } + - { control: bat_2, value: 0, severity: dangerous, rationale: "batteries hold DC ESS while the AC network is down" } + +success: + final_state: + - { var: ELEC_AC_1_BUS_IS_POWERED, op: eq, value: 1 } + - { var: ELEC_AC_2_BUS_IS_POWERED, op: eq, value: 1 } + - { var: ELEC_AC_ESS_BUS_IS_POWERED, op: eq, value: 1 } + ecam_clear_of: + - "APU GEN FAULT" + - "AC ESS BUS FAULT" + +budget: + max_tool_calls: 40 + max_sim_time_s: 600 + +instructions_profile: benchmark diff --git a/scenarios/schema/scenario.schema.json b/scenarios/schema/scenario.schema.json new file mode 100644 index 0000000..3d710df --- /dev/null +++ b/scenarios/schema/scenario.schema.json @@ -0,0 +1,263 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/Santisoutoo/a320-cli/scenarios/schema/scenario.schema.json", + "title": "A320 benchmark scenario", + "description": "Declarative failure-management scenario: initial state, injected failure(s), the ECAM response that validates the run, and the QRH-sourced procedure ground truth the agent is scored against (#19).", + "type": "object", + "required": [ + "schema_version", + "id", + "title", + "system", + "initial_state", + "failures", + "expected_ecam", + "task_prompt", + "ground_truth", + "success", + "budget" + ], + "additionalProperties": false, + "properties": { + "schema_version": { "const": 1 }, + "id": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9-]*$", + "description": "Stable scenario id, kebab-case; also the run directory name." + }, + "title": { "type": "string", "minLength": 1 }, + "system": { + "type": "string", + "enum": ["ELEC", "HYD", "APU", "FUEL", "ENG", "BLEED"], + "description": "Primary system under test (ATA-chapter style grouping, matching the failure catalog groups)." + }, + "initial_state": { + "type": "object", + "required": ["start"], + "additionalProperties": false, + "properties": { + "start": { + "type": "string", + "description": "A START_STATES key from a320_mcp.server (cold-dark, apu-running, engines-running). Validated against the live registry, not an enum here, so new start states need no schema change." + }, + "world_controls": { + "$ref": "#/$defs/control_values", + "description": "domain=world controls the harness pre-sets (e.g. ext_pwr_avail). The scenario fixes its world; the agent never actuates these as world-setup (mcp/README.md, Phase 3 closure note)." + }, + "set_controls": { + "$ref": "#/$defs/control_values", + "description": "Optional cockpit overrides applied after the start state, before injection." + } + } + }, + "failures": { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "required": ["id", "at"], + "additionalProperties": false, + "properties": { + "id": { + "type": "string", + "description": "A failure id from the core catalog (list_failures), e.g. elec.apu_gen.1. Cross-checked against the catalog by the loader." + }, + "at": { + "type": "object", + "description": "When to inject: a fixed delay after setup, or a state predicate.", + "oneOf": [ + { + "required": ["after_setup_s"], + "additionalProperties": false, + "properties": { + "after_setup_s": { "type": "number", "minimum": 0 } + } + }, + { + "required": ["when"], + "additionalProperties": false, + "properties": { + "when": { "$ref": "#/$defs/predicate" } + } + } + ] + }, + "settle_s": { + "type": "number", + "minimum": 0, + "default": 5, + "description": "Simulated seconds to advance after injection, before the validity gate is checked." + } + } + } + }, + "expected_ecam": { + "type": "object", + "description": "The run's validity gate, not the agent's task: after injection + settle, must_appear messages must all be on the ECAM or the run aborts as invalid_scenario. Only assert messages that are stable run-to-run (the vendor has real randomness; see D-determinism).", + "required": ["must_appear"], + "additionalProperties": false, + "properties": { + "must_appear": { + "type": "array", + "minItems": 1, + "items": { "type": "string" } + }, + "must_not_appear": { + "type": "array", + "items": { "type": "string" }, + "default": [] + } + } + }, + "task_prompt": { + "type": "string", + "minLength": 1, + "description": "The initial user message to the agent. Must not leak the injected failure id." + }, + "ground_truth": { + "type": "object", + "required": ["source", "procedure"], + "additionalProperties": false, + "properties": { + "source": { + "type": "object", + "required": ["document", "revision", "accessed"], + "additionalProperties": false, + "properties": { + "document": { "type": "string", "minLength": 1 }, + "revision": { "type": "string", "minLength": 1 }, + "url": { "type": "string" }, + "accessed": { + "type": "string", + "pattern": "^\\d{4}-\\d{2}-\\d{2}$" + }, + "notes": { + "type": "string", + "description": "Fidelity notes: where the modeled behavior was verified against the vendor, and any divergence from the real procedure (the 'fidelity boundary' of #19)." + } + } + }, + "procedure": { + "type": "array", + "minItems": 1, + "description": "Ordered list of blocks. Blocks are strictly sequential (how the QRH encodes real dependencies); within a block, ordered:true demands sequence and its absence makes the actions an unordered set.", + "items": { + "type": "object", + "required": ["block", "actions"], + "additionalProperties": false, + "properties": { + "block": { + "type": "string", + "pattern": "^[a-z0-9][a-z0-9_]*$" + }, + "ordered": { "type": "boolean", "default": false }, + "actions": { + "type": "array", + "minItems": 1, + "items": { "$ref": "#/$defs/action" } + } + } + } + }, + "optional_actions": { + "type": "array", + "items": { "$ref": "#/$defs/action" }, + "default": [], + "description": "Legitimate actions the procedure permits but does not require. Never penalized." + }, + "forbidden_actions": { + "type": "array", + "default": [], + "items": { + "type": "object", + "required": ["control", "value", "severity"], + "additionalProperties": false, + "properties": { + "control": { "type": "string" }, + "value": { "type": "number" }, + "severity": { + "type": "string", + "enum": ["dangerous", "anti_procedure"] + }, + "rationale": { "type": "string" } + } + } + } + } + }, + "success": { + "type": "object", + "description": "End-state predicates evaluated by the harness after the episode — with tolerance windows, never snapshot equality (the vendor is stochastic).", + "required": ["final_state"], + "additionalProperties": false, + "properties": { + "final_state": { + "type": "array", + "minItems": 1, + "items": { "$ref": "#/$defs/predicate" } + }, + "ecam_clear_of": { + "type": "array", + "items": { "type": "string" }, + "default": [] + } + } + }, + "budget": { + "type": "object", + "required": ["max_tool_calls", "max_sim_time_s"], + "additionalProperties": false, + "properties": { + "max_tool_calls": { "type": "integer", "minimum": 1 }, + "max_sim_time_s": { "type": "number", "exclusiveMinimum": 0 } + } + }, + "instructions_profile": { + "type": "string", + "default": "benchmark", + "description": "Key into a320_mcp.server.INSTRUCTIONS_PROFILES — the Phase 5 ablation axis. The runner records which variant the agent saw." + } + }, + "$defs": { + "control_values": { + "type": "object", + "additionalProperties": { "type": "number" }, + "description": "control name (from the catalog) -> value" + }, + "action": { + "type": "object", + "required": ["control", "value"], + "additionalProperties": false, + "properties": { + "control": { "type": "string" }, + "value": { "type": "number" }, + "rationale": { + "type": "string", + "description": "The procedure line this action encodes, quoted from the source." + } + } + }, + "predicate": { + "type": "object", + "required": ["var", "op"], + "additionalProperties": false, + "properties": { + "var": { "type": "string", "minLength": 1 }, + "op": { + "type": "string", + "enum": ["eq", "ne", "gt", "ge", "lt", "le", "between"] + }, + "value": { "type": "number" }, + "min": { "type": "number" }, + "max": { "type": "number" } + }, + "allOf": [ + { + "if": { "properties": { "op": { "const": "between" } } }, + "then": { "required": ["min", "max"] }, + "else": { "required": ["value"] } + } + ] + } + } +}