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 "<root>"
+        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/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"] }
+        }
+      ]
+    }
+  }
+}