debug: green-eye porting automation - new-device and driver-version port scripts, simulator packaging, porting docs; control-discovery fix

4 files changed, 3058 insertions, 2 deletions

diff --git a/debug/risc-v-gsp-rm-vm/harness/control_discovery.py b/debug/risc-v-gsp-rm-vm/harness/control_discovery.py
index eecf082..b44636a 100644
--- a/debug/risc-v-gsp-rm-vm/harness/control_discovery.py
+++ b/debug/risc-v-gsp-rm-vm/harness/control_discovery.py
@@ -115,6 +115,10 @@ SETTLE_MARKER = "SELFTEST DONE"
 SETTLE_SLICE = 0.02
 RUN_SLICE = 0.02
 TRAP_GRACE = 0.05
+#: Bound on instruction-driven single-steps used to retire a dispatcher's
+#: store sequence after the wall-clock observation window expires mid-epilogue
+#: (keeps findings byte-reproducible instead of timer-dependent).
+EPILOGUE_MAX_STEPS = 64
 #: gdb signals that carry a hazard (illegal, trap, bus, segv); SIGINT from the
 #: budget interrupt is wall-clock dependent and must never enter a finding.
 HAZARD_SIGNALS = (4, 5, 7, 11)
@@ -662,12 +666,16 @@ class ProbeRunner:
 
         Completion is a ``result`` store (the dispatcher writes it after every
         handler) or a trap/grace for ids that never return to the dispatcher.
+        The wall-clock loop is the budget for the no-progress/hung cases; when
+        a dispatch has been seen the store sequence is then retired with
+        bounded single-steps so the recorded effects cannot depend on where
+        the timer happened to expire.
         """
         deadline = time.monotonic() + self.options.timeout
         handler_seen_at: float | None = None
         while time.monotonic() < deadline:
             if not session.target.alive():
-                return
+                break
             remaining = max(0.005, min(RUN_SLICE, deadline - time.monotonic()))
             session.run(duration=remaining, max_stops=64)
             if handlers and handler_seen_at is None:
@@ -675,9 +683,37 @@ class ProbeRunner:
             result_written = any(hit.get("field") == "result" for hit in observer.hits)
             trapped = any(sig.get("signal") == 5 for sig in signals) or session.pc < 0x100
             if dispatches and (result_written or trapped):
-                return
+                break
             if handler_seen_at is not None and time.monotonic() - handler_seen_at >= TRAP_GRACE:
                 # Slow/hung handler: one grace window after the entry, then stop.
+                break
+        self._retire_dispatch(session, dispatches, signals, observer)
+
+    def _retire_dispatch(self, session, dispatches, signals, observer) -> None:
+        """Finish a dispatched command's store sequence with single-steps.
+
+        The demo dispatcher stores ``result``, ``last_rpc`` and finally
+        ``rpc_count``; a wall-clock observation window can expire between
+        those stores, which made findings timing-dependent.  Once a dispatch
+        has been seen, step (bounded and instruction-driven) until the final
+        watched store retires, a trap is seen, or the budget is exhausted.
+        """
+        if not dispatches or not any(watched.name == "rpc_count" for watched in observer.watched):
+            return
+        if any(hit.get("field") == "rpc_count" for hit in observer.hits):
+            return
+        if any(sig.get("signal") == 5 for sig in signals):
+            return
+        for _ in range(EPILOGUE_MAX_STEPS):
+            if not session.target.alive():
+                return
+            try:
+                session.step()
+            except (RspTimeout, GreenEyeError):
+                return
+            if any(hit.get("field") == "rpc_count" for hit in observer.hits):
+                return
+            if any(sig.get("signal") == 5 for sig in signals):
                 return
 
     def _keep(self, outcome: str) -> bool:
diff --git a/debug/risc-v-gsp-rm-vm/harness/package_sim.py b/debug/risc-v-gsp-rm-vm/harness/package_sim.py
new file mode 100644
index 0000000..28f54c4
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/package_sim.py
@@ -0,0 +1,1139 @@
+#!/usr/bin/env python3
+"""Package the native green-eye 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
+# green-eye 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: green-eye 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 = [
+        "# green-eye 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 green-eye",
+        "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 green-eye 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 green-eye 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 green-eye 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 = "GREEN-EYE 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 = ["# green-eye 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 green-eye 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())
diff --git a/debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py b/debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py
new file mode 100644
index 0000000..e751d42
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py
@@ -0,0 +1,1155 @@
+#!/usr/bin/env python3
+"""Driver-version port sweep (roadmap H4.1/R6, docs/28).
+
+Given a version list (``--versions 540,550,560,570``) or a JSON config, find
+each version's firmware target under ``--input-dir`` (driver runfiles/firmware
+blobs already on disk -- this tool never downloads), then per version:
+
+1. run ``harness/version_diff.py`` against a chosen baseline;
+2. run ``harness/portability_matrix.py`` on the baseline + candidate pair;
+3. run a bounded ``harness/mutation_harness.py`` smoke;
+4. assemble a per-version evidence bundle and a cross-version summary
+   (markdown + JSON) with the docs/06 M1-M6 skeleton.
+
+``--dry-run`` is the default: it prints the plan (inputs found/missing, steps,
+expected artifacts) and exits 0. ``--run`` executes, is resume-safe (a version
+whose ``version.json`` already reports ok/partial is skipped unless --force)
+and exits non-zero when any executed version has an errored step. The RM 610
+gate from ``version_diff.py`` (docs/23) is enforced before any target is
+opened, in dry-run too: versions >= 610 require ``--approval-token
+RM610-APPROVED`` or ``GREENEYE_RM610_APPROVED=1``.
+
+Known driver branch cadence (edit DEFAULT_VERSIONS or pass --versions/--config;
+no image is ever invented or downloaded):
+
+    branch   expected image
+    540      earlier branch (cadence only)
+    550      earlier branch (cadence only)
+    560      earlier branch (cadence only)
+    570      current anchor, e.g. 570.144
+    580      expected cadence, no local image assumed
+    610      RM 610 gate, refused without approval (docs/23)
+
+Run from the tool root:
+
+    python3 harness/port_driver_versions.py --input-dir ../../firmware/57X-GSP
+    python3 harness/port_driver_versions.py --run --input-dir ../../firmware/57X-GSP \\
+        --versions 540,550,560,570 --out dumps/sweep-570
+"""
+from __future__ import annotations
+
+import argparse
+import contextlib
+import json
+import os
+import re
+import signal
+import subprocess
+import sys
+import threading
+from dataclasses import dataclass, field
+
+try:
+    from .common import (
+        DEMO_ELF,
+        MAX_LOG_BYTES,
+        ensure_dir,
+        qemu_available,
+        sha256_file,
+        write_json,
+        write_text,
+    )
+    from .mutation_harness import HarnessOptions, run_harness
+    from .portability_matrix import (
+        METRIC_DEFS,
+        NO_REFERENCE,
+        MatrixOptions,
+        TargetSpec,
+        render_markdown as render_matrix_markdown,
+        run_matrix,
+    )
+    from .version_diff import (
+        RM610_TOKEN,
+        Rm610GateError,
+        VersionDiffOptions,
+        check_rm610_gate,
+        is_rm610_label,
+        run_version_diff,
+    )
+except ImportError:  # direct script execution (python3 harness/port_driver_versions.py)
+    from common import (  # type: ignore[no-redef]
+        DEMO_ELF,
+        MAX_LOG_BYTES,
+        ensure_dir,
+        qemu_available,
+        sha256_file,
+        write_json,
+        write_text,
+    )
+    from mutation_harness import HarnessOptions, run_harness  # type: ignore[no-redef]
+    from portability_matrix import (  # type: ignore[no-redef]
+        METRIC_DEFS,
+        NO_REFERENCE,
+        MatrixOptions,
+        TargetSpec,
+        render_markdown as render_matrix_markdown,
+        run_matrix,
+    )
+    from version_diff import (  # type: ignore[no-redef]
+        RM610_TOKEN,
+        Rm610GateError,
+        VersionDiffOptions,
+        check_rm610_gate,
+        is_rm610_label,
+        run_version_diff,
+    )
+
+FORMAT = "greeneye-port-sweep/1"
+PLAN_FORMAT = "greeneye-port-sweep-plan/1"
+VERSION_FORMAT = "greeneye-port-version/1"
+BUNDLE_FORMAT = "greeneye-release-bundle/1"
+CONFIG_FORMAT = "greeneye-port-sweep-config/1"
+
+#: Known driver branch cadence; edit here or pass --versions/--config. No image
+#: is downloaded by this tool: branches without local images become no-input.
+DEFAULT_VERSIONS = ("540", "550", "560", "570")
+RESUMABLE_STATUSES = ("ok", "partial")
+TARGET_SUFFIXES = (".fwimage.bin", ".fwimage", ".elf", ".bin", ".run")
+VERSION_TOKEN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+
+
+# ---------------------------------------------------------------------------
+# options + version helpers
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class SweepTarget:
+    version: str
+    paths: list[str] = field(default_factory=list)
+    kind: str = "none"  # fwimage | elf | raw | runfile | explicit | none
+    source: str = "none"  # explicit | discovered | none
+    matches: list[str] = field(default_factory=list)
+    runfiles: list[str] = field(default_factory=list)
+    notes: list[str] = field(default_factory=list)
+    symbols: str | None = None
+    ghidra_exports: str | None = None
+
+    @property
+    def path(self) -> str | None:
+        return self.paths[0] if self.paths else None
+
+
+@dataclass
+class SweepOptions:
+    versions: list[str] = field(default_factory=list)
+    baseline: str | None = None
+    input_dir: str | None = None
+    out: str | None = None
+    config: str | None = None
+    targets: dict[str, list[str]] = field(default_factory=dict)
+    symbols: dict[str, str] = field(default_factory=dict)
+    ghidra_exports: dict[str, str] = field(default_factory=dict)
+    run: bool = False
+    force: bool = False
+    extract_runfiles: bool = False
+    approval_token: str | None = None
+    probe_instructions: int = 2000
+    probe_duration: float = 0.5
+    scenario_duration: float = 0.4
+    max_stops: int = 5000
+    max_rows: int = 40
+    mutation_kernel: str = DEMO_ELF
+    mutation_iterations: int = 8
+    mutation_seed: int = 7
+    mutation_timeout: float = 0.25
+    mutation_settle: float = 0.1
+    allow_foreign_mutation: bool = False
+    step_timeout: float = 600.0
+    log_bytes: int = MAX_LOG_BYTES
+
+    @property
+    def dry_run(self) -> bool:
+        return not self.run
+
+
+def parse_versions(text: str) -> list[str]:
+    """Parse ``540,550,560`` into a validated, de-duplicated order-preserving list."""
+    versions: list[str] = []
+    for token in str(text).replace(";", ",").split(","):
+        value = token.strip()
+        if not value:
+            continue
+        if not VERSION_TOKEN_RE.match(value):
+            raise ValueError(f"invalid version token: {value!r}")
+        if value not in versions:
+            versions.append(value)
+    return versions
+
+
+def version_sort_key(value: str) -> tuple:
+    parts = re.findall(r"\d+", value)
+    return (tuple(int(part) for part in parts) if parts else (0,), value)
+
+
+def version_ge_610(value: str) -> bool:
+    match = re.match(r"\s*(\d+)", str(value))
+    return bool(match) and int(match.group(1)) >= 610
+
+
+def _is_numeric_firmware_match(version: str, text: str) -> bool:
+    pattern = re.compile(r"(?<!\d)" + re.escape(version) + r"(?:\.\d+)*(?!\d)")
+    return pattern.search(text) is not None
+
+
+def _gate_version_labels(versions: list[str]) -> list[str]:
+    """Numeric versions >= 610 must trigger the rm610-prefixed gate regex."""
+    return [f"rm610-{value}" if version_ge_610(value) else value
+            for value in versions if version_ge_610(value) or is_rm610_label(value)]
+
+
+def gate_labels(versions: list[str], targets: dict[str, SweepTarget]) -> list[str]:
+    labels = _gate_version_labels(versions)
+    for target in targets.values():
+        labels.append(target.version)
+        labels.extend(os.path.basename(path) for path in target.paths)
+        labels.extend(os.path.basename(path) for path in target.runfiles)
+    return labels
+
+
+# ---------------------------------------------------------------------------
+# config loading
+# ---------------------------------------------------------------------------
+
+
+def load_config(path: str) -> dict:
+    """Read a sweep config; strict enough that typos do not silently pass."""
+    if not os.path.isfile(path):
+        raise ValueError(f"config not found: {path}")
+    try:
+        with open(path, "r", encoding="utf-8") as fh:
+            data = json.load(fh)
+    except (OSError, json.JSONDecodeError) as exc:
+        raise ValueError(f"config {path}: {type(exc).__name__}: {exc}") from exc
+    if not isinstance(data, dict):
+        raise ValueError(f"config {path}: expected a JSON object")
+    known = {"format", "versions", "baseline", "input_dir", "targets", "symbols",
+             "ghidra_exports", "notes"}
+    unknown = sorted(set(data) - known)
+    if unknown:
+        raise ValueError(f"config {path}: unknown key(s): {', '.join(unknown)}")
+    if "format" in data and data["format"] != CONFIG_FORMAT:
+        raise ValueError(f"config {path}: unsupported format {data['format']!r}")
+    if "versions" in data:
+        if not isinstance(data["versions"], list):
+            raise ValueError(f"config {path}: versions must be a list")
+        data["versions"] = [str(value) for value in data["versions"]]
+    for key in ("targets", "symbols", "ghidra_exports"):
+        if key not in data:
+            continue
+        if not isinstance(data[key], dict):
+            raise ValueError(f"config {path}: {key} must be an object")
+        normalised: dict[str, object] = {}
+        for version, value in data[key].items():
+            if key == "targets":
+                if isinstance(value, str):
+                    value = [value]
+                if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
+                    raise ValueError(f"config {path}: targets[{version}] must be a path or list of paths")
+            elif not isinstance(value, str):
+                raise ValueError(f"config {path}: {key}[{version}] must be a string path")
+            normalised[str(version)] = value
+        data[key] = normalised
+    return data
+
+
+def options_from_config(options: SweepOptions, config: dict) -> None:
+    if config.get("versions"):
+        options.versions = list(config["versions"])
+    if config.get("baseline") and not options.baseline:
+        options.baseline = str(config["baseline"])
+    if config.get("input_dir") and not options.input_dir:
+        options.input_dir = str(config["input_dir"])
+    for version, paths in (config.get("targets") or {}).items():
+        options.targets.setdefault(str(version), []).extend(paths)
+    for version, path in (config.get("symbols") or {}).items():
+        options.symbols.setdefault(str(version), path)
+    for version, path in (config.get("ghidra_exports") or {}).items():
+        options.ghidra_exports.setdefault(str(version), path)
+
+
+# ---------------------------------------------------------------------------
+# target discovery (local disk only)
+# ---------------------------------------------------------------------------
+
+
+def _kind_for(name: str) -> str | None:
+    lower = name.lower()
+    if not any(lower.endswith(suffix) for suffix in TARGET_SUFFIXES):
+        return None
+    if lower.endswith(".run"):
+        return "runfile"
+    if lower.endswith(".fwimage.bin") or lower.endswith(".fwimage"):
+        return "fwimage"
+    if lower.endswith(".elf"):
+        return "elf"
+    if lower.endswith(".bin"):
+        return "raw"
+    return None
+
+
+def discover_version(input_dir: str, version: str) -> SweepTarget:
+    """Find the best local firmware target for one version under ``input_dir``."""
+    target = SweepTarget(version=version)
+    if not input_dir:
+        target.notes.append("no --input-dir supplied and no explicit target")
+        return target
+    if not os.path.isdir(input_dir):
+        target.notes.append(f"input directory not found: {input_dir}")
+        return target
+    ranked: list[tuple[int, int, str, str]] = []
+    for root, dirs, names in os.walk(input_dir):
+        dirs.sort()
+        for name in sorted(names):
+            kind = _kind_for(name)
+            if kind is None:
+                continue
+            path = os.path.join(root, name)
+            rel = os.path.relpath(path, input_dir)
+            parent = os.path.basename(root)
+            if name and _is_numeric_firmware_match(version, name):
+                scope = 0
+            elif parent and _is_numeric_firmware_match(version, parent):
+                scope = 1
+            elif _is_numeric_firmware_match(version, rel):
+                scope = 2
+            else:
+                continue
+            kind_rank = {"fwimage": 0, "elf": 1, "raw": 2, "runfile": 3}[kind]
+            ranked.append((kind_rank, int(scope), path, kind))
+    ranked.sort()
+    for _, _, path, kind in ranked:
+        if kind == "runfile":
+            target.runfiles.append(path)
+        else:
+            target.paths.append(path)
+            target.matches.append(path)
+            if target.kind == "none":
+                target.kind = kind
+                target.source = "discovered"
+    if not target.paths and target.runfiles:
+        target.notes.append(
+            "driver runfile(s) present but not extracted; pass --extract-runfiles "
+            "(local extraction only, nothing is downloaded) or extract with the "
+            "driver runfile's own --extract-only"
+        )
+    if not target.paths and not target.runfiles:
+        target.notes.append(f"no local image matching {version!r} under {input_dir}")
+    return target
+
+
+def resolve_targets(options: SweepOptions) -> dict[str, SweepTarget]:
+    """Resolve every requested version plus an out-of-list baseline when given."""
+    versions = list(options.versions)
+    if options.baseline and options.baseline not in versions:
+        versions.append(options.baseline)
+    resolved: dict[str, SweepTarget] = {}
+    for version in versions:
+        explicit = [path for path in options.targets.get(version, [])]
+        target = SweepTarget(version=version)
+        target.symbols = options.symbols.get(version)
+        target.ghidra_exports = options.ghidra_exports.get(version)
+        if explicit:
+            target.source = "explicit"
+            for path in explicit:
+                if os.path.isfile(path):
+                    target.paths.append(os.path.abspath(path))
+                    target.matches.append(path)
+                else:
+                    target.notes.append(f"explicit target not found: {path}")
+            if target.paths:
+                target.kind = _kind_for(os.path.basename(target.paths[0])) or "explicit"
+        else:
+            target = discover_version(options.input_dir or "", version)
+            target.symbols = options.symbols.get(version)
+            target.ghidra_exports = options.ghidra_exports.get(version)
+        resolved[version] = target
+    return resolved
+
+
+def choose_baseline(options: SweepOptions, resolved: dict[str, SweepTarget]) -> str | None:
+    if options.baseline:
+        return options.baseline
+    found = [version for version in options.versions if resolved.get(version) and resolved[version].paths]
+    return min(found, key=version_sort_key) if found else None
+
+
+# ---------------------------------------------------------------------------
+# bounded steps
+# ---------------------------------------------------------------------------
+
+
+@contextlib.contextmanager
+def _deadline(seconds: float, label: str):
+    """Wall-clock safety net around one step (main thread only)."""
+    usable = (
+        seconds > 0
+        and hasattr(signal, "SIGALRM")
+        and threading.current_thread() is threading.main_thread()
+    )
+    if not usable:
+        yield
+        return
+
+    def _raise(signum, frame):  # pragma: no cover - signal trampoline
+        raise TimeoutError(f"{label} exceeded the {seconds:g}s step budget")
+
+    previous = signal.signal(signal.SIGALRM, _raise)
+    previous_timer = signal.setitimer(signal.ITIMER_REAL, seconds)
+    try:
+        yield
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, *previous_timer)
+        signal.signal(signal.SIGALRM, previous)
+
+
+def _run_step(name: str, fn, *args, timeout: float = 0.0, **kwargs) -> dict:
+    try:
+        with _deadline(timeout, name):
+            detail = fn(*args, **kwargs)
+    except TimeoutError as exc:
+        return {"name": name, "status": "error", "reason": str(exc), "timeout": True}
+    except Exception as exc:  # a failed step must not hide the others
+        return {"name": name, "status": "error", "reason": f"{type(exc).__name__}: {exc}"}
+    entry = {"name": name}
+    entry.update(detail)
+    return entry
+
+
+# ---------------------------------------------------------------------------
+# per-version steps
+# ---------------------------------------------------------------------------
+
+
+def step_version_diff(version: str, baseline: str, target: SweepTarget, baseline_target: SweepTarget,
+                      options: SweepOptions, out_dir: str) -> dict:
+    if target.path is None:
+        return {"status": "skipped", "reason": "no target"}
+    if baseline_target.path is None:
+        return {"status": "error", "reason": f"baseline {baseline!r} has no target"}
+    report = run_version_diff(
+        VersionDiffOptions(
+            old=baseline_target.path,
+            new=target.path,
+            out=out_dir,
+            old_label=baseline,
+            new_label=version,
+            old_symbols=baseline_target.symbols,
+            new_symbols=target.symbols,
+            old_ghidra_exports=baseline_target.ghidra_exports,
+            new_ghidra_exports=target.ghidra_exports,
+            probe_instructions=options.probe_instructions,
+            probe_duration=options.probe_duration,
+            behavior_duration=options.scenario_duration,
+            max_stops=options.max_stops,
+            max_rows=options.max_rows,
+            log_bytes=options.log_bytes,
+            approval_token=options.approval_token,
+        )
+    )
+    sections = {name: section["status"] for name, section in report["sections"].items()}
+    behavior = report["sections"]["behavior"]
+    metrics = behavior.get("metrics") or []
+    if not metrics:
+        metrics = [
+            {"id": metric_id, "name": name, "status": "not measured", "reason": NO_REFERENCE,
+             "a_observed": None, "b_observed": None}
+            for metric_id, name in METRIC_DEFS
+        ]
+    return {
+        "status": "error" if "error" in sections.values() else "ok",
+        "report": "version-diff/version-diff.json",
+        "markdown": "version-diff/version-diff.md",
+        "sections": sections,
+        "all_zero": behavior.get("all_zero"),
+        "metrics": metrics,
+    }
+
+
+def step_portability_matrix(version: str, baseline: str, target: SweepTarget, baseline_target: SweepTarget,
+                            options: SweepOptions, out_dir: str) -> dict:
+    if target.path is None:
+        return {"status": "skipped", "reason": "no target"}
+    if baseline_target.path is None:
+        return {"status": "error", "reason": f"baseline {baseline!r} has no target"}
+    matrix = run_matrix(
+        [TargetSpec(path=baseline_target.path, name=baseline),
+         TargetSpec(path=target.path, name=version)],
+        MatrixOptions(
+            out=out_dir,
+            probe_instructions=options.probe_instructions,
+            probe_duration=options.probe_duration,
+            scenario_duration=options.scenario_duration,
+            max_stops=options.max_stops,
+            max_rows=options.max_rows,
+            log_bytes=options.log_bytes,
+        ),
+    )
+    write_json(os.path.join(out_dir, "portability-report.json"), matrix)
+    write_text(
+        os.path.join(out_dir, "portability-report.md"),
+        render_matrix_markdown(matrix, MatrixOptions(out=out_dir, max_rows=options.max_rows)),
+    )
+    statuses = {row["name"]: row["status"] for row in matrix["targets"]}
+    comparisons = [
+        {"a": row["a"], "b": row["b"], "status": row["status"], "all_zero": row["all_zero"]}
+        for row in matrix["comparisons"]
+    ]
+    return {
+        "status": "error" if "error" in statuses.values() else "ok",
+        "report": "portability-matrix/portability-report.json",
+        "markdown": "portability-matrix/portability-report.md",
+        "targets": statuses,
+        "comparisons": comparisons,
+    }
+
+
+def step_mutation_smoke(target: SweepTarget, options: SweepOptions, out_dir: str) -> dict:
+    if not qemu_available():
+        return {"status": "skipped", "reason": "qemu-system-riscv64 not found"}
+    kernel = options.mutation_kernel
+    if not os.path.isfile(kernel):
+        return {"status": "skipped", "reason": f"mutation kernel not found: {kernel}"}
+    findings = run_harness(
+        HarnessOptions(
+            kernel=os.path.abspath(kernel),
+            out=out_dir,
+            iterations=options.mutation_iterations,
+            seed=options.mutation_seed,
+            timeout=options.mutation_timeout,
+            settle=options.mutation_settle,
+            allow_foreign_elf=options.allow_foreign_mutation,
+        )
+    )
+    outcomes: dict[str, int] = {}
+    for finding in findings:
+        outcomes[finding["outcome"]] = outcomes.get(finding["outcome"], 0) + 1
+    return {
+        "status": "ok",
+        "report": "mutation-smoke/findings.jsonl",
+        "markdown": "mutation-smoke/summary.md",
+        "iterations": len(findings),
+        "outcomes": dict(sorted(outcomes.items())),
+    }
+
+
+def version_status(steps: list[dict]) -> str:
+    statuses = [step["status"] for step in steps]
+    if "error" in statuses:
+        return "error"
+    if all(status == "ok" for status in statuses):
+        return "ok"
+    return "partial"
+
+
+def build_bundle(out_dir: str, tool: str) -> dict:
+    files: list[dict] = []
+    for root, dirs, names in os.walk(out_dir):
+        dirs.sort()
+        for name in sorted(names):
+            path = os.path.join(root, name)
+            rel = os.path.relpath(path, out_dir)
+            if rel == "bundle-manifest.json":
+                continue
+            files.append({"path": rel, "bytes": os.path.getsize(path), "sha256": sha256_file(path)})
+    files.sort(key=lambda row: row["path"])
+    return {"format": BUNDLE_FORMAT, "tool": tool, "files": files}
+
+
+def completed_status(version_dir: str) -> str | None:
+    """Return the recorded status of a finished version (resume support)."""
+    path = os.path.join(version_dir, "version.json")
+    if not os.path.isfile(path):
+        return None
+    try:
+        with open(path, "r", encoding="utf-8") as fh:
+            data = json.load(fh)
+    except (OSError, json.JSONDecodeError):
+        return None
+    status = data.get("status")
+    return status if status in RESUMABLE_STATUSES else None
+
+
+# ---------------------------------------------------------------------------
+# sweep execution
+# ---------------------------------------------------------------------------
+
+
+def run_version(version: str, baseline: str, target: SweepTarget, baseline_target: SweepTarget,
+                options: SweepOptions, out_root: str) -> dict:
+    version_dir = os.path.join(out_root, version)
+    entry: dict = {
+        "format": VERSION_FORMAT,
+        "tool": "harness/port_driver_versions.py",
+        "version": version,
+        "baseline": baseline,
+        "target": target.path,
+        "target_kind": target.kind,
+        "target_source": target.source,
+        "target_matches": target.matches,
+        "runfiles": target.runfiles,
+        "notes": target.notes,
+        "symbols": target.symbols,
+        "steps": [],
+        "metrics": [],
+        "status": "no-input",
+    }
+    if target.path is None:
+        entry["reason"] = (
+            "no local firmware target found; "
+            + ("; ".join(target.notes) if target.notes else "supply --input-dir or --target VERSION=PATH")
+        )
+        ensure_dir(version_dir)
+        write_json(os.path.join(version_dir, "version.json"), entry)
+        write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
+        return entry
+
+    steps = [
+        _run_step("version_diff", step_version_diff, version, baseline, target, baseline_target,
+                  options, os.path.join(version_dir, "version-diff"), timeout=options.step_timeout),
+        _run_step("portability_matrix", step_portability_matrix, version, baseline, target,
+                  baseline_target, options, os.path.join(version_dir, "portability-matrix"),
+                  timeout=options.step_timeout),
+        _run_step("mutation_smoke", step_mutation_smoke, target, options,
+                  os.path.join(version_dir, "mutation-smoke"), timeout=options.step_timeout),
+    ]
+    entry["steps"] = steps
+    entry["status"] = version_status(steps)
+    for step in steps:
+        if step["name"] == "version_diff":
+            entry["metrics"] = step.get("metrics") or []
+            entry["all_zero"] = step.get("all_zero")
+    entry["target_sha256"] = sha256_file(target.path)
+    entry["evidence"] = {"manifest": os.path.join(version, "bundle-manifest.json")}
+    ensure_dir(version_dir)
+    write_json(os.path.join(version_dir, "version.json"), entry)
+    write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
+    manifest = build_bundle(version_dir, "harness/port_driver_versions.py")
+    write_json(os.path.join(version_dir, "bundle-manifest.json"), manifest)
+    return entry
+
+
+def run_sweep(options: SweepOptions) -> dict:
+    check_rm610_gate(
+        _gate_version_labels(options.versions),
+        approval_token=options.approval_token,
+    )
+    resolved = resolve_targets(options)
+    baseline = choose_baseline(options, resolved)
+    labels = gate_labels(list(options.versions), resolved)
+    check_rm610_gate(labels, approval_token=options.approval_token)
+    if not options.out:
+        raise ValueError("--out is required unless --dry-run is used")
+    out_root = os.path.abspath(options.out)
+    ensure_dir(out_root)
+    versions: list[dict] = []
+    for version in options.versions:
+        target = resolved[version]
+        version_dir = os.path.join(out_root, version)
+        previous = None if options.force else completed_status(version_dir)
+        if previous is not None:
+            versions.append({
+                "format": VERSION_FORMAT,
+                "tool": "harness/port_driver_versions.py",
+                "version": version,
+                "baseline": baseline,
+                "status": previous,
+                "resumed": True,
+                "target": target.path,
+                "evidence": {"manifest": os.path.relpath(
+                    os.path.join(version_dir, "bundle-manifest.json"), out_root)},
+            })
+            continue
+        if target.path is None and options.extract_runfiles and target.runfiles:
+            extract_dir = os.path.join(version_dir, "runfile-extract")
+            try:
+                target = extract_runfiles(version, target, extract_dir, options)
+            except Exception as exc:
+                entry = {
+                    "format": VERSION_FORMAT,
+                    "tool": "harness/port_driver_versions.py",
+                    "version": version,
+                    "baseline": baseline,
+                    "target": None,
+                    "steps": [],
+                    "metrics": [],
+                    "status": "error",
+                    "reason": f"runfile extraction failed: {type(exc).__name__}: {exc}",
+                }
+                ensure_dir(version_dir)
+                write_json(os.path.join(version_dir, "version.json"), entry)
+                write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
+                versions.append(entry)
+                continue
+        baseline_target = resolved.get(baseline) if baseline else None
+        if baseline_target is None:
+            entry = {
+                "format": VERSION_FORMAT,
+                "tool": "harness/port_driver_versions.py",
+                "version": version,
+                "baseline": None,
+                "target": target.path,
+                "target_kind": target.kind,
+                "target_source": target.source,
+                "steps": [],
+                "metrics": [],
+                "status": "error",
+                "reason": "no baseline target: pass --baseline or ensure one version has an image",
+            }
+            ensure_dir(version_dir)
+            write_json(os.path.join(version_dir, "version.json"), entry)
+            write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
+            versions.append(entry)
+            continue
+        versions.append(run_version(version, baseline, target, baseline_target, options, out_root))
+
+    failed = [row["version"] for row in versions if row["status"] == "error"]
+    not_run = [row["version"] for row in versions if row["status"] == "no-input"]
+    ran = len(versions) - len(not_run)
+    if failed:
+        interpretation = (
+            "fail: " + ", ".join(failed)
+            + " had an errored step; inspect each version.json and the step reports"
+        )
+    elif ran == 0:
+        interpretation = (
+            "not-run: no version had a local target; nothing was executed and no evidence was produced"
+        )
+    else:
+        interpretation = (
+            "pass: every executed version completed with no errored step; "
+            "skipped steps and resume decisions are listed per version"
+        )
+    report = {
+        "format": FORMAT,
+        "tool": "harness/port_driver_versions.py",
+        "dry_run": False,
+        "rm610_gate": {"status": "passed", "labels": sorted(set(labels))},
+        "baseline": {"version": baseline,
+                     "path": resolved[baseline].path if baseline and baseline in resolved else None},
+        "input_dir": options.input_dir,
+        "versions": versions,
+        "ran": ran,
+        "failed": failed,
+        "not_run": not_run,
+        "interpretation": interpretation,
+        "summary": "sweep-summary.json",
+        "markdown": "sweep-summary.md",
+        "manifest": "bundle-manifest.json",
+    }
+    write_json(os.path.join(out_root, "sweep-summary.json"), report)
+    write_text(os.path.join(out_root, "sweep-summary.md"), render_sweep_markdown(report))
+    manifest = build_bundle(out_root, "harness/port_driver_versions.py")
+    write_json(os.path.join(out_root, "bundle-manifest.json"), manifest)
+    return report
+
+
+def extract_runfiles(version: str, target: SweepTarget, extract_dir: str, options: SweepOptions) -> SweepTarget:
+    """Extract local driver runfile(s) with their own --extract-only (no download)."""
+    ensure_dir(extract_dir)
+    for runfile in target.runfiles:
+        argv = ["sh", os.path.abspath(runfile), "--extract-only", "--target", extract_dir]
+        subprocess.run(argv, check=True, timeout=options.step_timeout,
+                       stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+    discovered = discover_version(extract_dir, version)
+    discovered.symbols = target.symbols
+    discovered.ghidra_exports = target.ghidra_exports
+    discovered.notes = target.notes + [f"extracted from {len(target.runfiles)} runfile(s)"]
+    discovered.runfiles = target.runfiles
+    return discovered
+
+
+# ---------------------------------------------------------------------------
+# plan (dry run)
+# ---------------------------------------------------------------------------
+
+
+def _expected_artifacts(out: str, version: str) -> list[str]:
+    base = os.path.join(out, version)
+    return [
+        os.path.join(base, "version.json"),
+        os.path.join(base, "version.md"),
+        os.path.join(base, "bundle-manifest.json"),
+        os.path.join(base, "version-diff", "version-diff.json"),
+        os.path.join(base, "version-diff", "version-diff.md"),
+        os.path.join(base, "portability-matrix", "portability-report.json"),
+        os.path.join(base, "portability-matrix", "portability-report.md"),
+        os.path.join(base, "mutation-smoke", "findings.jsonl"),
+        os.path.join(base, "mutation-smoke", "summary.md"),
+    ]
+
+
+def plan_sweep(options: SweepOptions) -> dict:
+    check_rm610_gate(
+        _gate_version_labels(options.versions),
+        approval_token=options.approval_token,
+    )
+    resolved = resolve_targets(options)
+    baseline = choose_baseline(options, resolved)
+    labels = gate_labels(list(options.versions), resolved)
+    check_rm610_gate(labels, approval_token=options.approval_token)
+    out = os.path.abspath(options.out) if options.out else "<out>"
+    versions = []
+    found = missing = 0
+    for version in options.versions:
+        target = resolved[version]
+        entry = {
+            "version": version,
+            "status": "found" if target.path else ("runfile" if target.runfiles else "no-input"),
+            "target": target.path,
+            "kind": target.kind,
+            "source": target.source,
+            "matches": target.matches,
+            "runfiles": target.runfiles,
+            "notes": target.notes,
+            "baseline": version == baseline,
+            "symbols": target.symbols,
+            "steps": [
+                {"name": "version_diff", "out": os.path.join(out, version, "version-diff"),
+                 "old": resolved[baseline].path if baseline and baseline in resolved else None,
+                 "new": target.path},
+                {"name": "portability_matrix", "out": os.path.join(out, version, "portability-matrix"),
+                 "targets": [baseline, version] if baseline else [version]},
+                {"name": "mutation_smoke", "out": os.path.join(out, version, "mutation-smoke"),
+                 "kernel": options.mutation_kernel},
+            ],
+            "expected_artifacts": _expected_artifacts(out, version),
+        }
+        if options.extract_runfiles and target.runfiles and not target.path:
+            entry["extract"] = [
+                {"argv": ["sh", os.path.abspath(runfile), "--extract-only", "--target",
+                          os.path.join(out, version, "runfile-extract")]}
+                for runfile in target.runfiles
+            ]
+        if target.path:
+            found += 1
+        else:
+            missing += 1
+        versions.append(entry)
+    return {
+        "format": PLAN_FORMAT,
+        "tool": "harness/port_driver_versions.py",
+        "dry_run": True,
+        "rm610_gate": {"status": "passed", "labels": sorted(set(labels))},
+        "input_dir": options.input_dir,
+        "out": out,
+        "baseline": baseline,
+        "versions": versions,
+        "found": found,
+        "missing": missing,
+        "expected_artifacts": [
+            os.path.join(out, "sweep-summary.json"),
+            os.path.join(out, "sweep-summary.md"),
+            os.path.join(out, "bundle-manifest.json"),
+        ],
+        "interpretation": (
+            f"dry-run: {found} input(s) found, {missing} missing; "
+            "pass --run to execute, --force to re-run completed versions"
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+# rendering
+# ---------------------------------------------------------------------------
+
+
+def _table(headers: list[str], rows: list[list]) -> list[str]:
+    lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"]
+    for row in rows:
+        lines.append("| " + " | ".join(str(cell) for cell in row) + " |")
+    return lines
+
+
+def render_version_markdown(entry: dict) -> str:
+    lines = [
+        f"# green-eye port version {entry['version']}",
+        "",
+        "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_driver_versions.py`.",
+        "",
+        f"* status: **{entry['status']}**",
+        f"* baseline: `{entry.get('baseline')}`",
+        f"* target: `{entry.get('target')}`",
+    ]
+    if entry.get("target_sha256"):
+        lines.append(f"* target sha256: `{entry['target_sha256']}`")
+    if entry.get("reason"):
+        lines.append(f"* reason: {entry['reason']}")
+    lines += ["", "## Steps", ""]
+    lines += _table(
+        ["step", "status", "report", "reason"],
+        [[step["name"], step["status"], step.get("report") or "-", step.get("reason") or "-"]
+         for step in entry.get("steps") or []] or [["-", "-", "-", "not run"]],
+    )
+    lines += ["", "## M1-M6 (docs/06)", "", f"> {NO_REFERENCE}", ""]
+    lines += _table(
+        ["#", "metric", "status", "observed"],
+        [[row["id"], row["name"], row["status"],
+          json.dumps({"a": row.get("a_observed"), "b": row.get("b_observed")}, sort_keys=True)]
+         for row in entry.get("metrics") or []],
+    )
+    lines += [""]
+    return "\n".join(lines)
+
+
+def render_sweep_markdown(report: dict) -> str:
+    lines = [
+        "# green-eye driver-version port sweep",
+        "",
+        "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_driver_versions.py`.",
+        "",
+        f"* format: `{report['format']}`",
+        f"* baseline: `{report['baseline']['version']}` (`{report['baseline']['path']}`)",
+        f"* input dir: `{report.get('input_dir')}`",
+        f"* ran: {report['ran']}, failed: {len(report['failed'])}, no input: {len(report['not_run'])}",
+        "",
+        f"> **{report['interpretation']}**",
+        "",
+        "## Versions",
+        "",
+    ]
+    lines += _table(
+        ["version", "status", "target", "version-diff", "matrix", "mutation", "evidence"],
+        [
+            [
+                row["version"],
+                row["status"] + (" (resumed)" if row.get("resumed") else ""),
+                f"`{(row.get('target') or '-')}`",
+                _step_status(row, "version_diff"),
+                _step_status(row, "portability_matrix"),
+                _step_status(row, "mutation_smoke"),
+                row.get("evidence", {}).get("manifest", "-"),
+            ]
+            for row in report["versions"]
+        ],
+    )
+    lines += ["", "## M1-M6 across versions", "", f"> {NO_REFERENCE}", ""]
+    metric_ids = [metric_id for metric_id, _ in METRIC_DEFS]
+    lines += _table(
+        ["version"] + metric_ids,
+        [
+            [row["version"]] + [
+                next((m["status"] for m in row.get("metrics") or [] if m["id"] == metric_id), "-")
+                for metric_id in metric_ids
+            ]
+            for row in report["versions"]
+        ],
+    )
+    lines += ["", "## Failed / not run", ""]
+    if not report["failed"] and not report["not_run"]:
+        lines += ["_none_", ""]
+    for version in report["failed"]:
+        lines += [f"* failed `{version}`: see `{version}/version.json` and the step reports"]
+    for version in report["not_run"]:
+        row = next(item for item in report["versions"] if item["version"] == version)
+        lines += [f"* no input `{version}`: {row.get('reason', '-')}"]
+    lines += [
+        "",
+        "## Evidence",
+        "",
+        f"* sweep summary: `{report['summary']}` / `{report['markdown']}`",
+        f"* bundle manifest: `{report['manifest']}` (sha256 of every artifact)",
+        "* per-version tree: `version.json`, `version.md`, `version-diff/`, "
+        "`portability-matrix/`, `mutation-smoke/`, `bundle-manifest.json`",
+        "",
+    ]
+    return "\n".join(lines)
+
+
+def _step_status(row: dict, name: str) -> str:
+    for step in row.get("steps") or []:
+        if step["name"] == name:
+            return step["status"]
+    return "-"
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def _parse_mapping(value: str, flag: str) -> tuple[str, str]:
+    if "=" not in value:
+        raise argparse.ArgumentTypeError(f"{flag} expects VERSION=PATH, got {value!r}")
+    version, path = value.split("=", 1)
+    version, path = version.strip(), path.strip()
+    if not version or not path:
+        raise argparse.ArgumentTypeError(f"{flag} expects VERSION=PATH, got {value!r}")
+    return version, path
+
+
+_HELP_EPILOG = """\
+invocation:
+  dry-run first (default; reads input dirs only, never touches a target):
+    python3 harness/port_driver_versions.py --input-dir ../../firmware/57X-GSP
+  execute, resume-safe:
+    python3 harness/port_driver_versions.py --run --input-dir ../../firmware/57X-GSP \\
+        --versions 540,550,560,570 --out dumps/sweep-570
+  explicit local targets (no discovery):
+    python3 harness/port_driver_versions.py --run --versions 570.144 \\
+        --target 570.144=../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\
+        --baseline 570.144 --out dumps/sweep-570
+
+cron (weekly, resume-safe; nothing is downloaded):
+  17 3 * * 1 cd <tool-root> && python3 harness/port_driver_versions.py --run \\
+      --input-dir ../../firmware/57X-GSP --versions 540,550,560,570 \\
+      --out dumps/sweep-570 >> dumps/sweep-570/cron.log 2>&1
+
+agent (MCP, docs/20): read the dry-run plan, then call --run; read
+sweep-summary.json and each bundle-manifest.json instead of recomputing.
+RM 610 (docs/23): versions >= 610 require --approval-token RM610-APPROVED or
+GREENEYE_RM610_APPROVED=1; the gate fires before any file is opened.
+"""
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="port_driver_versions",
+        description=__doc__.splitlines()[0],
+        epilog=_HELP_EPILOG,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("--versions", help="comma-separated version list (default: known cadence: "
+                                           + ",".join(DEFAULT_VERSIONS) + ")")
+    parser.add_argument("--config", help="JSON sweep config (versions, baseline, input_dir, targets, symbols)")
+    parser.add_argument("--baseline", help="baseline version for version_diff (default: earliest found)")
+    parser.add_argument("--input-dir", help="directory to discover local firmware/driver images in (recursive)")
+    parser.add_argument("--target", action="append", default=[], metavar="VERSION=PATH",
+                        help="explicit local target for a version (repeatable; skips discovery)")
+    parser.add_argument("--symbols", action="append", default=[], metavar="VERSION=PATH",
+                        help="green-eye symbol DB JSON for a version (repeatable)")
+    parser.add_argument("--ghidra-exports", action="append", default=[], metavar="VERSION=PATH",
+                        help="Ghidra exports TSV for a version (repeatable)")
+    parser.add_argument("--out", help="sweep output directory (required for --run)")
+    parser.add_argument("--run", dest="run", action="store_true",
+                        help="execute the sweep (default is the dry-run plan)")
+    parser.add_argument("--dry-run", dest="run", action="store_false",
+                        help="print the plan and exit 0 (default)")
+    parser.set_defaults(run=False)
+    parser.add_argument("--force", action="store_true",
+                        help="re-run versions whose version.json already reports ok/partial")
+    parser.add_argument("--extract-runfiles", action="store_true",
+                        help="locally extract driver runfiles with --extract-only (never downloads)")
+    parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})")
+    parser.add_argument("--probe-instructions", type=int, default=2000,
+                        help="probe step budget per target (0 = wall-clock --probe-duration)")
+    parser.add_argument("--probe-duration", type=float, default=0.5)
+    parser.add_argument("--scenario-duration", type=float, default=0.4,
+                        help="budget per version_diff/matrix scenario phase")
+    parser.add_argument("--max-stops", type=int, default=5000)
+    parser.add_argument("--max-rows", type=int, default=40)
+    parser.add_argument("--mutation-kernel", default=DEMO_ELF,
+                        help="kernel for the mutation smoke (default: bundled demo)")
+    parser.add_argument("--mutation-iterations", type=int, default=8)
+    parser.add_argument("--mutation-seed", type=int, default=7)
+    parser.add_argument("--mutation-timeout", type=float, default=0.25)
+    parser.add_argument("--mutation-settle", type=float, default=0.1)
+    parser.add_argument("--allow-foreign-mutation", action="store_true",
+                        help="permit a non-demo mutation kernel (QEMU only)")
+    parser.add_argument("--step-timeout", type=float, default=600.0,
+                        help="wall-clock safety net per step, seconds (0 disables)")
+    parser.add_argument("--log-bytes", type=int, default=MAX_LOG_BYTES)
+    return parser
+
+
+def options_from_args(args: argparse.Namespace) -> SweepOptions:
+    options = SweepOptions()
+    if args.config:
+        options.config = args.config
+        options_from_config(options, load_config(args.config))
+    if args.versions:
+        options.versions = parse_versions(args.versions)
+    if not options.versions:
+        options.versions = list(DEFAULT_VERSIONS)
+    options.baseline = args.baseline
+    options.input_dir = args.input_dir
+    options.out = args.out
+    for value in args.target:
+        version, path = _parse_mapping(value, "--target")
+        options.targets.setdefault(version, []).append(path)
+    for value in args.symbols:
+        version, path = _parse_mapping(value, "--symbols")
+        options.symbols[version] = path
+    for value in args.ghidra_exports:
+        version, path = _parse_mapping(value, "--ghidra-exports")
+        options.ghidra_exports[version] = path
+    options.run = args.run
+    options.force = args.force
+    options.extract_runfiles = args.extract_runfiles
+    options.approval_token = args.approval_token
+    options.probe_instructions = args.probe_instructions
+    options.probe_duration = args.probe_duration
+    options.scenario_duration = args.scenario_duration
+    options.max_stops = args.max_stops
+    options.max_rows = args.max_rows
+    options.mutation_kernel = args.mutation_kernel
+    options.mutation_iterations = args.mutation_iterations
+    options.mutation_seed = args.mutation_seed
+    options.mutation_timeout = args.mutation_timeout
+    options.mutation_settle = args.mutation_settle
+    options.allow_foreign_mutation = args.allow_foreign_mutation
+    options.step_timeout = args.step_timeout
+    options.log_bytes = args.log_bytes
+    return options
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    try:
+        options = options_from_args(args)
+    except (ValueError, argparse.ArgumentTypeError) as exc:
+        parser.error(str(exc))
+    try:
+        check_rm610_gate(
+            _gate_version_labels(options.versions),
+            approval_token=options.approval_token,
+        )
+        check_rm610_gate(
+            gate_labels(options.versions, resolve_targets(options)),
+            approval_token=options.approval_token,
+        )
+    except Rm610GateError as exc:
+        print(f"[sweep] RM610 gate refused: {exc}", file=sys.stderr)
+        return 3
+    if options.dry_run:
+        try:
+            plan = plan_sweep(options)
+        except (ValueError, FileNotFoundError) as exc:
+            print(f"[sweep] error: {exc}", file=sys.stderr)
+            return 2
+        print(json.dumps(plan, indent=2, sort_keys=True))
+        print(f"[sweep] dry-run: {plan['found']} input(s) found, {plan['missing']} missing; no target touched")
+        return 0
+    if not options.out:
+        parser.error("--out is required unless --dry-run is used")
+    try:
+        report = run_sweep(options)
+    except (ValueError, FileNotFoundError) as exc:
+        print(f"[sweep] error: {exc}", file=sys.stderr)
+        return 2
+    except Rm610GateError as exc:  # pragma: no cover - main checks first
+        print(f"[sweep] RM610 gate refused: {exc}", file=sys.stderr)
+        return 3
+    for row in report["versions"]:
+        reason = f" reason={row['reason']}" if row.get("reason") else ""
+        resumed = " resumed" if row.get("resumed") else ""
+        print(f"[sweep] {row['version']}: status={row['status']}{resumed}{reason}")
+    print(f"[sweep] {report['interpretation']}")
+    print(f"[sweep] wrote {os.path.join(os.path.abspath(options.out), 'sweep-summary.json')}")
+    print(f"[sweep] wrote {os.path.join(os.path.abspath(options.out), 'bundle-manifest.json')}")
+    return 1 if report["failed"] or report["ran"] == 0 else 0
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(main())
diff --git a/debug/risc-v-gsp-rm-vm/harness/port_new_device.py b/debug/risc-v-gsp-rm-vm/harness/port_new_device.py
new file mode 100644
index 0000000..56522ec
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/port_new_device.py
@@ -0,0 +1,726 @@
+#!/usr/bin/env python3
+"""New-device port scaffold (roadmap H4.1/R6, docs/07 P2, docs/22 §3, docs/28).
+
+``--device <name>`` creates or uses a profile JSON holding the user-supplied
+device facts (PCI IDs, BARs, FB geometry, engine inventory, interrupt topology,
+ROM/VBIOS notes) and then drives, on the local QEMU backend only:
+
+1. probe-first-touch on the device's firmware (first unmapped access, bounded);
+2. a symbol inventory (ELF .symtab + optional Ghidra/green-eye DB sources);
+3. a portability matrix run of the device against a reference firmware;
+4. a port checklist report against the docs/07 evidence checklist.
+
+``--dry-run`` is the default: it prints the plan (profile action, missing
+profile fields, steps, expected artifacts) and exits 0 without creating files.
+``--run`` writes an evidence bundle under ``--out`` with a sha256 manifest.
+The RM 610 gate from ``version_diff.py`` (docs/23) is enforced before any file
+is opened, in dry-run too.
+
+This tool never touches hardware: no /dev, no kernel module, no mdev, no
+guest, no physical GPU. Physical validation is a separate supervised step
+(docs/06-portability/hardware-software-matrix.md §5.1).
+
+Run from the tool root:
+
+    python3 harness/port_new_device.py --device rtx4090 --profile profiles/rtx4090.json
+    python3 harness/port_new_device.py --run --device rtx4090 \\
+        --firmware ../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\
+        --reference firmware/uproc_demo/build/uproc_demo.elf --out dumps/port-rtx4090
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from dataclasses import dataclass
+
+try:
+    from .common import (
+        MAX_LOG_BYTES,
+        SymbolDB,
+        ensure_dir,
+        qemu_available,
+        sha256_file,
+        write_json,
+        write_text,
+    )
+    from .portability_matrix import (
+        MatrixOptions,
+        TargetSpec,
+        render_markdown as render_matrix_markdown,
+        run_matrix,
+    )
+    from .version_diff import (
+        RM610_TOKEN,
+        Rm610GateError,
+        VersionDiffOptions,
+        check_rm610_gate,
+        load_any_target,
+        load_symbol_sources,
+        run_probe,
+    )
+except ImportError:  # direct script execution (python3 harness/port_new_device.py)
+    from common import (  # type: ignore[no-redef]
+        MAX_LOG_BYTES,
+        SymbolDB,
+        ensure_dir,
+        qemu_available,
+        sha256_file,
+        write_json,
+        write_text,
+    )
+    from portability_matrix import (  # type: ignore[no-redef]
+        MatrixOptions,
+        TargetSpec,
+        render_markdown as render_matrix_markdown,
+        run_matrix,
+    )
+    from version_diff import (  # type: ignore[no-redef]
+        RM610_TOKEN,
+        Rm610GateError,
+        VersionDiffOptions,
+        check_rm610_gate,
+        load_any_target,
+        load_symbol_sources,
+        run_probe,
+    )
+
+FORMAT = "greeneye-device-port/1"
+PLAN_FORMAT = "greeneye-device-port-plan/1"
+PROFILE_FORMAT = "greeneye-device-profile/1"
+BUNDLE_FORMAT = "greeneye-release-bundle/1"
+DEVICE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+
+#: Where each user-supplied profile section comes from (docs/22 §3.1).
+PROFILE_HELP = {
+    "pci": "lspci / vendor docs, config/hardware/*.yaml is a design target only",
+    "bars": "guest lspci -vvv, host sysfs",
+    "fb": "host oracle: rmtool mem stats --format=json",
+    "engines": "host oracle: rmtool engine list --format=json; validate per driver profile",
+    "interrupts": "MSI-X tables, guest lspci -vvv",
+    "rom": "ROM/VBIOS read; laptop/Surface overrides",
+    "firmware": "extracted GSP .fwimage or re-hosted ELF (user-supplied; never downloaded)",
+    "pairing": "verify image family + FMC manifest + version string before debugging (docs/07 P2.2)",
+    "physical": "supervised physical validation only; this tool is simulator-only (docs/22 §3.5)",
+}
+
+
+@dataclass
+class DeviceOptions:
+    device: str
+    profile: str | None = None
+    firmware: str | None = None
+    reference: str | None = None
+    symbols: str | None = None
+    ghidra_exports: str | None = None
+    ghidra_bias: int = 0
+    out: str | None = None
+    run: bool = False
+    approval_token: str | None = None
+    probe_instructions: int = 2000
+    probe_duration: float = 0.5
+    scenario_duration: float = 0.4
+    max_stops: int = 5000
+    max_rows: int = 40
+    step_timeout: float = 600.0
+    log_bytes: int = MAX_LOG_BYTES
+
+    @property
+    def dry_run(self) -> bool:
+        return not self.run
+
+
+# ---------------------------------------------------------------------------
+# profile
+# ---------------------------------------------------------------------------
+
+
+def scaffold_profile(name: str) -> dict:
+    """Empty user-supplied profile; every field documents where it comes from."""
+    return {
+        "format": PROFILE_FORMAT,
+        "name": name,
+        "vendor": "",
+        "device": "",
+        "pci": {"vendor_id": "0x10de", "device_id": "", "subsystem_vendor_id": "",
+                "subsystem_device_id": ""},
+        "bars": [],
+        "fb": {"vram_mib": None, "bar1_mib": None, "ecc": None, "notes": ""},
+        "engines": [],
+        "interrupts": {"msix_vectors": None, "notes": ""},
+        "rom": {"vblos_sha256": "", "notes": ""},
+        "firmware": "",
+        "firmware_sha256": "",
+        "firmware_family": "",
+        "firmware_version": "",
+        "reference": "",
+        "physical_validation": {"status": "pending", "date": "", "by": "", "evidence": "", "notes": ""},
+        "notes": "",
+        "where_to_fill": dict(PROFILE_HELP),
+        "references": [
+            "docs/07-PORTABILITY-PLAYBOOKS.md P2",
+            "docs/22-PORTING-CUDA-AND-DEVICES.md 3",
+            "docs/06-portability/hardware-software-matrix.md 5.1",
+        ],
+    }
+
+
+def load_profile(path: str) -> tuple[dict, bool]:
+    """Return ``(profile, created)``; an unreadable profile is a clean error."""
+    if not os.path.isfile(path):
+        return {}, False
+    with open(path, "r", encoding="utf-8") as fh:
+        data = json.load(fh)
+    if not isinstance(data, dict):
+        raise ValueError(f"profile {path}: expected a JSON object")
+    return data, True
+
+
+def profile_missing_fields(profile: dict) -> list[str]:
+    missing: list[str] = []
+    pci = profile.get("pci") or {}
+    if not any(str(pci.get(key, "")).strip() for key in ("vendor_id", "device_id", "subsystem_device_id")):
+        missing.append("pci")
+    if not profile.get("bars"):
+        missing.append("bars")
+    fb = profile.get("fb") or {}
+    if not any(fb.get(key) not in (None, "", 0) for key in ("vram_mib", "bar1_mib", "notes")):
+        missing.append("fb")
+    if not profile.get("engines"):
+        missing.append("engines")
+    interrupts = profile.get("interrupts") or {}
+    if not any(interrupts.get(key) not in (None, "", 0) for key in ("msix_vectors", "notes")):
+        missing.append("interrupts")
+    rom = profile.get("rom") or {}
+    if not any(str(rom.get(key, "")).strip() for key in ("vblos_sha256", "notes")):
+        missing.append("rom")
+    return missing
+
+
+def profile_labels(profile: dict, device: str, profile_path: str | None) -> list[str]:
+    labels = [device, os.path.basename(profile_path or "")]
+    for key in ("name", "device", "firmware", "reference", "firmware_family", "firmware_version"):
+        value = profile.get(key)
+        if isinstance(value, str) and value:
+            labels.append(os.path.basename(value))
+    return labels
+
+
+def merge_profile(profile: dict, options: DeviceOptions) -> dict:
+    merged = json.loads(json.dumps(profile))
+    if options.firmware:
+        merged["firmware"] = options.firmware
+    if options.reference:
+        merged["reference"] = options.reference
+    return merged
+
+
+# ---------------------------------------------------------------------------
+# plan (dry run)
+# ---------------------------------------------------------------------------
+
+
+def _expected_artifacts(out: str, device: str) -> list[str]:
+    base = os.path.join(out, device)
+    return [
+        os.path.join(base, "device-port.json"),
+        os.path.join(base, "device-port.md"),
+        os.path.join(base, "port-checklist.json"),
+        os.path.join(base, "port-checklist.md"),
+        os.path.join(base, "symbols", f"{device}.json"),
+        os.path.join(base, "sessions"),
+        os.path.join(base, "portability-matrix", "portability-report.json"),
+        os.path.join(base, "portability-matrix", "portability-report.md"),
+        os.path.join(base, "bundle-manifest.json"),
+    ]
+
+
+def build_plan(options: DeviceOptions) -> dict:
+    profile_path = options.profile or (os.path.join(options.out or "<out>", "profiles",
+                                                     f"{options.device}.json"))
+    profile, created = load_profile(profile_path) if os.path.isfile(profile_path) else ({}, False)
+    merged = merge_profile(profile if created else scaffold_profile(options.device), options)
+    firmware = merged.get("firmware") or ""
+    reference = merged.get("reference") or ""
+    check_rm610_gate(
+        profile_labels(profile if created else {}, options.device, profile_path)
+        + [os.path.basename(firmware), os.path.basename(reference)],
+        approval_token=options.approval_token,
+    )
+    out = os.path.abspath(options.out) if options.out else "<out>"
+    return {
+        "format": PLAN_FORMAT,
+        "tool": "harness/port_new_device.py",
+        "dry_run": True,
+        "rm610_gate": {"status": "passed", "labels": sorted(set(
+            profile_labels(profile if created else {}, options.device, profile_path)
+            + [os.path.basename(firmware), os.path.basename(reference)]
+        ))},
+        "device": options.device,
+        "profile": {
+            "path": profile_path,
+            "action": "use" if created else "would-create-scaffold",
+            "missing_fields": profile_missing_fields(merged),
+            "where_to_fill": PROFILE_HELP,
+        },
+        "firmware": {"path": firmware or None, "found": bool(firmware) and os.path.isfile(firmware)},
+        "reference": {"path": reference or None,
+                      "found": bool(reference) and os.path.isfile(reference)},
+        "out": out,
+        "steps": [
+            {"name": "probe_first_touch", "target": firmware or None,
+             "out": os.path.join(out, options.device, "sessions", f"{options.device}-probe")},
+            {"name": "symbol_inventory", "sources": [firmware or None, options.symbols]},
+            {"name": "portability_matrix", "targets": [reference or None, firmware or None]},
+            {"name": "port_checklist", "out": os.path.join(out, options.device, "port-checklist.json")},
+        ],
+        "expected_artifacts": _expected_artifacts(out, options.device),
+        "interpretation": (
+            "dry-run: fill the profile fields, then pass --run; physical validation "
+            "is a separate supervised step and is never done by this tool"
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+# steps
+# ---------------------------------------------------------------------------
+
+
+def _run_step(name: str, fn, *args, **kwargs) -> dict:
+    try:
+        detail = fn(*args, **kwargs)
+    except Exception as exc:  # a failed step must not hide the others
+        return {"name": name, "status": "error", "reason": f"{type(exc).__name__}: {exc}"}
+    entry = {"name": name}
+    entry.update(detail)
+    return entry
+
+
+def step_probe(loaded, db: SymbolDB, options: DeviceOptions, out_dir: str, qemu_ok: bool) -> dict:
+    vd = VersionDiffOptions(
+        old=loaded.path,
+        new=loaded.path,
+        out=out_dir,
+        probe_instructions=options.probe_instructions,
+        probe_duration=options.probe_duration,
+        log_bytes=options.log_bytes,
+        approval_token=options.approval_token,
+    )
+    result = run_probe(loaded, db, vd, out_dir, qemu_ok)
+    return {
+        "status": result["status"],
+        "reason": result.get("reason"),
+        "session": result.get("session"),
+        "stopped_pc": result.get("stopped_pc"),
+        "stopped_symbol": result.get("stopped_symbol"),
+        "fault_count": result.get("fault_count"),
+        "first_fault": result.get("first_fault"),
+    }
+
+
+def step_symbols(loaded, db: SymbolDB, sources: list[str], notes: list[str],
+                 options: DeviceOptions, out_dir: str) -> dict:
+    records = [
+        {"name": symbol.name, "addr": symbol.addr, "size": symbol.size, "source": symbol.source}
+        for symbol in db
+    ]
+    rel = None
+    if records:
+        rel = os.path.join("symbols", f"{options.device}.json")
+        path = os.path.join(out_dir, rel)
+        ensure_dir(os.path.dirname(path))
+        db.save_json(path, meta={
+            "device": options.device,
+            "firmware": os.path.basename(loaded.path),
+            "bias": loaded.bias,
+            "sources": sources,
+            "notes": notes,
+        })
+    return {
+        "status": "ok" if records else "skipped",
+        "reason": None if records else "no symbols: no ELF .symtab and no Ghidra/JSON sources supplied",
+        "count": len(records),
+        "sources": sources,
+        "notes": notes,
+        "db": rel,
+    }
+
+
+def step_matrix(firmware: str, reference: str | None, options: DeviceOptions, out_dir: str) -> dict:
+    specs = []
+    if reference and os.path.isfile(reference):
+        specs.append(TargetSpec(path=os.path.abspath(reference), name="reference"))
+    specs.append(TargetSpec(path=os.path.abspath(firmware), name=options.device))
+    matrix = run_matrix(
+        specs,
+        MatrixOptions(
+            out=out_dir,
+            probe_instructions=options.probe_instructions,
+            probe_duration=options.probe_duration,
+            scenario_duration=options.scenario_duration,
+            max_stops=options.max_stops,
+            max_rows=options.max_rows,
+            log_bytes=options.log_bytes,
+        ),
+    )
+    write_json(os.path.join(out_dir, "portability-report.json"), matrix)
+    write_text(
+        os.path.join(out_dir, "portability-report.md"),
+        render_matrix_markdown(matrix, MatrixOptions(out=out_dir, max_rows=options.max_rows)),
+    )
+    statuses = {row["name"]: row["status"] for row in matrix["targets"]}
+    return {
+        "status": "error" if "error" in statuses.values() else "ok",
+        "report": "portability-matrix/portability-report.json",
+        "targets": statuses,
+        "comparisons": [
+            {"a": row["a"], "b": row["b"], "status": row["status"], "all_zero": row["all_zero"]}
+            for row in matrix["comparisons"]
+        ],
+    }
+
+
+def build_checklist(profile: dict, firmware: str, reference: str | None, results: dict) -> dict:
+    """docs/07 evidence checklist + docs/22 §3.1 profile items."""
+    missing = profile_missing_fields(profile)
+    pairing_ok = bool(str(profile.get("firmware_family", "")).strip()
+                      and str(profile.get("firmware_version", "")).strip())
+    physical = profile.get("physical_validation") or {}
+    physical_status = str(physical.get("status", "pending")).lower()
+    probe = results.get("probe") or {}
+    symbols = results.get("symbols") or {}
+    matrix = results.get("matrix") or {}
+    items = [
+        {
+            "id": "profile",
+            "item": "device profile fields recorded (PCI IDs, BARs, FB, engines, interrupts, ROM)",
+            "status": "ok" if not missing else "pending",
+            "detail": "missing: " + ", ".join(missing) if missing else "all profile sections populated",
+        },
+        {
+            "id": "pairing",
+            "item": "firmware hash + family/version pairing recorded before debugging (docs/07 P2.2)",
+            "status": "ok" if pairing_ok else "pending",
+            "detail": f"family={profile.get('firmware_family') or '-'} version={profile.get('firmware_version') or '-'}",
+        },
+        {
+            "id": "hash",
+            "item": "firmware sha256 recorded",
+            "status": "ok" if firmware and os.path.isfile(firmware) else "missing",
+            "detail": sha256_file(firmware) if firmware and os.path.isfile(firmware) else "firmware not found",
+        },
+        {
+            "id": "probe",
+            "item": "probe first-touch captured and explained as profile vs logic",
+            "status": "ok" if probe.get("status") == "ok" else "pending",
+            "detail": f"stopped_pc={probe.get('stopped_pc')} faults={probe.get('fault_count')}",
+        },
+        {
+            "id": "symbols",
+            "item": "symbol inventory built with provenance",
+            "status": "ok" if symbols.get("status") == "ok" else "pending",
+            "detail": f"count={symbols.get('count')} sources={symbols.get('sources')}",
+        },
+        {
+            "id": "matrix",
+            "item": "portability matrix run vs the reference firmware",
+            "status": "ok" if matrix.get("status") == "ok" and reference else "pending",
+            "detail": f"targets={matrix.get('targets')}",
+        },
+        {
+            "id": "docs07",
+            "item": "docs/07 evidence checklist attachable (hash, probe, symbols, sessions, compare, M1-M6)",
+            "status": "ok" if firmware else "missing",
+            "detail": "bundle-manifest.json hashes every artifact",
+        },
+        {
+            "id": "physical",
+            "item": "physical/supervised validation done (outside this simulator-only tool)",
+            "status": "ok" if physical_status in ("passed", "ok", "done") else "pending",
+            "detail": f"status={physical_status}; docs/22 §3.5: simulator-only evidence never justifies the support claim",
+        },
+    ]
+    required_pending = [row["id"] for row in items
+                        if row["status"] != "ok" and row["id"] != "physical"]
+    if required_pending:
+        recommendation = "hold: " + ", ".join(required_pending) + " not complete"
+    elif physical_status in ("passed", "ok", "done"):
+        recommendation = "go: simulator evidence complete and physical validation recorded"
+    else:
+        recommendation = "simulator-evidence-complete; physical validation pending (supervised step)"
+    return {
+        "format": "greeneye-device-checklist/1",
+        "tool": "harness/port_new_device.py",
+        "device": profile.get("name") or "-",
+        "firmware": firmware,
+        "reference": reference,
+        "profile_missing_fields": missing,
+        "items": items,
+        "recommendation": recommendation,
+        "safety": "simulator-only: no /dev, no kernel module, no mdev, no guest, no physical GPU",
+    }
+
+
+def render_checklist_markdown(checklist: dict) -> str:
+    lines = [
+        f"# green-eye port checklist: {checklist['device']}",
+        "",
+        "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_new_device.py`.",
+        "",
+        f"* firmware: `{checklist['firmware']}`",
+        f"* reference: `{checklist['reference']}`",
+        f"* recommendation: **{checklist['recommendation']}**",
+        f"* safety: {checklist['safety']}",
+        "",
+        "| id | item | status | detail |",
+        "|---|---|---|---|",
+    ]
+    for row in checklist["items"]:
+        lines.append(f"| {row['id']} | {row['item']} | {row['status']} | {row['detail']} |")
+    lines += ["", "Evidence: `port-checklist.json`, `bundle-manifest.json`, `portability-matrix/`.", ""]
+    return "\n".join(lines)
+
+
+def build_bundle(out_dir: str) -> dict:
+    files: list[dict] = []
+    for root, dirs, names in os.walk(out_dir):
+        dirs.sort()
+        for name in sorted(names):
+            path = os.path.join(root, name)
+            rel = os.path.relpath(path, out_dir)
+            if rel == "bundle-manifest.json":
+                continue
+            files.append({"path": rel, "bytes": os.path.getsize(path), "sha256": sha256_file(path)})
+    files.sort(key=lambda row: row["path"])
+    return {"format": BUNDLE_FORMAT, "tool": "harness/port_new_device.py", "files": files}
+
+
+def run_device_port(options: DeviceOptions) -> dict:
+    if options.out is None:
+        raise ValueError("--out is required unless --dry-run is used")
+    out_root = os.path.abspath(options.out)
+    device_dir = os.path.join(out_root, options.device)
+    profile_path = options.profile or os.path.join(out_root, "profiles", f"{options.device}.json")
+    profile, created = load_profile(profile_path)
+    created_now = not created
+    if not created:
+        profile = scaffold_profile(options.device)
+    profile = merge_profile(profile, options)
+    labels = profile_labels(profile, options.device, profile_path)
+    firmware = profile.get("firmware") or ""
+    reference = profile.get("reference") or ""
+    check_rm610_gate(labels + [os.path.basename(firmware), os.path.basename(reference)],
+                     approval_token=options.approval_token)
+    if created_now:
+        ensure_dir(os.path.dirname(profile_path))
+        write_json(profile_path, profile)
+    if not firmware:
+        raise ValueError(
+            "no device firmware: pass --firmware PATH or set \"firmware\" in the profile "
+            f"({profile_path}); see docs/07 P2"
+        )
+    loaded = load_any_target(firmware)
+    if not loaded.ok:
+        raise ValueError(f"firmware not usable: {loaded.path}: {loaded.reason or loaded.kind}")
+    if created_now:
+        profile["firmware_sha256"] = loaded.sha256
+        write_json(profile_path, profile)
+
+    ensure_dir(device_dir)
+    db, sources, notes = load_symbol_sources(
+        loaded,
+        symbols_json=options.symbols,
+        ghidra_exports=options.ghidra_exports,
+        ghidra_bias=options.ghidra_bias,
+    )
+    qemu_ok = qemu_available()
+    results: dict = {}
+    steps = [
+        _run_step("probe_first_touch", step_probe, loaded, db, options, device_dir, qemu_ok),
+        _run_step("symbol_inventory", step_symbols, loaded, db, sources, notes, options, device_dir),
+        _run_step("portability_matrix", step_matrix, firmware, reference, options,
+                  os.path.join(device_dir, "portability-matrix")),
+    ]
+    results["probe"] = steps[0]
+    results["symbols"] = steps[1]
+    results["matrix"] = steps[2]
+    checklist = build_checklist(profile, firmware, reference, results)
+    write_json(os.path.join(device_dir, "port-checklist.json"), checklist)
+    write_text(os.path.join(device_dir, "port-checklist.md"), render_checklist_markdown(checklist))
+    checklist_step = {
+        "name": "port_checklist",
+        "status": "ok",
+        "report": "port-checklist.json",
+        "recommendation": checklist["recommendation"],
+    }
+    steps.append(checklist_step)
+    failed = [step["name"] for step in steps if step["status"] == "error"]
+    report = {
+        "format": FORMAT,
+        "tool": "harness/port_new_device.py",
+        "dry_run": False,
+        "rm610_gate": {"status": "passed", "labels": sorted(set(labels))},
+        "device": options.device,
+        "profile": {
+            "path": os.path.relpath(profile_path, out_root),
+            "created": created_now,
+            "missing_fields": profile_missing_fields(profile),
+        },
+        "firmware": {"path": firmware, "kind": loaded.kind, "sha256": loaded.sha256},
+        "reference": {"path": reference, "found": bool(reference) and os.path.isfile(reference)},
+        "qemu_available": qemu_ok,
+        "steps": steps,
+        "checklist": checklist,
+        "failed": failed,
+        "manifest": "bundle-manifest.json",
+    }
+    write_json(os.path.join(device_dir, "device-port.json"), report)
+    write_text(os.path.join(device_dir, "device-port.md"), render_report_markdown(report))
+    write_json(os.path.join(device_dir, "bundle-manifest.json"), build_bundle(device_dir))
+    return report
+
+
+def render_report_markdown(report: dict) -> str:
+    lines = [
+        f"# green-eye device port: {report['device']}",
+        "",
+        "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_new_device.py`.",
+        "",
+        f"* format: `{report['format']}`",
+        f"* firmware: `{report['firmware']['path']}` ({report['firmware']['kind']}, "
+        f"sha256 `{report['firmware']['sha256'][:16]}`)",
+        f"* reference: `{report['reference']['path'] or '-'}`",
+        f"* qemu available: {report['qemu_available']}",
+        "",
+        "| step | status | report | reason |",
+        "|---|---|---|---|",
+    ]
+    for step in report["steps"]:
+        lines.append(f"| {step['name']} | {step['status']} | {step.get('report') or '-'} "
+                     f"| {step.get('reason') or '-'} |")
+    lines += [
+        "",
+        f"* checklist recommendation: **{report['checklist']['recommendation']}**",
+        f"* manifest: `{report['manifest']}`",
+        "",
+    ]
+    return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+_HELP_EPILOG = """\
+invocation:
+  dry-run first (default; nothing is created or touched):
+    python3 harness/port_new_device.py --device rtx4090 --profile profiles/rtx4090.json
+  execute (scaffolds the profile if absent; simulator only):
+    python3 harness/port_new_device.py --run --device rtx4090 \\
+        --firmware ../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\
+        --reference firmware/uproc_demo/build/uproc_demo.elf --out dumps/port-rtx4090
+
+agent (MCP, docs/20): read the dry-run plan, fill the profile, then --run;
+read device-port.json and port-checklist.json from the bundle.
+RM 610 (docs/23): rm610-labelled device/firmware labels are refused without
+--approval-token RM610-APPROVED or GREENEYE_RM610_APPROVED=1.
+Safety: no /dev, no kernel module, no mdev, no guest, no physical GPU.
+"""
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="port_new_device",
+        description=__doc__.splitlines()[0],
+        epilog=_HELP_EPILOG,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("--device", required=True, help="device name (profile key, e.g. rtx4090)")
+    parser.add_argument("--profile", help="profile JSON path (default: <out>/profiles/<device>.json)")
+    parser.add_argument("--firmware", help="device GSP firmware image or re-hosted ELF (overrides profile)")
+    parser.add_argument("--reference", help="reference firmware for the portability matrix")
+    parser.add_argument("--symbols", help="green-eye symbol DB JSON for the device firmware")
+    parser.add_argument("--ghidra-exports", help="Ghidra exports TSV for the device firmware")
+    parser.add_argument("--ghidra-bias", type=lambda value: int(value, 0), default=0)
+    parser.add_argument("--out", help="evidence bundle directory (required for --run)")
+    parser.add_argument("--run", dest="run", action="store_true",
+                        help="execute (default is the dry-run plan)")
+    parser.add_argument("--dry-run", dest="run", action="store_false",
+                        help="print the plan and exit 0 (default)")
+    parser.set_defaults(run=False)
+    parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})")
+    parser.add_argument("--probe-instructions", type=int, default=2000)
+    parser.add_argument("--probe-duration", type=float, default=0.5)
+    parser.add_argument("--scenario-duration", type=float, default=0.4)
+    parser.add_argument("--max-stops", type=int, default=5000)
+    parser.add_argument("--max-rows", type=int, default=40)
+    parser.add_argument("--step-timeout", type=float, default=600.0)
+    parser.add_argument("--log-bytes", type=int, default=MAX_LOG_BYTES)
+    return parser
+
+
+def options_from_args(args: argparse.Namespace) -> DeviceOptions:
+    if not DEVICE_RE.match(args.device):
+        raise ValueError(f"invalid device name: {args.device!r}")
+    return DeviceOptions(
+        device=args.device,
+        profile=args.profile,
+        firmware=args.firmware,
+        reference=args.reference,
+        symbols=args.symbols,
+        ghidra_exports=args.ghidra_exports,
+        ghidra_bias=args.ghidra_bias,
+        out=args.out,
+        run=args.run,
+        approval_token=args.approval_token,
+        probe_instructions=args.probe_instructions,
+        probe_duration=args.probe_duration,
+        scenario_duration=args.scenario_duration,
+        max_stops=args.max_stops,
+        max_rows=args.max_rows,
+        step_timeout=args.step_timeout,
+        log_bytes=args.log_bytes,
+    )
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    try:
+        options = options_from_args(args)
+    except ValueError as exc:
+        parser.error(str(exc))
+    try:
+        check_rm610_gate(
+            [options.device, os.path.basename(options.profile or ""),
+             os.path.basename(options.firmware or ""), os.path.basename(options.reference or "")],
+            approval_token=options.approval_token,
+        )
+        if options.dry_run:
+            plan = build_plan(options)
+            print(json.dumps(plan, indent=2, sort_keys=True))
+            print(f"[device] dry-run: profile {plan['profile']['action']}, "
+                  f"firmware found={plan['firmware']['found']}; nothing created or touched")
+            return 0
+        report = run_device_port(options)
+    except Rm610GateError as exc:
+        print(f"[device] RM610 gate refused: {exc}", file=sys.stderr)
+        return 3
+    except (ValueError, FileNotFoundError, json.JSONDecodeError) as exc:
+        print(f"[device] error: {exc}", file=sys.stderr)
+        return 2
+    for step in report["steps"]:
+        reason = f" reason={step['reason']}" if step.get("reason") else ""
+        print(f"[device] {step['name']}: status={step['status']}{reason}")
+    print(f"[device] checklist: {report['checklist']['recommendation']}")
+    print(f"[device] wrote {os.path.join(os.path.abspath(options.out), options.device, 'device-port.json')}")
+    print(f"[device] wrote {os.path.join(os.path.abspath(options.out), options.device, 'bundle-manifest.json')}")
+    return 1 if report["failed"] else 0
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(main())