path: root/debug/risc-v-gsp-rm-vm/harness/package_sim.py

#!/usr/bin/env python3
"""Package the native risc-v-gsp-rm-debugger simulator together with injected code.

Builds a self-contained trial directory (or deterministic ``.tar.gz``):

    README.md       what this is, how to run, what to expect, safety
    run.sh          single entry: native simulator + scenario, writes results
    manifest.json   format/version, hashes, component list, run contract
    payload/        target image/ELF, service blob, device map, preinject config
    scenario.py     the pre-scripted scenario as run (builtin or --script)
    results/        produced by run.sh (results.json, results.md, session)

The trial boots the target natively, loads/registers the injected service,
invokes it directly and through the firmware dispatcher, runs one hook/watch
pass, and writes machine-readable and human-readable results.  ``--pack-image``
additionally emits a combined ELF (demo flattened plus the service at a fixed
offset) and a preinject config, so the trial starts with the injected code
already part of the simulated system.

Honesty: the real GSP ``.fwimage`` is packaged with a lenient device map but
cannot boot on the native simulator yet (H2.2 device model absent); its trial
records ``status=blocked`` and a bounded first-touch probe, never a boot.

Everything is stdlib-only; builds are deterministic apart from the manifest
``created`` field (set ``SOURCE_DATE_EPOCH`` to make even that stable).
"""
from __future__ import annotations

import argparse
import gzip
import hashlib
import json
import os
import shutil
import struct
import subprocess
import sys
import tarfile
import time
from dataclasses import dataclass

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 import __version__ as GREENEYE_VERSION  # noqa: E402
from greeneye.elf import EM_RISCV, load_elf  # noqa: E402
from greeneye.errors import GreenEyeError  # noqa: E402
from greeneye.inject import ENTRY_SYMBOL  # noqa: E402

FORMAT = "greeneye-trial/1"
RESULTS_SCHEMA = "greeneye-trial-results/1"
PREINJECT_SCHEMA = "greeneye-preinject/1"
DEVICE_MAP_SCHEMA = "greeneye-device-map/1"

DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
DEFAULT_SERVICE = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf")
REPO_ROOT = os.path.dirname(os.path.dirname(TOOL_ROOT))
REAL_IMAGE = os.path.normpath(
    os.path.join(REPO_ROOT, "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin")
)

RAM_BASE = 0x80000000
REAL_IMAGE_BIAS = 0x80000000
#: Service placement mirrors greeneye.inject.load_service's default load address.
SERVICE_OFFSET = 0x01000000
DEFAULT_SLOT = 6
PING_XOR = 0xBEEF
IMAGE_BUDGET = 200_000
DEMO_BOOT_SECONDS = 0.25
DEMO_SETTLE_SECONDS = 0.25
DEFAULT_RUN_TIMEOUT = 180.0


# -- small deterministic helpers -------------------------------------------


def sha256_file(path: str) -> str:
    digest = hashlib.sha256()
    with open(path, "rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 20), b""):
            digest.update(chunk)
    return digest.hexdigest()


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _portable(path: str) -> str:
    """Manifest-friendly source name: repo-relative when possible, else basename."""
    absolute = os.path.abspath(path)
    relative = os.path.relpath(absolute, TOOL_ROOT)
    if relative.startswith(".."):
        return os.path.basename(absolute)
    return relative


def _created_iso() -> str:
    """Build timestamp; stable when SOURCE_DATE_EPOCH is set."""
    raw = os.environ.get("SOURCE_DATE_EPOCH")
    try:
        epoch = int(raw) if raw else int(time.time())
    except ValueError:
        epoch = 0
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(epoch))


def _write_bytes(path: str, data: bytes) -> str:
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "wb") as fh:
        fh.write(data)
    return path


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, indent: int = 2) -> str:
    return _write_text(path, json.dumps(payload, indent=indent, sort_keys=True) + "\n")


# -- combined demo+service ELF ---------------------------------------------


def _load_plan(image, bias: int) -> list[tuple[int, int, bytes]]:
    """``(vaddr, p_flags, memsz-padded bytes)`` for every non-empty PT_LOAD."""
    plan = []
    for seg in image.load_segments:
        data = bytearray(image.segment_bytes(seg))
        if len(data) < seg.p_memsz:
            data += b"\x00" * (seg.p_memsz - len(data))
        if data:
            plan.append((seg.p_vaddr + bias, seg.p_flags, bytes(data)))
    return plan


def write_combined_elf(
    demo_path: str,
    service_path: str,
    out_path: str,
    *,
    service_offset: int = SERVICE_OFFSET,
    slot: int = DEFAULT_SLOT,
) -> dict:
    """Merge a target ELF and a relocated service ELF into one runnable ELF64.

    The service is placed exactly as ``greeneye.inject.load_service`` would place
    it (segments at ``service_load_addr + p_vaddr``, entry at
    ``service_load_addr + e_entry``), so a pre-injected boot is address-for-address
    equivalent to the load/register path.  Returns the placement config.
    """
    demo = load_elf(demo_path)
    service = load_elf(service_path)
    if not demo.is_riscv or not service.is_riscv:
        raise GreenEyeError("combined image requires RISC-V ELF inputs")
    target_load = demo.min_vaddr
    service_load = target_load + service_offset
    demo_plan = _load_plan(demo, 0)
    service_plan = _load_plan(service, service_load)
    if not demo_plan:
        raise GreenEyeError(f"{demo_path}: no loadable segments")

    spans = sorted((vaddr, vaddr + len(data)) for vaddr, _flags, data in demo_plan + service_plan)
    for (a1, b1), (a2, b2) in zip(spans, spans[1:]):
        if a2 < b1:
            raise GreenEyeError(
                f"combined image overlap: [{a2:#x},{b2:#x}) collides with [{a1:#x},{b1:#x})"
            )

    segments = demo_plan + service_plan
    phnum = len(segments)
    ident = b"\x7fELF" + bytes([2, 1, 1, 0]) + b"\x00" * 8  # ELF64, LE, v1
    header = ident + struct.pack(
        "<HHIQQQIHHHHHH", 2, EM_RISCV, 1, demo.entry, 64, 0, 0, 64, 56, phnum, 64, 0, 0
    )
    offset = 64 + 56 * phnum
    phdrs = bytearray()
    body = bytearray()
    for vaddr, flags, data in segments:
        phdrs += struct.pack(
            "<IIQQQQQQ", 1, flags, offset + len(body), vaddr, vaddr, len(data), len(data), 0x1000
        )
        body += data
    _write_bytes(out_path, header + bytes(phdrs) + bytes(body))

    check = load_elf(out_path)
    if check.is_riscv is False or check.entry != demo.entry or len(check.load_segments) != phnum:
        raise GreenEyeError(f"combined ELF failed its own round-trip: {out_path}")

    info = {
        "target_load_addr": target_load,
        "target_entry": demo.entry,
        "service_load_addr": service_load,
        "service_offset": service_offset,
        "service_entry": service.entry,
        "entry_addr": service_load + service.entry,
        "slot": slot,
        "size": sum(len(data) for _v, _f, data in service_plan),
        "segments": phnum,
    }
    return info


# -- deterministic tar ------------------------------------------------------


def write_deterministic_tar(out_dir: str, tar_path: str) -> str:
    """Write ``out_dir`` as a gzip tar with zero timestamps and sorted members.

    Member mtimes, uid/gid, uname/gname and the gzip mtime are fixed, so two
    builds of identical content produce byte-identical archives.
    """
    names = []
    for root, dirs, files in os.walk(out_dir):
        dirs.sort()
        for name in sorted(files):
            names.append(os.path.relpath(os.path.join(root, name), out_dir))
    names.sort()
    with open(tar_path, "wb") as raw:
        with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as gz:
            with tarfile.open(fileobj=gz, mode="w", format=tarfile.GNU_FORMAT) as tar:
                for name in names:
                    full = os.path.join(out_dir, name)
                    info = tar.gettarinfo(full, arcname=name)
                    info.mtime = 0
                    info.uid = info.gid = 0
                    info.uname = info.gname = ""
                    info.mode = 0o755 if name == "run.sh" else 0o644
                    with open(full, "rb") as fh:
                        tar.addfile(info, fh)
    return tar_path


# -- request/build ----------------------------------------------------------

RUN_SH = """#!/bin/sh
# risc-v-gsp-rm-debugger trial runner (generated by harness/package_sim.py; inspect before use).
# Usage: ./run.sh [scenario args]     Environment: GREENEYE_HOME overrides the tool root.
set -eu
here=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
if [ -n "${GREENEYE_HOME:-}" ]; then
    root=$GREENEYE_HOME
else
    root=$here
    while [ "$root" != "/" ] && [ ! -f "$root/greeneye/__init__.py" ]; do
        root=$(dirname "$root")
    done
    if [ ! -f "$root/greeneye/__init__.py" ]; then
        echo "run.sh: risc-v-gsp-rm-debugger tool root not found; set GREENEYE_HOME" >&2
        exit 3
    fi
fi
export PYTHONPATH="$root${PYTHONPATH:+:$PYTHONPATH}"
exec python3 "$here/scenario.py" --package "$here" "$@"
"""


@dataclass
class PackageRequest:
    out: str
    target: str = "demo"  # demo | image
    image: str | None = None
    service: str | None = None
    device_map: str | None = None
    script: str | None = None
    slot: int = DEFAULT_SLOT
    pack_image: bool = False
    tar: bool = False


def _readme_text(
    *,
    target: str,
    target_name: str,
    target_sha: str,
    service_name: str,
    service_sha: str,
    slot: int,
    device_map_name: str | None,
    preinject: dict | None,
) -> str:
    lines = [
        "# risc-v-gsp-rm-debugger trial package",
        "",
        "Generated by `harness/package_sim.py` (`greeneye-trial/1`). This package",
        "bundles a target, an injected service blob and a pre-scripted scenario",
        "that runs on the deterministic native RV64 simulator in the risc-v-gsp-rm-debugger",
        "tool tree. It needs no QEMU, no GPU and no guest.",
        "",
        "## Contents",
        "",
        f"- target: `payload/{target_name}` (`{target}`), sha256 `{target_sha}`",
        f"- service: `payload/{service_name}`, slot {slot}, sha256 `{service_sha}`",
        f"- device map: `{('payload/' + device_map_name) if device_map_name else 'none'}`",
        f"- pre-injected combined image: `{preinject['file'] if preinject else 'none'}`",
        "- scenario: `scenario.py`; results are written to `results/`",
        "",
        "## How to run",
        "",
        "```sh",
        "./run.sh",
        "# image packages may bound the lenient probe explicitly:",
        "./run.sh --budget 200000",
        "```",
        "",
        "Requires Python 3.10+ and the risc-v-gsp-rm-debugger tool tree. `run.sh` finds the",
        "tree by walking up from this directory; set `GREENEYE_HOME=/path/to/risc-v-gsp-rm-vm`",
        "if the package was moved. Exit code 0 means the assertions passed (or an",
        "image package produced its documented blocked status), 1 means an",
        "assertion failure, 3 means the tool tree was not found.",
        "",
        "## What to expect",
        "",
    ]
    if target == "demo":
        lines += [
            "The demo uproc boots natively, the service is loaded and registered in",
            f"`gsp_ctrl.handlers[{slot}]`, then invoked directly and through mailbox",
            "command 16 + slot. `results/results.json` should report `status=ok` with",
            "matching call results, at least one `rpc_dispatch` hook hit, at least one",
            "watch hit on `gsp_ctrl.result`, and the UART banner/self-test lines.",
        ]
        if preinject:
            lines += [
                "",
                "This package uses `--pack-image`: the combined ELF",
                f"`{preinject['file']}` already contains the service at",
                f"{preinject['service_load_addr']:#x}, so no separate load step runs.",
            ]
    else:
        lines += [
            "**Honest status: the native simulator cannot boot the real GSP",
            "`.fwimage` yet** (the H2.2 device model is absent). The scenario runs a",
            "bounded lenient first-touch probe with the bundled device map and",
            "records `status=blocked`, `blocked_on_device_model=true` in",
            "`results/results.json`; it is not a boot and makes no hardware claim.",
            "No RM service registration point exists yet (M5), so the service blob",
            "is packaged but not loaded or registered.",
        ]
        if preinject:
            lines += [
                "",
                f"A combined artifact `{preinject['file']}` is present but marked",
                "`runnable: false` for the same reason; it is not executed.",
            ]
    lines += [
        "",
        "## How to read the results",
        "",
        "- `results/results.json` - machine-readable: status, call results, hook and",
        "  watch hits, control block, UART excerpt, session path, assertion list.",
        "- `results/results.md` - the same rendered for humans.",
        "- `results/session/` - the recorded greeneye session (`events.jsonl`,",
        "  `session.json`, snapshots) for `greeneye analyze`/`compare`.",
        "- `manifest.json` - format/version, input hashes and the component list;",
        "  every component hash covers the file as packaged.",
        "",
        "## Safety",
        "",
        "Run only packages from sources you trust. Injected code executes inside",
        "the simulator process with the simulator's full privileges. This package",
        "contains no hardware access and must never be used to inject into a guest,",
        "a shipped environment or real hardware.",
        "",
        "## Determinism",
        "",
        "Manifest component order and hashes are stable, and `--tar` archives use",
        "zeroed member timestamps. The only variable field is `manifest.created`;",
        "set `SOURCE_DATE_EPOCH` to freeze it and reproduce byte-identical builds.",
        "",
    ]
    return "\n".join(lines)


def build_package(request: PackageRequest) -> dict:
    """Build the trial package; returns the manifest that was written."""
    out = os.path.abspath(request.out)
    if os.path.isdir(out) and os.listdir(out):
        raise GreenEyeError(f"output directory is not empty: {out}")
    if request.target not in ("demo", "image"):
        raise GreenEyeError(f"target must be demo or image, got {request.target!r}")
    if not 0 <= request.slot <= 7:
        raise GreenEyeError(f"slot must be 0..7, got {request.slot}")

    if request.target == "demo":
        target_input = os.path.abspath(request.image or DEMO_ELF)
    else:
        target_input = os.path.abspath(request.image or REAL_IMAGE)
    if not os.path.isfile(target_input):
        raise GreenEyeError(f"target not found: {target_input}")
    service_input = os.path.abspath(request.service or DEFAULT_SERVICE)
    if not os.path.isfile(service_input):
        raise GreenEyeError(f"service blob not found: {service_input} (run: make sdk)")

    target_image = load_elf(target_input)
    if not target_image.is_riscv:
        raise GreenEyeError(f"{target_input}: not a RISC-V ELF image")
    service_image = load_elf(service_input)
    if not service_image.is_riscv:
        raise GreenEyeError(f"{service_input}: not a RISC-V ELF image")

    os.makedirs(out, exist_ok=True)
    payload_dir = os.path.join(out, "payload")
    os.makedirs(payload_dir, exist_ok=True)
    components: list[tuple[str, str]] = []

    target_name = os.path.basename(target_input)
    shutil.copyfile(target_input, os.path.join(payload_dir, target_name))
    components.append((f"payload/{target_name}", "target-image"))

    service_name = os.path.basename(service_input)
    shutil.copyfile(service_input, os.path.join(payload_dir, service_name))
    components.append((f"payload/{service_name}", "service-blob"))

    device_map_info: dict | None = None
    if request.device_map:
        dm_input = os.path.abspath(request.device_map)
        if not os.path.isfile(dm_input):
            raise GreenEyeError(f"device map not found: {dm_input}")
        dm_name = os.path.basename(dm_input)
        shutil.copyfile(dm_input, os.path.join(payload_dir, dm_name))
        components.append((f"payload/{dm_name}", "device-map"))
        device_map_info = {
            "file": f"payload/{dm_name}",
            "input": _portable(dm_input),
            "sha256": sha256_file(dm_input),
            "lenient": request.target == "image",
            "learn": False,
        }
    elif request.target == "image":
        dm_name = "device_map.json"
        _write_json(
            os.path.join(payload_dir, dm_name),
            {"schema": DEVICE_MAP_SCHEMA, "meta": {"bias": REAL_IMAGE_BIAS, "source": "lenient"}, "entries": []},
            indent=1,
        )
        components.append((f"payload/{dm_name}", "device-map"))
        device_map_info = {
            "file": f"payload/{dm_name}",
            "input": None,
            "sha256": sha256_file(os.path.join(payload_dir, dm_name)),
            "lenient": True,
            "learn": True,
        }

    preinject: dict | None = None
    if request.pack_image:
        service_offset = SERVICE_OFFSET
        if request.target == "image":
            span = target_image.max_vaddr - target_image.min_vaddr
            service_offset = max(SERVICE_OFFSET, ((span + 0xFFF) // 0x1000) * 0x1000)
        combined_rel = "payload/combined.elf"
        placement = write_combined_elf(
            target_input,
            service_input,
            os.path.join(out, combined_rel),
            service_offset=service_offset,
            slot=request.slot,
        )
        components.append((combined_rel, "combined-elf"))
        runnable = request.target == "demo"
        reason = (
            ""
            if runnable
            else "the native simulator cannot boot the real GSP image yet (H2.2 device model absent)"
        )
        preinject = {
            "file": combined_rel,
            "schema": PREINJECT_SCHEMA,
            "runnable": runnable,
            "reason": reason,
            **placement,
        }
        _write_json(os.path.join(payload_dir, "preinject.json"), preinject)
        components.append(("payload/preinject.json", "preinject-config"))

    if request.script:
        script_input = os.path.abspath(request.script)
        if not os.path.isfile(script_input):
            raise GreenEyeError(f"scenario script not found: {script_input}")
        shutil.copyfile(script_input, os.path.join(out, "scenario.py"))
        script_info = {
            "kind": "custom",
            "name": os.path.basename(script_input),
            "input": _portable(script_input),
            "sha256": sha256_file(script_input),
        }
    else:
        _write_text(os.path.join(out, "scenario.py"), BUILTIN_SCENARIO)
        script_info = {
            "kind": "builtin",
            "name": "builtin-demo-v1",
            "input": None,
            "sha256": sha256_bytes(BUILTIN_SCENARIO.encode("utf-8")),
        }
    components.append(("scenario.py", "scenario"))

    run_sh_path = os.path.join(out, "run.sh")
    _write_text(run_sh_path, RUN_SH)
    os.chmod(run_sh_path, 0o755)
    components.append(("run.sh", "runner"))

    _write_text(os.path.join(out, "results", ".gitkeep"), "")
    components.append(("results/.gitkeep", "results-placeholder"))

    if request.target == "image":
        if preinject and preinject.get("runnable"):
            end = max(
                target_image.max_vaddr + REAL_IMAGE_BIAS,
                preinject["service_load_addr"] + preinject["size"],
            )
        else:
            end = target_image.max_vaddr + REAL_IMAGE_BIAS
        need = end - RAM_BASE + (1 << 20)
        memory_mb = max(256, (need + (1 << 20) - 1) // (1 << 20))
        limitations = [
            "The native simulator cannot boot the real GSP .fwimage yet: the H2.2 device model is absent.",
            "The run is a bounded lenient first-touch probe, not a boot; results.json records status=blocked.",
            "No RM service registration point exists yet (M5): the blob is packaged but not loaded or registered.",
            "No L2 timing/concurrency and no L3 device fidelity is claimed (docs/06-ACCURACY.md).",
        ]
    else:
        memory_mb = 256
        limitations = [
            "The demo uproc is a stand-in for GSP-RM, not the real device; results describe the injection path only.",
            "Injected code runs with the simulator's full in-process privileges; only run trusted packages.",
        ]

    _write_text(
        os.path.join(out, "README.md"),
        _readme_text(
            target=request.target,
            target_name=target_name,
            target_sha=sha256_file(target_input),
            service_name=service_name,
            service_sha=sha256_file(service_input),
            slot=request.slot,
            device_map_name=device_map_info["file"].split("/", 1)[1] if device_map_info else None,
            preinject=preinject,
        ),
    )
    components.append(("README.md", "documentation"))

    manifest = {
        "format": FORMAT,
        "greeneye_version": GREENEYE_VERSION,
        "created": _created_iso(),
        "results_schema": RESULTS_SCHEMA,
        "target": {
            "kind": request.target,
            "file": f"payload/{target_name}",
            "input": _portable(target_input),
            "sha256": sha256_file(target_input),
            "entry": target_image.entry,
            "min_vaddr": target_image.min_vaddr,
            "max_vaddr": target_image.max_vaddr,
            "bias": REAL_IMAGE_BIAS if request.target == "image" else 0,
            "runnable": request.target == "demo",
            "limitations": limitations,
        },
        "service": {
            "file": f"payload/{service_name}",
            "input": _portable(service_input),
            "sha256": sha256_file(service_input),
            "entry": service_image.entry,
            "entry_symbol": ENTRY_SYMBOL,
            "slot": request.slot,
        },
        "device_map": device_map_info,
        "script": script_info,
        "preinject": preinject,
        "scenario": {
            "kind": script_info["kind"],
            "demo_boot_seconds": DEMO_BOOT_SECONDS,
            "demo_settle_seconds": DEMO_SETTLE_SECONDS,
            "image_budget_instructions": IMAGE_BUDGET,
            "memory_mb": memory_mb,
            "expected": {
                "cmd_ping": 1,
                "cmd_sum": 2,
                "ping_arg": 0x41,
                "sum_args": [0x10, 0x20],
                "ping_result": 0x41 ^ PING_XOR,
                "sum_result": 0x30,
            },
        },
        "run": {
            "command": "./run.sh",
            "results": ["results/results.json", "results/results.md"],
            "exit_nonzero_on": "assertion failure",
        },
        "safety": (
            "Local, development-only study tooling: no hardware access, no guest, "
            "never shipped. Run only packages from trusted sources."
        ),
        "determinism": (
            "Component order and hashes are stable; --tar members use mtime 0. "
            "manifest.created is the only timestamp (SOURCE_DATE_EPOCH freezes it)."
        ),
        "components": [],
    }
    for relpath, role in sorted(components):
        full = os.path.join(out, relpath)
        manifest["components"].append(
            {
                "path": relpath,
                "role": role,
                "bytes": os.path.getsize(full),
                "sha256": sha256_file(full),
            }
        )

    _write_json(os.path.join(out, "manifest.json"), manifest)
    if request.tar:
        write_deterministic_tar(out, out + ".tar.gz")
    return manifest


def run_package(out: str, *, timeout: float = DEFAULT_RUN_TIMEOUT, args: list[str] | None = None) -> int:
    """Run ``run.sh`` in a bounded subprocess; returns its exit code."""
    run_sh = os.path.join(os.path.abspath(out), "run.sh")
    if not os.path.isfile(run_sh):
        raise GreenEyeError(f"no run.sh in {out}; build the package first")
    environment = dict(os.environ)
    environment.setdefault("GREENEYE_HOME", TOOL_ROOT)
    try:
        proc = subprocess.run(
            ["/bin/sh", run_sh, *(args or [])],
            cwd=os.path.abspath(out),
            env=environment,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired as exc:
        print(f"[package-sim] run.sh timed out after {timeout}s", file=sys.stderr)
        print(exc.stdout or "", end="")
        print(exc.stderr or "", end="", file=sys.stderr)
        return 124
    print(proc.stdout, end="")
    print(proc.stderr, end="", file=sys.stderr)
    return proc.returncode


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="python3 harness/package_sim.py",
        description="Build a self-contained risc-v-gsp-rm-debugger trial package (simulator + injected service).",
    )
    parser.add_argument("--target", choices=("demo", "image"), default="demo",
                        help="demo uproc (default) or a real .fwimage")
    parser.add_argument("--image", help="target ELF/.fwimage path (image target; overrides the demo ELF)")
    parser.add_argument("--service", help=f"service blob ELF (default: {os.path.relpath(DEFAULT_SERVICE, TOOL_ROOT)})")
    parser.add_argument("--device-map", help="greeneye-device-map/1 JSON (lenient MMIO evidence)")
    parser.add_argument("--script", help="custom scenario script (default: builtin demo scenario)")
    parser.add_argument("--out", required=True, help="output directory (must be empty)")
    parser.add_argument("--slot", type=int, default=DEFAULT_SLOT, help="handler slot 0..7 (default 6)")
    parser.add_argument("--pack-image", action="store_true",
                        help="also emit payload/combined.elf + preinject config")
    parser.add_argument("--tar", action="store_true", help="write a deterministic <out>.tar.gz")
    parser.add_argument("--run", action="store_true", help="run the packaged trial after building")
    parser.add_argument("--run-timeout", type=float, default=DEFAULT_RUN_TIMEOUT, help="run.sh timeout seconds")
    args = parser.parse_args(argv)

    request = PackageRequest(
        out=args.out,
        target=args.target,
        image=args.image,
        service=args.service,
        device_map=args.device_map,
        script=args.script,
        slot=args.slot,
        pack_image=args.pack_image,
        tar=args.tar,
    )
    try:
        manifest = build_package(request)
    except GreenEyeError as exc:
        print(f"[package-sim] error: {exc}", file=sys.stderr)
        return 1
    print(
        f"[package-sim] built {args.out}: {len(manifest['components'])} components, "
        f"target={manifest['target']['kind']}"
    )
    if args.tar:
        print(f"[package-sim] archive {args.out}.tar.gz")
    if manifest["target"]["runnable"] is False:
        print("[package-sim] note: packaged target cannot boot yet; results will report status=blocked")
        if manifest.get("preinject"):
            print("[package-sim] note: preinject/unused combined artifact is informational only")
    if args.run:
        return run_package(args.out, timeout=args.run_timeout)
    return 0


# -- generated scenario -----------------------------------------------------

BUILTIN_SCENARIO = r'''#!/usr/bin/env python3
"""Pre-scripted risc-v-gsp-rm-debugger trial scenario (generated by harness/package_sim.py).

Boots the packaged target on the native simulator, loads and registers the
injected service, invokes it directly and through the firmware mailbox, runs
one hook/watch pass, and writes results/results.json + results/results.md.
Exit 0 = assertions passed (or the documented blocked image status), 1 = a
verification assertion failed.
"""
from __future__ import annotations

import argparse
import json
import os
import sys

CTRL_HANDLERS = 0x150
CTRL_RESULT = 0x40
CTRL_RPC_COUNT = 0x10
PING_ARG = 0x41
SUM_A0 = 0x10
SUM_A1 = 0x20
EXPECTED_PING = PING_ARG ^ 0xBEEF
EXPECTED_SUM = SUM_A0 + SUM_A1
UART_BANNER = "RISC-V GSP-RM DEBUGGER UPROC DEMO v1"
UART_SELFTEST = "SELFTEST DONE"


def bootstrap_tool_root(pkg):
    """Make `import greeneye` work when scenario.py is run directly."""
    try:
        import greeneye  # noqa: F401
        return
    except ImportError:
        pass
    candidates = []
    if os.environ.get("GREENEYE_HOME"):
        candidates.append(os.path.abspath(os.environ["GREENEYE_HOME"]))
    here = pkg
    while True:
        candidates.append(here)
        parent = os.path.dirname(here)
        if parent == here:
            break
        here = parent
    for candidate in candidates:
        if os.path.isfile(os.path.join(candidate, "greeneye", "__init__.py")):
            sys.path.insert(0, candidate)
            return


def load_json(path):
    with open(path, "r", encoding="utf-8") as fh:
        return json.load(fh)


def write_json(path, payload):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(payload, fh, indent=2, sort_keys=True)
        fh.write("\n")


def write_text(path, text):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)


def component(manifest, role):
    for comp in manifest.get("components", ()):
        if comp.get("role") == role:
            return comp["path"]
    return None


def uart_excerpt(text, first=4, last=6, cap=14):
    lines = [line for line in text.splitlines() if line.strip()]
    if len(lines) > cap:
        return lines[:first] + ["..."] + lines[-last:]
    return lines


def hexv(value):
    return f"0x{int(value):x}"


def render_md(manifest, data):
    target = data.get("target", {})
    lines = ["# risc-v-gsp-rm-debugger trial results", ""]
    lines.append(f"- status: **{data.get('status')}**")
    lines.append(f"- target: `{target.get('file', '?')}` ({target.get('kind', '?')})")
    service = data.get("service")
    if service:
        lines.append(
            f"- service: `{service.get('file', '?')}` slot {service.get('slot', '?')} "
            f"pre-injected {'yes' if service.get('preinjected') else 'no'}"
        )
    lines.append(f"- backend: {data.get('backend', 'native')}")
    if data.get("error"):
        lines += ["", "## Error", "", "```", str(data["error"]), "```"]
    if data.get("status") == "blocked":
        lines += [
            "",
            "## Honest status",
            "",
            str(data.get("blocked_reason", "the native simulator cannot boot this target yet")),
            "",
            f"- instructions executed: {data.get('instructions_executed')}",
            f"- stopped: pc {hexv(data.get('stopped', {}).get('pc', 0))} "
            f"(firmware {hexv(data.get('stopped', {}).get('fw_pc', 0))})",
            f"- unmodelled MMIO accesses served leniently: {data.get('mmio_unknown')}",
        ]
    if data.get("calls"):
        calls = data["calls"]
        expected = calls.get("expected", {})
        lines += [
            "",
            "## Injected call results",
            "",
            "| route | cmd | result | expected | ok |",
            "|---|---|---|---|---|",
        ]
        for route, pair in (("remote (direct)", "ping"), ("remote (direct)", "sum"),
                            ("via firmware mailbox", "ping"), ("via firmware mailbox", "sum")):
            key = "remote" if route.startswith("remote") else "via_firmware"
            actual = calls.get(key, {}).get(pair)
            want = expected.get(pair)
            lines.append(f"| {route} | {pair} | {hexv(actual)} | {hexv(want)} | "
                         f"{'yes' if actual == want else 'no'} |")
        lines.append("")
        lines.append(f"routes agree: {'yes' if calls.get('agree') else 'no'}")
    if data.get("hooks"):
        hooks = data["hooks"]["rpc_dispatch"]
        lines += ["", "## Hook / watch pass", "",
                  f"- `rpc_dispatch` hook hits: {hooks['hits']}"]
        for hit in hooks.get("first", [])[:4]:
            lines.append(f"  - pc {hit['pc']} cmd {hit['cmd']} a0 {hit['a0']} a1 {hit['a1']}")
    if data.get("watches"):
        watch = data["watches"]["gsp_ctrl.result"]
        lines.append(f"- `gsp_ctrl.result` watch hits: {watch['hits']}")
        for hit in watch.get("first", [])[:4]:
            lines.append(f"  - pc {hit['pc']} ({hit.get('symbol')})")
    if data.get("control"):
        control = data["control"]
        lines.append(
            f"- control block: result {hexv(control['result'])}, rpc_count {control['rpc_count']}"
        )
    lines.append(f"- handlers slot value: {hexv(data.get('handlers_slot', 0))}")
    if data.get("uart_excerpt"):
        lines += ["", "## UART excerpt", "", "```"]
        lines += list(data["uart_excerpt"])
        lines += ["```"]
    lines += ["", "## Session", "", f"`{data.get('session', 'results/session')}`"]
    if data.get("learned_device_map"):
        lines.append(f"`{data['learned_device_map']}`")
    assertions = data.get("assertions", {})
    lines += ["", "## Assertions", ""]
    for name in sorted(assertions.get("checks", {})):
        lines.append(f"- [{'x' if assertions['checks'][name] else ' '}] {name}")
    failed = assertions.get("failed", [])
    lines.append("")
    lines.append("all checks passed" if not failed else f"FAILED: {', '.join(failed)}")
    if data.get("limitations"):
        lines += ["", "## Limits", ""]
        lines += [f"- {item}" for item in data["limitations"]]
    lines.append("")
    return "\n".join(lines)


def run_demo(pkg, manifest):
    from greeneye.elf import load_elf
    from greeneye.inject import InjectedService, invoke_remote, invoke_via_firmware, load_service
    from greeneye.sim_backend import launch_native

    results_dir = os.path.join(pkg, "results")
    target_rel = component(manifest, "target-image")
    service_rel = component(manifest, "service-blob")
    dm_rel = component(manifest, "device-map")
    combined_rel = component(manifest, "combined-elf")
    preinject_rel = component(manifest, "preinject-config")
    target_path = os.path.join(pkg, target_rel)
    service_path = os.path.join(pkg, service_rel)
    device_map_path = os.path.join(pkg, dm_rel) if dm_rel else None
    preinject = load_json(os.path.join(pkg, preinject_rel)) if preinject_rel else None
    preinjected = bool(preinject and preinject.get("runnable", True))
    kernel = os.path.join(pkg, combined_rel) if preinjected else target_path

    dispatch_total = [0]
    watch_total = [0]
    dispatch_hits = []
    watch_hits = []
    data = {
        "schema": "greeneye-trial-results/1",
        "backend": "native",
        "target": {"kind": "demo", "file": target_rel, "sha256": manifest["target"]["sha256"]},
        "session": "results/session",
        "service": {
            "file": service_rel,
            "sha256": manifest["service"]["sha256"],
            "slot": manifest["service"]["slot"],
            "preinjected": preinjected,
        },
    }
    checks = {}
    session = None
    try:
        session = launch_native(
            kernel=kernel,
            record=os.path.join(results_dir, "session"),
            name="trial-demo",
            memory_mb=int(manifest["scenario"].get("memory_mb", 256)),
            device_map=device_map_path,
            unknown_mmio="lenient" if device_map_path else "fault",
        )
        if preinjected:
            session.symbols.load_elf(load_elf(target_path))
        ctrl = session.symbols.lookup("gsp_ctrl")
        if preinjected:
            session.memory.write_u64(
                ctrl + CTRL_HANDLERS + int(preinject["slot"]) * 8, int(preinject["entry_addr"])
            )
            service = InjectedService(
                path=service_rel,
                name=os.path.basename(service_rel),
                load_addr=int(preinject["service_load_addr"]),
                entry_addr=int(preinject["entry_addr"]),
                dispatch_offset=int(preinject["service_entry"]),
                slot=int(preinject["slot"]),
                size=int(preinject["size"]),
            )
            session.recorder.event(
                "service_injected",
                path=service.path,
                name=service.name,
                load_addr=service.load_addr,
                entry_addr=service.entry_addr,
                slot=service.slot,
                size=service.size,
                preinjected=True,
            )
        else:
            service = load_service(session, service_path, slot=int(manifest["service"]["slot"]))

        def on_dispatch(hit):
            dispatch_total[0] += 1
            if len(dispatch_hits) < 64:
                dispatch_hits.append(
                    {
                        "pc": hexv(hit.pc),
                        "cmd": hexv(hit.arg(0)),
                        "a0": hexv(hit.arg(1)),
                        "a1": hexv(hit.arg(2)),
                    }
                )

        def on_result(hit):
            watch_total[0] += 1
            if len(watch_hits) < 64:
                watch_hits.append(
                    {"pc": hexv(hit.pc), "symbol": hit.symbol.name if hit.symbol else None}
                )

        session.hook("rpc_dispatch", on_enter=on_dispatch)
        session.watch(ctrl + CTRL_RESULT, 8, "w", callback=on_result)
        session.run(duration=float(manifest["scenario"]["demo_boot_seconds"]))

        remote_ping = invoke_remote(session, service, 1, PING_ARG)
        remote_sum = invoke_remote(session, service, 2, SUM_A0, SUM_A1)
        fw_ping = invoke_via_firmware(session, service, 1, PING_ARG)
        fw_sum = invoke_via_firmware(session, service, 2, SUM_A0, SUM_A1)

        handlers = session.memory.read_u64(ctrl + CTRL_HANDLERS + service.slot * 8)
        result_value = session.memory.read_u64(ctrl + CTRL_RESULT)
        rpc_count = session.memory.read_u64(ctrl + CTRL_RPC_COUNT)
        uart = session.target.read_uart()

        data["service"].update(
            {"load_addr": hexv(service.load_addr), "entry_addr": hexv(service.entry_addr)}
        )
        data["calls"] = {
            "remote": {"ping": remote_ping, "sum": remote_sum},
            "via_firmware": {"ping": fw_ping, "sum": fw_sum},
            "expected": {"ping": EXPECTED_PING, "sum": EXPECTED_SUM},
            "agree": remote_ping == fw_ping and remote_sum == fw_sum,
        }
        data["handlers_slot"] = handlers
        data["control"] = {"result": result_value, "rpc_count": rpc_count}
        data["hooks"] = {"rpc_dispatch": {"hits": dispatch_total[0], "first": dispatch_hits}}
        data["watches"] = {"gsp_ctrl.result": {"hits": watch_total[0], "first": watch_hits}}
        data["uart_excerpt"] = uart_excerpt(uart)
        checks = {
            "handlers_slot_registered": handlers == service.entry_addr,
            "remote_ping": remote_ping == EXPECTED_PING,
            "remote_sum": remote_sum == EXPECTED_SUM,
            "via_firmware_ping": fw_ping == EXPECTED_PING,
            "via_firmware_sum": fw_sum == EXPECTED_SUM,
            "routes_agree": remote_ping == fw_ping and remote_sum == fw_sum,
            "dispatch_hook_hits": dispatch_total[0] >= 1,
            "result_watch_hits": watch_total[0] >= 1,
            "uart_banner": UART_BANNER in uart,
            "uart_selftest": UART_SELFTEST in uart,
        }
    except Exception as exc:  # record the honest failure, never fake a pass
        data["error"] = f"{type(exc).__name__}: {exc}"
        checks = {"no_exception": False}
    finally:
        if session is not None:
            try:
                session.close()
            except Exception:
                pass

    failures = sorted(name for name, ok in checks.items() if not ok)
    data["status"] = "ok" if not failures else "failed"
    data["success"] = not failures
    data["assertions"] = {"checks": checks, "failed": failures}
    data["limitations"] = manifest["target"].get("limitations", [])
    write_json(os.path.join(results_dir, "results.json"), data)
    write_text(os.path.join(results_dir, "results.md"), render_md(manifest, data))
    print(f"trial: {data['status']} (results/results.json)")
    return 0 if not failures else 1


def run_image(pkg, manifest, budget=None):
    from greeneye.sim_backend import launch_native

    results_dir = os.path.join(pkg, "results")
    target_rel = component(manifest, "target-image")
    service_rel = component(manifest, "service-blob")
    dm_rel = component(manifest, "device-map")
    scenario = manifest["scenario"]
    target_path = os.path.join(pkg, target_rel)
    dm_path = os.path.join(pkg, dm_rel) if dm_rel else None
    budget = int(budget if budget is not None else scenario["image_budget_instructions"])
    bias = int(manifest["target"].get("bias", 0x80000000))
    learn = bool(manifest.get("device_map") and manifest["device_map"].get("learn"))

    data = {
        "schema": "greeneye-trial-results/1",
        "backend": "native",
        "target": {
            "kind": "image",
            "file": target_rel,
            "sha256": manifest["target"]["sha256"],
            "entry": manifest["target"]["entry"],
            "bias": bias,
            "runnable": False,
        },
        "service": {
            "file": service_rel,
            "sha256": manifest["service"]["sha256"],
            "loaded": False,
            "reason": (
                "no RM service registration point exists yet (M5); loading the blob "
                "would overwrite the image's own RAM"
            ),
        },
        "session": "results/session",
        "budget_instructions": budget,
        "preinject": manifest.get("preinject"),
        "limitations": manifest["target"].get("limitations", []),
        "blocked_reason": (
            "the native simulator cannot boot the real GSP image yet: the H2.2 device "
            "model is absent. This is a bounded lenient first-touch probe, not a boot."
        ),
    }
    checks = {"bounded_lenient_run": False}
    session = None
    try:
        session = launch_native(
            kernel=target_path,
            memory_mb=int(scenario.get("memory_mb", 256)),
            record=os.path.join(results_dir, "session"),
            name="trial-image",
            device_map=dm_path,
            unknown_mmio="lenient",
            learn_mmio=learn,
            bias=bias,
        )
        session.run(duration=budget / 1_000_000.0)
        machine = session.target.machine
        pc = machine.pc
        data["instructions_executed"] = machine.insns
        data["stopped"] = {"pc": pc, "fw_pc": pc - bias if pc >= bias else pc}
        data["mmio_unknown"] = machine.mmio_unknown
        data["first_unknown_mmio"] = [
            {"pc": hexv(entry["pc"]), "addr": hexv(entry["addr"]),
             "size": entry["size"], "rw": entry["rw"]}
            for entry in machine.mmio_log[:8]
        ]
        data["uart_excerpt"] = uart_excerpt(session.target.read_uart())
        if learn and machine.device_map is not None:
            machine.device_map.save(os.path.join(results_dir, "learned_device_map.json"))
            data["learned_device_map"] = "results/learned_device_map.json"
        checks = {"bounded_lenient_run": True, "booted": False}
    except Exception as exc:
        data["error"] = f"{type(exc).__name__}: {exc}"
    finally:
        if session is not None:
            try:
                session.close()
            except Exception:
                pass

    data["status"] = "blocked"
    data["success"] = False
    data["blocked_on_device_model"] = True
    data["assertions"] = {"checks": checks, "failed": []}
    write_json(os.path.join(results_dir, "results.json"), data)
    write_text(os.path.join(results_dir, "results.md"), render_md(manifest, data))
    print(f"trial: blocked (results/results.json): {data['blocked_reason']}")
    return 0 if "error" not in data else 1


def main(argv=None):
    parser = argparse.ArgumentParser(description="Run a packaged risc-v-gsp-rm-debugger trial.")
    parser.add_argument("--package", default=os.path.dirname(os.path.abspath(__file__)))
    parser.add_argument("--budget", type=int, default=None,
                        help="image probe budget in instructions (default from manifest)")
    args = parser.parse_args(argv)
    pkg = os.path.abspath(args.package)
    bootstrap_tool_root(pkg)
    manifest = load_json(os.path.join(pkg, "manifest.json"))
    if manifest["target"]["kind"] == "image":
        return run_image(pkg, manifest, budget=args.budget)
    return run_demo(pkg, manifest)


if __name__ == "__main__":
    raise SystemExit(main())
'''


if __name__ == "__main__":
    raise SystemExit(main())