diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/harness/sim_qemu_line.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/harness/sim_qemu_line.py | 1247 |
1 files changed, 1247 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/harness/sim_qemu_line.py b/debug/risc-v-gsp-rm-vm/harness/sim_qemu_line.py new file mode 100644 index 0000000..b47a028 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/harness/sim_qemu_line.py @@ -0,0 +1,1247 @@ +#!/usr/bin/env python3 +"""S4 evidence harness: is the native GSP simulator above the QEMU line? + +Runs one A/B pair on the real TU104 570.144 GSP image: + +* **QEMU baseline** - the v0.1/v0.2 probe path (flat relocated load, bounded + wall-clock run, ``guest_errors,unimp,int`` log): first exception, address and + ladder rung, recorded as a normal ``greeneye-session/1``. +* **Native baseline** - ``sim.gsp_profile`` (deterministic RV64IMAC, modeled + cycles): instructions, cycles, first modeled MMIO, stop reason and rung. + If S1's device model/profile is present (``sim/profiles/``, ``sim/devices/``, + or a Python ``sim.devices`` entry point) it is discovered defensively and + used as the MMIO map; otherwise the run is the lenient B0 baseline. + +The rung ladder is docs/35-GREENEYE-SIMULATOR-DESIGN.md section 3.3 +(``pre-B0`` < B0 < B1 < ... < B5). "Above the QEMU line" means the native +rung ordinal is at least one past QEMU's, twice identically, within the frozen +wall-clock threshold, and with a discovered S1 device model (docs/35 1.1); +requiring the device model means a lenient run reports ``above_lenient_only``, +not ``above``. The frozen record and the first measured results live in +``docs/37-SIM-QEMU-LINE.md``. + +The harness also exercises the native feature matrix (hooks/breakpoints/ +watchpoints, memory/struct, sessions, compare, trace, injection) on the bundled +demo firmware. MCP is deliberately ``not_run``: the native MCP surface is S3's +deliverable (see docs/37 section "Not verified"). + +Everything is stdlib-only, bounded (``<= 2M`` instructions per native run, +``<= 10 s`` of QEMU) and touches no hardware, GPU, guest, module or network. + +CLI:: + + python3 harness/sim_qemu_line.py + python3 harness/sim_qemu_line.py --no-qemu --instructions 50000 + python3 harness/sim_qemu_line.py --device-map sim/profiles/tu104-570.144.json +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import shutil +import subprocess +import sys +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.dirname(HERE) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from greeneye.analyze import compare_sessions, parse_qemu_faults # noqa: E402 +from greeneye.device_map import ( # noqa: E402 + QEMU_REFERENCE, + DeviceEntry, + DeviceMap, + firmware_sha256, + harvest_from_text, +) +from greeneye.errors import GreenEyeError # noqa: E402 +from harness.common import ( # noqa: E402 + load_target, + open_flat_session, + qemu_available, + read_log_window, +) +from sim.gsp_profile import DEFAULT_IMAGE, run_profile # noqa: E402 + +SCHEMA = "greeneye-sim-qemu-line/1" + +#: Boot-ladder rungs (docs/35 section 3.3), lowest first. ``pre-B0`` is the +#: QEMU fault position: image loaded, first MMIO observed *as a fault*. +RUNG_ORDER = ("pre-B0", "B0", "B1", "B2", "B3", "B4", "B5") +RUNG_INDEX = {name: index for index, name in enumerate(RUNG_ORDER)} +QEMU_RUNG = "pre-B0" +NATIVE_LENIENT_RUNG = "B0" + +#: Default frozen thresholds (recorded 2026-10-03, before the first +#: measurement). The measured run and rationale live in +#: docs/37-SIM-QEMU-LINE.md section 3; the CLI uses these values when no +#: override is passed. Never move silently. +FROZEN_THRESHOLDS = { + "frozen_on": "2026-10-03", + "native_rungs_above_qemu": 1, + "identical_runs": 2, + "native_pair_wall_s": 20.0, + "qemu_first_fault_fw": QEMU_REFERENCE["firmware_vaddr"], + "qemu_first_touch": QEMU_REFERENCE["access_addr"], + # docs/35 1.1 requires the milestone to be reached "with the device model + # and shims documented": lenient MMIO is observation, not a device model, + # so it cannot satisfy the line contract on its own. + "device_model_required": True, +} + +#: Bounds (eng-ops / task discipline). +MAX_INSTRUCTIONS = 2_000_000 +MAX_QEMU_DURATION_S = 10.0 +DEFAULT_INSTRUCTIONS = 200_000 +DEFAULT_QEMU_DURATION_S = 0.5 + +#: Default report directory (new, S4-owned). +DEFAULT_OUT = os.path.join(HERE, "qemu_line") + +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +DEMO_SCHEMA = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "gsp_ctrl.schema.json") +SDK_ELF = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf") + +#: S1 device-model discovery: JSON conventions first, then a duck-typed Python +#: API. See docs/37 section "Re-running when S1 lands". +S1_JSON_DIRS = ("sim/profiles", "sim/devices") +S1_PY_API_NAMES = ( + "device_map_for_firmware", + "profile_for_firmware", + "load_device_map", + "load_profile", + "profile_for", +) +_RUNG_KEYS = ("rung", "highest_rung", "boot_ladder_rung", "milestone") + + +class HarnessError(RuntimeError): + """A harness precondition failed in a way the caller should report.""" + + +# -- small helpers --------------------------------------------------------- + + +def _relpath(path: str | None) -> str | None: + if path is None: + return None + return os.path.relpath(os.path.abspath(path), TOOL_ROOT) + + +def _rm_tree(path: str) -> None: + shutil.rmtree(path, ignore_errors=True) + + +def _write_text(path: str, text: str) -> str: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as fh: + fh.write(text) + return path + + +def _write_json(path: str, payload: dict) -> str: + os.makedirs(os.path.dirname(path), exist_ok=True) + tmp = f"{path}.tmp-{os.getpid()}" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(payload, fh, indent=2, sort_keys=True) + fh.write("\n") + os.replace(tmp, path) + return path + + +def session_hashes(session_dir: str | None) -> dict: + """sha256 of a session's core evidence files ({} when the dir is absent).""" + if not session_dir or not os.path.isdir(session_dir): + return {} + hashes: dict = {} + for name in ("session.json", "events.jsonl", "summary.md"): + path = os.path.join(session_dir, name) + if not os.path.isfile(path): + continue + digest = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + digest.update(chunk) + hashes[name] = digest.hexdigest() + return hashes + + +def _x86_vms_running() -> int | None: + """Count other agents' x86 QEMU VMs (never disturbed; recorded only).""" + try: + proc = subprocess.run( + ["pgrep", "-f", "qemu-system-x86_64"], + capture_output=True, + text=True, + timeout=5.0, + ) + except (OSError, subprocess.SubprocessError): + return None + if proc.returncode not in (0, 1): + return None + pids = [line for line in proc.stdout.splitlines() if line.strip().isdigit()] + return len(pids) + + +def rung_ordinal(rung: str | None) -> int | None: + if rung in RUNG_INDEX: + return RUNG_INDEX[rung] + return None + + +def above_qemu_line(qemu_rung: str | None, native_rung: str | None, minimum: int = 1) -> bool: + """True when ``native_rung`` is at least ``minimum`` rungs past QEMU's.""" + q = rung_ordinal(qemu_rung) + n = rung_ordinal(native_rung) + if q is None or n is None: + return False + return n >= q + minimum + + +def _summary_fault(fault: dict, bias: int) -> dict: + access = fault.get("access") or {} + epc = int(fault.get("epc", 0) or 0) + return { + "desc": fault.get("desc"), + "cause": fault.get("cause"), + "epc": epc, + "epc_fw": (epc - bias) if (bias and epc and epc >= bias) else epc, + "access_addr": access.get("access_addr"), + "size": access.get("size"), + "rw": access.get("rw"), + "reason": access.get("reason"), + } + + +# -- QEMU baseline --------------------------------------------------------- + + +def run_qemu_baseline( + image: str = DEFAULT_IMAGE, + out_dir: str = DEFAULT_OUT, + duration: float = DEFAULT_QEMU_DURATION_S, +) -> dict: + """Probe the real image under QEMU; skip cleanly when unavailable/absent.""" + result: dict = { + "backend": "qemu", + "ran": False, + "reason": None, + "rung": None, + "session": None, + "wall_s": 0.0, + } + if duration <= 0 or duration > MAX_QEMU_DURATION_S: + raise HarnessError(f"qemu duration must be in (0, {MAX_QEMU_DURATION_S}], got {duration}") + if not os.path.isfile(image): + result["reason"] = f"firmware absent: {image}" + return result + if not qemu_available(): + result["reason"] = "qemu-system-riscv64 not installed" + return result + + start = time.monotonic() + image_obj = load_target(image) + session_dir = os.path.join(out_dir, "sessions", "qemu-probe") + _rm_tree(session_dir) + with open_flat_session(image_obj, image, record=session_dir, name="s4-qemu-probe") as flat: + flat.session.run(duration=duration) + pc = flat.session.pc + bias = flat.bias + # Read the log window once: `-d int` trap loops flood the file, so the + # first exception has to be parsed from the bounded head window. + log_text, capped = read_log_window(flat.target.qemu_log) + faults = parse_qemu_faults(log_text) + first = faults[0] if faults else None + device_map = harvest_from_text(log_text, bias=bias) + meta = { + "s4_qemu_rung": QEMU_RUNG, + "qemu_fault_count_window": len(faults), + "qemu_log_window_capped": capped, + "qemu_first_fault": first, + } + recorder = getattr(flat.session, "recorder", None) + if recorder is not None and hasattr(recorder, "meta"): + recorder.meta.update(meta) + if log_text and hasattr(recorder, "blob"): + recorder.meta["qemu_log_blob"] = recorder.blob( + "qemu.log", log_text.encode("utf-8") + ) + result.update( + { + "ran": True, + "reason": "ok", + "rung": QEMU_RUNG, + "session": _relpath(session_dir), + "session_hashes": session_hashes(session_dir), + "stopped_pc": pc, + "fault_count_window": len(faults), + "log_window_capped": capped, + "first_fault": _summary_fault(first, bias) if first else None, + "first_touch_match": bool( + first + and (first.get("access") or {}).get("access_addr") == QEMU_REFERENCE["access_addr"] + ), + "device_map_entries": len(device_map), + "device_map": device_map, + } + ) + result["wall_s"] = round(time.monotonic() - start, 4) + return result + + +# -- S1 integration (defensive) ------------------------------------------- + + +def _coerce_device_map(obj) -> DeviceMap | None: + """Accept DeviceMap objects, paths, or device-map/profile dicts; else None.""" + if isinstance(obj, DeviceMap): + return obj + if isinstance(obj, (str, os.PathLike)): + path = os.fspath(obj) + if os.path.isfile(path): + try: + return DeviceMap.load(path) + except GreenEyeError: + return None + return None + if isinstance(obj, dict): + if obj.get("schema") == "greeneye-device-map/1": + try: + return DeviceMap.from_dict(obj) + except (KeyError, TypeError, ValueError, GreenEyeError): + return None + if "device_map" in obj: + return _coerce_device_map(obj["device_map"]) + entries = obj.get("entries") + if isinstance(entries, list): + dm = DeviceMap(meta=dict(obj.get("meta") or {})) + try: + for raw in entries: + dm.add(DeviceEntry.from_dict(raw)) + except (KeyError, TypeError, ValueError): + return None + return dm + return None + + +def _json_rung(data) -> str | None: + if not isinstance(data, dict): + return None + for key in _RUNG_KEYS: + value = data.get(key) + if isinstance(value, str) and value in RUNG_INDEX: + return value + if isinstance(value, dict): + inner = value.get("highest") or value.get("rung") + if isinstance(inner, str) and inner in RUNG_INDEX: + return inner + return None + + +def discover_s1_device_model(image: str, sha: str | None = None) -> dict: + """Find S1's device model/profile without assuming it exists. + + Returns ``{"device_map": DeviceMap|None, "profile": path|None, + "declared_rung": str|None, "source": str|None, "notes": [...]}``. + Discovery order: an explicit ``sim.devices`` Python entry point, then JSON + files under ``sim/profiles/`` and ``sim/devices/`` (a + ``greeneye-sim-profile/1`` profile, a ``greeneye-device-map/1`` file, a + profile with a ``device_map`` key, or a raw ``entries`` list). A profile + is the semantic device model (``sim.gsp_boot`` runs it); a device map only + replays observed accesses. A candidate is preferred when its firmware + hash matches the image; otherwise a single candidate is used and noted as + unverified pairing. + """ + notes: list[str] = [] + declared_rung: str | None = None + + try: + import sim.devices as devices # type: ignore + + for name in S1_PY_API_NAMES: + fn = getattr(devices, name, None) + if not callable(fn): + continue + for args in ((sha,), (image,), ()): + if sha is None and args == (sha,): + continue + try: + candidate = fn(*args) + except TypeError: + continue + except (OSError, UnicodeError): + # A path-taking helper handed a digest (or a helper handed + # the binary firmware) simply does not apply; not an error. + break + except Exception as exc: # discovery must never break the run + notes.append(f"sim.devices.{name}{args!r} failed: {exc}") + break + if candidate is None: + continue + dm = _coerce_device_map(candidate) + rung = getattr(candidate, "rung", None) + if isinstance(rung, str) and rung in RUNG_INDEX: + declared_rung = rung + if dm is not None: + notes.append(f"device map via sim.devices.{name}{args!r}") + return { + "device_map": dm, + "profile": None, + "declared_rung": declared_rung, + "source": f"sim.devices.{name}", + "notes": notes, + } + except ImportError: + notes.append("sim.devices package absent") + except Exception as exc: + notes.append(f"sim.devices import failed: {exc}") + + candidates: list[tuple[str, DeviceMap, str | None, str | None]] = [] + profiles: list[tuple[str, str, str | None]] = [] + for rel in S1_JSON_DIRS: + root = os.path.join(TOOL_ROOT, rel) + if not os.path.isdir(root): + notes.append(f"{rel}/ absent") + continue + for name in sorted(os.listdir(root)): + if not name.endswith(".json"): + continue + path = os.path.join(root, name) + try: + with open(path, "r", encoding="utf-8") as fh: + data = json.load(fh) + except (OSError, ValueError) as exc: + notes.append(f"{rel}/{name}: unreadable ({exc})") + continue + if isinstance(data, dict) and data.get("schema") == "greeneye-sim-profile/1": + firmware = data.get("firmware") or {} + profiles.append((f"{rel}/{name}", path, + str(firmware.get("sha256") or "") or None)) + continue + dm = _coerce_device_map(data) + rung = _json_rung(data) + if rung and declared_rung is None: + declared_rung = rung + if dm is not None: + candidates.append((f"{rel}/{name}", dm, rung, None)) + + exact_profiles = [p for p in profiles if sha and p[2] == sha] + if exact_profiles: + rel, path, _digest = exact_profiles[0] + notes.append(f"using {rel} (greeneye-sim-profile/1, firmware_sha256 matched)") + return { + "device_map": None, + "profile": path, + "declared_rung": declared_rung, + "source": rel, + "notes": notes, + } + + if not candidates: + if len(profiles) == 1: + rel, path, _digest = profiles[0] + notes.append(f"using {rel} (single profile, pairing unverified)") + return { + "device_map": None, + "profile": path, + "declared_rung": declared_rung, + "source": rel, + "notes": notes, + } + notes.append("no S1 device-map/profile JSON found") + return {"device_map": None, "profile": None, "declared_rung": declared_rung, + "source": None, "notes": notes} + + exact = [ + c for c in candidates + if sha and str((c[1].meta or {}).get("firmware_sha256", "")) == sha + ] + chosen = exact[0] if exact else candidates[0] + note = "firmware_sha256 matched" if exact else "single candidate, pairing unverified" + notes.append(f"using {chosen[0]} ({note})") + return { + "device_map": chosen[1], + "profile": None, + "declared_rung": chosen[2] or declared_rung, + "source": chosen[0], + "notes": notes, + } + + +# -- native baseline ------------------------------------------------------- + + +def estimate_native_rung(run: dict, *, device_model_used: bool, + declared_rung: str | None = None) -> tuple[str, str]: + """Conservative rung estimate from a bounded native run. + + * matching lenient first touch -> B0 (observed, no device semantics); + * a device model that serves the QEMU first-touch and lets execution + continue -> B1 (frontier moved past the QEMU fault); + * a higher rung only when S1 declares it in profile data (honoured as + ``rung_source: declared``). + + B2+ need boot-stage milestones, not just progress; absence of a fault is + not evidence of a boot stage. See docs/37 sections 3-4. + """ + stop = run.get("stop") or {} + stop_reason = stop.get("reason") + first = run.get("first_unknown_mmio") + reached_b0 = bool( + (first and first.get("addr") == QEMU_REFERENCE["access_addr"]) + or run.get("instructions", 0) > 0 + ) + + inferred = None + basis = None + if device_model_used and reached_b0: + unmodelled_at_qemu = bool(first and first.get("addr") == QEMU_REFERENCE["access_addr"]) + if not unmodelled_at_qemu: + inferred = "B1" + basis = ( + "device model served the QEMU first-touch and execution continued " + "past the QEMU fault to the run bound" + ) + else: + inferred = "B0" + basis = "device model present but the QEMU first-touch was still served leniently" + elif reached_b0: + inferred = NATIVE_LENIENT_RUNG + basis = "lenient first-touch observation; execution continued past the QEMU fault" + + if inferred is None: + return "pre-B0", "no native progress observed" + + rung, source = inferred, "inferred" + declared_index = rung_ordinal(declared_rung) + if declared_index is not None and declared_index > RUNG_INDEX[inferred]: + rung, source = declared_rung, "declared" + basis = f"S1 profile declares {declared_rung}; inference was {inferred} ({basis})" + return rung, f"{basis} [rung_source={source}, stop={stop_reason}]" + + +def run_native_baseline( + image: str = DEFAULT_IMAGE, + out_dir: str = DEFAULT_OUT, + instructions: int = DEFAULT_INSTRUCTIONS, + repeat: int = 2, + device_map=None, + declared_rung: str | None = None, + profile: str | None = None, + use_shim: bool = True, +) -> dict: + """Run the bounded deterministic native baseline ``repeat`` times. + + ``profile`` selects the S1 semantic device model (``sim.gsp_boot`` with + the versioned profile and the recorded boot shim); without it the run is + the lenient ``sim.gsp_profile`` baseline. ``device_map`` remains a + legacy/observational MMIO map for the lenient path. + """ + if instructions <= 0 or instructions > MAX_INSTRUCTIONS: + raise HarnessError(f"instructions must be in (0, {MAX_INSTRUCTIONS}], got {instructions}") + if repeat < 1: + raise HarnessError(f"repeat must be >= 1, got {repeat}") + if not os.path.isfile(image): + return { + "backend": "native", + "ran": False, + "reason": f"firmware absent: {image}", + "runs": [], + } + + s1_model = bool(profile) or device_map is not None + runs: list[dict] = [] + start_all = time.monotonic() + for index in range(repeat): + session_dir = os.path.join(out_dir, "sessions", f"native-{index + 1}") + _rm_tree(session_dir) + start = time.monotonic() + if profile: + from sim.gsp_boot import run_boot # deferred: only the S1 path needs it + + boot = run_boot( + image=image, + profile_path=profile, + instructions=instructions, + record=session_dir, + name=f"s4-native-{index + 1}", + use_shim=use_shim, + ) + wall = time.monotonic() - start + if not boot.ran: + raise HarnessError(f"native boot refused to run: {boot.reason}") + run = { + "instructions": boot.instructions, + "cycles": boot.cycles, + "cycle_model": boot.cycle_model, + "stop": { + "reason": boot.stop_reason, + "pc": boot.stop_pc, + "pc_fw": boot.stop_pc_fw, + }, + "first_unknown_mmio": boot.first_unknown_mmio, + "mmio_unknown": boot.mmio_unknown, + "device_unhandled": boot.device_unhandled, + "device_accesses": boot.device_accesses, + "device_state_sha256": boot.device_state_sha256, + "milestones": list(boot.milestones), + "session": _relpath(session_dir), + "session_hashes": session_hashes(session_dir), + "wall_s": round(wall, 4), + } + run["rung"] = boot.rung + run["rung_basis"] = ( + "observed S1 boot-ladder rung from sim.gsp_boot " + f"(milestones={list(boot.milestones)}, mmio_unknown={boot.mmio_unknown}, " + f"device_unhandled={boot.device_unhandled})" + ) + else: + result = run_profile( + image=image, + instructions=instructions, + record=session_dir, + device_map=device_map, + name=f"s4-native-{index + 1}", + ) + wall = time.monotonic() - start + if not result.ran: + raise HarnessError(f"native profile refused to run: {result.reason}") + run = { + "instructions": result.instructions, + "cycles": result.cycles, + "cycle_model": result.cycle_model, + "stop": { + "reason": result.stop_reason, + "pc": result.stop_pc, + "pc_fw": result.stop_pc_fw, + }, + "first_unknown_mmio": result.first_unknown_mmio, + "mmio_unknown": result.mmio_unknown, + "session": _relpath(session_dir), + "session_hashes": session_hashes(session_dir), + "wall_s": round(wall, 4), + } + run["rung"], run["rung_basis"] = estimate_native_rung( + run, device_model_used=s1_model, declared_rung=declared_rung + ) + runs.append(run) + + keys = ( + "instructions", + "cycles", + "stop", + "first_unknown_mmio", + "mmio_unknown", + "rung", + "device_state_sha256", + "milestones", + ) + identical = all( + all(a.get(key) == runs[0].get(key) for key in keys) for a in runs[1:] + ) + compare = {"ran": False} + if repeat >= 2: + try: + text = compare_sessions( + os.path.join(out_dir, "sessions", "native-1"), + os.path.join(out_dir, "sessions", "native-2"), + ) + compare = { + "ran": True, + "chars": len(text), + "headers": [ + line[3:].strip() + for line in text.splitlines() + if line.startswith("## ") + ], + } + except Exception as exc: # compare is evidence, never a fatal step + compare = {"ran": False, "error": f"{type(exc).__name__}: {exc}"} + + return { + "backend": "native", + "ran": True, + "reason": "ok", + "instructions_budget": instructions, + "repeat": repeat, + "identical": identical, + "rung": runs[0]["rung"], + "rung_basis": runs[0]["rung_basis"], + "runs": runs, + "pair_wall_s": round(time.monotonic() - start_all, 4), + "compare": compare, + "device_model_used": s1_model, + "profile": _relpath(profile), + } + + +# -- native feature matrix ------------------------------------------------- + + +def _events_summary(session_dir: str) -> dict: + counts: dict[str, int] = {} + path = os.path.join(session_dir, "events.jsonl") + try: + with open(path, "r", encoding="utf-8") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + kind = json.loads(line).get("type") + except ValueError: + continue + if kind: + counts[kind] = counts.get(kind, 0) + 1 + except OSError: + pass + return dict(sorted(counts.items())) + + +def run_feature_matrix(out_dir: str = DEFAULT_OUT) -> dict: + """Exercise the native debugger feature set on the bundled demo firmware.""" + import greeneye as ge + + matrix: dict = {} + if not os.path.isfile(DEMO_ELF): + return { + "ran": False, + "reason": f"demo firmware absent: {DEMO_ELF} (run make demo)", + "features": {}, + } + + session_dir = os.path.join(out_dir, "sessions", "features") + _rm_tree(session_dir) + session = ge.launch(kernel=DEMO_ELF, backend="native", record=session_dir, name="s4-features") + try: + ctrl = session.symbols.lookup("gsp_ctrl") + session.run(duration=0.2) + magic = session.memory.read_u32(ctrl) + matrix["memory"] = { + "status": "pass" if magic == 0x47535031 else "fail", + "detail": {"ctrl": ctrl, "magic": magic}, + } + + with open(DEMO_SCHEMA, "r", encoding="utf-8") as fh: + schema = ge.schema_from_json(json.load(fh)) + decoded = schema.decode(session.memory.read(ctrl, schema.size), base=0) + matrix["struct"] = { + "status": "pass" if decoded.get("magic") == 0x47535031 else "fail", + "detail": {"schema": os.path.basename(DEMO_SCHEMA), "fields": len(decoded)}, + } + + dispatches: list[int] = [] + stores: list[int] = [] + session.hook("rpc_dispatch", on_enter=lambda hit: dispatches.append(hit.arg(0))) + session.watch(ctrl + 0x40, size=8, kind="w", callback=lambda hit: stores.append(hit.pc)) + session.memory.write_u64(ctrl + 0x28, 0x1234) + session.memory.write_u64(ctrl + 0x20, 1) + session.run(duration=0.25) + matrix["hooks"] = { + "status": "pass" if dispatches else "fail", + "detail": {"rpc_dispatch_hits": len(dispatches), "cmds": dispatches}, + } + matrix["watch"] = { + "status": "pass" if stores else "fail", + "detail": {"result_field_stores": len(stores), "pcs": stores[:4]}, + } + + session.memory.write_u64(ctrl + 0x28, 0x55) + session.memory.write_u64(ctrl + 0x20, 2) + hit = session.run_until("rpc_dispatch", timeout=1.0) + bp = session.breakpoint("rpc_dispatch") + attached = bool(bp.attached) + session.remove_breakpoint(bp) + matrix["breakpoints"] = { + "status": "pass" if hit is not None and attached and not bp.attached else "fail", + "detail": { + "run_until_hit": hit.addr if hit is not None else None, + "attached_then_removed": attached and not bp.attached, + "kind": "sw", + }, + } + finally: + session.close() + + events = _events_summary(session_dir) + matrix["sessions"] = { + "status": "pass" + if os.path.isfile(os.path.join(session_dir, "session.json")) + and os.path.isfile(os.path.join(session_dir, "events.jsonl")) + else "fail", + "detail": { + "dir": _relpath(session_dir), + "event_counts": events, + "hashes": session_hashes(session_dir), + }, + } + + trace_dir = os.path.join(out_dir, "sessions", "features-trace") + _rm_tree(trace_dir) + trace_session = ge.launch( + kernel=DEMO_ELF, + backend="native", + record=trace_dir, + name="s4-features-trace", + trace={"exec": True, "max_events": 4000}, + ) + try: + trace_session.run(duration=0.1) + finally: + trace_session.close() + trace_path = os.path.join(trace_dir, "blobs", "trace.jsonl") + trace_lines = 0 + if os.path.isfile(trace_path): + with open(trace_path, "r", encoding="utf-8", errors="replace") as fh: + trace_lines = sum(1 for _ in fh) + matrix["trace"] = { + "status": "pass" if trace_lines > 0 else "fail", + "detail": { + "dir": _relpath(trace_dir), + "blob": _relpath(trace_path), + "events": trace_lines, + "max_events": 4000, + "hashes": session_hashes(trace_dir), + }, + } + + if os.path.isfile(SDK_ELF): + from greeneye.inject import invoke_remote, load_service + + inject_dir = os.path.join(out_dir, "sessions", "features-inject") + _rm_tree(inject_dir) + inject_session = ge.launch( + kernel=DEMO_ELF, backend="native", record=inject_dir, name="s4-features-inject" + ) + try: + inject_session.run(duration=0.25) + service = load_service(inject_session, SDK_ELF, slot=6) + value = invoke_remote(inject_session, service, 1, 0x41) + finally: + inject_session.close() + matrix["injection"] = { + "status": "pass" if value == (0x41 ^ 0xBEEF) else "fail", + "detail": { + "dir": _relpath(inject_dir), + "blob": _relpath(SDK_ELF), + "slot": 6, + "ping_result": value, + "expected": 0x41 ^ 0xBEEF, + "hashes": session_hashes(inject_dir), + }, + } + else: + matrix["injection"] = { + "status": "not_run", + "reason": "sdk/build/injected_demo.elf not built (run make sdk)", + } + + matrix["mcp"] = { + "status": "not_run", + "reason": ( + "native MCP surface is S3's deliverable (plugins/mcp_server.py); " + "S4 must not race that work. Re-run after S3 lands (docs/37)." + ), + } + analyze_dir = os.path.join(out_dir, "sessions", "native-1") + compare_dir = os.path.join(out_dir, "sessions", "native-2") + if os.path.isfile(os.path.join(analyze_dir, "session.json")) and os.path.isfile( + os.path.join(compare_dir, "session.json") + ): + try: + from greeneye.analyze import render_summary + + summary = render_summary(analyze_dir) + delta_text = compare_sessions(analyze_dir, compare_dir) + matrix["analyze_compare"] = { + "status": "pass" if summary and delta_text else "fail", + "detail": { + "analyze_session": _relpath(analyze_dir), + "analyze_chars": len(summary), + "compare_pair": [_relpath(analyze_dir), _relpath(compare_dir)], + "compare_chars": len(delta_text), + "compare_headers": [ + line[3:].strip() + for line in delta_text.splitlines() + if line.startswith("## ") + ], + }, + } + except Exception as exc: # evidence step, never fatal to the matrix + matrix["analyze_compare"] = { + "status": "fail", + "reason": f"{type(exc).__name__}: {exc}", + } + else: + matrix["analyze_compare"] = { + "status": "not_run", + "reason": "native-1/native-2 sessions absent", + } + return {"ran": True, "reason": "ok", "features": matrix} + + +# -- report ---------------------------------------------------------------- + + +def threshold_checks(report: dict, thresholds: dict | None = None) -> dict: + """Evaluate the frozen thresholds against a report; returns check -> bool.""" + th = dict(FROZEN_THRESHOLDS) + th.update(thresholds or {}) + qemu = report.get("qemu") or {} + native = report.get("native") or {} + checks: dict = { + "qemu_first_fault_reproduced": bool( + qemu.get("ran") and qemu.get("first_touch_match") + ), + "native_above_qemu": above_qemu_line( + qemu.get("rung"), native.get("rung"), int(th["native_rungs_above_qemu"]) + ), + "identical_runs": bool( + native.get("ran") + and native.get("repeat", 0) >= int(th["identical_runs"]) + and native.get("identical") + ), + "native_pair_wall_s": bool( + native.get("ran") + and float(native.get("pair_wall_s", 1e9)) <= float(th["native_pair_wall_s"]) + ), + "device_model_documented": bool( + not th.get("device_model_required", True) or native.get("device_model_used") + ), + } + checks["all"] = all(checks.values()) + return checks + + +def build_report( + *, + image: str, + qemu: dict, + native: dict, + features: dict | None, + s1: dict, + thresholds: dict, +) -> dict: + checks = threshold_checks({"qemu": qemu, "native": native}, thresholds) + rung_progress = bool(checks["native_above_qemu"] and checks["identical_runs"]) + if not qemu.get("ran"): + verdict = "unverified_qemu_absent" + elif not checks["qemu_first_fault_reproduced"]: + verdict = "unverified_qemu_mismatch" + elif checks["all"]: + verdict = "above" + elif rung_progress and not checks["device_model_documented"]: + # Rung progress exists, but docs/35 1.1 also requires working device + # semantics; lenient MMIO cannot satisfy the line contract. Underclaim. + verdict = "above_lenient_only" + else: + verdict = "not_above" + return { + "schema": SCHEMA, + "harness": "harness/sim_qemu_line.py", + "line_definition": "docs/35-GREENEYE-SIMULATOR-DESIGN.md 1.1 + 3.3", + "firmware": { + "name": os.path.basename(image), + "sha256": firmware_sha256(image) if os.path.isfile(image) else None, + }, + "rung_order": list(RUNG_ORDER), + "thresholds": thresholds, + "threshold_checks": checks, + "verdict": verdict, + "verdict_explanation": { + "above": "all frozen checks passed: QEMU fault reproduced, native >= QEMU + 1 rung, identical runs, wall bound, documented device model", + "above_lenient_only": "native progressed past the QEMU fault, but lenient MMIO is not a documented device model (docs/35 1.1) - contract NOT satisfied", + "not_above": "one or more frozen checks failed and no rung-progress excuse applies", + "unverified_qemu_absent": "QEMU baseline unavailable; no A/B comparison is possible", + "unverified_qemu_mismatch": "QEMU ran but did not reproduce its documented first fault; the rungs are not comparable", + }.get(verdict, ""), + "rung_progress": rung_progress, + "above_qemu_line": verdict == "above", + "qemu": {k: v for k, v in qemu.items() if k != "device_map"}, + "native": native, + "features": features, + "s1_integration": { + "source": s1.get("source"), + "profile": _relpath(s1.get("profile")), + "declared_rung": s1.get("declared_rung"), + "notes": s1.get("notes", []), + }, + "x86_vms_running": _x86_vms_running(), + "not_verified": [ + "lenient MMIO (B0) is observation, not device semantics", + "B3/B4 are not reached: the S1 device model stops at the B2 DMEM-runspace unknown", + "the boot shim values (bootargs/WPR meta/DMEM mirror) are declared assumptions", + "MCP on native is S3-owned and not exercised here", + "L2 timing: cycles are modeled (docs/32), never silicon claims", + "no hardware/GPU/guest/module involvement by design", + ], + "rerun_when_s1_lands": [ + "python3 harness/sim_qemu_line.py", + "S1 landed (2026-10-03): sim/profiles/*.json is discovered and the native side runs sim.gsp_boot", + "sim/devices/*.json with greeneye-device-map/1 (or a device_map key) is still picked up as an observational map", + ], + } + + +def render_markdown(report: dict) -> str: + qemu = report.get("qemu") or {} + native = report.get("native") or {} + lines = [ + "# Simulator vs the QEMU line - measured report", + "", + f"* schema: `{report.get('schema')}`", + f"* harness: `{report.get('harness')}`", + f"* line: {report.get('line_definition')}", + f"* firmware: `{(report.get('firmware') or {}).get('name')}` " + f"sha256 `{((report.get('firmware') or {}).get('sha256') or '')[:16]}...`", + f"* verdict: **{report.get('verdict')}** " + f"(above_qemu_line={report.get('above_qemu_line')})", + f"* verdict meaning: {report.get('verdict_explanation') or '-'}", + "", + "## Rung table", + "", + "| side | ran | rung | instructions | cycles (modeled) | stop | wall s | evidence |", + "|---|---|---|---:|---:|---|---:|---|", + ] + runs = native.get("runs") or [] + lines.append( + f"| QEMU probe | {qemu.get('ran')} | {qemu.get('rung')} | - | - | " + f"{'fault' if qemu.get('first_fault') else '-'} | {qemu.get('wall_s')} | " + f"`{qemu.get('session') or qemu.get('reason')}` |" + ) + for index, run in enumerate(runs, start=1): + stop = run.get("stop") or {} + pc_fw = stop.get("pc_fw") + pc_fw_text = f"{pc_fw:#x}" if pc_fw is not None else "-" + lines.append( + f"| native run {index} | True | {run.get('rung')} | {run.get('instructions')} | " + f"{run.get('cycles')} | {stop.get('reason')}@{pc_fw_text} | " + f"{run.get('wall_s')} | `{run.get('session')}` |" + ) + if not runs: + lines.append(f"| native | {native.get('ran')} | - | - | - | - | - | {native.get('reason')} |") + + first = qemu.get("first_fault") or {} + lines += [ + "", + "### QEMU first fault", + "", + f"* {first.get('desc')} at fw vaddr `{first.get('epc_fw') if first.get('epc_fw') is not None else 0:#x}` " + f"accessing `{(first.get('access_addr') or 0):#x}` ({first.get('size')} bytes, {first.get('rw')})", + f"* first-touch matches the documented reference: {qemu.get('first_touch_match')}", + f"* fault records in the capped log window: {qemu.get('fault_count_window')}", + "", + "## Frozen thresholds", + "", + "| check | pass |", + "|---|---|", + ] + for name, value in (report.get("threshold_checks") or {}).items(): + lines.append(f"| {name} | {value} |") + lines += [ + "", + f"Thresholds (frozen {report['thresholds'].get('frozen_on')}): " + f"native >= QEMU + {report['thresholds'].get('native_rungs_above_qemu')} rung(s); " + f"{report['thresholds'].get('identical_runs')} identical runs; " + f"pair wall <= {report['thresholds'].get('native_pair_wall_s')} s.", + "", + "## Determinism", + "", + f"* identical across {native.get('repeat')} native runs: {native.get('identical')}", + f"* session compare ran: {(native.get('compare') or {}).get('ran')} " + f"({(native.get('compare') or {}).get('headers')})", + "", + "## Session hashes (sha256)", + "", + "| session | file | sha256 |", + "|---|---|---|", + ] + hash_rows = [("qemu-probe", qemu.get("session"), qemu.get("session_hashes") or {})] + for index, run in enumerate(runs, start=1): + hash_rows.append((f"native-{index}", run.get("session"), run.get("session_hashes") or {})) + features_detail = ((report.get("features") or {}).get("features") or {}) + for feature in ("sessions", "trace", "injection"): + detail = (features_detail.get(feature) or {}).get("detail") or {} + if isinstance(detail, dict): + hashes = detail.get("hashes") or {} + if hashes: + hash_rows.append((feature, detail.get("dir") or detail.get("blob"), hashes)) + for label, session, hashes in hash_rows: + if not hashes: + lines.append(f"| {label} | - | (no hashes recorded) |") + continue + for name in sorted(hashes): + lines.append(f"| {label} | {name} | `{hashes[name]}` |") + lines += [ + "", + "## Native feature matrix (demo firmware)", + "", + "| feature | status | detail |", + "|---|---|---|", + ] + feature_block = (report.get("features") or {}).get("features") or {} + for name in sorted(feature_block): + entry = feature_block[name] + detail = entry.get("detail") or entry.get("reason") or "" + lines.append(f"| {name} | {entry.get('status')} | {detail} |") + if not feature_block: + lines.append(f"| - | not_run | {(report.get('features') or {}).get('reason')} |") + lines += [ + "", + "## S1 integration", + "", + f"* device source: {report['s1_integration'].get('source') or 'lenient (no S1 device model found)'}", + f"* profile: {report['s1_integration'].get('profile') or '-'}", + f"* declared rung: {report['s1_integration'].get('declared_rung')}", + ] + for note in report["s1_integration"].get("notes") or []: + lines.append(f"* note: {note}") + lines += [ + "", + "## Not verified", + "", + ] + for item in report.get("not_verified") or []: + lines.append(f"* {item}") + lines += [ + "", + "## S1 integration notes", + "", + ] + for item in report.get("rerun_when_s1_lands") or []: + lines.append(f"* `{item}`" if item.startswith("python3") else f"* {item}") + lines += [ + "", + f"* other agents' x86 QEMU VMs observed (never touched): {report.get('x86_vms_running')}", + "", + ] + return "\n".join(lines) + + +def run_harness( + *, + image: str = DEFAULT_IMAGE, + out_dir: str = DEFAULT_OUT, + instructions: int = DEFAULT_INSTRUCTIONS, + qemu_duration: float = DEFAULT_QEMU_DURATION_S, + repeat: int = 2, + device_map: str | None = None, + features: bool = True, + thresholds: dict | None = None, + run_qemu: bool = True, +) -> dict: + """Run the full A/B harness and write ``sim-qemu-line.json`` + ``.md``.""" + os.makedirs(out_dir, exist_ok=True) + frozen = dict(FROZEN_THRESHOLDS) + frozen.update(thresholds or {}) + + if device_map is not None: + dm = DeviceMap.load(device_map) + s1 = { + "device_map": dm, + "profile": None, + "declared_rung": None, + "source": _relpath(device_map), + "notes": ["explicit --device-map override"], + } + else: + sha = firmware_sha256(image) if os.path.isfile(image) else None + s1 = discover_s1_device_model(image, sha) + + qemu = ( + run_qemu_baseline(image=image, out_dir=out_dir, duration=qemu_duration) + if run_qemu + else {"backend": "qemu", "ran": False, "reason": "qemu=False", "rung": None} + ) + + native = run_native_baseline( + image=image, + out_dir=out_dir, + instructions=instructions, + repeat=repeat, + device_map=s1.get("device_map"), + declared_rung=s1.get("declared_rung"), + profile=s1.get("profile"), + ) + if not native.get("ran"): + report = build_report( + image=image, qemu=qemu, native=native, features=None, s1=s1, thresholds=frozen + ) + else: + feature_result = run_feature_matrix(out_dir) if features else None + report = build_report( + image=image, + qemu=qemu, + native=native, + features=feature_result, + s1=s1, + thresholds=frozen, + ) + + json_path = _write_json(os.path.join(out_dir, "sim-qemu-line.json"), report) + md_path = _write_text(os.path.join(out_dir, "sim-qemu-line.md"), render_markdown(report)) + report["report_json"] = _relpath(json_path) + report["report_md"] = _relpath(md_path) + return report + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="python3 harness/sim_qemu_line.py", + description=( + "S4 A/B: QEMU probe vs native GSP simulator on the real 570.144 image; " + "emits sim-qemu-line.json + .md. Bounded, no hardware." + ), + ) + parser.add_argument("--image", default=DEFAULT_IMAGE, help="flat GSP firmware image") + parser.add_argument("--out", default=DEFAULT_OUT, help="report directory") + parser.add_argument( + "--instructions", type=int, default=DEFAULT_INSTRUCTIONS, + help=f"native instruction budget per run (<= {MAX_INSTRUCTIONS})", + ) + parser.add_argument( + "--qemu-duration", type=float, default=DEFAULT_QEMU_DURATION_S, + help=f"QEMU wall-clock probe bound (<= {MAX_QEMU_DURATION_S}s)", + ) + parser.add_argument("--repeat", type=int, default=2, help="native runs for determinism") + parser.add_argument("--device-map", help="explicit greeneye-device-map/1 file (skips discovery)") + parser.add_argument("--no-qemu", action="store_true", help="native only (no baseline claim)") + parser.add_argument("--no-features", action="store_true", help="skip the feature matrix") + parser.add_argument("--json", action="store_true", help="print the report JSON") + args = parser.parse_args(argv) + + try: + report = run_harness( + image=args.image, + out_dir=args.out, + instructions=args.instructions, + qemu_duration=args.qemu_duration, + repeat=args.repeat, + device_map=args.device_map, + features=not args.no_features, + run_qemu=not args.no_qemu, + ) + except (HarnessError, GreenEyeError) as exc: + print(f"[sim-qemu-line] error: {exc}", file=sys.stderr) + return 1 + + if not os.path.isfile(args.image): + print(f"[sim-qemu-line] skipped: firmware absent: {args.image}", file=sys.stderr) + return 2 + if args.json: + print(json.dumps(report, indent=1, sort_keys=True, default=str)) + native = report.get("native") or {} + print(f"[sim-qemu-line] qemu rung: {(report.get('qemu') or {}).get('rung')}") + print(f"[sim-qemu-line] native rung: {native.get('rung')} (identical={native.get('identical')})") + print(f"[sim-qemu-line] verdict: {report.get('verdict')}") + print(f"[sim-qemu-line] thresholds: {report.get('threshold_checks')}") + print(f"[sim-qemu-line] report: {report.get('report_json')} + {report.get('report_md')}") + return 0 if report.get("above_qemu_line") else 1 + + +if __name__ == "__main__": + sys.exit(main()) |