v0.5.0: Phase 5 benchmark infrastructure complete (serve, hardened loader, 0.5.0 versions)

bench/README.md

@@ -33,7 +33,16 @@ pip install -e bench/
   self-contained (scenario embedded, tool surface, vendor pin, versions) so
   the #20 scorer never re-simulates.
 - `a320_bench/providers/` — agent adapters; `scripted` needs no LLM and is
-  what CI runs (litellm adapter arrives in slice D, #71).
+  what CI runs; `litellm_adapter` (pinned litellm, `[providers]` extra) is
+  the canonical real-model path.
+- `a320_bench/serve.py` — `a320-bench serve --scenario X --result out.json`:
+  the same privileged setup + injection + validity gate, then the benchmark
+  tool surface on **stdio** for an external MCP client that owns its own
+  agent loop — above all `claude -p` on a Claude subscription. The harness's
+  verdict lands in `--result` on shutdown. Demo/dev path: the trajectory is
+  the client's transcript (`--output-format stream-json`), and the client's
+  system prompt is a confound for model-vs-model baselines — those go
+  through `a320-bench run`.
 
 ## Tests
 

bench/a320_bench/cli.py

@@ -63,23 +63,44 @@ def build_parser() -> argparse.ArgumentParser:
         default=None,
         help='JSON dict passed to litellm.completion verbatim, e.g. \'{"temperature": 0}\'',
     )
+
+    serve = sub.add_parser(
+        "serve",
+        help="serve one scenario over stdio (benchmark profile) for an external "
+        "MCP client, e.g. claude -p on a subscription",
+    )
+    serve.add_argument("--scenario", required=True, help="path to a scenario YAML")
+    serve.add_argument(
+        "--result",
+        default=None,
+        help="path to write the harness's success evaluation JSON on shutdown",
+    )
     return parser
 
 
 def main(argv: "list[str] | None" = None) -> int:
     args = build_parser().parse_args(argv)
 
-    # Imported here, not at module top: the CLI is the only piece that needs
-    # litellm, and the error message tells the user exactly what to install.
     try:
-        from a320_bench.providers.litellm_adapter import LiteLLMAdapter
-    except ImportError as exc:
+        scenario = load_scenario(args.scenario)
+    except ScenarioError as exc:
         print(f"a320-bench: {exc}", file=sys.stderr)
         return 2
 
+    if args.command == "serve":
+        from pathlib import Path
+
+        from a320_bench.serve import serve_scenario
+
+        return serve_scenario(
+            scenario, result_path=Path(args.result) if args.result else None
+        )
+
+    # Imported here, not at module top: `run` is the only piece that needs
+    # litellm, and the error message tells the user exactly what to install.
     try:
-        scenario = load_scenario(args.scenario)
-    except ScenarioError as exc:
+        from a320_bench.providers.litellm_adapter import LiteLLMAdapter
+    except ImportError as exc:
         print(f"a320-bench: {exc}", file=sys.stderr)
         return 2
 

bench/a320_bench/scenario.py

@@ -161,14 +161,26 @@ def evaluate_predicate(pred: Predicate, value: float) -> bool:
 
 
 # --- catalogs (cached: building a Sim costs ~1 s) ------------------------------
+@functools.lru_cache(maxsize=1)
+def _catalog_sim() -> "Any":
+    """A throwaway Sim used only as the catalog/validation oracle.
+
+    Never stepped, never observed: `set` may be called on it to let the core
+    validate a (control, value) pair — writes land in its variable store but
+    the aircraft is never advanced, so nothing accumulates.
+    """
+    import a320_sim
+
+    return a320_sim.Sim()
+
+
 @functools.lru_cache(maxsize=1)
 def _catalogs() -> tuple[dict[str, str], frozenset[str], frozenset[str], frozenset[str]]:
     """(control name -> domain, failure ids, START_STATES keys, instructions
     profiles), from the live core and the MCP server module."""
-    import a320_sim
     from a320_mcp.server import INSTRUCTIONS_PROFILES, START_STATES
 
-    sim = a320_sim.Sim()
+    sim = _catalog_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), frozenset(INSTRUCTIONS_PROFILES)
@@ -325,7 +337,14 @@ def _cross_check(scenario: Scenario, path: Path) -> None:
                 f"see list_failures())"
             )
 
-    def check_control(name: str, where: str, *, must_be_world: bool = False) -> None:
+    def check_control(
+        name: str,
+        where: str,
+        *,
+        must_be_world: bool = False,
+        cockpit_only: bool = False,
+        value: "float | None" = None,
+    ) -> None:
         if name not in domains:
             raise ScenarioError(
                 f"{path}: unknown control '{name}' in {where} (not in the core catalog; "
@@ -337,15 +356,33 @@ def check_control(name: str, where: str, *, must_be_world: bool = False) -> None
                 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")
+        if cockpit_only and domains[name] == "world":
+            raise ScenarioError(
+                f"{path}: control '{name}' in {where} has domain 'world' — world "
+                f"state belongs in world_controls, not among cockpit overrides"
+            )
+        if value is not None:
+            # The core is the oracle for valid values (same philosophy as
+            # D-017): its catalog strings ('0 (off) or 1 (on)', 'one of
+            # [...]') are for humans, `set` is the machine-checkable truth.
+            # The catalog Sim is never advanced, so the write is inert.
+            import a320_sim
+
+            try:
+                _catalog_sim().set(name, value)
+            except a320_sim.SimError as exc:
+                raise ScenarioError(f"{path}: in {where}: {exc}") from exc
+
+    for name, value in scenario.initial_state.world_controls.items():
+        check_control(name, "initial_state.world_controls", must_be_world=True, value=value)
+    for name, value in scenario.initial_state.set_controls.items():
+        check_control(name, "initial_state.set_controls", cockpit_only=True, value=value)
     for block in scenario.ground_truth.procedure:
         for action in block.actions:
-            check_control(action.control, f"procedure block '{block.block}'")
+            check_control(
+                action.control, f"procedure block '{block.block}'", value=action.value
+            )
     for action in scenario.ground_truth.optional_actions:
-        check_control(action.control, "optional_actions")
+        check_control(action.control, "optional_actions", value=action.value)
     for forbidden in scenario.ground_truth.forbidden_actions:
-        check_control(forbidden.control, "forbidden_actions")
+        check_control(forbidden.control, "forbidden_actions", value=forbidden.value)

bench/a320_bench/serve.py

@@ -0,0 +1,72 @@
+"""``a320-bench serve``: a scenario served over stdio, benchmark profile.
+
+The bridge for agents the runner cannot drive as a ProviderAdapter — above
+all **Claude Code on a subscription** (``claude -p`` is its own agent loop
+with its own MCP client, so it connects here instead of being called by
+``run_episode``). This does the privileged half of an episode in-process
+(setup, world controls, injection, validity gate), then serves the benchmark
+tool surface on stdio; on shutdown it writes the harness's own success
+evaluation to ``--result`` so the outer process can judge the run.
+
+    a320-bench serve --scenario scenarios/elec/apu_gen_fault.yaml \
+                     --result result.json
+
+    claude -p "<task prompt>" --mcp-config <cfg pointing at the line above> ...
+
+Trade-off, stated honestly: unlike ``a320-bench run``, the trajectory is NOT
+recorded here (the client owns the loop; use its transcript, e.g.
+``claude -p --output-format stream-json``), and the client's own system
+prompt is a confound for model-vs-model comparisons — this is the demo/dev
+path, not the canonical baseline path.
+
+stdout is the MCP transport: everything diagnostic goes to stderr.
+"""
+
+import atexit
+import json
+import sys
+from pathlib import Path
+
+import a320_sim
+from a320_mcp.server import INSTRUCTIONS_PROFILES, create_server
+
+from a320_bench.episode import _evaluate_success, _inject, _setup, _validity_gate
+from a320_bench.scenario import Scenario
+
+
+def serve_scenario(scenario: Scenario, *, result_path: "Path | None" = None) -> int:
+    """Set up, inject, gate — then serve on stdio until the client hangs up."""
+    sim = a320_sim.Sim()
+    print(f"serve: preparing '{scenario.id}' ({scenario.initial_state.start})...", file=sys.stderr)
+    _setup(sim, scenario)
+    error = _inject(sim, scenario)
+    gate = {"passed": False, "error": error} if error else _validity_gate(sim, scenario)
+    print(f"serve: validity gate: {gate}", file=sys.stderr)
+    if not gate["passed"]:
+        print("serve: invalid scenario, refusing to serve", file=sys.stderr)
+        return 1
+
+    if result_path is not None:
+
+        def dump_result() -> None:
+            result = {
+                "scenario": scenario.id,
+                "success_eval": _evaluate_success(sim, scenario),
+                "final_ecam": [w["message"] for w in sim.read_ecam()],
+                "active_failures": sim.active_failures(),
+                "sim_time": sim.sim_time(),
+            }
+            result_path.parent.mkdir(parents=True, exist_ok=True)
+            result_path.write_text(json.dumps(result, indent=2), encoding="utf-8")
+            print(f"serve: result written to {result_path}", file=sys.stderr)
+
+        atexit.register(dump_result)
+
+    server = create_server(
+        sim,
+        instructions=INSTRUCTIONS_PROFILES[scenario.instructions_profile],
+        profile="benchmark",
+    )
+    print(f"serve: benchmark profile on stdio (t={sim.sim_time():.1f}s)", file=sys.stderr)
+    server.run()
+    return 0

bench/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "a320-bench"
-version = "0.1.0"
+version = "0.5.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"

bench/tests/test_scenario_schema.py

@@ -160,6 +160,25 @@ def test_unknown_start_state_is_rejected():
     _expect_error(data, "hangar")
 
 
+def test_out_of_range_action_value_is_rejected_by_the_core_oracle():
+    """Control values are validated by the core's own `set` (D-017 spirit).
+
+    The catalog's `valid_values` strings are for humans; the machine truth is
+    whether the core accepts the write. apu_gen only takes 0/1 — a 7 in a
+    procedure block must fail at load time, not mid-episode.
+    """
+    data = _base()
+    data["ground_truth"]["procedure"][0]["actions"][0]["value"] = 7
+    _expect_error(data, "reset_attempt")
+
+
+def test_world_control_in_set_controls_is_rejected():
+    """set_controls is for cockpit overrides; world state goes in world_controls."""
+    data = _base()
+    data["initial_state"]["set_controls"] = {"ext_pwr_avail": 1}
+    _expect_error(data, "ext_pwr_avail", "world_controls")
+
+
 # --- predicates ------------------------------------------------------------------
 def test_predicate_evaluation():
     assert evaluate_predicate(Predicate(var="v", op="eq", value=1.0), 1.0)

bench/tests/test_serve.py

@@ -0,0 +1,88 @@
+"""``a320-bench serve`` over the real stdio protocol (#19).
+
+Same pattern as mcp/tests/test_server.py: spawn the command as a subprocess
+and talk to it with the SDK client — what is verified is what an external
+agent (claude -p on a subscription) would actually see.
+"""
+
+import asyncio
+import json
+import sys
+import tempfile
+from pathlib import Path
+
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+from a320_bench.scenario import REPO_ROOT
+
+SCENARIO = REPO_ROOT / "scenarios" / "elec" / "apu_gen_fault.yaml"
+
+
+def test_serve_hands_an_injected_aircraft_to_an_external_client():
+    """The client sees the benchmark surface and the failed aircraft; the
+    result JSON judges the run after the client hangs up."""
+    with tempfile.TemporaryDirectory() as tmp:
+        result_path = Path(tmp) / "result.json"
+        server = StdioServerParameters(
+            command=sys.executable,
+            args=[
+                "-m",
+                "a320_bench.cli",
+                "serve",
+                "--scenario",
+                str(SCENARIO),
+                "--result",
+                str(result_path),
+            ],
+            cwd=str(REPO_ROOT),
+        )
+
+        async def drive(session):
+            tools = {t.name for t in (await session.list_tools()).tools}
+            ecam = await session.call_tool("read_ecam", {})
+            # Manage the failure the blunt way: alternate source on.
+            await session.call_tool("set_control", {"control": "ext_pwr", "value": 1})
+            await session.call_tool("advance", {"seconds": 5})
+            done = await session.call_tool(
+                "report_done", {"diagnosis": "apu gen down", "actions_summary": "ext pwr on"}
+            )
+            return tools, ecam, done
+
+        async def check():
+            async with stdio_client(server) as (read, write):
+                async with ClientSession(read, write) as session:
+                    await session.initialize()
+                    return await drive(session)
+
+        tools, ecam, done = asyncio.run(check())
+
+        assert "report_done" in tools and "clear_failure" not in tools
+        messages = [w["message"] for w in ecam.structuredContent["result"]]
+        assert "APU GEN FAULT" in messages, messages
+        assert not done.isError
+
+        # The subprocess exited when the session closed; atexit wrote the verdict.
+        result = json.loads(result_path.read_text(encoding="utf-8"))
+        assert result["scenario"] == "elec-apu-gen-fault"
+        assert result["success_eval"]["all_passed"] is True
+        assert result["active_failures"] == ["elec.apu_gen.1"]
+
+
+def test_serve_parser_and_run_still_parse():
+    from a320_bench.cli import build_parser
+
+    serve = build_parser().parse_args(["serve", "--scenario", "s.yaml", "--result", "r.json"])
+    assert serve.command == "serve" and serve.result == "r.json"
+    run = build_parser().parse_args(["run", "--scenario", "s.yaml", "--model", "m"])
+    assert run.command == "run"
+
+
+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)} serve tests passed.")

mcp/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "a320-mcp"
-version = "0.1.0"
+version = "0.5.0"
 description = "MCP server exposing the headless A320 systems core to LLM agents"
 readme = "README.md"
 # El SDK de MCP exige >=3.10; el resto de paquetes del repo declaran >=3.9.