"""AICC conformance test runner. Loads declarative scenarios (see https://github.com/emil28092005/AICC-Protocol conformance/scenarios/) and executes them against a bridge, reporting pass/fail per scenario. Usage: from aicc.conformance import run_scenarios, summarize results = run_scenarios(bridge, "/path/to/scenarios") print(summarize(results)) """ from __future__ import annotations import asyncio import json import re import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Any from aicc.bridge import Bridge from aicc.transport.in_process import InProcessTransport UUID_RE = re.compile( r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$" ) @dataclass class ScenarioResult: id: str title: str passed: bool failures: list[str] = field(default_factory=list) error: str | None = None def _match_value(expect: Any, actual: Any, path: str, failures: list[str]) -> None: """Subset matcher with special `$` matchers for conformance assertions.""" if isinstance(expect, dict): # Special matchers if "$kind" in expect and len(expect) == 1: kind = expect["$kind"] ok = { "uuid": isinstance(actual, str) and bool(UUID_RE.match(actual)), "integer": isinstance(actual, int) and not isinstance(actual, bool), "number": isinstance(actual, (int, float)) and not isinstance(actual, bool), "string": isinstance(actual, str), "boolean": isinstance(actual, bool), "array": isinstance(actual, list), "object": isinstance(actual, dict), }.get(kind, False) if not ok: failures.append(f"{path}: expected {kind}, got {type(actual).__name__}") return if "$gt" in expect and len(expect) == 1: if not (isinstance(actual, (int, float)) and actual > expect["$gt"]): failures.append(f"{path}: expected > {expect['$gt']}, got {actual!r}") return if "$enum" in expect and len(expect) == 1: if actual not in expect["$enum"]: failures.append(f"{path}: expected one of {expect['$enum']}, got {actual!r}") return if "$required" in expect and len(expect) == 1: if not isinstance(actual, dict): failures.append(f"{path}: expected object, got {type(actual).__name__}") return for key in expect["$required"]: if key not in actual: failures.append(f"{path}: missing required key {key!r}") return if not isinstance(actual, dict): failures.append(f"{path}: expected object, got {type(actual).__name__}") return for key, sub in expect.items(): if key not in actual: failures.append(f"{path}: missing key {key!r}") continue _match_value(sub, actual[key], f"{path}.{key}", failures) return if isinstance(expect, list): if not isinstance(actual, list): failures.append(f"{path}: expected array, got {type(actual).__name__}") return if len(expect) != len(actual): failures.append(f"{path}: expected array len {len(expect)}, got {len(actual)}") return for i, (sub, act) in enumerate(zip(expect, actual)): _match_value(sub, act, f"{path}[{i}]", failures) return if expect != actual: failures.append(f"{path}: expected {expect!r}, got {actual!r}") def _validate_scenario(sc: dict[str, Any]) -> None: if "id" not in sc: raise ValueError("Scenario missing 'id'") if "steps" not in sc or not isinstance(sc["steps"], list): raise ValueError(f"Scenario {sc['id']} missing 'steps' list") for step in sc["steps"]: if "send" not in step and "expect" not in step: raise ValueError(f"Scenario {sc['id']} has step without send/expect") async def _run_scenario(bridge: Bridge, sc: dict[str, Any]) -> ScenarioResult: result = ScenarioResult(id=sc["id"], title=sc.get("title", sc["id"]), passed=True) try: _validate_scenario(sc) t = InProcessTransport.start(bridge) async with t: # Handshake: connect pushes session_init into the receive queue. manifest_raw = await asyncio.wait_for(t.receive(), timeout=10.0) if manifest_raw.get("type") != "session_init": result.passed = False result.error = ( f"expected session_init on connect, got {manifest_raw.get('type')!r}" ) return result session_id = manifest_raw["session_id"] last_call_id: str | None = None # Pending message already consumed from the wire (the handshake # session_init). Scenarios that expect session_init as their # first step consume it here instead of receiving again. pending: dict[str, Any] | None = manifest_raw for step in sc["steps"]: if "send" in step: msg = dict(step["send"]) msg.setdefault("session_id", session_id) msg.setdefault("message_id", str(uuid.uuid4())) if msg.get("type") == "tool_call": last_call_id = msg.get("call_id") or f"tc_{uuid.uuid4().hex[:12]}" msg["call_id"] = last_call_id await t.send(msg) if "expect" in step: expected = step["expect"] if pending is not None and expected.get("type") == "session_init": raw = pending pending = None else: raw = await asyncio.wait_for(t.receive(), timeout=10.0) failures: list[str] = [] if ( expected.get("call_id_echo") and last_call_id is not None and raw.get("call_id") != last_call_id ): failures.append( f"call_id_echo: expected {last_call_id!r}, " f"got {raw.get('call_id')!r}" ) if expected.get("session_id_echo") and raw.get("session_id") != session_id: failures.append( f"session_id_echo: expected {session_id!r}, " f"got {raw.get('session_id')!r}" ) remaining = { k: v for k, v in expected.items() if k not in ("call_id_echo", "session_id_echo") } _match_value(remaining, raw, "msg", failures) if failures: result.passed = False result.failures.extend(failures) return result except Exception as exc: # noqa: BLE001 result.passed = False result.error = f"{type(exc).__name__}: {exc}" return result def load_scenarios(scenario_dir: str | Path) -> list[dict[str, Any]]: """Load all *.json scenario files from a directory, sorted by id.""" d = Path(scenario_dir) files = sorted(d.glob("*.json")) scenarios = [] for f in files: with open(f) as fh: scenarios.append(json.load(fh)) return scenarios async def run_scenarios(bridge: Bridge, scenario_dir: str | Path) -> list[ScenarioResult]: """Run all scenarios in a directory against a bridge. One session per scenario.""" results = [] for sc in load_scenarios(scenario_dir): results.append(await _run_scenario(bridge, sc)) return results def summarize(results: list[ScenarioResult]) -> str: lines = [] passed = sum(1 for r in results if r.passed) for r in results: status = "PASS" if r.passed else "FAIL" lines.append(f"[{status}] {r.id}: {r.title}") for f in r.failures: lines.append(f" - {f}") if r.error: lines.append(f" - error: {r.error}") lines.append(f"\n{passed}/{len(results)} scenarios passed") return "\n".join(lines) def reference_bridge() -> Bridge: """Reference bridge exposing the tools the core scenarios expect. Tools required by scenarios: - echo: returns input value - boom: raises -> execution_failed - bump: returns ok and emits a collision event """ bridge = Bridge(name="conformance-reference", kind="3d", tick_rate_hz=10.0) @bridge.tool(description="Echo a value back to the caller.") async def echo(value: str = "") -> dict: return {"value": value} @bridge.tool(description="Always raises an exception.") async def boom() -> dict: raise RuntimeError("kaboom") @bridge.tool(description="Returns ok and emits a collision event.") async def bump() -> dict: return {"bumped": True} # Emit a collision event when bump completes. Hook into tool handling: # after a successful bump call, push the event into the session queue. _orig = bridge._handle_tool_call async def _handle_with_event(session_id: str, call): resp = await _orig(session_id, call) if call.tool == "bump" and resp.get("ok"): bridge.emit_event( session_id, "collision", {"other": "wall", "impulse": 1.0}, tick=1, ) return resp bridge._handle_tool_call = _handle_with_event # type: ignore[method-assign] return bridge def main() -> None: """CLI entry: python -m aicc.conformance """ import sys if len(sys.argv) < 2: print("usage: python -m aicc.conformance ") sys.exit(2) bridge = reference_bridge() results = asyncio.run(run_scenarios(bridge, sys.argv[1])) print(summarize(results)) sys.exit(0 if all(r.passed for r in results) else 1) if __name__ == "__main__": main()