From 4467eb66a4c8a776e35e9644960eba699dc85e6f Mon Sep 17 00:00:00 2001 From: Arthur Hanson Rasmusson Date: Fri, 2 Oct 2026 11:30:46 -0400 Subject: debug: green-eye porting automation - new-device and driver-version port scripts, simulator packaging, porting docs; control-discovery fix --- debug/risc-v-gsp-rm-vm/README.md | 6 + debug/risc-v-gsp-rm-vm/docs/00-OVERVIEW.md | 8 + .../docs/21-CI-RELEASE-PIPELINE.md | 16 +- .../docs/22-PORTING-CUDA-AND-DEVICES.md | 19 +- debug/risc-v-gsp-rm-vm/docs/23-RM610-READINESS.md | 31 +- debug/risc-v-gsp-rm-vm/docs/26-ARCHITECTURE.md | 15 +- debug/risc-v-gsp-rm-vm/docs/27-TOOL-AUDIT.md | 15 + .../risc-v-gsp-rm-vm/harness/control_discovery.py | 40 +- debug/risc-v-gsp-rm-vm/harness/package_sim.py | 1139 +++++++++++++++++++ .../harness/port_driver_versions.py | 1155 ++++++++++++++++++++ debug/risc-v-gsp-rm-vm/harness/port_new_device.py | 726 ++++++++++++ debug/risc-v-gsp-rm-vm/tests/test_package_sim.py | 216 ++++ .../tests/test_porting_automation.py | 301 +++++ 13 files changed, 3658 insertions(+), 29 deletions(-) create mode 100644 debug/risc-v-gsp-rm-vm/harness/package_sim.py create mode 100644 debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py create mode 100644 debug/risc-v-gsp-rm-vm/harness/port_new_device.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_package_sim.py create mode 100644 debug/risc-v-gsp-rm-vm/tests/test_porting_automation.py diff --git a/debug/risc-v-gsp-rm-vm/README.md b/debug/risc-v-gsp-rm-vm/README.md index eab691f..627e1d5 100644 --- a/debug/risc-v-gsp-rm-vm/README.md +++ b/debug/risc-v-gsp-rm-vm/README.md @@ -104,4 +104,10 @@ dumps/ recorded sessions (gitignored) | [25-VALUE-VALIDATION](docs/25-VALUE-VALIDATION.md) | value-validation demos (s01–s10) | | [26-ARCHITECTURE](docs/26-ARCHITECTURE.md) | full architecture/API surface map | | [27-TOOL-AUDIT](docs/27-TOOL-AUDIT.md) | tool audit: purpose, tests, usage, status | +| [28-PORTING-AUTOMATION](docs/28-PORTING-AUTOMATION.md) | driver-version sweep and new-device scaffolds (pilot, dry-run default) | +| [29-PACKAGING-AND-TRIALING](docs/29-PACKAGING-AND-TRIALING.md) | trial simulation packages (pilot) | | [GREEN-EYE-ROADMAP](GREEN-EYE-ROADMAP.md) | the long-term program (H1–H7) | + + diff --git a/debug/risc-v-gsp-rm-vm/docs/00-OVERVIEW.md b/debug/risc-v-gsp-rm-vm/docs/00-OVERVIEW.md index 3b56512..e4906cc 100644 --- a/debug/risc-v-gsp-rm-vm/docs/00-OVERVIEW.md +++ b/debug/risc-v-gsp-rm-vm/docs/00-OVERVIEW.md @@ -52,6 +52,7 @@ whole stack scriptable, and QEMU is just the first execution backend. | Host RM oracle harness (`rmharness`) | wrappers + capture working; device tables depend on driver permissions (17) | | First-touch device map + lenient native MMIO | working (24) | | MCP server (`plugins/mcp_server.py`) | working (20); stdlib-only, read-only by default | +| Porting automation + trial packaging (28/29) | scaffolds + tests present; unadmitted pilots (dry-run default) | | Native M2+ device model | roadmap only (08, 16) | ## Quick start @@ -102,6 +103,8 @@ Interactively: | [25-VALUE-VALIDATION](25-VALUE-VALIDATION.md) | value-validation demos s01–s10 | | [26-ARCHITECTURE](26-ARCHITECTURE.md) | full architecture and API surface map | | [27-TOOL-AUDIT](27-TOOL-AUDIT.md) | tool audit: purpose, tests, usage, status | +| [28-PORTING-AUTOMATION](28-PORTING-AUTOMATION.md) | driver-version sweep and new-device scaffolds (pilot) | +| [29-PACKAGING-AND-TRIALING](29-PACKAGING-AND-TRIALING.md) | trial simulation packages (pilot) | | [GREEN-EYE-ROADMAP](../GREEN-EYE-ROADMAP.md) | long-term program H1–H7 | ## Where this fits in the Open-IOV workflow @@ -120,3 +123,8 @@ The intended accuracy loop is: run the same workload against the real GSP-RM (hardware or mdev), record its RPC/state behaviour, then run the firmware in green-eye, record the same events, and compare with `compare` (see 06 and 07). + + diff --git a/debug/risc-v-gsp-rm-vm/docs/21-CI-RELEASE-PIPELINE.md b/debug/risc-v-gsp-rm-vm/docs/21-CI-RELEASE-PIPELINE.md index d687cca..aa71b54 100644 --- a/debug/risc-v-gsp-rm-vm/docs/21-CI-RELEASE-PIPELINE.md +++ b/debug/risc-v-gsp-rm-vm/docs/21-CI-RELEASE-PIPELINE.md @@ -249,11 +249,13 @@ the docs/14 §6 pack; the docs/23 §4 step 7 expected-artifact list maps 1:1. * runs `python3 -m unittest discover -s tests -t .` from the tool root (QEMU integration tests run; if QEMU were absent they skip cleanly). -The intended release gates this workflow covers are the fast inventory smoke -from docs/19 §6 (`control_discovery --max-id 32 --patterns zero`), the harness -suites, and the docs/21 dry-run. A full `ci_release` run against real firmware -is driven manually/agentically because it needs the versioned inputs and -symbol DBs; CI-adjacent automation should call it with `--dry-run` first. +The release gates this workflow reaches are the harness suites: the docs/19 §6 +fast inventory smoke runs inside `tests/test_discovery.py`, and the docs/21 +dry-run is covered by `tests/test_version_diff.py`. The workflow does not run +`control_discovery` or `ci_release` as standalone steps. A full `ci_release` +run against real firmware is driven manually/agentically because it needs the +versioned inputs and symbol DBs; CI-adjacent automation should call it with +`--dry-run` first. --- @@ -318,3 +320,7 @@ or the demo ELF is absent. | **V1 — metrics/golden data** | deterministic self-diff and stable JSON reports support golden comparisons | | **docs/23 C5/C7** | release pipeline implemented and exercisable; dry-run rehearsal path | | **H7 — agentic harness** | self-describing bundle + docs/20 MCP tools for monitored runs | + + diff --git a/debug/risc-v-gsp-rm-vm/docs/22-PORTING-CUDA-AND-DEVICES.md b/debug/risc-v-gsp-rm-vm/docs/22-PORTING-CUDA-AND-DEVICES.md index 30f4720..ad8a5e7 100644 --- a/debug/risc-v-gsp-rm-vm/docs/22-PORTING-CUDA-AND-DEVICES.md +++ b/debug/risc-v-gsp-rm-vm/docs/22-PORTING-CUDA-AND-DEVICES.md @@ -71,7 +71,7 @@ tools/cudatool.py delta /run/libcuda.so. /run/libcuda.so. | Step | Existing project tooling | green-eye addition | |---|---|---| | Export/interface diff | `tools/cudatool.py delta` | — | -| Function matching / name-transfer measurement | `tools/cudatool.py fdematch --names ` | — | +| Function matching / name-transfer measurement | `tools/cudatool.py fdematch --names ` | — | | Corpus-level diff (added/removed/resized functions) | `tools/diff_corpus.py ` | — | | Name transfer + corpus annotation | `ApplyNameMap.java` (`--dry-run` first), `tools/annotate_corpus.py` | `./greeneye.sh symbols export` builds a runtime-address symbol DB from the same index/names | | Driver-API smoke | `tools/test/cuda_drv_test.c` compiled as `/tmp/cdt` (prints `ALL OK`) | `./greeneye.sh symbols resolve --db ... ` spot-checks transferred addresses per call path | @@ -241,9 +241,11 @@ The project's support ladder is `docs/06-portability/hardware-software-matrix.md 1. Run `harness/portability_matrix.py` across the reference and candidate device firmware with the same scenario options (docs/18 §1); keep `/portability-report.json`, per-pair compare reports and sessions. -2. The planned version-diff/CI pipeline (docs/21-CI-RELEASE-PIPELINE.md, - planned) is where those reports become release-gated artifacts; until it - exists, attach them to the matrix process manually. +2. The version-diff/CI pipeline + ([`docs/21`](21-CI-RELEASE-PIPELINE.md)) packages those reports into + release-gated artifacts; the porting scaffolds + ([`docs/28`](28-PORTING-AUTOMATION.md)) wrap it per version as unadmitted + pilots (evaluation in future-eng-ops F2/F3). 3. Update the device profile (documentation today; `config/hardware/*.yaml` is not read by code — do not claim configuration-driven portability). 4. Validate on the physical device with the supervised harness before claiming @@ -345,9 +347,14 @@ Architecture (P3): | Differential execution and reimplementation loop | H4.5 | every port lands with A/B sessions and the M1–M6 table | reference-capture pipeline | | Adjacent-controller work for GB10-class bring-up | H5 | PMU/SEC2/ACR/SOE models, cross-controller sessions | H2.2 | | Multi-target orchestration | R6 / docs/18 | one report across firmware/device targets | v0.1 stable | -| Version diff and release gating | planned (docs/21) | CI pipeline that gates a port on the evidence bundle | pipeline implemented | -| Control discovery | planned (docs/19) | systematic control/class inventory across versions | document exists | +| Version diff and release gating | implemented (docs/21) | CI pipeline that gates a port on the evidence bundle | pipeline implemented | +| Control discovery | implemented (docs/19) | systematic control/class inventory across versions | document exists | Program-level success criterion served here: a new GSP version (or device) can be triaged — inventory diff + behaviour diff + metric table — in **hours**, with recorded evidence. + + diff --git a/debug/risc-v-gsp-rm-vm/docs/23-RM610-READINESS.md b/debug/risc-v-gsp-rm-vm/docs/23-RM610-READINESS.md index 18d93bd..72e0e62 100644 --- a/debug/risc-v-gsp-rm-vm/docs/23-RM610-READINESS.md +++ b/debug/risc-v-gsp-rm-vm/docs/23-RM610-READINESS.md @@ -53,13 +53,13 @@ closed. | # | Criterion | How it is verified | Evidence | Status today | |---|---|---|---|---| -| C1 | All tools implemented and tested on the current driver/firmware | full test suites green (documents/demo/probe/inject/Ghidra/harness) | test logs attached to the request | partial (MCP/CI missing) | -| C2 | Tests green | `make test` and the harness/rmharness suites pass | transcript with date and commit | green at v0.1 (55 tests) | +| C1 | All tools implemented and tested on the current driver/firmware | full test suites green (documents/demo/probe/inject/Ghidra/harness) | test logs attached to the request | tools implemented (docs/00); fresh full-suite run pending | +| C2 | Tests green | `make test` and the harness/rmharness suites pass | transcript with date and commit | green at v0.1; re-run pending (155 tests, docs/27) | | C3 | Demo, probe, inject, Ghidra work | `./greeneye.sh demo`, `probe`, `inject`, `ghidra` all demonstrated | session dirs + transcript | working per docs/00 | -| C4 | MCP monitoring works | MCP plugin implemented and exercised (roadmap R3, docs/08; docs/15) | tool list + a monitored run | **not implemented (planned R3)** | -| C5 | CI pipeline works | release/CI pipeline implemented and exercised (planned docs/21) | pipeline run with artifacts | **not implemented (planned docs/21)** | +| C4 | MCP monitoring works | MCP plugin implemented and exercised (roadmap R3, docs/08; docs/15) | tool list + a monitored run | implemented (docs/20); monitored run not yet attached | +| C5 | CI pipeline works | release/CI pipeline implemented and exercised (docs/21) | pipeline run with artifacts | implemented (docs/21); current-version run pending | | C6 | Docs complete and accurate | doc review checklist (§3.2) signed | review record | to complete | -| C7 | Discovery/version-diff/CI exercised on the current version | control discovery (planned docs/19) + `harness/portability_matrix.py` + `compare` + pipeline dry-run on 570.144 | report + session hashes | partial (matrix/compare exist; docs/19 planned) | +| C7 | Discovery/version-diff/CI exercised on the current version | control discovery (docs/19) + `harness/portability_matrix.py` + `compare` + pipeline dry-run on 570.144 | report + session hashes | partial (docs/19/21 implemented; current-version run pending) | | C8 | Harness produces a full evidence bundle | docs/07 checklist + docs/14 §6 pack produced end to end for a current-version port rehearsal | bundle directory with SHA-256 inventory | to complete | | C9 | Human has approved | signed message (§6) | approval ID recorded in the request thread | pending | @@ -116,9 +116,9 @@ Required results: tables, test counts); * [ ] `docs/22` porting guide is accurate for the current tools; planned items are labelled `planned (docs/xx)`; -* [ ] `docs/19-CONTROL-DISCOVERY.md` (planned) exists and has been run once on - the current version, or C7 is explicitly waived; -* [ ] `docs/21-CI-RELEASE-PIPELINE.md` (planned) exists and has run once on the +* [ ] `docs/19-CONTROL-DISCOVERY.md` exists and has been run once on the + current version, or C7 is explicitly waived; +* [ ] `docs/21-CI-RELEASE-PIPELINE.md` exists and has run once on the current version, or C5 is explicitly waived; * [ ] `GREEN-EYE-ROADMAP.md` and `docs/08` status claims match reality; * [ ] no local source paths, no invented command names, no unsupported @@ -158,10 +158,11 @@ must not happen beforehand. `harness/portability_matrix.py` with the 610 and 570.144 targets and identical scenario options; run the mutation harness once per version with the same seed; archive both reports and every session. -6. **Pipeline dry-run, then real.** Run the release pipeline (planned - docs/21-CI-RELEASE-PIPELINE.md) in its dry-run mode first; inspect the - planned artifact list without publishing. Only after the dry-run is clean, - run the real pipeline on the 610 evidence bundle. +6. **Pipeline dry-run, then real.** Run the release pipeline + ([`21-CI-RELEASE-PIPELINE.md`](21-CI-RELEASE-PIPELINE.md)) in its dry-run + mode first; inspect the planned artifact list without publishing. Only + after the dry-run is clean, run the real pipeline on the 610 evidence + bundle. 7. **Expected artifacts.** * approval record and input hashes; * 610 symbol DB + Ghidra export provenance; @@ -257,3 +258,9 @@ Approval ID: ; Date: ; Scope limits: After approval, the 610 port proceeds only through the docs/22 workflow and the docs/05-operations §4.1 version discipline. Until that message exists, this repository is a 610 no-fly zone. + + diff --git a/debug/risc-v-gsp-rm-vm/docs/26-ARCHITECTURE.md b/debug/risc-v-gsp-rm-vm/docs/26-ARCHITECTURE.md index 326533b..ab8414a 100644 --- a/debug/risc-v-gsp-rm-vm/docs/26-ARCHITECTURE.md +++ b/debug/risc-v-gsp-rm-vm/docs/26-ARCHITECTURE.md @@ -259,9 +259,9 @@ argument schemas for all 24 tools are in 20 §5. ## 9. Automation harness `harness/` turns the playbooks into repeatable, self-describing runs. Modules -bootstrap the tool root and use the QEMU backend by default (18, 19, 21); -`common.py` provides flat-image relocation, session helpers and bounded log -reads. +bootstrap the tool root and use the QEMU backend by default (18, 19, 21, 28, +29); `common.py` provides flat-image relocation, session helpers and bounded +log reads. ```mermaid flowchart TB @@ -293,6 +293,9 @@ flowchart TB | `version_diff.py` | old vs new image: shape, symbols, corpus hints, probe, behavior, M1–M6 | `version-diff.json/.md`, `corpus/`, `compare/` | 21 §2 | | `ci_release.py` | version_diff → matrix → mutation smoke, then hash the bundle | `release-report.json`, `bundle-manifest.json` | 21 §4 | | `value_demos/run_all.py` | ten fixed scenarios (reply, counters, discovery, reach, diff, inject, version, mutation, Ghidra, evidence) | `artifacts/results.json` + per-demo JSON | 25 | +| `port_driver_versions.py` | version-list sweep of version_diff + matrix + mutation (dry-run default; unadmitted pilot) | per-version bundles + `sweep-summary.json/.md` | 28 §2 | +| `port_new_device.py` | device-port scaffold: profile, probe, symbols, matrix, checklist (dry-run default; unadmitted pilot) | `device-port.json/.md`, `port-checklist.json/.md`, bundle | 28 §3 | +| `package_sim.py` | build/run a trial simulation package (unadmitted pilot) | package tree + `manifest.json`, `results.json` | 29 | | `common.py` | shared setup (relocate to `0x80000000`, open session, bounded logs) | library | 18 §1.2 | Harness limits are intentional: QEMU/re-hosted/demo code only; no hardware, @@ -442,8 +445,12 @@ Invariants (enforced by tests or review): Section diagrams and their backing docs: §2 suite (00, 01, README); §3 dispatch (01, 03); §4 backends (01 §8, 16 §4–5); §5 recording (01 §6, 09, 12 §6, 16 §10, 17 §5); §6 injection (11, 12); §7 Ghidra (05, 13); §8 MCP (20); -§9 harness (18, 19, 21); §10 device map (24, 16 §7); §11 OODA (06, 07 P4, 25 +§9 harness (18, 19, 21, 28, 29); §10 device map (24, 16 §7); §11 OODA (06, 07 P4, 25 §7). Roadmap keys: R3 = MCP (08), R5 = mutation (18), R6 = multi-version (18, 21), H1.3 = RM service injection (GREEN-EYE-ROADMAP), H2.2 = QEMU device model, H3/M1–M6 = native simulator (16), H4 = adversarial navigation (14), H7 = agentic platform (15, 20). + + diff --git a/debug/risc-v-gsp-rm-vm/docs/27-TOOL-AUDIT.md b/debug/risc-v-gsp-rm-vm/docs/27-TOOL-AUDIT.md index e208253..86f2b06 100644 --- a/debug/risc-v-gsp-rm-vm/docs/27-TOOL-AUDIT.md +++ b/debug/risc-v-gsp-rm-vm/docs/27-TOOL-AUDIT.md @@ -250,3 +250,18 @@ Files changed by this audit: Verification after the changes: 155/155 unit tests OK; docs-link check OK (55 files); every smoke in section 0 re-run green. + +## 11. Post-audit additions (not yet audited) + +`harness/port_driver_versions.py`, `harness/port_new_device.py` and +`harness/package_sim.py` with docs 28/29 landed after this audit; their tests +are `tests/test_porting_automation.py` and `tests/test_package_sim.py` (main +suite). They are unadmitted pilots under `eng-ops.md` §2 and have not had a +tool-audit pass, so treat their status as `needs improvement (open)` until one +does; docs 28/29 carry the exit-code and failure readings that the audit would +otherwise record. + + 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( + " 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 .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"(? 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 "" + 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 && 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 `` 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 "", "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 "" + 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: /profiles/.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()) diff --git a/debug/risc-v-gsp-rm-vm/tests/test_package_sim.py b/debug/risc-v-gsp-rm-vm/tests/test_package_sim.py new file mode 100644 index 0000000..c6df626 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_package_sim.py @@ -0,0 +1,216 @@ +"""Tests for the packaging/trialing feature (docs/29, harness/package_sim.py). + +Builds real packages from the bundled demo plus the SDK service, checks the +layout and manifest hashes, runs ``run.sh`` in a bounded subprocess and asserts +the injected call results, round-trips the ``--pack-image`` combined artifact, +and checks manifest/archive determinism. Skips cleanly when the demo firmware +or the SDK blob has not been built (``make demo && make sdk``); the real-image +test additionally skips when the 570.144 fwimage is absent. +""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.normpath(os.path.join(HERE, "..")) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +import harness.package_sim as pk # noqa: E402 +from greeneye.elf import load_elf # noqa: E402 + +DEMO_ELF = pk.DEMO_ELF +SERVICE_ELF = pk.DEFAULT_SERVICE +REAL_IMAGE = pk.REAL_IMAGE +EXPECTED_PING = 0x41 ^ pk.PING_XOR +EXPECTED_SUM = 0x30 + +def artifacts_available() -> bool: + return os.path.isfile(DEMO_ELF) and os.path.isfile(SERVICE_ELF) + + +def run_package(out: str, args: tuple = (), timeout: float = 120.0): + env = dict(os.environ) + env["GREENEYE_HOME"] = TOOL_ROOT + return subprocess.run( + ["/bin/sh", os.path.join(out, "run.sh"), *args], + cwd=out, + env=env, + capture_output=True, + text=True, + timeout=timeout, + ) + + +def load_results(out: str) -> dict: + with open(os.path.join(out, "results", "results.json"), "r", encoding="utf-8") as fh: + return json.load(fh) + + +@unittest.skipUnless(artifacts_available(), "demo firmware or SDK service not built (make demo && make sdk)") +class PackageTests(unittest.TestCase): + def test_layout_and_manifest_hashes(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "trial") + manifest = pk.build_package(pk.PackageRequest(out=out)) + + for relpath in ( + "README.md", + "run.sh", + "manifest.json", + "scenario.py", + "payload/uproc_demo.elf", + "payload/injected_demo.elf", + "results/.gitkeep", + ): + self.assertTrue(os.path.isfile(os.path.join(out, relpath)), relpath) + + self.assertEqual(manifest["format"], pk.FORMAT) + self.assertEqual(manifest["target"]["kind"], "demo") + self.assertTrue(manifest["target"]["runnable"]) + self.assertEqual(manifest["target"]["sha256"], pk.sha256_file(DEMO_ELF)) + self.assertEqual(manifest["service"]["sha256"], pk.sha256_file(SERVICE_ELF)) + self.assertEqual(manifest["service"]["entry_symbol"], "ge_demo_service_dispatch") + + paths = [comp["path"] for comp in manifest["components"]] + self.assertEqual(paths, sorted(paths), "component order must be stable") + for comp in manifest["components"]: + full = os.path.join(out, comp["path"]) + self.assertEqual(comp["sha256"], pk.sha256_file(full), comp["path"]) + self.assertEqual(comp["bytes"], os.path.getsize(full), comp["path"]) + script = next(comp for comp in manifest["components"] if comp["path"] == "scenario.py") + self.assertEqual(script["sha256"], pk.sha256_bytes(pk.BUILTIN_SCENARIO.encode("utf-8"))) + + def test_run_sh_writes_passing_results(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "trial") + manifest = pk.build_package(pk.PackageRequest(out=out)) + proc = run_package(out) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + + results = load_results(out) + self.assertEqual(results["status"], "ok") + self.assertTrue(results["success"]) + self.assertEqual(results["calls"]["remote"]["ping"], EXPECTED_PING) + self.assertEqual(results["calls"]["remote"]["sum"], EXPECTED_SUM) + self.assertEqual(results["calls"]["via_firmware"], results["calls"]["remote"]) + self.assertTrue(results["calls"]["agree"]) + self.assertEqual(results["assertions"]["failed"], []) + self.assertGreaterEqual(results["hooks"]["rpc_dispatch"]["hits"], 1) + self.assertGreaterEqual(results["watches"]["gsp_ctrl.result"]["hits"], 1) + self.assertTrue(results["assertions"]["checks"]["uart_banner"]) + self.assertFalse(results["service"]["preinjected"]) + # Non-preinjected packages use greeneye.inject's default load address. + self.assertEqual(results["handlers_slot"], 0x81000000 + manifest["service"]["entry"]) + self.assertTrue(os.path.isfile(os.path.join(out, "results", "session", "session.json"))) + self.assertTrue(os.path.isfile(os.path.join(out, "results", "results.md"))) + + def test_pack_image_round_trip(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "trial") + manifest = pk.build_package(pk.PackageRequest(out=out, pack_image=True)) + self.assertIsNotNone(manifest["preinject"]) + self.assertTrue(manifest["preinject"]["runnable"]) + combined = os.path.join(out, "payload", "combined.elf") + preinject = os.path.join(out, "payload", "preinject.json") + self.assertTrue(os.path.isfile(combined)) + self.assertTrue(os.path.isfile(preinject)) + + demo = load_elf(DEMO_ELF) + service = load_elf(SERVICE_ELF) + merged = load_elf(combined) + self.assertEqual(merged.entry, demo.entry) + self.assertEqual( + len(merged.load_segments), + len(demo.load_segments) + len(service.load_segments), + ) + + proc = run_package(out) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + results = load_results(out) + self.assertEqual(results["status"], "ok") + self.assertTrue(results["service"]["preinjected"]) + self.assertEqual(results["calls"]["remote"], results["calls"]["via_firmware"]) + self.assertEqual(results["calls"]["remote"]["ping"], EXPECTED_PING) + self.assertEqual(results["calls"]["remote"]["sum"], EXPECTED_SUM) + + def test_manifest_and_tar_deterministic(self): + with mock.patch.dict(os.environ, {"SOURCE_DATE_EPOCH": "1700000000"}): + with tempfile.TemporaryDirectory() as tmp_a, tempfile.TemporaryDirectory() as tmp_b: + out_a = os.path.join(tmp_a, "trial") + out_b = os.path.join(tmp_b, "trial") + pk.build_package(pk.PackageRequest(out=out_a, pack_image=True, tar=True)) + pk.build_package(pk.PackageRequest(out=out_b, pack_image=True, tar=True)) + with open(os.path.join(out_a, "manifest.json"), "rb") as fh_a: + manifest_a = fh_a.read() + with open(os.path.join(out_b, "manifest.json"), "rb") as fh_b: + manifest_b = fh_b.read() + self.assertEqual(manifest_a, manifest_b) + with open(out_a + ".tar.gz", "rb") as fh_a: + tar_a = fh_a.read() + with open(out_b + ".tar.gz", "rb") as fh_b: + tar_b = fh_b.read() + self.assertEqual(tar_a, tar_b) + + def test_cli_build_and_run(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "trial") + proc = subprocess.run( + [ + sys.executable, + os.path.join(TOOL_ROOT, "harness", "package_sim.py"), + "--pack-image", + "--tar", + "--run", + "--out", + out, + ], + cwd=TOOL_ROOT, + capture_output=True, + text=True, + timeout=180.0, + ) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertTrue(os.path.isfile(out + ".tar.gz")) + self.assertEqual(load_results(out)["status"], "ok") + + +@unittest.skipUnless(artifacts_available(), "demo firmware or SDK service not built") +@unittest.skipUnless(os.path.isfile(REAL_IMAGE), "570.144 fwimage absent") +class ImagePackageTests(unittest.TestCase): + def test_image_package_is_bounded_and_honest(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "trial") + manifest = pk.build_package(pk.PackageRequest(out=out, target="image")) + self.assertEqual(manifest["target"]["kind"], "image") + self.assertFalse(manifest["target"]["runnable"]) + self.assertTrue(os.path.isfile(os.path.join(out, "payload", os.path.basename(REAL_IMAGE)))) + self.assertTrue(os.path.isfile(os.path.join(out, "payload", "injected_demo.elf"))) + self.assertTrue(os.path.isfile(os.path.join(out, "payload", "device_map.json"))) + self.assertTrue(manifest["device_map"]["learn"]) + with open(os.path.join(out, "README.md"), "r", encoding="utf-8") as fh: + self.assertIn("cannot boot", fh.read()) + + proc = run_package(out, args=("--budget", "20000")) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + results = load_results(out) + self.assertEqual(results["status"], "blocked") + self.assertTrue(results["blocked_on_device_model"]) + self.assertFalse(results["success"]) + self.assertGreater(results["instructions_executed"], 0) + self.assertGreaterEqual(results["mmio_unknown"], 1) + self.assertTrue(results["first_unknown_mmio"]) + self.assertFalse(results["service"]["loaded"]) + self.assertTrue(os.path.isfile(os.path.join(out, "results", "learned_device_map.json"))) + with open(os.path.join(out, "results", "results.md"), "r", encoding="utf-8") as fh: + self.assertIn("cannot boot", fh.read()) + + +if __name__ == "__main__": + unittest.main() diff --git a/debug/risc-v-gsp-rm-vm/tests/test_porting_automation.py b/debug/risc-v-gsp-rm-vm/tests/test_porting_automation.py new file mode 100644 index 0000000..5c7e795 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/tests/test_porting_automation.py @@ -0,0 +1,301 @@ +"""Tests for the porting automation chain (docs/28, roadmap H4.1/R6). + +Pure/CLI tests need neither QEMU nor the demo; the one-version sweep test +boots the demo under QEMU and skips cleanly when qemu-system-riscv64 or the +demo ELF is absent. Everything else runs from a temp directory and is +side-effect free. +""" +from __future__ import annotations + +import contextlib +import io +import json +import os +import sys +import tempfile +import unittest +from unittest import mock + +HERE = os.path.dirname(os.path.abspath(__file__)) +TOOL_ROOT = os.path.abspath(os.path.join(HERE, "..")) +DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf") +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from harness import port_driver_versions as pdv # noqa: E402 +from harness import port_new_device as pnd # noqa: E402 +from harness.common import qemu_available # noqa: E402 +from harness.version_diff import check_rm610_gate # noqa: E402 + +NOP = b"\x13\x00\x00\x00" # addi x0, x0, 0 + + +def _run_main(main_fn, argv: list[str]) -> tuple[int, str, str]: + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = main_fn(argv) + return rc, out.getvalue(), err.getvalue() + + +def _plan_from_stdout(stdout: str) -> dict: + """The dry-run prints one JSON plan followed by [tool] status lines.""" + marker = stdout.find("\n[sweep]") + if marker < 0: + marker = stdout.find("\n[device]") + return json.loads(stdout[:marker] if marker >= 0 else stdout) + + +class VersionHelperTests(unittest.TestCase): + def test_parse_versions(self): + self.assertEqual(pdv.parse_versions("540,550,560,570"), ["540", "550", "560", "570"]) + self.assertEqual(pdv.parse_versions(" 570.144 , 570.144 "), ["570.144"]) + self.assertEqual(pdv.parse_versions(""), []) + with self.assertRaises(ValueError): + pdv.parse_versions("540$") + + def test_version_ge_610(self): + self.assertTrue(pdv.version_ge_610("610")) + self.assertTrue(pdv.version_ge_610("610.1")) + self.assertTrue(pdv.version_ge_610("620")) + self.assertFalse(pdv.version_ge_610("570.144")) + self.assertFalse(pdv.version_ge_610("609.99")) + + def test_default_cadence_is_documented_and_not_invented(self): + self.assertIn("540", pdv.DEFAULT_VERSIONS) + self.assertIn("570", pdv.DEFAULT_VERSIONS) + self.assertNotIn("610", pdv.DEFAULT_VERSIONS) + self.assertIn("DEFAULT_VERSIONS", pdv.__doc__ or "") + + def test_version_sort_key(self): + values = sorted(["560", "570.144", "540", "550"], key=pdv.version_sort_key) + self.assertEqual(values, ["540", "550", "560", "570.144"]) + + def test_non_rm610_labels_pass_the_gate(self): + check_rm610_gate(["570.144", "uproc_demo"], environ={}) + + +class SweepDryRunTests(unittest.TestCase): + def test_help_is_clean(self): + for main_fn in (pdv.main, pnd.main): + with contextlib.redirect_stdout(io.StringIO()) as buf, self.assertRaises(SystemExit) as ctx: + main_fn(["--help"]) + self.assertEqual(ctx.exception.code, 0) + self.assertIn("usage", buf.getvalue()) + + def test_help_documents_cron_and_agent_invocation(self): + with contextlib.redirect_stdout(io.StringIO()) as buf, self.assertRaises(SystemExit): + pdv.main(["--help"]) + text = buf.getvalue() + self.assertIn("cron", text) + self.assertIn("agent", text) + self.assertIn("RM610-APPROVED", text) + + def test_dry_run_explicit_target_exits_zero_and_creates_nothing(self): + with tempfile.TemporaryDirectory() as tmp: + image = os.path.join(tmp, "gsp-570.144.bin") + with open(image, "wb") as fh: + fh.write(NOP * 4) + out = os.path.join(tmp, "out") + rc, stdout, _ = _run_main( + pdv.main, + ["--versions", "570.144", "--target", f"570.144={image}", "--out", out], + ) + self.assertEqual(rc, 0) + plan = _plan_from_stdout(stdout) + self.assertEqual(plan["format"], pdv.PLAN_FORMAT) + self.assertEqual(plan["found"], 1) + self.assertEqual(plan["versions"][0]["status"], "found") + self.assertFalse(os.path.exists(out), "dry-run must not create --out") + + def test_dry_run_missing_input_dir_is_clean(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + rc, stdout, _ = _run_main( + pdv.main, + ["--versions", "540,550", "--input-dir", os.path.join(tmp, "absent"), "--out", out], + ) + self.assertEqual(rc, 0) + plan = _plan_from_stdout(stdout) + self.assertEqual(plan["missing"], 2) + self.assertTrue(all(row["status"] == "no-input" for row in plan["versions"])) + self.assertFalse(os.path.exists(out)) + + def test_dry_run_rm610_refused_before_out(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + rc, _, err = _run_main(pdv.main, ["--versions", "610", "--out", out]) + self.assertEqual(rc, 3) + self.assertIn("docs/23", err) + self.assertFalse(os.path.exists(out)) + + def test_dry_run_rm610_passes_with_token(self): + rc, stdout, _ = _run_main( + pdv.main, ["--versions", "610", "--approval-token", "RM610-APPROVED"] + ) + self.assertEqual(rc, 0) + self.assertEqual(_plan_from_stdout(stdout)["rm610_gate"]["status"], "passed") + + def test_config_file_drives_the_plan(self): + with tempfile.TemporaryDirectory() as tmp: + image = os.path.join(tmp, "gsp-550.bin") + with open(image, "wb") as fh: + fh.write(NOP * 4) + config = os.path.join(tmp, "sweep.json") + with open(config, "w", encoding="utf-8") as fh: + json.dump({ + "format": pdv.CONFIG_FORMAT, + "versions": ["550"], + "baseline": "550", + "targets": {"550": [image]}, + }, fh) + rc, stdout, _ = _run_main(pdv.main, ["--config", config]) + self.assertEqual(rc, 0) + plan = _plan_from_stdout(stdout) + self.assertEqual(plan["baseline"], "550") + self.assertEqual(plan["found"], 1) + + def test_config_rejects_unknown_keys(self): + with tempfile.TemporaryDirectory() as tmp: + config = os.path.join(tmp, "bad.json") + with open(config, "w", encoding="utf-8") as fh: + fh.write('{"versons": ["540"]}') + with self.assertRaises(ValueError): + pdv.load_config(config) + + +class SweepRunTests(unittest.TestCase): + def test_help_lists_exit_interpretation(self): + with contextlib.redirect_stdout(io.StringIO()) as buf, self.assertRaises(SystemExit): + pdv.main(["--help"]) + self.assertIn("--run", buf.getvalue()) + + def test_run_without_inputs_fails_cleanly(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + rc, stdout, err = _run_main( + pdv.main, + ["--run", "--versions", "540,550", "--input-dir", os.path.join(tmp, "absent"), + "--out", out], + ) + self.assertEqual(rc, 1) + self.assertIn("no baseline target", stdout) + self.assertNotIn("Traceback", err) + self.assertTrue(os.path.exists(os.path.join(out, "sweep-summary.json"))) + + def test_completed_status_only_accepts_resumable_states(self): + with tempfile.TemporaryDirectory() as tmp: + for state, expected in (("ok", "ok"), ("partial", "partial"), + ("error", None), ("no-input", None)): + version_dir = os.path.join(tmp, state) + os.makedirs(version_dir, exist_ok=True) + with open(os.path.join(version_dir, "version.json"), "w", encoding="utf-8") as fh: + json.dump({"status": state}, fh) + self.assertEqual(pdv.completed_status(version_dir), expected, state) + + def test_run_resumes_completed_versions(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + for version in ("1", "2"): + version_dir = os.path.join(out, version) + os.makedirs(version_dir) + with open(os.path.join(version_dir, "version.json"), "w", encoding="utf-8") as fh: + json.dump({"status": "ok"}, fh) + fake = { + version: pdv.SweepTarget(version=version, paths=[os.path.join(tmp, f"{version}.bin")], + kind="raw", source="explicit") + for version in ("1", "2") + } + options = pdv.SweepOptions(versions=["1", "2"], out=out, run=True, step_timeout=5) + with mock.patch.object(pdv, "resolve_targets", return_value=fake): + report = pdv.run_sweep(options) + self.assertEqual(report["failed"], []) + self.assertEqual(report["ran"], 2) + self.assertTrue(all(row.get("resumed") for row in report["versions"])) + self.assertTrue(os.path.exists(os.path.join(out, "bundle-manifest.json"))) + + +class DevicePortDryRunTests(unittest.TestCase): + def test_dry_run_exits_zero_and_creates_nothing(self): + with tempfile.TemporaryDirectory() as tmp: + firmware = os.path.join(tmp, "gsp-570.144.bin") + with open(firmware, "wb") as fh: + fh.write(NOP * 4) + out = os.path.join(tmp, "out") + rc, stdout, _ = _run_main( + pnd.main, ["--device", "rtx4090", "--firmware", firmware, "--out", out] + ) + self.assertEqual(rc, 0) + plan = _plan_from_stdout(stdout) + self.assertEqual(plan["format"], pnd.PLAN_FORMAT) + self.assertEqual(plan["profile"]["action"], "would-create-scaffold") + self.assertTrue(plan["firmware"]["found"]) + self.assertFalse(os.path.exists(out)) + + def test_dry_run_rm610_refused(self): + rc, _, err = _run_main(pnd.main, ["--device", "rm610-box"]) + self.assertEqual(rc, 3) + self.assertIn("docs/23", err) + + def test_dry_run_rm610_passes_with_env(self): + with mock.patch.dict(os.environ, {"GREENEYE_RM610_APPROVED": "1"}): + rc, stdout, _ = _run_main(pnd.main, ["--device", "rm610-box"]) + self.assertEqual(rc, 0) + self.assertEqual(_plan_from_stdout(stdout)["rm610_gate"]["status"], "passed") + + def test_run_without_firmware_fails_cleanly(self): + with tempfile.TemporaryDirectory() as tmp: + out = os.path.join(tmp, "out") + rc, _, err = _run_main(pnd.main, ["--run", "--device", "testdev", "--out", out]) + self.assertEqual(rc, 2) + self.assertIn("no device firmware", err) + self.assertNotIn("Traceback", err) + + def test_scaffold_profile_has_all_sections(self): + profile = pnd.scaffold_profile("testdev") + self.assertEqual(profile["format"], pnd.PROFILE_FORMAT) + for key in ("pci", "bars", "fb", "engines", "interrupts", "rom", "physical_validation"): + self.assertIn(key, profile) + self.assertEqual(sorted(pnd.profile_missing_fields(profile)), + ["bars", "engines", "fb", "interrupts", "rom"]) + + +@unittest.skipUnless(qemu_available(), "qemu-system-riscv64 not installed") +@unittest.skipUnless(os.path.exists(DEMO_ELF), "demo firmware not built (make demo)") +class OneVersionSweepQemuTests(unittest.TestCase): + def test_one_version_sweep_writes_the_evidence_bundle(self): + with tempfile.TemporaryDirectory(prefix="port-sweep-") as tmp: + out = os.path.join(tmp, "out") + rc, stdout, err = _run_main( + pdv.main, + ["--run", "--versions", "demo", "--target", f"demo={DEMO_ELF}", + "--baseline", "demo", "--out", out, + "--probe-instructions", "16", "--probe-duration", "0.05", + "--scenario-duration", "0.05", "--max-stops", "50", + "--mutation-iterations", "1", "--mutation-seed", "1", + "--mutation-timeout", "0.05", "--mutation-settle", "0.01", + "--step-timeout", "180"], + ) + self.assertEqual(rc, 0, err + stdout) + version_dir = os.path.join(out, "demo") + with open(os.path.join(version_dir, "version.json"), encoding="utf-8") as fh: + version = json.load(fh) + self.assertEqual(version["status"], "ok", version) + self.assertEqual(pdv.completed_status(version_dir), "ok") + for rel in ("version-diff/version-diff.json", + "portability-matrix/portability-report.json", + "mutation-smoke/findings.jsonl", + "bundle-manifest.json"): + self.assertTrue(os.path.exists(os.path.join(version_dir, rel)), rel) + with open(os.path.join(out, "sweep-summary.json"), encoding="utf-8") as fh: + summary = json.load(fh) + self.assertEqual(summary["failed"], []) + self.assertEqual(summary["ran"], 1) + self.assertTrue(summary["interpretation"].startswith("pass")) + with open(os.path.join(version_dir, "bundle-manifest.json"), encoding="utf-8") as fh: + manifest = json.load(fh) + paths = {row["path"] for row in manifest["files"]} + self.assertIn("version-diff/version-diff.json", paths) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3