summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorArthur Hanson Rasmusson <arthur@vgpu.io>2026-10-02 11:30:46 -0400
committerArthur Hanson Rasmusson <arthur@vgpu.io>2026-10-02 11:30:46 -0400
commit2fd653044127a665e4677a2cdfb58120d7b3e16b (patch)
tree5adaeaecb30c4c553e79cd0d674d7b4029905937
parentf02ba2b6b0059e55e4777cb3a2e6fa87dc6438f3 (diff)
downloadopen-iov-2fd653044127a665e4677a2cdfb58120d7b3e16b.tar.gz
open-iov-2fd653044127a665e4677a2cdfb58120d7b3e16b.zip
debug: green-eye porting automation - new-device and driver-version port scripts, simulator packaging, porting docs; control-discovery fix
Diffstat
-rw-r--r--debug/risc-v-gsp-rm-vm/README.md6
-rw-r--r--debug/risc-v-gsp-rm-vm/docs/00-OVERVIEW.md8
-rw-r--r--debug/risc-v-gsp-rm-vm/docs/21-CI-RELEASE-PIPELINE.md16
-rw-r--r--debug/risc-v-gsp-rm-vm/docs/22-PORTING-CUDA-AND-DEVICES.md19
-rw-r--r--debug/risc-v-gsp-rm-vm/docs/23-RM610-READINESS.md31
-rw-r--r--debug/risc-v-gsp-rm-vm/docs/26-ARCHITECTURE.md15
-rw-r--r--debug/risc-v-gsp-rm-vm/docs/27-TOOL-AUDIT.md15
-rw-r--r--debug/risc-v-gsp-rm-vm/harness/control_discovery.py40
-rw-r--r--debug/risc-v-gsp-rm-vm/harness/package_sim.py1139
-rw-r--r--debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py1155
-rw-r--r--debug/risc-v-gsp-rm-vm/harness/port_new_device.py726
-rw-r--r--debug/risc-v-gsp-rm-vm/tests/test_package_sim.py216
-rw-r--r--debug/risc-v-gsp-rm-vm/tests/test_porting_automation.py301
13 files changed, 3658 insertions, 29 deletions
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) |
+
+<!-- change (2026-10-02): add docs 28/29 to the documentation map; scope: doc
+map only; benefit: new tool-tree documents are linked from the tree front
+door. -->
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).
+
+<!-- change (2026-10-02): add docs 28/29 to the status table and document map
+and mark their scripts as unadmitted pilots; scope: status table and document
+map only; benefit: the newest tool tree docs are discoverable and not mistaken
+for admitted tooling. -->
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 |
+
+<!-- change (2026-10-02): state exactly which release gates the CI workflow
+runs (unit suites) instead of implying standalone control_discovery/ci_release
+steps; scope: §5 only; benefit: no false claim about CI coverage. -->
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 <old>/run/libcuda.so.<old> <new>/run/libcuda.so.<new>
| Step | Existing project tooling | green-eye addition |
|---|---|---|
| Export/interface diff | `tools/cudatool.py delta` | — |
-| Function matching / name-transfer measurement | `tools/cudatool.py fdematch --names <names.tsv>` | — |
+| Function matching / name-transfer measurement | `tools/cudatool.py fdematch <old-lib> <new-lib> --names <names.tsv>` | — |
| Corpus-level diff (added/removed/resized functions) | `tools/diff_corpus.py <old_index> <new_index>` | — |
| 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 ... <addr>` 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
`<out>/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.
+
+<!-- change (2026-10-02): fix the fdematch invocation (it needs two libraries),
+and replace "planned docs/19/21" with their implemented status plus the docs/28
+scaffolds; scope: §2.2, §3.5, §6 only; benefit: commands run as written and
+status tables match the tree. -->
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: <id>; Date: <date>; Scope limits: <text>
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.
+
+<!-- change (2026-10-02): correct C1/C2/C4/C5/C7 status and drop "planned" for
+docs/19/21, which are implemented in the tree; scope: readiness table, §3.2
+checklist and §4 step 6 only; benefit: the gate record no longer claims
+implemented tooling is missing, and the still-open items (C6/C8/C9 and the
+current-version exercise) stay visible. -->
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).
+
+<!-- change (2026-10-02): list the docs 28/29 harness scripts in the automation
+table and index so the architecture map matches the tree; scope: §9, §14 only;
+benefit: the newest entry points are discoverable and labelled as pilots. -->
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.
+
+<!-- change (2026-10-02): record the post-audit porting/packaging scaffolds as
+unaudited rather than leaving the audit silent about them; scope: new §11 only;
+benefit: the audit's scope statement stays true for the commit it describes
+and the gap to the new code is explicit. -->
diff --git a/debug/risc-v-gsp-rm-vm/harness/control_discovery.py b/debug/risc-v-gsp-rm-vm/harness/control_discovery.py
index eecf082..b44636a 100644
--- a/debug/risc-v-gsp-rm-vm/harness/control_discovery.py
+++ b/debug/risc-v-gsp-rm-vm/harness/control_discovery.py
@@ -115,6 +115,10 @@ SETTLE_MARKER = "SELFTEST DONE"
SETTLE_SLICE = 0.02
RUN_SLICE = 0.02
TRAP_GRACE = 0.05
+#: Bound on instruction-driven single-steps used to retire a dispatcher's
+#: store sequence after the wall-clock observation window expires mid-epilogue
+#: (keeps findings byte-reproducible instead of timer-dependent).
+EPILOGUE_MAX_STEPS = 64
#: gdb signals that carry a hazard (illegal, trap, bus, segv); SIGINT from the
#: budget interrupt is wall-clock dependent and must never enter a finding.
HAZARD_SIGNALS = (4, 5, 7, 11)
@@ -662,12 +666,16 @@ class ProbeRunner:
Completion is a ``result`` store (the dispatcher writes it after every
handler) or a trap/grace for ids that never return to the dispatcher.
+ The wall-clock loop is the budget for the no-progress/hung cases; when
+ a dispatch has been seen the store sequence is then retired with
+ bounded single-steps so the recorded effects cannot depend on where
+ the timer happened to expire.
"""
deadline = time.monotonic() + self.options.timeout
handler_seen_at: float | None = None
while time.monotonic() < deadline:
if not session.target.alive():
- return
+ break
remaining = max(0.005, min(RUN_SLICE, deadline - time.monotonic()))
session.run(duration=remaining, max_stops=64)
if handlers and handler_seen_at is None:
@@ -675,9 +683,37 @@ class ProbeRunner:
result_written = any(hit.get("field") == "result" for hit in observer.hits)
trapped = any(sig.get("signal") == 5 for sig in signals) or session.pc < 0x100
if dispatches and (result_written or trapped):
- return
+ break
if handler_seen_at is not None and time.monotonic() - handler_seen_at >= TRAP_GRACE:
# Slow/hung handler: one grace window after the entry, then stop.
+ break
+ self._retire_dispatch(session, dispatches, signals, observer)
+
+ def _retire_dispatch(self, session, dispatches, signals, observer) -> None:
+ """Finish a dispatched command's store sequence with single-steps.
+
+ The demo dispatcher stores ``result``, ``last_rpc`` and finally
+ ``rpc_count``; a wall-clock observation window can expire between
+ those stores, which made findings timing-dependent. Once a dispatch
+ has been seen, step (bounded and instruction-driven) until the final
+ watched store retires, a trap is seen, or the budget is exhausted.
+ """
+ if not dispatches or not any(watched.name == "rpc_count" for watched in observer.watched):
+ return
+ if any(hit.get("field") == "rpc_count" for hit in observer.hits):
+ return
+ if any(sig.get("signal") == 5 for sig in signals):
+ return
+ for _ in range(EPILOGUE_MAX_STEPS):
+ if not session.target.alive():
+ return
+ try:
+ session.step()
+ except (RspTimeout, GreenEyeError):
+ return
+ if any(hit.get("field") == "rpc_count" for hit in observer.hits):
+ return
+ if any(sig.get("signal") == 5 for sig in signals):
return
def _keep(self, outcome: str) -> bool:
diff --git a/debug/risc-v-gsp-rm-vm/harness/package_sim.py b/debug/risc-v-gsp-rm-vm/harness/package_sim.py
new file mode 100644
index 0000000..28f54c4
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/package_sim.py
@@ -0,0 +1,1139 @@
+#!/usr/bin/env python3
+"""Package the native green-eye simulator together with injected code.
+
+Builds a self-contained trial directory (or deterministic ``.tar.gz``):
+
+ README.md what this is, how to run, what to expect, safety
+ run.sh single entry: native simulator + scenario, writes results
+ manifest.json format/version, hashes, component list, run contract
+ payload/ target image/ELF, service blob, device map, preinject config
+ scenario.py the pre-scripted scenario as run (builtin or --script)
+ results/ produced by run.sh (results.json, results.md, session)
+
+The trial boots the target natively, loads/registers the injected service,
+invokes it directly and through the firmware dispatcher, runs one hook/watch
+pass, and writes machine-readable and human-readable results. ``--pack-image``
+additionally emits a combined ELF (demo flattened plus the service at a fixed
+offset) and a preinject config, so the trial starts with the injected code
+already part of the simulated system.
+
+Honesty: the real GSP ``.fwimage`` is packaged with a lenient device map but
+cannot boot on the native simulator yet (H2.2 device model absent); its trial
+records ``status=blocked`` and a bounded first-touch probe, never a boot.
+
+Everything is stdlib-only; builds are deterministic apart from the manifest
+``created`` field (set ``SOURCE_DATE_EPOCH`` to make even that stable).
+"""
+from __future__ import annotations
+
+import argparse
+import gzip
+import hashlib
+import json
+import os
+import shutil
+import struct
+import subprocess
+import sys
+import tarfile
+import time
+from dataclasses import dataclass
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+TOOL_ROOT = os.path.dirname(HERE)
+if TOOL_ROOT not in sys.path:
+ sys.path.insert(0, TOOL_ROOT)
+
+from greeneye import __version__ as GREENEYE_VERSION # noqa: E402
+from greeneye.elf import EM_RISCV, load_elf # noqa: E402
+from greeneye.errors import GreenEyeError # noqa: E402
+from greeneye.inject import ENTRY_SYMBOL # noqa: E402
+
+FORMAT = "greeneye-trial/1"
+RESULTS_SCHEMA = "greeneye-trial-results/1"
+PREINJECT_SCHEMA = "greeneye-preinject/1"
+DEVICE_MAP_SCHEMA = "greeneye-device-map/1"
+
+DEMO_ELF = os.path.join(TOOL_ROOT, "firmware", "uproc_demo", "build", "uproc_demo.elf")
+DEFAULT_SERVICE = os.path.join(TOOL_ROOT, "sdk", "build", "injected_demo.elf")
+REPO_ROOT = os.path.dirname(os.path.dirname(TOOL_ROOT))
+REAL_IMAGE = os.path.normpath(
+ os.path.join(REPO_ROOT, "firmware", "57X-GSP", "570.144", "gsp-570.144.fwimage.bin")
+)
+
+RAM_BASE = 0x80000000
+REAL_IMAGE_BIAS = 0x80000000
+#: Service placement mirrors greeneye.inject.load_service's default load address.
+SERVICE_OFFSET = 0x01000000
+DEFAULT_SLOT = 6
+PING_XOR = 0xBEEF
+IMAGE_BUDGET = 200_000
+DEMO_BOOT_SECONDS = 0.25
+DEMO_SETTLE_SECONDS = 0.25
+DEFAULT_RUN_TIMEOUT = 180.0
+
+
+# -- small deterministic helpers -------------------------------------------
+
+
+def sha256_file(path: str) -> str:
+ digest = hashlib.sha256()
+ with open(path, "rb") as fh:
+ for chunk in iter(lambda: fh.read(1 << 20), b""):
+ digest.update(chunk)
+ return digest.hexdigest()
+
+
+def sha256_bytes(data: bytes) -> str:
+ return hashlib.sha256(data).hexdigest()
+
+
+def _portable(path: str) -> str:
+ """Manifest-friendly source name: repo-relative when possible, else basename."""
+ absolute = os.path.abspath(path)
+ relative = os.path.relpath(absolute, TOOL_ROOT)
+ if relative.startswith(".."):
+ return os.path.basename(absolute)
+ return relative
+
+
+def _created_iso() -> str:
+ """Build timestamp; stable when SOURCE_DATE_EPOCH is set."""
+ raw = os.environ.get("SOURCE_DATE_EPOCH")
+ try:
+ epoch = int(raw) if raw else int(time.time())
+ except ValueError:
+ epoch = 0
+ return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(epoch))
+
+
+def _write_bytes(path: str, data: bytes) -> str:
+ os.makedirs(os.path.dirname(path), exist_ok=True)
+ with open(path, "wb") as fh:
+ fh.write(data)
+ return path
+
+
+def _write_text(path: str, text: str) -> str:
+ os.makedirs(os.path.dirname(path), exist_ok=True)
+ with open(path, "w", encoding="utf-8") as fh:
+ fh.write(text)
+ return path
+
+
+def _write_json(path: str, payload, indent: int = 2) -> str:
+ return _write_text(path, json.dumps(payload, indent=indent, sort_keys=True) + "\n")
+
+
+# -- combined demo+service ELF ---------------------------------------------
+
+
+def _load_plan(image, bias: int) -> list[tuple[int, int, bytes]]:
+ """``(vaddr, p_flags, memsz-padded bytes)`` for every non-empty PT_LOAD."""
+ plan = []
+ for seg in image.load_segments:
+ data = bytearray(image.segment_bytes(seg))
+ if len(data) < seg.p_memsz:
+ data += b"\x00" * (seg.p_memsz - len(data))
+ if data:
+ plan.append((seg.p_vaddr + bias, seg.p_flags, bytes(data)))
+ return plan
+
+
+def write_combined_elf(
+ demo_path: str,
+ service_path: str,
+ out_path: str,
+ *,
+ service_offset: int = SERVICE_OFFSET,
+ slot: int = DEFAULT_SLOT,
+) -> dict:
+ """Merge a target ELF and a relocated service ELF into one runnable ELF64.
+
+ The service is placed exactly as ``greeneye.inject.load_service`` would place
+ it (segments at ``service_load_addr + p_vaddr``, entry at
+ ``service_load_addr + e_entry``), so a pre-injected boot is address-for-address
+ equivalent to the load/register path. Returns the placement config.
+ """
+ demo = load_elf(demo_path)
+ service = load_elf(service_path)
+ if not demo.is_riscv or not service.is_riscv:
+ raise GreenEyeError("combined image requires RISC-V ELF inputs")
+ target_load = demo.min_vaddr
+ service_load = target_load + service_offset
+ demo_plan = _load_plan(demo, 0)
+ service_plan = _load_plan(service, service_load)
+ if not demo_plan:
+ raise GreenEyeError(f"{demo_path}: no loadable segments")
+
+ spans = sorted((vaddr, vaddr + len(data)) for vaddr, _flags, data in demo_plan + service_plan)
+ for (a1, b1), (a2, b2) in zip(spans, spans[1:]):
+ if a2 < b1:
+ raise GreenEyeError(
+ f"combined image overlap: [{a2:#x},{b2:#x}) collides with [{a1:#x},{b1:#x})"
+ )
+
+ segments = demo_plan + service_plan
+ phnum = len(segments)
+ ident = b"\x7fELF" + bytes([2, 1, 1, 0]) + b"\x00" * 8 # ELF64, LE, v1
+ header = ident + struct.pack(
+ "<HHIQQQIHHHHHH", 2, EM_RISCV, 1, demo.entry, 64, 0, 0, 64, 56, phnum, 64, 0, 0
+ )
+ offset = 64 + 56 * phnum
+ phdrs = bytearray()
+ body = bytearray()
+ for vaddr, flags, data in segments:
+ phdrs += struct.pack(
+ "<IIQQQQQQ", 1, flags, offset + len(body), vaddr, vaddr, len(data), len(data), 0x1000
+ )
+ body += data
+ _write_bytes(out_path, header + bytes(phdrs) + bytes(body))
+
+ check = load_elf(out_path)
+ if check.is_riscv is False or check.entry != demo.entry or len(check.load_segments) != phnum:
+ raise GreenEyeError(f"combined ELF failed its own round-trip: {out_path}")
+
+ info = {
+ "target_load_addr": target_load,
+ "target_entry": demo.entry,
+ "service_load_addr": service_load,
+ "service_offset": service_offset,
+ "service_entry": service.entry,
+ "entry_addr": service_load + service.entry,
+ "slot": slot,
+ "size": sum(len(data) for _v, _f, data in service_plan),
+ "segments": phnum,
+ }
+ return info
+
+
+# -- deterministic tar ------------------------------------------------------
+
+
+def write_deterministic_tar(out_dir: str, tar_path: str) -> str:
+ """Write ``out_dir`` as a gzip tar with zero timestamps and sorted members.
+
+ Member mtimes, uid/gid, uname/gname and the gzip mtime are fixed, so two
+ builds of identical content produce byte-identical archives.
+ """
+ names = []
+ for root, dirs, files in os.walk(out_dir):
+ dirs.sort()
+ for name in sorted(files):
+ names.append(os.path.relpath(os.path.join(root, name), out_dir))
+ names.sort()
+ with open(tar_path, "wb") as raw:
+ with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as gz:
+ with tarfile.open(fileobj=gz, mode="w", format=tarfile.GNU_FORMAT) as tar:
+ for name in names:
+ full = os.path.join(out_dir, name)
+ info = tar.gettarinfo(full, arcname=name)
+ info.mtime = 0
+ info.uid = info.gid = 0
+ info.uname = info.gname = ""
+ info.mode = 0o755 if name == "run.sh" else 0o644
+ with open(full, "rb") as fh:
+ tar.addfile(info, fh)
+ return tar_path
+
+
+# -- request/build ----------------------------------------------------------
+
+RUN_SH = """#!/bin/sh
+# green-eye trial runner (generated by harness/package_sim.py; inspect before use).
+# Usage: ./run.sh [scenario args] Environment: GREENEYE_HOME overrides the tool root.
+set -eu
+here=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
+if [ -n "${GREENEYE_HOME:-}" ]; then
+ root=$GREENEYE_HOME
+else
+ root=$here
+ while [ "$root" != "/" ] && [ ! -f "$root/greeneye/__init__.py" ]; do
+ root=$(dirname "$root")
+ done
+ if [ ! -f "$root/greeneye/__init__.py" ]; then
+ echo "run.sh: green-eye tool root not found; set GREENEYE_HOME" >&2
+ exit 3
+ fi
+fi
+export PYTHONPATH="$root${PYTHONPATH:+:$PYTHONPATH}"
+exec python3 "$here/scenario.py" --package "$here" "$@"
+"""
+
+
+@dataclass
+class PackageRequest:
+ out: str
+ target: str = "demo" # demo | image
+ image: str | None = None
+ service: str | None = None
+ device_map: str | None = None
+ script: str | None = None
+ slot: int = DEFAULT_SLOT
+ pack_image: bool = False
+ tar: bool = False
+
+
+def _readme_text(
+ *,
+ target: str,
+ target_name: str,
+ target_sha: str,
+ service_name: str,
+ service_sha: str,
+ slot: int,
+ device_map_name: str | None,
+ preinject: dict | None,
+) -> str:
+ lines = [
+ "# green-eye trial package",
+ "",
+ "Generated by `harness/package_sim.py` (`greeneye-trial/1`). This package",
+ "bundles a target, an injected service blob and a pre-scripted scenario",
+ "that runs on the deterministic native RV64 simulator in the green-eye",
+ "tool tree. It needs no QEMU, no GPU and no guest.",
+ "",
+ "## Contents",
+ "",
+ f"- target: `payload/{target_name}` (`{target}`), sha256 `{target_sha}`",
+ f"- service: `payload/{service_name}`, slot {slot}, sha256 `{service_sha}`",
+ f"- device map: `{('payload/' + device_map_name) if device_map_name else 'none'}`",
+ f"- pre-injected combined image: `{preinject['file'] if preinject else 'none'}`",
+ "- scenario: `scenario.py`; results are written to `results/`",
+ "",
+ "## How to run",
+ "",
+ "```sh",
+ "./run.sh",
+ "# image packages may bound the lenient probe explicitly:",
+ "./run.sh --budget 200000",
+ "```",
+ "",
+ "Requires Python 3.10+ and the green-eye tool tree. `run.sh` finds the",
+ "tree by walking up from this directory; set `GREENEYE_HOME=/path/to/risc-v-gsp-rm-vm`",
+ "if the package was moved. Exit code 0 means the assertions passed (or an",
+ "image package produced its documented blocked status), 1 means an",
+ "assertion failure, 3 means the tool tree was not found.",
+ "",
+ "## What to expect",
+ "",
+ ]
+ if target == "demo":
+ lines += [
+ "The demo uproc boots natively, the service is loaded and registered in",
+ f"`gsp_ctrl.handlers[{slot}]`, then invoked directly and through mailbox",
+ "command 16 + slot. `results/results.json` should report `status=ok` with",
+ "matching call results, at least one `rpc_dispatch` hook hit, at least one",
+ "watch hit on `gsp_ctrl.result`, and the UART banner/self-test lines.",
+ ]
+ if preinject:
+ lines += [
+ "",
+ "This package uses `--pack-image`: the combined ELF",
+ f"`{preinject['file']}` already contains the service at",
+ f"{preinject['service_load_addr']:#x}, so no separate load step runs.",
+ ]
+ else:
+ lines += [
+ "**Honest status: the native simulator cannot boot the real GSP",
+ "`.fwimage` yet** (the H2.2 device model is absent). The scenario runs a",
+ "bounded lenient first-touch probe with the bundled device map and",
+ "records `status=blocked`, `blocked_on_device_model=true` in",
+ "`results/results.json`; it is not a boot and makes no hardware claim.",
+ "No RM service registration point exists yet (M5), so the service blob",
+ "is packaged but not loaded or registered.",
+ ]
+ if preinject:
+ lines += [
+ "",
+ f"A combined artifact `{preinject['file']}` is present but marked",
+ "`runnable: false` for the same reason; it is not executed.",
+ ]
+ lines += [
+ "",
+ "## How to read the results",
+ "",
+ "- `results/results.json` - machine-readable: status, call results, hook and",
+ " watch hits, control block, UART excerpt, session path, assertion list.",
+ "- `results/results.md` - the same rendered for humans.",
+ "- `results/session/` - the recorded greeneye session (`events.jsonl`,",
+ " `session.json`, snapshots) for `greeneye analyze`/`compare`.",
+ "- `manifest.json` - format/version, input hashes and the component list;",
+ " every component hash covers the file as packaged.",
+ "",
+ "## Safety",
+ "",
+ "Run only packages from sources you trust. Injected code executes inside",
+ "the simulator process with the simulator's full privileges. This package",
+ "contains no hardware access and must never be used to inject into a guest,",
+ "a shipped environment or real hardware.",
+ "",
+ "## Determinism",
+ "",
+ "Manifest component order and hashes are stable, and `--tar` archives use",
+ "zeroed member timestamps. The only variable field is `manifest.created`;",
+ "set `SOURCE_DATE_EPOCH` to freeze it and reproduce byte-identical builds.",
+ "",
+ ]
+ return "\n".join(lines)
+
+
+def build_package(request: PackageRequest) -> dict:
+ """Build the trial package; returns the manifest that was written."""
+ out = os.path.abspath(request.out)
+ if os.path.isdir(out) and os.listdir(out):
+ raise GreenEyeError(f"output directory is not empty: {out}")
+ if request.target not in ("demo", "image"):
+ raise GreenEyeError(f"target must be demo or image, got {request.target!r}")
+ if not 0 <= request.slot <= 7:
+ raise GreenEyeError(f"slot must be 0..7, got {request.slot}")
+
+ if request.target == "demo":
+ target_input = os.path.abspath(request.image or DEMO_ELF)
+ else:
+ target_input = os.path.abspath(request.image or REAL_IMAGE)
+ if not os.path.isfile(target_input):
+ raise GreenEyeError(f"target not found: {target_input}")
+ service_input = os.path.abspath(request.service or DEFAULT_SERVICE)
+ if not os.path.isfile(service_input):
+ raise GreenEyeError(f"service blob not found: {service_input} (run: make sdk)")
+
+ target_image = load_elf(target_input)
+ if not target_image.is_riscv:
+ raise GreenEyeError(f"{target_input}: not a RISC-V ELF image")
+ service_image = load_elf(service_input)
+ if not service_image.is_riscv:
+ raise GreenEyeError(f"{service_input}: not a RISC-V ELF image")
+
+ os.makedirs(out, exist_ok=True)
+ payload_dir = os.path.join(out, "payload")
+ os.makedirs(payload_dir, exist_ok=True)
+ components: list[tuple[str, str]] = []
+
+ target_name = os.path.basename(target_input)
+ shutil.copyfile(target_input, os.path.join(payload_dir, target_name))
+ components.append((f"payload/{target_name}", "target-image"))
+
+ service_name = os.path.basename(service_input)
+ shutil.copyfile(service_input, os.path.join(payload_dir, service_name))
+ components.append((f"payload/{service_name}", "service-blob"))
+
+ device_map_info: dict | None = None
+ if request.device_map:
+ dm_input = os.path.abspath(request.device_map)
+ if not os.path.isfile(dm_input):
+ raise GreenEyeError(f"device map not found: {dm_input}")
+ dm_name = os.path.basename(dm_input)
+ shutil.copyfile(dm_input, os.path.join(payload_dir, dm_name))
+ components.append((f"payload/{dm_name}", "device-map"))
+ device_map_info = {
+ "file": f"payload/{dm_name}",
+ "input": _portable(dm_input),
+ "sha256": sha256_file(dm_input),
+ "lenient": request.target == "image",
+ "learn": False,
+ }
+ elif request.target == "image":
+ dm_name = "device_map.json"
+ _write_json(
+ os.path.join(payload_dir, dm_name),
+ {"schema": DEVICE_MAP_SCHEMA, "meta": {"bias": REAL_IMAGE_BIAS, "source": "lenient"}, "entries": []},
+ indent=1,
+ )
+ components.append((f"payload/{dm_name}", "device-map"))
+ device_map_info = {
+ "file": f"payload/{dm_name}",
+ "input": None,
+ "sha256": sha256_file(os.path.join(payload_dir, dm_name)),
+ "lenient": True,
+ "learn": True,
+ }
+
+ preinject: dict | None = None
+ if request.pack_image:
+ service_offset = SERVICE_OFFSET
+ if request.target == "image":
+ span = target_image.max_vaddr - target_image.min_vaddr
+ service_offset = max(SERVICE_OFFSET, ((span + 0xFFF) // 0x1000) * 0x1000)
+ combined_rel = "payload/combined.elf"
+ placement = write_combined_elf(
+ target_input,
+ service_input,
+ os.path.join(out, combined_rel),
+ service_offset=service_offset,
+ slot=request.slot,
+ )
+ components.append((combined_rel, "combined-elf"))
+ runnable = request.target == "demo"
+ reason = (
+ ""
+ if runnable
+ else "the native simulator cannot boot the real GSP image yet (H2.2 device model absent)"
+ )
+ preinject = {
+ "file": combined_rel,
+ "schema": PREINJECT_SCHEMA,
+ "runnable": runnable,
+ "reason": reason,
+ **placement,
+ }
+ _write_json(os.path.join(payload_dir, "preinject.json"), preinject)
+ components.append(("payload/preinject.json", "preinject-config"))
+
+ if request.script:
+ script_input = os.path.abspath(request.script)
+ if not os.path.isfile(script_input):
+ raise GreenEyeError(f"scenario script not found: {script_input}")
+ shutil.copyfile(script_input, os.path.join(out, "scenario.py"))
+ script_info = {
+ "kind": "custom",
+ "name": os.path.basename(script_input),
+ "input": _portable(script_input),
+ "sha256": sha256_file(script_input),
+ }
+ else:
+ _write_text(os.path.join(out, "scenario.py"), BUILTIN_SCENARIO)
+ script_info = {
+ "kind": "builtin",
+ "name": "builtin-demo-v1",
+ "input": None,
+ "sha256": sha256_bytes(BUILTIN_SCENARIO.encode("utf-8")),
+ }
+ components.append(("scenario.py", "scenario"))
+
+ run_sh_path = os.path.join(out, "run.sh")
+ _write_text(run_sh_path, RUN_SH)
+ os.chmod(run_sh_path, 0o755)
+ components.append(("run.sh", "runner"))
+
+ _write_text(os.path.join(out, "results", ".gitkeep"), "")
+ components.append(("results/.gitkeep", "results-placeholder"))
+
+ if request.target == "image":
+ if preinject and preinject.get("runnable"):
+ end = max(
+ target_image.max_vaddr + REAL_IMAGE_BIAS,
+ preinject["service_load_addr"] + preinject["size"],
+ )
+ else:
+ end = target_image.max_vaddr + REAL_IMAGE_BIAS
+ need = end - RAM_BASE + (1 << 20)
+ memory_mb = max(256, (need + (1 << 20) - 1) // (1 << 20))
+ limitations = [
+ "The native simulator cannot boot the real GSP .fwimage yet: the H2.2 device model is absent.",
+ "The run is a bounded lenient first-touch probe, not a boot; results.json records status=blocked.",
+ "No RM service registration point exists yet (M5): the blob is packaged but not loaded or registered.",
+ "No L2 timing/concurrency and no L3 device fidelity is claimed (docs/06-ACCURACY.md).",
+ ]
+ else:
+ memory_mb = 256
+ limitations = [
+ "The demo uproc is a stand-in for GSP-RM, not the real device; results describe the injection path only.",
+ "Injected code runs with the simulator's full in-process privileges; only run trusted packages.",
+ ]
+
+ _write_text(
+ os.path.join(out, "README.md"),
+ _readme_text(
+ target=request.target,
+ target_name=target_name,
+ target_sha=sha256_file(target_input),
+ service_name=service_name,
+ service_sha=sha256_file(service_input),
+ slot=request.slot,
+ device_map_name=device_map_info["file"].split("/", 1)[1] if device_map_info else None,
+ preinject=preinject,
+ ),
+ )
+ components.append(("README.md", "documentation"))
+
+ manifest = {
+ "format": FORMAT,
+ "greeneye_version": GREENEYE_VERSION,
+ "created": _created_iso(),
+ "results_schema": RESULTS_SCHEMA,
+ "target": {
+ "kind": request.target,
+ "file": f"payload/{target_name}",
+ "input": _portable(target_input),
+ "sha256": sha256_file(target_input),
+ "entry": target_image.entry,
+ "min_vaddr": target_image.min_vaddr,
+ "max_vaddr": target_image.max_vaddr,
+ "bias": REAL_IMAGE_BIAS if request.target == "image" else 0,
+ "runnable": request.target == "demo",
+ "limitations": limitations,
+ },
+ "service": {
+ "file": f"payload/{service_name}",
+ "input": _portable(service_input),
+ "sha256": sha256_file(service_input),
+ "entry": service_image.entry,
+ "entry_symbol": ENTRY_SYMBOL,
+ "slot": request.slot,
+ },
+ "device_map": device_map_info,
+ "script": script_info,
+ "preinject": preinject,
+ "scenario": {
+ "kind": script_info["kind"],
+ "demo_boot_seconds": DEMO_BOOT_SECONDS,
+ "demo_settle_seconds": DEMO_SETTLE_SECONDS,
+ "image_budget_instructions": IMAGE_BUDGET,
+ "memory_mb": memory_mb,
+ "expected": {
+ "cmd_ping": 1,
+ "cmd_sum": 2,
+ "ping_arg": 0x41,
+ "sum_args": [0x10, 0x20],
+ "ping_result": 0x41 ^ PING_XOR,
+ "sum_result": 0x30,
+ },
+ },
+ "run": {
+ "command": "./run.sh",
+ "results": ["results/results.json", "results/results.md"],
+ "exit_nonzero_on": "assertion failure",
+ },
+ "safety": (
+ "Local, development-only study tooling: no hardware access, no guest, "
+ "never shipped. Run only packages from trusted sources."
+ ),
+ "determinism": (
+ "Component order and hashes are stable; --tar members use mtime 0. "
+ "manifest.created is the only timestamp (SOURCE_DATE_EPOCH freezes it)."
+ ),
+ "components": [],
+ }
+ for relpath, role in sorted(components):
+ full = os.path.join(out, relpath)
+ manifest["components"].append(
+ {
+ "path": relpath,
+ "role": role,
+ "bytes": os.path.getsize(full),
+ "sha256": sha256_file(full),
+ }
+ )
+
+ _write_json(os.path.join(out, "manifest.json"), manifest)
+ if request.tar:
+ write_deterministic_tar(out, out + ".tar.gz")
+ return manifest
+
+
+def run_package(out: str, *, timeout: float = DEFAULT_RUN_TIMEOUT, args: list[str] | None = None) -> int:
+ """Run ``run.sh`` in a bounded subprocess; returns its exit code."""
+ run_sh = os.path.join(os.path.abspath(out), "run.sh")
+ if not os.path.isfile(run_sh):
+ raise GreenEyeError(f"no run.sh in {out}; build the package first")
+ environment = dict(os.environ)
+ environment.setdefault("GREENEYE_HOME", TOOL_ROOT)
+ try:
+ proc = subprocess.run(
+ ["/bin/sh", run_sh, *(args or [])],
+ cwd=os.path.abspath(out),
+ env=environment,
+ capture_output=True,
+ text=True,
+ timeout=timeout,
+ )
+ except subprocess.TimeoutExpired as exc:
+ print(f"[package-sim] run.sh timed out after {timeout}s", file=sys.stderr)
+ print(exc.stdout or "", end="")
+ print(exc.stderr or "", end="", file=sys.stderr)
+ return 124
+ print(proc.stdout, end="")
+ print(proc.stderr, end="", file=sys.stderr)
+ return proc.returncode
+
+
+def main(argv: list[str] | None = None) -> int:
+ parser = argparse.ArgumentParser(
+ prog="python3 harness/package_sim.py",
+ description="Build a self-contained green-eye trial package (simulator + injected service).",
+ )
+ parser.add_argument("--target", choices=("demo", "image"), default="demo",
+ help="demo uproc (default) or a real .fwimage")
+ parser.add_argument("--image", help="target ELF/.fwimage path (image target; overrides the demo ELF)")
+ parser.add_argument("--service", help=f"service blob ELF (default: {os.path.relpath(DEFAULT_SERVICE, TOOL_ROOT)})")
+ parser.add_argument("--device-map", help="greeneye-device-map/1 JSON (lenient MMIO evidence)")
+ parser.add_argument("--script", help="custom scenario script (default: builtin demo scenario)")
+ parser.add_argument("--out", required=True, help="output directory (must be empty)")
+ parser.add_argument("--slot", type=int, default=DEFAULT_SLOT, help="handler slot 0..7 (default 6)")
+ parser.add_argument("--pack-image", action="store_true",
+ help="also emit payload/combined.elf + preinject config")
+ parser.add_argument("--tar", action="store_true", help="write a deterministic <out>.tar.gz")
+ parser.add_argument("--run", action="store_true", help="run the packaged trial after building")
+ parser.add_argument("--run-timeout", type=float, default=DEFAULT_RUN_TIMEOUT, help="run.sh timeout seconds")
+ args = parser.parse_args(argv)
+
+ request = PackageRequest(
+ out=args.out,
+ target=args.target,
+ image=args.image,
+ service=args.service,
+ device_map=args.device_map,
+ script=args.script,
+ slot=args.slot,
+ pack_image=args.pack_image,
+ tar=args.tar,
+ )
+ try:
+ manifest = build_package(request)
+ except GreenEyeError as exc:
+ print(f"[package-sim] error: {exc}", file=sys.stderr)
+ return 1
+ print(
+ f"[package-sim] built {args.out}: {len(manifest['components'])} components, "
+ f"target={manifest['target']['kind']}"
+ )
+ if args.tar:
+ print(f"[package-sim] archive {args.out}.tar.gz")
+ if manifest["target"]["runnable"] is False:
+ print("[package-sim] note: packaged target cannot boot yet; results will report status=blocked")
+ if manifest.get("preinject"):
+ print("[package-sim] note: preinject/unused combined artifact is informational only")
+ if args.run:
+ return run_package(args.out, timeout=args.run_timeout)
+ return 0
+
+
+# -- generated scenario -----------------------------------------------------
+
+BUILTIN_SCENARIO = r'''#!/usr/bin/env python3
+"""Pre-scripted green-eye trial scenario (generated by harness/package_sim.py).
+
+Boots the packaged target on the native simulator, loads and registers the
+injected service, invokes it directly and through the firmware mailbox, runs
+one hook/watch pass, and writes results/results.json + results/results.md.
+Exit 0 = assertions passed (or the documented blocked image status), 1 = a
+verification assertion failed.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+
+CTRL_HANDLERS = 0x150
+CTRL_RESULT = 0x40
+CTRL_RPC_COUNT = 0x10
+PING_ARG = 0x41
+SUM_A0 = 0x10
+SUM_A1 = 0x20
+EXPECTED_PING = PING_ARG ^ 0xBEEF
+EXPECTED_SUM = SUM_A0 + SUM_A1
+UART_BANNER = "GREEN-EYE UPROC DEMO v1"
+UART_SELFTEST = "SELFTEST DONE"
+
+
+def bootstrap_tool_root(pkg):
+ """Make `import greeneye` work when scenario.py is run directly."""
+ try:
+ import greeneye # noqa: F401
+ return
+ except ImportError:
+ pass
+ candidates = []
+ if os.environ.get("GREENEYE_HOME"):
+ candidates.append(os.path.abspath(os.environ["GREENEYE_HOME"]))
+ here = pkg
+ while True:
+ candidates.append(here)
+ parent = os.path.dirname(here)
+ if parent == here:
+ break
+ here = parent
+ for candidate in candidates:
+ if os.path.isfile(os.path.join(candidate, "greeneye", "__init__.py")):
+ sys.path.insert(0, candidate)
+ return
+
+
+def load_json(path):
+ with open(path, "r", encoding="utf-8") as fh:
+ return json.load(fh)
+
+
+def write_json(path, payload):
+ os.makedirs(os.path.dirname(path), exist_ok=True)
+ with open(path, "w", encoding="utf-8") as fh:
+ json.dump(payload, fh, indent=2, sort_keys=True)
+ fh.write("\n")
+
+
+def write_text(path, text):
+ os.makedirs(os.path.dirname(path), exist_ok=True)
+ with open(path, "w", encoding="utf-8") as fh:
+ fh.write(text)
+
+
+def component(manifest, role):
+ for comp in manifest.get("components", ()):
+ if comp.get("role") == role:
+ return comp["path"]
+ return None
+
+
+def uart_excerpt(text, first=4, last=6, cap=14):
+ lines = [line for line in text.splitlines() if line.strip()]
+ if len(lines) > cap:
+ return lines[:first] + ["..."] + lines[-last:]
+ return lines
+
+
+def hexv(value):
+ return f"0x{int(value):x}"
+
+
+def render_md(manifest, data):
+ target = data.get("target", {})
+ lines = ["# green-eye trial results", ""]
+ lines.append(f"- status: **{data.get('status')}**")
+ lines.append(f"- target: `{target.get('file', '?')}` ({target.get('kind', '?')})")
+ service = data.get("service")
+ if service:
+ lines.append(
+ f"- service: `{service.get('file', '?')}` slot {service.get('slot', '?')} "
+ f"pre-injected {'yes' if service.get('preinjected') else 'no'}"
+ )
+ lines.append(f"- backend: {data.get('backend', 'native')}")
+ if data.get("error"):
+ lines += ["", "## Error", "", "```", str(data["error"]), "```"]
+ if data.get("status") == "blocked":
+ lines += [
+ "",
+ "## Honest status",
+ "",
+ str(data.get("blocked_reason", "the native simulator cannot boot this target yet")),
+ "",
+ f"- instructions executed: {data.get('instructions_executed')}",
+ f"- stopped: pc {hexv(data.get('stopped', {}).get('pc', 0))} "
+ f"(firmware {hexv(data.get('stopped', {}).get('fw_pc', 0))})",
+ f"- unmodelled MMIO accesses served leniently: {data.get('mmio_unknown')}",
+ ]
+ if data.get("calls"):
+ calls = data["calls"]
+ expected = calls.get("expected", {})
+ lines += [
+ "",
+ "## Injected call results",
+ "",
+ "| route | cmd | result | expected | ok |",
+ "|---|---|---|---|---|",
+ ]
+ for route, pair in (("remote (direct)", "ping"), ("remote (direct)", "sum"),
+ ("via firmware mailbox", "ping"), ("via firmware mailbox", "sum")):
+ key = "remote" if route.startswith("remote") else "via_firmware"
+ actual = calls.get(key, {}).get(pair)
+ want = expected.get(pair)
+ lines.append(f"| {route} | {pair} | {hexv(actual)} | {hexv(want)} | "
+ f"{'yes' if actual == want else 'no'} |")
+ lines.append("")
+ lines.append(f"routes agree: {'yes' if calls.get('agree') else 'no'}")
+ if data.get("hooks"):
+ hooks = data["hooks"]["rpc_dispatch"]
+ lines += ["", "## Hook / watch pass", "",
+ f"- `rpc_dispatch` hook hits: {hooks['hits']}"]
+ for hit in hooks.get("first", [])[:4]:
+ lines.append(f" - pc {hit['pc']} cmd {hit['cmd']} a0 {hit['a0']} a1 {hit['a1']}")
+ if data.get("watches"):
+ watch = data["watches"]["gsp_ctrl.result"]
+ lines.append(f"- `gsp_ctrl.result` watch hits: {watch['hits']}")
+ for hit in watch.get("first", [])[:4]:
+ lines.append(f" - pc {hit['pc']} ({hit.get('symbol')})")
+ if data.get("control"):
+ control = data["control"]
+ lines.append(
+ f"- control block: result {hexv(control['result'])}, rpc_count {control['rpc_count']}"
+ )
+ lines.append(f"- handlers slot value: {hexv(data.get('handlers_slot', 0))}")
+ if data.get("uart_excerpt"):
+ lines += ["", "## UART excerpt", "", "```"]
+ lines += list(data["uart_excerpt"])
+ lines += ["```"]
+ lines += ["", "## Session", "", f"`{data.get('session', 'results/session')}`"]
+ if data.get("learned_device_map"):
+ lines.append(f"`{data['learned_device_map']}`")
+ assertions = data.get("assertions", {})
+ lines += ["", "## Assertions", ""]
+ for name in sorted(assertions.get("checks", {})):
+ lines.append(f"- [{'x' if assertions['checks'][name] else ' '}] {name}")
+ failed = assertions.get("failed", [])
+ lines.append("")
+ lines.append("all checks passed" if not failed else f"FAILED: {', '.join(failed)}")
+ if data.get("limitations"):
+ lines += ["", "## Limits", ""]
+ lines += [f"- {item}" for item in data["limitations"]]
+ lines.append("")
+ return "\n".join(lines)
+
+
+def run_demo(pkg, manifest):
+ from greeneye.elf import load_elf
+ from greeneye.inject import InjectedService, invoke_remote, invoke_via_firmware, load_service
+ from greeneye.sim_backend import launch_native
+
+ results_dir = os.path.join(pkg, "results")
+ target_rel = component(manifest, "target-image")
+ service_rel = component(manifest, "service-blob")
+ dm_rel = component(manifest, "device-map")
+ combined_rel = component(manifest, "combined-elf")
+ preinject_rel = component(manifest, "preinject-config")
+ target_path = os.path.join(pkg, target_rel)
+ service_path = os.path.join(pkg, service_rel)
+ device_map_path = os.path.join(pkg, dm_rel) if dm_rel else None
+ preinject = load_json(os.path.join(pkg, preinject_rel)) if preinject_rel else None
+ preinjected = bool(preinject and preinject.get("runnable", True))
+ kernel = os.path.join(pkg, combined_rel) if preinjected else target_path
+
+ dispatch_total = [0]
+ watch_total = [0]
+ dispatch_hits = []
+ watch_hits = []
+ data = {
+ "schema": "greeneye-trial-results/1",
+ "backend": "native",
+ "target": {"kind": "demo", "file": target_rel, "sha256": manifest["target"]["sha256"]},
+ "session": "results/session",
+ "service": {
+ "file": service_rel,
+ "sha256": manifest["service"]["sha256"],
+ "slot": manifest["service"]["slot"],
+ "preinjected": preinjected,
+ },
+ }
+ checks = {}
+ session = None
+ try:
+ session = launch_native(
+ kernel=kernel,
+ record=os.path.join(results_dir, "session"),
+ name="trial-demo",
+ memory_mb=int(manifest["scenario"].get("memory_mb", 256)),
+ device_map=device_map_path,
+ unknown_mmio="lenient" if device_map_path else "fault",
+ )
+ if preinjected:
+ session.symbols.load_elf(load_elf(target_path))
+ ctrl = session.symbols.lookup("gsp_ctrl")
+ if preinjected:
+ session.memory.write_u64(
+ ctrl + CTRL_HANDLERS + int(preinject["slot"]) * 8, int(preinject["entry_addr"])
+ )
+ service = InjectedService(
+ path=service_rel,
+ name=os.path.basename(service_rel),
+ load_addr=int(preinject["service_load_addr"]),
+ entry_addr=int(preinject["entry_addr"]),
+ dispatch_offset=int(preinject["service_entry"]),
+ slot=int(preinject["slot"]),
+ size=int(preinject["size"]),
+ )
+ session.recorder.event(
+ "service_injected",
+ path=service.path,
+ name=service.name,
+ load_addr=service.load_addr,
+ entry_addr=service.entry_addr,
+ slot=service.slot,
+ size=service.size,
+ preinjected=True,
+ )
+ else:
+ service = load_service(session, service_path, slot=int(manifest["service"]["slot"]))
+
+ def on_dispatch(hit):
+ dispatch_total[0] += 1
+ if len(dispatch_hits) < 64:
+ dispatch_hits.append(
+ {
+ "pc": hexv(hit.pc),
+ "cmd": hexv(hit.arg(0)),
+ "a0": hexv(hit.arg(1)),
+ "a1": hexv(hit.arg(2)),
+ }
+ )
+
+ def on_result(hit):
+ watch_total[0] += 1
+ if len(watch_hits) < 64:
+ watch_hits.append(
+ {"pc": hexv(hit.pc), "symbol": hit.symbol.name if hit.symbol else None}
+ )
+
+ session.hook("rpc_dispatch", on_enter=on_dispatch)
+ session.watch(ctrl + CTRL_RESULT, 8, "w", callback=on_result)
+ session.run(duration=float(manifest["scenario"]["demo_boot_seconds"]))
+
+ remote_ping = invoke_remote(session, service, 1, PING_ARG)
+ remote_sum = invoke_remote(session, service, 2, SUM_A0, SUM_A1)
+ fw_ping = invoke_via_firmware(session, service, 1, PING_ARG)
+ fw_sum = invoke_via_firmware(session, service, 2, SUM_A0, SUM_A1)
+
+ handlers = session.memory.read_u64(ctrl + CTRL_HANDLERS + service.slot * 8)
+ result_value = session.memory.read_u64(ctrl + CTRL_RESULT)
+ rpc_count = session.memory.read_u64(ctrl + CTRL_RPC_COUNT)
+ uart = session.target.read_uart()
+
+ data["service"].update(
+ {"load_addr": hexv(service.load_addr), "entry_addr": hexv(service.entry_addr)}
+ )
+ data["calls"] = {
+ "remote": {"ping": remote_ping, "sum": remote_sum},
+ "via_firmware": {"ping": fw_ping, "sum": fw_sum},
+ "expected": {"ping": EXPECTED_PING, "sum": EXPECTED_SUM},
+ "agree": remote_ping == fw_ping and remote_sum == fw_sum,
+ }
+ data["handlers_slot"] = handlers
+ data["control"] = {"result": result_value, "rpc_count": rpc_count}
+ data["hooks"] = {"rpc_dispatch": {"hits": dispatch_total[0], "first": dispatch_hits}}
+ data["watches"] = {"gsp_ctrl.result": {"hits": watch_total[0], "first": watch_hits}}
+ data["uart_excerpt"] = uart_excerpt(uart)
+ checks = {
+ "handlers_slot_registered": handlers == service.entry_addr,
+ "remote_ping": remote_ping == EXPECTED_PING,
+ "remote_sum": remote_sum == EXPECTED_SUM,
+ "via_firmware_ping": fw_ping == EXPECTED_PING,
+ "via_firmware_sum": fw_sum == EXPECTED_SUM,
+ "routes_agree": remote_ping == fw_ping and remote_sum == fw_sum,
+ "dispatch_hook_hits": dispatch_total[0] >= 1,
+ "result_watch_hits": watch_total[0] >= 1,
+ "uart_banner": UART_BANNER in uart,
+ "uart_selftest": UART_SELFTEST in uart,
+ }
+ except Exception as exc: # record the honest failure, never fake a pass
+ data["error"] = f"{type(exc).__name__}: {exc}"
+ checks = {"no_exception": False}
+ finally:
+ if session is not None:
+ try:
+ session.close()
+ except Exception:
+ pass
+
+ failures = sorted(name for name, ok in checks.items() if not ok)
+ data["status"] = "ok" if not failures else "failed"
+ data["success"] = not failures
+ data["assertions"] = {"checks": checks, "failed": failures}
+ data["limitations"] = manifest["target"].get("limitations", [])
+ write_json(os.path.join(results_dir, "results.json"), data)
+ write_text(os.path.join(results_dir, "results.md"), render_md(manifest, data))
+ print(f"trial: {data['status']} (results/results.json)")
+ return 0 if not failures else 1
+
+
+def run_image(pkg, manifest, budget=None):
+ from greeneye.sim_backend import launch_native
+
+ results_dir = os.path.join(pkg, "results")
+ target_rel = component(manifest, "target-image")
+ service_rel = component(manifest, "service-blob")
+ dm_rel = component(manifest, "device-map")
+ scenario = manifest["scenario"]
+ target_path = os.path.join(pkg, target_rel)
+ dm_path = os.path.join(pkg, dm_rel) if dm_rel else None
+ budget = int(budget if budget is not None else scenario["image_budget_instructions"])
+ bias = int(manifest["target"].get("bias", 0x80000000))
+ learn = bool(manifest.get("device_map") and manifest["device_map"].get("learn"))
+
+ data = {
+ "schema": "greeneye-trial-results/1",
+ "backend": "native",
+ "target": {
+ "kind": "image",
+ "file": target_rel,
+ "sha256": manifest["target"]["sha256"],
+ "entry": manifest["target"]["entry"],
+ "bias": bias,
+ "runnable": False,
+ },
+ "service": {
+ "file": service_rel,
+ "sha256": manifest["service"]["sha256"],
+ "loaded": False,
+ "reason": (
+ "no RM service registration point exists yet (M5); loading the blob "
+ "would overwrite the image's own RAM"
+ ),
+ },
+ "session": "results/session",
+ "budget_instructions": budget,
+ "preinject": manifest.get("preinject"),
+ "limitations": manifest["target"].get("limitations", []),
+ "blocked_reason": (
+ "the native simulator cannot boot the real GSP image yet: the H2.2 device "
+ "model is absent. This is a bounded lenient first-touch probe, not a boot."
+ ),
+ }
+ checks = {"bounded_lenient_run": False}
+ session = None
+ try:
+ session = launch_native(
+ kernel=target_path,
+ memory_mb=int(scenario.get("memory_mb", 256)),
+ record=os.path.join(results_dir, "session"),
+ name="trial-image",
+ device_map=dm_path,
+ unknown_mmio="lenient",
+ learn_mmio=learn,
+ bias=bias,
+ )
+ session.run(duration=budget / 1_000_000.0)
+ machine = session.target.machine
+ pc = machine.pc
+ data["instructions_executed"] = machine.insns
+ data["stopped"] = {"pc": pc, "fw_pc": pc - bias if pc >= bias else pc}
+ data["mmio_unknown"] = machine.mmio_unknown
+ data["first_unknown_mmio"] = [
+ {"pc": hexv(entry["pc"]), "addr": hexv(entry["addr"]),
+ "size": entry["size"], "rw": entry["rw"]}
+ for entry in machine.mmio_log[:8]
+ ]
+ data["uart_excerpt"] = uart_excerpt(session.target.read_uart())
+ if learn and machine.device_map is not None:
+ machine.device_map.save(os.path.join(results_dir, "learned_device_map.json"))
+ data["learned_device_map"] = "results/learned_device_map.json"
+ checks = {"bounded_lenient_run": True, "booted": False}
+ except Exception as exc:
+ data["error"] = f"{type(exc).__name__}: {exc}"
+ finally:
+ if session is not None:
+ try:
+ session.close()
+ except Exception:
+ pass
+
+ data["status"] = "blocked"
+ data["success"] = False
+ data["blocked_on_device_model"] = True
+ data["assertions"] = {"checks": checks, "failed": []}
+ write_json(os.path.join(results_dir, "results.json"), data)
+ write_text(os.path.join(results_dir, "results.md"), render_md(manifest, data))
+ print(f"trial: blocked (results/results.json): {data['blocked_reason']}")
+ return 0 if "error" not in data else 1
+
+
+def main(argv=None):
+ parser = argparse.ArgumentParser(description="Run a packaged green-eye trial.")
+ parser.add_argument("--package", default=os.path.dirname(os.path.abspath(__file__)))
+ parser.add_argument("--budget", type=int, default=None,
+ help="image probe budget in instructions (default from manifest)")
+ args = parser.parse_args(argv)
+ pkg = os.path.abspath(args.package)
+ bootstrap_tool_root(pkg)
+ manifest = load_json(os.path.join(pkg, "manifest.json"))
+ if manifest["target"]["kind"] == "image":
+ return run_image(pkg, manifest, budget=args.budget)
+ return run_demo(pkg, manifest)
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
+'''
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py b/debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py
new file mode 100644
index 0000000..e751d42
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/port_driver_versions.py
@@ -0,0 +1,1155 @@
+#!/usr/bin/env python3
+"""Driver-version port sweep (roadmap H4.1/R6, docs/28).
+
+Given a version list (``--versions 540,550,560,570``) or a JSON config, find
+each version's firmware target under ``--input-dir`` (driver runfiles/firmware
+blobs already on disk -- this tool never downloads), then per version:
+
+1. run ``harness/version_diff.py`` against a chosen baseline;
+2. run ``harness/portability_matrix.py`` on the baseline + candidate pair;
+3. run a bounded ``harness/mutation_harness.py`` smoke;
+4. assemble a per-version evidence bundle and a cross-version summary
+ (markdown + JSON) with the docs/06 M1-M6 skeleton.
+
+``--dry-run`` is the default: it prints the plan (inputs found/missing, steps,
+expected artifacts) and exits 0. ``--run`` executes, is resume-safe (a version
+whose ``version.json`` already reports ok/partial is skipped unless --force)
+and exits non-zero when any executed version has an errored step. The RM 610
+gate from ``version_diff.py`` (docs/23) is enforced before any target is
+opened, in dry-run too: versions >= 610 require ``--approval-token
+RM610-APPROVED`` or ``GREENEYE_RM610_APPROVED=1``.
+
+Known driver branch cadence (edit DEFAULT_VERSIONS or pass --versions/--config;
+no image is ever invented or downloaded):
+
+ branch expected image
+ 540 earlier branch (cadence only)
+ 550 earlier branch (cadence only)
+ 560 earlier branch (cadence only)
+ 570 current anchor, e.g. 570.144
+ 580 expected cadence, no local image assumed
+ 610 RM 610 gate, refused without approval (docs/23)
+
+Run from the tool root:
+
+ python3 harness/port_driver_versions.py --input-dir ../../firmware/57X-GSP
+ python3 harness/port_driver_versions.py --run --input-dir ../../firmware/57X-GSP \\
+ --versions 540,550,560,570 --out dumps/sweep-570
+"""
+from __future__ import annotations
+
+import argparse
+import contextlib
+import json
+import os
+import re
+import signal
+import subprocess
+import sys
+import threading
+from dataclasses import dataclass, field
+
+try:
+ from .common import (
+ DEMO_ELF,
+ MAX_LOG_BYTES,
+ ensure_dir,
+ qemu_available,
+ sha256_file,
+ write_json,
+ write_text,
+ )
+ from .mutation_harness import HarnessOptions, run_harness
+ from .portability_matrix import (
+ METRIC_DEFS,
+ NO_REFERENCE,
+ MatrixOptions,
+ TargetSpec,
+ render_markdown as render_matrix_markdown,
+ run_matrix,
+ )
+ from .version_diff import (
+ RM610_TOKEN,
+ Rm610GateError,
+ VersionDiffOptions,
+ check_rm610_gate,
+ is_rm610_label,
+ run_version_diff,
+ )
+except ImportError: # direct script execution (python3 harness/port_driver_versions.py)
+ from common import ( # type: ignore[no-redef]
+ DEMO_ELF,
+ MAX_LOG_BYTES,
+ ensure_dir,
+ qemu_available,
+ sha256_file,
+ write_json,
+ write_text,
+ )
+ from mutation_harness import HarnessOptions, run_harness # type: ignore[no-redef]
+ from portability_matrix import ( # type: ignore[no-redef]
+ METRIC_DEFS,
+ NO_REFERENCE,
+ MatrixOptions,
+ TargetSpec,
+ render_markdown as render_matrix_markdown,
+ run_matrix,
+ )
+ from version_diff import ( # type: ignore[no-redef]
+ RM610_TOKEN,
+ Rm610GateError,
+ VersionDiffOptions,
+ check_rm610_gate,
+ is_rm610_label,
+ run_version_diff,
+ )
+
+FORMAT = "greeneye-port-sweep/1"
+PLAN_FORMAT = "greeneye-port-sweep-plan/1"
+VERSION_FORMAT = "greeneye-port-version/1"
+BUNDLE_FORMAT = "greeneye-release-bundle/1"
+CONFIG_FORMAT = "greeneye-port-sweep-config/1"
+
+#: Known driver branch cadence; edit here or pass --versions/--config. No image
+#: is downloaded by this tool: branches without local images become no-input.
+DEFAULT_VERSIONS = ("540", "550", "560", "570")
+RESUMABLE_STATUSES = ("ok", "partial")
+TARGET_SUFFIXES = (".fwimage.bin", ".fwimage", ".elf", ".bin", ".run")
+VERSION_TOKEN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+
+
+# ---------------------------------------------------------------------------
+# options + version helpers
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class SweepTarget:
+ version: str
+ paths: list[str] = field(default_factory=list)
+ kind: str = "none" # fwimage | elf | raw | runfile | explicit | none
+ source: str = "none" # explicit | discovered | none
+ matches: list[str] = field(default_factory=list)
+ runfiles: list[str] = field(default_factory=list)
+ notes: list[str] = field(default_factory=list)
+ symbols: str | None = None
+ ghidra_exports: str | None = None
+
+ @property
+ def path(self) -> str | None:
+ return self.paths[0] if self.paths else None
+
+
+@dataclass
+class SweepOptions:
+ versions: list[str] = field(default_factory=list)
+ baseline: str | None = None
+ input_dir: str | None = None
+ out: str | None = None
+ config: str | None = None
+ targets: dict[str, list[str]] = field(default_factory=dict)
+ symbols: dict[str, str] = field(default_factory=dict)
+ ghidra_exports: dict[str, str] = field(default_factory=dict)
+ run: bool = False
+ force: bool = False
+ extract_runfiles: bool = False
+ approval_token: str | None = None
+ probe_instructions: int = 2000
+ probe_duration: float = 0.5
+ scenario_duration: float = 0.4
+ max_stops: int = 5000
+ max_rows: int = 40
+ mutation_kernel: str = DEMO_ELF
+ mutation_iterations: int = 8
+ mutation_seed: int = 7
+ mutation_timeout: float = 0.25
+ mutation_settle: float = 0.1
+ allow_foreign_mutation: bool = False
+ step_timeout: float = 600.0
+ log_bytes: int = MAX_LOG_BYTES
+
+ @property
+ def dry_run(self) -> bool:
+ return not self.run
+
+
+def parse_versions(text: str) -> list[str]:
+ """Parse ``540,550,560`` into a validated, de-duplicated order-preserving list."""
+ versions: list[str] = []
+ for token in str(text).replace(";", ",").split(","):
+ value = token.strip()
+ if not value:
+ continue
+ if not VERSION_TOKEN_RE.match(value):
+ raise ValueError(f"invalid version token: {value!r}")
+ if value not in versions:
+ versions.append(value)
+ return versions
+
+
+def version_sort_key(value: str) -> tuple:
+ parts = re.findall(r"\d+", value)
+ return (tuple(int(part) for part in parts) if parts else (0,), value)
+
+
+def version_ge_610(value: str) -> bool:
+ match = re.match(r"\s*(\d+)", str(value))
+ return bool(match) and int(match.group(1)) >= 610
+
+
+def _is_numeric_firmware_match(version: str, text: str) -> bool:
+ pattern = re.compile(r"(?<!\d)" + re.escape(version) + r"(?:\.\d+)*(?!\d)")
+ return pattern.search(text) is not None
+
+
+def _gate_version_labels(versions: list[str]) -> list[str]:
+ """Numeric versions >= 610 must trigger the rm610-prefixed gate regex."""
+ return [f"rm610-{value}" if version_ge_610(value) else value
+ for value in versions if version_ge_610(value) or is_rm610_label(value)]
+
+
+def gate_labels(versions: list[str], targets: dict[str, SweepTarget]) -> list[str]:
+ labels = _gate_version_labels(versions)
+ for target in targets.values():
+ labels.append(target.version)
+ labels.extend(os.path.basename(path) for path in target.paths)
+ labels.extend(os.path.basename(path) for path in target.runfiles)
+ return labels
+
+
+# ---------------------------------------------------------------------------
+# config loading
+# ---------------------------------------------------------------------------
+
+
+def load_config(path: str) -> dict:
+ """Read a sweep config; strict enough that typos do not silently pass."""
+ if not os.path.isfile(path):
+ raise ValueError(f"config not found: {path}")
+ try:
+ with open(path, "r", encoding="utf-8") as fh:
+ data = json.load(fh)
+ except (OSError, json.JSONDecodeError) as exc:
+ raise ValueError(f"config {path}: {type(exc).__name__}: {exc}") from exc
+ if not isinstance(data, dict):
+ raise ValueError(f"config {path}: expected a JSON object")
+ known = {"format", "versions", "baseline", "input_dir", "targets", "symbols",
+ "ghidra_exports", "notes"}
+ unknown = sorted(set(data) - known)
+ if unknown:
+ raise ValueError(f"config {path}: unknown key(s): {', '.join(unknown)}")
+ if "format" in data and data["format"] != CONFIG_FORMAT:
+ raise ValueError(f"config {path}: unsupported format {data['format']!r}")
+ if "versions" in data:
+ if not isinstance(data["versions"], list):
+ raise ValueError(f"config {path}: versions must be a list")
+ data["versions"] = [str(value) for value in data["versions"]]
+ for key in ("targets", "symbols", "ghidra_exports"):
+ if key not in data:
+ continue
+ if not isinstance(data[key], dict):
+ raise ValueError(f"config {path}: {key} must be an object")
+ normalised: dict[str, object] = {}
+ for version, value in data[key].items():
+ if key == "targets":
+ if isinstance(value, str):
+ value = [value]
+ if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
+ raise ValueError(f"config {path}: targets[{version}] must be a path or list of paths")
+ elif not isinstance(value, str):
+ raise ValueError(f"config {path}: {key}[{version}] must be a string path")
+ normalised[str(version)] = value
+ data[key] = normalised
+ return data
+
+
+def options_from_config(options: SweepOptions, config: dict) -> None:
+ if config.get("versions"):
+ options.versions = list(config["versions"])
+ if config.get("baseline") and not options.baseline:
+ options.baseline = str(config["baseline"])
+ if config.get("input_dir") and not options.input_dir:
+ options.input_dir = str(config["input_dir"])
+ for version, paths in (config.get("targets") or {}).items():
+ options.targets.setdefault(str(version), []).extend(paths)
+ for version, path in (config.get("symbols") or {}).items():
+ options.symbols.setdefault(str(version), path)
+ for version, path in (config.get("ghidra_exports") or {}).items():
+ options.ghidra_exports.setdefault(str(version), path)
+
+
+# ---------------------------------------------------------------------------
+# target discovery (local disk only)
+# ---------------------------------------------------------------------------
+
+
+def _kind_for(name: str) -> str | None:
+ lower = name.lower()
+ if not any(lower.endswith(suffix) for suffix in TARGET_SUFFIXES):
+ return None
+ if lower.endswith(".run"):
+ return "runfile"
+ if lower.endswith(".fwimage.bin") or lower.endswith(".fwimage"):
+ return "fwimage"
+ if lower.endswith(".elf"):
+ return "elf"
+ if lower.endswith(".bin"):
+ return "raw"
+ return None
+
+
+def discover_version(input_dir: str, version: str) -> SweepTarget:
+ """Find the best local firmware target for one version under ``input_dir``."""
+ target = SweepTarget(version=version)
+ if not input_dir:
+ target.notes.append("no --input-dir supplied and no explicit target")
+ return target
+ if not os.path.isdir(input_dir):
+ target.notes.append(f"input directory not found: {input_dir}")
+ return target
+ ranked: list[tuple[int, int, str, str]] = []
+ for root, dirs, names in os.walk(input_dir):
+ dirs.sort()
+ for name in sorted(names):
+ kind = _kind_for(name)
+ if kind is None:
+ continue
+ path = os.path.join(root, name)
+ rel = os.path.relpath(path, input_dir)
+ parent = os.path.basename(root)
+ if name and _is_numeric_firmware_match(version, name):
+ scope = 0
+ elif parent and _is_numeric_firmware_match(version, parent):
+ scope = 1
+ elif _is_numeric_firmware_match(version, rel):
+ scope = 2
+ else:
+ continue
+ kind_rank = {"fwimage": 0, "elf": 1, "raw": 2, "runfile": 3}[kind]
+ ranked.append((kind_rank, int(scope), path, kind))
+ ranked.sort()
+ for _, _, path, kind in ranked:
+ if kind == "runfile":
+ target.runfiles.append(path)
+ else:
+ target.paths.append(path)
+ target.matches.append(path)
+ if target.kind == "none":
+ target.kind = kind
+ target.source = "discovered"
+ if not target.paths and target.runfiles:
+ target.notes.append(
+ "driver runfile(s) present but not extracted; pass --extract-runfiles "
+ "(local extraction only, nothing is downloaded) or extract with the "
+ "driver runfile's own --extract-only"
+ )
+ if not target.paths and not target.runfiles:
+ target.notes.append(f"no local image matching {version!r} under {input_dir}")
+ return target
+
+
+def resolve_targets(options: SweepOptions) -> dict[str, SweepTarget]:
+ """Resolve every requested version plus an out-of-list baseline when given."""
+ versions = list(options.versions)
+ if options.baseline and options.baseline not in versions:
+ versions.append(options.baseline)
+ resolved: dict[str, SweepTarget] = {}
+ for version in versions:
+ explicit = [path for path in options.targets.get(version, [])]
+ target = SweepTarget(version=version)
+ target.symbols = options.symbols.get(version)
+ target.ghidra_exports = options.ghidra_exports.get(version)
+ if explicit:
+ target.source = "explicit"
+ for path in explicit:
+ if os.path.isfile(path):
+ target.paths.append(os.path.abspath(path))
+ target.matches.append(path)
+ else:
+ target.notes.append(f"explicit target not found: {path}")
+ if target.paths:
+ target.kind = _kind_for(os.path.basename(target.paths[0])) or "explicit"
+ else:
+ target = discover_version(options.input_dir or "", version)
+ target.symbols = options.symbols.get(version)
+ target.ghidra_exports = options.ghidra_exports.get(version)
+ resolved[version] = target
+ return resolved
+
+
+def choose_baseline(options: SweepOptions, resolved: dict[str, SweepTarget]) -> str | None:
+ if options.baseline:
+ return options.baseline
+ found = [version for version in options.versions if resolved.get(version) and resolved[version].paths]
+ return min(found, key=version_sort_key) if found else None
+
+
+# ---------------------------------------------------------------------------
+# bounded steps
+# ---------------------------------------------------------------------------
+
+
+@contextlib.contextmanager
+def _deadline(seconds: float, label: str):
+ """Wall-clock safety net around one step (main thread only)."""
+ usable = (
+ seconds > 0
+ and hasattr(signal, "SIGALRM")
+ and threading.current_thread() is threading.main_thread()
+ )
+ if not usable:
+ yield
+ return
+
+ def _raise(signum, frame): # pragma: no cover - signal trampoline
+ raise TimeoutError(f"{label} exceeded the {seconds:g}s step budget")
+
+ previous = signal.signal(signal.SIGALRM, _raise)
+ previous_timer = signal.setitimer(signal.ITIMER_REAL, seconds)
+ try:
+ yield
+ finally:
+ signal.setitimer(signal.ITIMER_REAL, *previous_timer)
+ signal.signal(signal.SIGALRM, previous)
+
+
+def _run_step(name: str, fn, *args, timeout: float = 0.0, **kwargs) -> dict:
+ try:
+ with _deadline(timeout, name):
+ detail = fn(*args, **kwargs)
+ except TimeoutError as exc:
+ return {"name": name, "status": "error", "reason": str(exc), "timeout": True}
+ except Exception as exc: # a failed step must not hide the others
+ return {"name": name, "status": "error", "reason": f"{type(exc).__name__}: {exc}"}
+ entry = {"name": name}
+ entry.update(detail)
+ return entry
+
+
+# ---------------------------------------------------------------------------
+# per-version steps
+# ---------------------------------------------------------------------------
+
+
+def step_version_diff(version: str, baseline: str, target: SweepTarget, baseline_target: SweepTarget,
+ options: SweepOptions, out_dir: str) -> dict:
+ if target.path is None:
+ return {"status": "skipped", "reason": "no target"}
+ if baseline_target.path is None:
+ return {"status": "error", "reason": f"baseline {baseline!r} has no target"}
+ report = run_version_diff(
+ VersionDiffOptions(
+ old=baseline_target.path,
+ new=target.path,
+ out=out_dir,
+ old_label=baseline,
+ new_label=version,
+ old_symbols=baseline_target.symbols,
+ new_symbols=target.symbols,
+ old_ghidra_exports=baseline_target.ghidra_exports,
+ new_ghidra_exports=target.ghidra_exports,
+ probe_instructions=options.probe_instructions,
+ probe_duration=options.probe_duration,
+ behavior_duration=options.scenario_duration,
+ max_stops=options.max_stops,
+ max_rows=options.max_rows,
+ log_bytes=options.log_bytes,
+ approval_token=options.approval_token,
+ )
+ )
+ sections = {name: section["status"] for name, section in report["sections"].items()}
+ behavior = report["sections"]["behavior"]
+ metrics = behavior.get("metrics") or []
+ if not metrics:
+ metrics = [
+ {"id": metric_id, "name": name, "status": "not measured", "reason": NO_REFERENCE,
+ "a_observed": None, "b_observed": None}
+ for metric_id, name in METRIC_DEFS
+ ]
+ return {
+ "status": "error" if "error" in sections.values() else "ok",
+ "report": "version-diff/version-diff.json",
+ "markdown": "version-diff/version-diff.md",
+ "sections": sections,
+ "all_zero": behavior.get("all_zero"),
+ "metrics": metrics,
+ }
+
+
+def step_portability_matrix(version: str, baseline: str, target: SweepTarget, baseline_target: SweepTarget,
+ options: SweepOptions, out_dir: str) -> dict:
+ if target.path is None:
+ return {"status": "skipped", "reason": "no target"}
+ if baseline_target.path is None:
+ return {"status": "error", "reason": f"baseline {baseline!r} has no target"}
+ matrix = run_matrix(
+ [TargetSpec(path=baseline_target.path, name=baseline),
+ TargetSpec(path=target.path, name=version)],
+ MatrixOptions(
+ out=out_dir,
+ probe_instructions=options.probe_instructions,
+ probe_duration=options.probe_duration,
+ scenario_duration=options.scenario_duration,
+ max_stops=options.max_stops,
+ max_rows=options.max_rows,
+ log_bytes=options.log_bytes,
+ ),
+ )
+ write_json(os.path.join(out_dir, "portability-report.json"), matrix)
+ write_text(
+ os.path.join(out_dir, "portability-report.md"),
+ render_matrix_markdown(matrix, MatrixOptions(out=out_dir, max_rows=options.max_rows)),
+ )
+ statuses = {row["name"]: row["status"] for row in matrix["targets"]}
+ comparisons = [
+ {"a": row["a"], "b": row["b"], "status": row["status"], "all_zero": row["all_zero"]}
+ for row in matrix["comparisons"]
+ ]
+ return {
+ "status": "error" if "error" in statuses.values() else "ok",
+ "report": "portability-matrix/portability-report.json",
+ "markdown": "portability-matrix/portability-report.md",
+ "targets": statuses,
+ "comparisons": comparisons,
+ }
+
+
+def step_mutation_smoke(target: SweepTarget, options: SweepOptions, out_dir: str) -> dict:
+ if not qemu_available():
+ return {"status": "skipped", "reason": "qemu-system-riscv64 not found"}
+ kernel = options.mutation_kernel
+ if not os.path.isfile(kernel):
+ return {"status": "skipped", "reason": f"mutation kernel not found: {kernel}"}
+ findings = run_harness(
+ HarnessOptions(
+ kernel=os.path.abspath(kernel),
+ out=out_dir,
+ iterations=options.mutation_iterations,
+ seed=options.mutation_seed,
+ timeout=options.mutation_timeout,
+ settle=options.mutation_settle,
+ allow_foreign_elf=options.allow_foreign_mutation,
+ )
+ )
+ outcomes: dict[str, int] = {}
+ for finding in findings:
+ outcomes[finding["outcome"]] = outcomes.get(finding["outcome"], 0) + 1
+ return {
+ "status": "ok",
+ "report": "mutation-smoke/findings.jsonl",
+ "markdown": "mutation-smoke/summary.md",
+ "iterations": len(findings),
+ "outcomes": dict(sorted(outcomes.items())),
+ }
+
+
+def version_status(steps: list[dict]) -> str:
+ statuses = [step["status"] for step in steps]
+ if "error" in statuses:
+ return "error"
+ if all(status == "ok" for status in statuses):
+ return "ok"
+ return "partial"
+
+
+def build_bundle(out_dir: str, tool: str) -> dict:
+ files: list[dict] = []
+ for root, dirs, names in os.walk(out_dir):
+ dirs.sort()
+ for name in sorted(names):
+ path = os.path.join(root, name)
+ rel = os.path.relpath(path, out_dir)
+ if rel == "bundle-manifest.json":
+ continue
+ files.append({"path": rel, "bytes": os.path.getsize(path), "sha256": sha256_file(path)})
+ files.sort(key=lambda row: row["path"])
+ return {"format": BUNDLE_FORMAT, "tool": tool, "files": files}
+
+
+def completed_status(version_dir: str) -> str | None:
+ """Return the recorded status of a finished version (resume support)."""
+ path = os.path.join(version_dir, "version.json")
+ if not os.path.isfile(path):
+ return None
+ try:
+ with open(path, "r", encoding="utf-8") as fh:
+ data = json.load(fh)
+ except (OSError, json.JSONDecodeError):
+ return None
+ status = data.get("status")
+ return status if status in RESUMABLE_STATUSES else None
+
+
+# ---------------------------------------------------------------------------
+# sweep execution
+# ---------------------------------------------------------------------------
+
+
+def run_version(version: str, baseline: str, target: SweepTarget, baseline_target: SweepTarget,
+ options: SweepOptions, out_root: str) -> dict:
+ version_dir = os.path.join(out_root, version)
+ entry: dict = {
+ "format": VERSION_FORMAT,
+ "tool": "harness/port_driver_versions.py",
+ "version": version,
+ "baseline": baseline,
+ "target": target.path,
+ "target_kind": target.kind,
+ "target_source": target.source,
+ "target_matches": target.matches,
+ "runfiles": target.runfiles,
+ "notes": target.notes,
+ "symbols": target.symbols,
+ "steps": [],
+ "metrics": [],
+ "status": "no-input",
+ }
+ if target.path is None:
+ entry["reason"] = (
+ "no local firmware target found; "
+ + ("; ".join(target.notes) if target.notes else "supply --input-dir or --target VERSION=PATH")
+ )
+ ensure_dir(version_dir)
+ write_json(os.path.join(version_dir, "version.json"), entry)
+ write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
+ return entry
+
+ steps = [
+ _run_step("version_diff", step_version_diff, version, baseline, target, baseline_target,
+ options, os.path.join(version_dir, "version-diff"), timeout=options.step_timeout),
+ _run_step("portability_matrix", step_portability_matrix, version, baseline, target,
+ baseline_target, options, os.path.join(version_dir, "portability-matrix"),
+ timeout=options.step_timeout),
+ _run_step("mutation_smoke", step_mutation_smoke, target, options,
+ os.path.join(version_dir, "mutation-smoke"), timeout=options.step_timeout),
+ ]
+ entry["steps"] = steps
+ entry["status"] = version_status(steps)
+ for step in steps:
+ if step["name"] == "version_diff":
+ entry["metrics"] = step.get("metrics") or []
+ entry["all_zero"] = step.get("all_zero")
+ entry["target_sha256"] = sha256_file(target.path)
+ entry["evidence"] = {"manifest": os.path.join(version, "bundle-manifest.json")}
+ ensure_dir(version_dir)
+ write_json(os.path.join(version_dir, "version.json"), entry)
+ write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
+ manifest = build_bundle(version_dir, "harness/port_driver_versions.py")
+ write_json(os.path.join(version_dir, "bundle-manifest.json"), manifest)
+ return entry
+
+
+def run_sweep(options: SweepOptions) -> dict:
+ check_rm610_gate(
+ _gate_version_labels(options.versions),
+ approval_token=options.approval_token,
+ )
+ resolved = resolve_targets(options)
+ baseline = choose_baseline(options, resolved)
+ labels = gate_labels(list(options.versions), resolved)
+ check_rm610_gate(labels, approval_token=options.approval_token)
+ if not options.out:
+ raise ValueError("--out is required unless --dry-run is used")
+ out_root = os.path.abspath(options.out)
+ ensure_dir(out_root)
+ versions: list[dict] = []
+ for version in options.versions:
+ target = resolved[version]
+ version_dir = os.path.join(out_root, version)
+ previous = None if options.force else completed_status(version_dir)
+ if previous is not None:
+ versions.append({
+ "format": VERSION_FORMAT,
+ "tool": "harness/port_driver_versions.py",
+ "version": version,
+ "baseline": baseline,
+ "status": previous,
+ "resumed": True,
+ "target": target.path,
+ "evidence": {"manifest": os.path.relpath(
+ os.path.join(version_dir, "bundle-manifest.json"), out_root)},
+ })
+ continue
+ if target.path is None and options.extract_runfiles and target.runfiles:
+ extract_dir = os.path.join(version_dir, "runfile-extract")
+ try:
+ target = extract_runfiles(version, target, extract_dir, options)
+ except Exception as exc:
+ entry = {
+ "format": VERSION_FORMAT,
+ "tool": "harness/port_driver_versions.py",
+ "version": version,
+ "baseline": baseline,
+ "target": None,
+ "steps": [],
+ "metrics": [],
+ "status": "error",
+ "reason": f"runfile extraction failed: {type(exc).__name__}: {exc}",
+ }
+ ensure_dir(version_dir)
+ write_json(os.path.join(version_dir, "version.json"), entry)
+ write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
+ versions.append(entry)
+ continue
+ baseline_target = resolved.get(baseline) if baseline else None
+ if baseline_target is None:
+ entry = {
+ "format": VERSION_FORMAT,
+ "tool": "harness/port_driver_versions.py",
+ "version": version,
+ "baseline": None,
+ "target": target.path,
+ "target_kind": target.kind,
+ "target_source": target.source,
+ "steps": [],
+ "metrics": [],
+ "status": "error",
+ "reason": "no baseline target: pass --baseline or ensure one version has an image",
+ }
+ ensure_dir(version_dir)
+ write_json(os.path.join(version_dir, "version.json"), entry)
+ write_text(os.path.join(version_dir, "version.md"), render_version_markdown(entry))
+ versions.append(entry)
+ continue
+ versions.append(run_version(version, baseline, target, baseline_target, options, out_root))
+
+ failed = [row["version"] for row in versions if row["status"] == "error"]
+ not_run = [row["version"] for row in versions if row["status"] == "no-input"]
+ ran = len(versions) - len(not_run)
+ if failed:
+ interpretation = (
+ "fail: " + ", ".join(failed)
+ + " had an errored step; inspect each version.json and the step reports"
+ )
+ elif ran == 0:
+ interpretation = (
+ "not-run: no version had a local target; nothing was executed and no evidence was produced"
+ )
+ else:
+ interpretation = (
+ "pass: every executed version completed with no errored step; "
+ "skipped steps and resume decisions are listed per version"
+ )
+ report = {
+ "format": FORMAT,
+ "tool": "harness/port_driver_versions.py",
+ "dry_run": False,
+ "rm610_gate": {"status": "passed", "labels": sorted(set(labels))},
+ "baseline": {"version": baseline,
+ "path": resolved[baseline].path if baseline and baseline in resolved else None},
+ "input_dir": options.input_dir,
+ "versions": versions,
+ "ran": ran,
+ "failed": failed,
+ "not_run": not_run,
+ "interpretation": interpretation,
+ "summary": "sweep-summary.json",
+ "markdown": "sweep-summary.md",
+ "manifest": "bundle-manifest.json",
+ }
+ write_json(os.path.join(out_root, "sweep-summary.json"), report)
+ write_text(os.path.join(out_root, "sweep-summary.md"), render_sweep_markdown(report))
+ manifest = build_bundle(out_root, "harness/port_driver_versions.py")
+ write_json(os.path.join(out_root, "bundle-manifest.json"), manifest)
+ return report
+
+
+def extract_runfiles(version: str, target: SweepTarget, extract_dir: str, options: SweepOptions) -> SweepTarget:
+ """Extract local driver runfile(s) with their own --extract-only (no download)."""
+ ensure_dir(extract_dir)
+ for runfile in target.runfiles:
+ argv = ["sh", os.path.abspath(runfile), "--extract-only", "--target", extract_dir]
+ subprocess.run(argv, check=True, timeout=options.step_timeout,
+ stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+ discovered = discover_version(extract_dir, version)
+ discovered.symbols = target.symbols
+ discovered.ghidra_exports = target.ghidra_exports
+ discovered.notes = target.notes + [f"extracted from {len(target.runfiles)} runfile(s)"]
+ discovered.runfiles = target.runfiles
+ return discovered
+
+
+# ---------------------------------------------------------------------------
+# plan (dry run)
+# ---------------------------------------------------------------------------
+
+
+def _expected_artifacts(out: str, version: str) -> list[str]:
+ base = os.path.join(out, version)
+ return [
+ os.path.join(base, "version.json"),
+ os.path.join(base, "version.md"),
+ os.path.join(base, "bundle-manifest.json"),
+ os.path.join(base, "version-diff", "version-diff.json"),
+ os.path.join(base, "version-diff", "version-diff.md"),
+ os.path.join(base, "portability-matrix", "portability-report.json"),
+ os.path.join(base, "portability-matrix", "portability-report.md"),
+ os.path.join(base, "mutation-smoke", "findings.jsonl"),
+ os.path.join(base, "mutation-smoke", "summary.md"),
+ ]
+
+
+def plan_sweep(options: SweepOptions) -> dict:
+ check_rm610_gate(
+ _gate_version_labels(options.versions),
+ approval_token=options.approval_token,
+ )
+ resolved = resolve_targets(options)
+ baseline = choose_baseline(options, resolved)
+ labels = gate_labels(list(options.versions), resolved)
+ check_rm610_gate(labels, approval_token=options.approval_token)
+ out = os.path.abspath(options.out) if options.out else "<out>"
+ versions = []
+ found = missing = 0
+ for version in options.versions:
+ target = resolved[version]
+ entry = {
+ "version": version,
+ "status": "found" if target.path else ("runfile" if target.runfiles else "no-input"),
+ "target": target.path,
+ "kind": target.kind,
+ "source": target.source,
+ "matches": target.matches,
+ "runfiles": target.runfiles,
+ "notes": target.notes,
+ "baseline": version == baseline,
+ "symbols": target.symbols,
+ "steps": [
+ {"name": "version_diff", "out": os.path.join(out, version, "version-diff"),
+ "old": resolved[baseline].path if baseline and baseline in resolved else None,
+ "new": target.path},
+ {"name": "portability_matrix", "out": os.path.join(out, version, "portability-matrix"),
+ "targets": [baseline, version] if baseline else [version]},
+ {"name": "mutation_smoke", "out": os.path.join(out, version, "mutation-smoke"),
+ "kernel": options.mutation_kernel},
+ ],
+ "expected_artifacts": _expected_artifacts(out, version),
+ }
+ if options.extract_runfiles and target.runfiles and not target.path:
+ entry["extract"] = [
+ {"argv": ["sh", os.path.abspath(runfile), "--extract-only", "--target",
+ os.path.join(out, version, "runfile-extract")]}
+ for runfile in target.runfiles
+ ]
+ if target.path:
+ found += 1
+ else:
+ missing += 1
+ versions.append(entry)
+ return {
+ "format": PLAN_FORMAT,
+ "tool": "harness/port_driver_versions.py",
+ "dry_run": True,
+ "rm610_gate": {"status": "passed", "labels": sorted(set(labels))},
+ "input_dir": options.input_dir,
+ "out": out,
+ "baseline": baseline,
+ "versions": versions,
+ "found": found,
+ "missing": missing,
+ "expected_artifacts": [
+ os.path.join(out, "sweep-summary.json"),
+ os.path.join(out, "sweep-summary.md"),
+ os.path.join(out, "bundle-manifest.json"),
+ ],
+ "interpretation": (
+ f"dry-run: {found} input(s) found, {missing} missing; "
+ "pass --run to execute, --force to re-run completed versions"
+ ),
+ }
+
+
+# ---------------------------------------------------------------------------
+# rendering
+# ---------------------------------------------------------------------------
+
+
+def _table(headers: list[str], rows: list[list]) -> list[str]:
+ lines = ["| " + " | ".join(headers) + " |", "|" + "|".join("---" for _ in headers) + "|"]
+ for row in rows:
+ lines.append("| " + " | ".join(str(cell) for cell in row) + " |")
+ return lines
+
+
+def render_version_markdown(entry: dict) -> str:
+ lines = [
+ f"# green-eye port version {entry['version']}",
+ "",
+ "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_driver_versions.py`.",
+ "",
+ f"* status: **{entry['status']}**",
+ f"* baseline: `{entry.get('baseline')}`",
+ f"* target: `{entry.get('target')}`",
+ ]
+ if entry.get("target_sha256"):
+ lines.append(f"* target sha256: `{entry['target_sha256']}`")
+ if entry.get("reason"):
+ lines.append(f"* reason: {entry['reason']}")
+ lines += ["", "## Steps", ""]
+ lines += _table(
+ ["step", "status", "report", "reason"],
+ [[step["name"], step["status"], step.get("report") or "-", step.get("reason") or "-"]
+ for step in entry.get("steps") or []] or [["-", "-", "-", "not run"]],
+ )
+ lines += ["", "## M1-M6 (docs/06)", "", f"> {NO_REFERENCE}", ""]
+ lines += _table(
+ ["#", "metric", "status", "observed"],
+ [[row["id"], row["name"], row["status"],
+ json.dumps({"a": row.get("a_observed"), "b": row.get("b_observed")}, sort_keys=True)]
+ for row in entry.get("metrics") or []],
+ )
+ lines += [""]
+ return "\n".join(lines)
+
+
+def render_sweep_markdown(report: dict) -> str:
+ lines = [
+ "# green-eye driver-version port sweep",
+ "",
+ "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_driver_versions.py`.",
+ "",
+ f"* format: `{report['format']}`",
+ f"* baseline: `{report['baseline']['version']}` (`{report['baseline']['path']}`)",
+ f"* input dir: `{report.get('input_dir')}`",
+ f"* ran: {report['ran']}, failed: {len(report['failed'])}, no input: {len(report['not_run'])}",
+ "",
+ f"> **{report['interpretation']}**",
+ "",
+ "## Versions",
+ "",
+ ]
+ lines += _table(
+ ["version", "status", "target", "version-diff", "matrix", "mutation", "evidence"],
+ [
+ [
+ row["version"],
+ row["status"] + (" (resumed)" if row.get("resumed") else ""),
+ f"`{(row.get('target') or '-')}`",
+ _step_status(row, "version_diff"),
+ _step_status(row, "portability_matrix"),
+ _step_status(row, "mutation_smoke"),
+ row.get("evidence", {}).get("manifest", "-"),
+ ]
+ for row in report["versions"]
+ ],
+ )
+ lines += ["", "## M1-M6 across versions", "", f"> {NO_REFERENCE}", ""]
+ metric_ids = [metric_id for metric_id, _ in METRIC_DEFS]
+ lines += _table(
+ ["version"] + metric_ids,
+ [
+ [row["version"]] + [
+ next((m["status"] for m in row.get("metrics") or [] if m["id"] == metric_id), "-")
+ for metric_id in metric_ids
+ ]
+ for row in report["versions"]
+ ],
+ )
+ lines += ["", "## Failed / not run", ""]
+ if not report["failed"] and not report["not_run"]:
+ lines += ["_none_", ""]
+ for version in report["failed"]:
+ lines += [f"* failed `{version}`: see `{version}/version.json` and the step reports"]
+ for version in report["not_run"]:
+ row = next(item for item in report["versions"] if item["version"] == version)
+ lines += [f"* no input `{version}`: {row.get('reason', '-')}"]
+ lines += [
+ "",
+ "## Evidence",
+ "",
+ f"* sweep summary: `{report['summary']}` / `{report['markdown']}`",
+ f"* bundle manifest: `{report['manifest']}` (sha256 of every artifact)",
+ "* per-version tree: `version.json`, `version.md`, `version-diff/`, "
+ "`portability-matrix/`, `mutation-smoke/`, `bundle-manifest.json`",
+ "",
+ ]
+ return "\n".join(lines)
+
+
+def _step_status(row: dict, name: str) -> str:
+ for step in row.get("steps") or []:
+ if step["name"] == name:
+ return step["status"]
+ return "-"
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+def _parse_mapping(value: str, flag: str) -> tuple[str, str]:
+ if "=" not in value:
+ raise argparse.ArgumentTypeError(f"{flag} expects VERSION=PATH, got {value!r}")
+ version, path = value.split("=", 1)
+ version, path = version.strip(), path.strip()
+ if not version or not path:
+ raise argparse.ArgumentTypeError(f"{flag} expects VERSION=PATH, got {value!r}")
+ return version, path
+
+
+_HELP_EPILOG = """\
+invocation:
+ dry-run first (default; reads input dirs only, never touches a target):
+ python3 harness/port_driver_versions.py --input-dir ../../firmware/57X-GSP
+ execute, resume-safe:
+ python3 harness/port_driver_versions.py --run --input-dir ../../firmware/57X-GSP \\
+ --versions 540,550,560,570 --out dumps/sweep-570
+ explicit local targets (no discovery):
+ python3 harness/port_driver_versions.py --run --versions 570.144 \\
+ --target 570.144=../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\
+ --baseline 570.144 --out dumps/sweep-570
+
+cron (weekly, resume-safe; nothing is downloaded):
+ 17 3 * * 1 cd <tool-root> && python3 harness/port_driver_versions.py --run \\
+ --input-dir ../../firmware/57X-GSP --versions 540,550,560,570 \\
+ --out dumps/sweep-570 >> dumps/sweep-570/cron.log 2>&1
+
+agent (MCP, docs/20): read the dry-run plan, then call --run; read
+sweep-summary.json and each bundle-manifest.json instead of recomputing.
+RM 610 (docs/23): versions >= 610 require --approval-token RM610-APPROVED or
+GREENEYE_RM610_APPROVED=1; the gate fires before any file is opened.
+"""
+
+
+def build_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(
+ prog="port_driver_versions",
+ description=__doc__.splitlines()[0],
+ epilog=_HELP_EPILOG,
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ )
+ parser.add_argument("--versions", help="comma-separated version list (default: known cadence: "
+ + ",".join(DEFAULT_VERSIONS) + ")")
+ parser.add_argument("--config", help="JSON sweep config (versions, baseline, input_dir, targets, symbols)")
+ parser.add_argument("--baseline", help="baseline version for version_diff (default: earliest found)")
+ parser.add_argument("--input-dir", help="directory to discover local firmware/driver images in (recursive)")
+ parser.add_argument("--target", action="append", default=[], metavar="VERSION=PATH",
+ help="explicit local target for a version (repeatable; skips discovery)")
+ parser.add_argument("--symbols", action="append", default=[], metavar="VERSION=PATH",
+ help="green-eye symbol DB JSON for a version (repeatable)")
+ parser.add_argument("--ghidra-exports", action="append", default=[], metavar="VERSION=PATH",
+ help="Ghidra exports TSV for a version (repeatable)")
+ parser.add_argument("--out", help="sweep output directory (required for --run)")
+ parser.add_argument("--run", dest="run", action="store_true",
+ help="execute the sweep (default is the dry-run plan)")
+ parser.add_argument("--dry-run", dest="run", action="store_false",
+ help="print the plan and exit 0 (default)")
+ parser.set_defaults(run=False)
+ parser.add_argument("--force", action="store_true",
+ help="re-run versions whose version.json already reports ok/partial")
+ parser.add_argument("--extract-runfiles", action="store_true",
+ help="locally extract driver runfiles with --extract-only (never downloads)")
+ parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})")
+ parser.add_argument("--probe-instructions", type=int, default=2000,
+ help="probe step budget per target (0 = wall-clock --probe-duration)")
+ parser.add_argument("--probe-duration", type=float, default=0.5)
+ parser.add_argument("--scenario-duration", type=float, default=0.4,
+ help="budget per version_diff/matrix scenario phase")
+ parser.add_argument("--max-stops", type=int, default=5000)
+ parser.add_argument("--max-rows", type=int, default=40)
+ parser.add_argument("--mutation-kernel", default=DEMO_ELF,
+ help="kernel for the mutation smoke (default: bundled demo)")
+ parser.add_argument("--mutation-iterations", type=int, default=8)
+ parser.add_argument("--mutation-seed", type=int, default=7)
+ parser.add_argument("--mutation-timeout", type=float, default=0.25)
+ parser.add_argument("--mutation-settle", type=float, default=0.1)
+ parser.add_argument("--allow-foreign-mutation", action="store_true",
+ help="permit a non-demo mutation kernel (QEMU only)")
+ parser.add_argument("--step-timeout", type=float, default=600.0,
+ help="wall-clock safety net per step, seconds (0 disables)")
+ parser.add_argument("--log-bytes", type=int, default=MAX_LOG_BYTES)
+ return parser
+
+
+def options_from_args(args: argparse.Namespace) -> SweepOptions:
+ options = SweepOptions()
+ if args.config:
+ options.config = args.config
+ options_from_config(options, load_config(args.config))
+ if args.versions:
+ options.versions = parse_versions(args.versions)
+ if not options.versions:
+ options.versions = list(DEFAULT_VERSIONS)
+ options.baseline = args.baseline
+ options.input_dir = args.input_dir
+ options.out = args.out
+ for value in args.target:
+ version, path = _parse_mapping(value, "--target")
+ options.targets.setdefault(version, []).append(path)
+ for value in args.symbols:
+ version, path = _parse_mapping(value, "--symbols")
+ options.symbols[version] = path
+ for value in args.ghidra_exports:
+ version, path = _parse_mapping(value, "--ghidra-exports")
+ options.ghidra_exports[version] = path
+ options.run = args.run
+ options.force = args.force
+ options.extract_runfiles = args.extract_runfiles
+ options.approval_token = args.approval_token
+ options.probe_instructions = args.probe_instructions
+ options.probe_duration = args.probe_duration
+ options.scenario_duration = args.scenario_duration
+ options.max_stops = args.max_stops
+ options.max_rows = args.max_rows
+ options.mutation_kernel = args.mutation_kernel
+ options.mutation_iterations = args.mutation_iterations
+ options.mutation_seed = args.mutation_seed
+ options.mutation_timeout = args.mutation_timeout
+ options.mutation_settle = args.mutation_settle
+ options.allow_foreign_mutation = args.allow_foreign_mutation
+ options.step_timeout = args.step_timeout
+ options.log_bytes = args.log_bytes
+ return options
+
+
+def main(argv: list[str] | None = None) -> int:
+ parser = build_parser()
+ args = parser.parse_args(argv)
+ try:
+ options = options_from_args(args)
+ except (ValueError, argparse.ArgumentTypeError) as exc:
+ parser.error(str(exc))
+ try:
+ check_rm610_gate(
+ _gate_version_labels(options.versions),
+ approval_token=options.approval_token,
+ )
+ check_rm610_gate(
+ gate_labels(options.versions, resolve_targets(options)),
+ approval_token=options.approval_token,
+ )
+ except Rm610GateError as exc:
+ print(f"[sweep] RM610 gate refused: {exc}", file=sys.stderr)
+ return 3
+ if options.dry_run:
+ try:
+ plan = plan_sweep(options)
+ except (ValueError, FileNotFoundError) as exc:
+ print(f"[sweep] error: {exc}", file=sys.stderr)
+ return 2
+ print(json.dumps(plan, indent=2, sort_keys=True))
+ print(f"[sweep] dry-run: {plan['found']} input(s) found, {plan['missing']} missing; no target touched")
+ return 0
+ if not options.out:
+ parser.error("--out is required unless --dry-run is used")
+ try:
+ report = run_sweep(options)
+ except (ValueError, FileNotFoundError) as exc:
+ print(f"[sweep] error: {exc}", file=sys.stderr)
+ return 2
+ except Rm610GateError as exc: # pragma: no cover - main checks first
+ print(f"[sweep] RM610 gate refused: {exc}", file=sys.stderr)
+ return 3
+ for row in report["versions"]:
+ reason = f" reason={row['reason']}" if row.get("reason") else ""
+ resumed = " resumed" if row.get("resumed") else ""
+ print(f"[sweep] {row['version']}: status={row['status']}{resumed}{reason}")
+ print(f"[sweep] {report['interpretation']}")
+ print(f"[sweep] wrote {os.path.join(os.path.abspath(options.out), 'sweep-summary.json')}")
+ print(f"[sweep] wrote {os.path.join(os.path.abspath(options.out), 'bundle-manifest.json')}")
+ return 1 if report["failed"] or report["ran"] == 0 else 0
+
+
+if __name__ == "__main__": # pragma: no cover
+ raise SystemExit(main())
diff --git a/debug/risc-v-gsp-rm-vm/harness/port_new_device.py b/debug/risc-v-gsp-rm-vm/harness/port_new_device.py
new file mode 100644
index 0000000..56522ec
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/harness/port_new_device.py
@@ -0,0 +1,726 @@
+#!/usr/bin/env python3
+"""New-device port scaffold (roadmap H4.1/R6, docs/07 P2, docs/22 §3, docs/28).
+
+``--device <name>`` creates or uses a profile JSON holding the user-supplied
+device facts (PCI IDs, BARs, FB geometry, engine inventory, interrupt topology,
+ROM/VBIOS notes) and then drives, on the local QEMU backend only:
+
+1. probe-first-touch on the device's firmware (first unmapped access, bounded);
+2. a symbol inventory (ELF .symtab + optional Ghidra/green-eye DB sources);
+3. a portability matrix run of the device against a reference firmware;
+4. a port checklist report against the docs/07 evidence checklist.
+
+``--dry-run`` is the default: it prints the plan (profile action, missing
+profile fields, steps, expected artifacts) and exits 0 without creating files.
+``--run`` writes an evidence bundle under ``--out`` with a sha256 manifest.
+The RM 610 gate from ``version_diff.py`` (docs/23) is enforced before any file
+is opened, in dry-run too.
+
+This tool never touches hardware: no /dev, no kernel module, no mdev, no
+guest, no physical GPU. Physical validation is a separate supervised step
+(docs/06-portability/hardware-software-matrix.md §5.1).
+
+Run from the tool root:
+
+ python3 harness/port_new_device.py --device rtx4090 --profile profiles/rtx4090.json
+ python3 harness/port_new_device.py --run --device rtx4090 \\
+ --firmware ../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\
+ --reference firmware/uproc_demo/build/uproc_demo.elf --out dumps/port-rtx4090
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from dataclasses import dataclass
+
+try:
+ from .common import (
+ MAX_LOG_BYTES,
+ SymbolDB,
+ ensure_dir,
+ qemu_available,
+ sha256_file,
+ write_json,
+ write_text,
+ )
+ from .portability_matrix import (
+ MatrixOptions,
+ TargetSpec,
+ render_markdown as render_matrix_markdown,
+ run_matrix,
+ )
+ from .version_diff import (
+ RM610_TOKEN,
+ Rm610GateError,
+ VersionDiffOptions,
+ check_rm610_gate,
+ load_any_target,
+ load_symbol_sources,
+ run_probe,
+ )
+except ImportError: # direct script execution (python3 harness/port_new_device.py)
+ from common import ( # type: ignore[no-redef]
+ MAX_LOG_BYTES,
+ SymbolDB,
+ ensure_dir,
+ qemu_available,
+ sha256_file,
+ write_json,
+ write_text,
+ )
+ from portability_matrix import ( # type: ignore[no-redef]
+ MatrixOptions,
+ TargetSpec,
+ render_markdown as render_matrix_markdown,
+ run_matrix,
+ )
+ from version_diff import ( # type: ignore[no-redef]
+ RM610_TOKEN,
+ Rm610GateError,
+ VersionDiffOptions,
+ check_rm610_gate,
+ load_any_target,
+ load_symbol_sources,
+ run_probe,
+ )
+
+FORMAT = "greeneye-device-port/1"
+PLAN_FORMAT = "greeneye-device-port-plan/1"
+PROFILE_FORMAT = "greeneye-device-profile/1"
+BUNDLE_FORMAT = "greeneye-release-bundle/1"
+DEVICE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+
+#: Where each user-supplied profile section comes from (docs/22 §3.1).
+PROFILE_HELP = {
+ "pci": "lspci / vendor docs, config/hardware/*.yaml is a design target only",
+ "bars": "guest lspci -vvv, host sysfs",
+ "fb": "host oracle: rmtool mem stats --format=json",
+ "engines": "host oracle: rmtool engine list --format=json; validate per driver profile",
+ "interrupts": "MSI-X tables, guest lspci -vvv",
+ "rom": "ROM/VBIOS read; laptop/Surface overrides",
+ "firmware": "extracted GSP .fwimage or re-hosted ELF (user-supplied; never downloaded)",
+ "pairing": "verify image family + FMC manifest + version string before debugging (docs/07 P2.2)",
+ "physical": "supervised physical validation only; this tool is simulator-only (docs/22 §3.5)",
+}
+
+
+@dataclass
+class DeviceOptions:
+ device: str
+ profile: str | None = None
+ firmware: str | None = None
+ reference: str | None = None
+ symbols: str | None = None
+ ghidra_exports: str | None = None
+ ghidra_bias: int = 0
+ out: str | None = None
+ run: bool = False
+ approval_token: str | None = None
+ probe_instructions: int = 2000
+ probe_duration: float = 0.5
+ scenario_duration: float = 0.4
+ max_stops: int = 5000
+ max_rows: int = 40
+ step_timeout: float = 600.0
+ log_bytes: int = MAX_LOG_BYTES
+
+ @property
+ def dry_run(self) -> bool:
+ return not self.run
+
+
+# ---------------------------------------------------------------------------
+# profile
+# ---------------------------------------------------------------------------
+
+
+def scaffold_profile(name: str) -> dict:
+ """Empty user-supplied profile; every field documents where it comes from."""
+ return {
+ "format": PROFILE_FORMAT,
+ "name": name,
+ "vendor": "",
+ "device": "",
+ "pci": {"vendor_id": "0x10de", "device_id": "", "subsystem_vendor_id": "",
+ "subsystem_device_id": ""},
+ "bars": [],
+ "fb": {"vram_mib": None, "bar1_mib": None, "ecc": None, "notes": ""},
+ "engines": [],
+ "interrupts": {"msix_vectors": None, "notes": ""},
+ "rom": {"vblos_sha256": "", "notes": ""},
+ "firmware": "",
+ "firmware_sha256": "",
+ "firmware_family": "",
+ "firmware_version": "",
+ "reference": "",
+ "physical_validation": {"status": "pending", "date": "", "by": "", "evidence": "", "notes": ""},
+ "notes": "",
+ "where_to_fill": dict(PROFILE_HELP),
+ "references": [
+ "docs/07-PORTABILITY-PLAYBOOKS.md P2",
+ "docs/22-PORTING-CUDA-AND-DEVICES.md 3",
+ "docs/06-portability/hardware-software-matrix.md 5.1",
+ ],
+ }
+
+
+def load_profile(path: str) -> tuple[dict, bool]:
+ """Return ``(profile, created)``; an unreadable profile is a clean error."""
+ if not os.path.isfile(path):
+ return {}, False
+ with open(path, "r", encoding="utf-8") as fh:
+ data = json.load(fh)
+ if not isinstance(data, dict):
+ raise ValueError(f"profile {path}: expected a JSON object")
+ return data, True
+
+
+def profile_missing_fields(profile: dict) -> list[str]:
+ missing: list[str] = []
+ pci = profile.get("pci") or {}
+ if not any(str(pci.get(key, "")).strip() for key in ("vendor_id", "device_id", "subsystem_device_id")):
+ missing.append("pci")
+ if not profile.get("bars"):
+ missing.append("bars")
+ fb = profile.get("fb") or {}
+ if not any(fb.get(key) not in (None, "", 0) for key in ("vram_mib", "bar1_mib", "notes")):
+ missing.append("fb")
+ if not profile.get("engines"):
+ missing.append("engines")
+ interrupts = profile.get("interrupts") or {}
+ if not any(interrupts.get(key) not in (None, "", 0) for key in ("msix_vectors", "notes")):
+ missing.append("interrupts")
+ rom = profile.get("rom") or {}
+ if not any(str(rom.get(key, "")).strip() for key in ("vblos_sha256", "notes")):
+ missing.append("rom")
+ return missing
+
+
+def profile_labels(profile: dict, device: str, profile_path: str | None) -> list[str]:
+ labels = [device, os.path.basename(profile_path or "")]
+ for key in ("name", "device", "firmware", "reference", "firmware_family", "firmware_version"):
+ value = profile.get(key)
+ if isinstance(value, str) and value:
+ labels.append(os.path.basename(value))
+ return labels
+
+
+def merge_profile(profile: dict, options: DeviceOptions) -> dict:
+ merged = json.loads(json.dumps(profile))
+ if options.firmware:
+ merged["firmware"] = options.firmware
+ if options.reference:
+ merged["reference"] = options.reference
+ return merged
+
+
+# ---------------------------------------------------------------------------
+# plan (dry run)
+# ---------------------------------------------------------------------------
+
+
+def _expected_artifacts(out: str, device: str) -> list[str]:
+ base = os.path.join(out, device)
+ return [
+ os.path.join(base, "device-port.json"),
+ os.path.join(base, "device-port.md"),
+ os.path.join(base, "port-checklist.json"),
+ os.path.join(base, "port-checklist.md"),
+ os.path.join(base, "symbols", f"{device}.json"),
+ os.path.join(base, "sessions"),
+ os.path.join(base, "portability-matrix", "portability-report.json"),
+ os.path.join(base, "portability-matrix", "portability-report.md"),
+ os.path.join(base, "bundle-manifest.json"),
+ ]
+
+
+def build_plan(options: DeviceOptions) -> dict:
+ profile_path = options.profile or (os.path.join(options.out or "<out>", "profiles",
+ f"{options.device}.json"))
+ profile, created = load_profile(profile_path) if os.path.isfile(profile_path) else ({}, False)
+ merged = merge_profile(profile if created else scaffold_profile(options.device), options)
+ firmware = merged.get("firmware") or ""
+ reference = merged.get("reference") or ""
+ check_rm610_gate(
+ profile_labels(profile if created else {}, options.device, profile_path)
+ + [os.path.basename(firmware), os.path.basename(reference)],
+ approval_token=options.approval_token,
+ )
+ out = os.path.abspath(options.out) if options.out else "<out>"
+ return {
+ "format": PLAN_FORMAT,
+ "tool": "harness/port_new_device.py",
+ "dry_run": True,
+ "rm610_gate": {"status": "passed", "labels": sorted(set(
+ profile_labels(profile if created else {}, options.device, profile_path)
+ + [os.path.basename(firmware), os.path.basename(reference)]
+ ))},
+ "device": options.device,
+ "profile": {
+ "path": profile_path,
+ "action": "use" if created else "would-create-scaffold",
+ "missing_fields": profile_missing_fields(merged),
+ "where_to_fill": PROFILE_HELP,
+ },
+ "firmware": {"path": firmware or None, "found": bool(firmware) and os.path.isfile(firmware)},
+ "reference": {"path": reference or None,
+ "found": bool(reference) and os.path.isfile(reference)},
+ "out": out,
+ "steps": [
+ {"name": "probe_first_touch", "target": firmware or None,
+ "out": os.path.join(out, options.device, "sessions", f"{options.device}-probe")},
+ {"name": "symbol_inventory", "sources": [firmware or None, options.symbols]},
+ {"name": "portability_matrix", "targets": [reference or None, firmware or None]},
+ {"name": "port_checklist", "out": os.path.join(out, options.device, "port-checklist.json")},
+ ],
+ "expected_artifacts": _expected_artifacts(out, options.device),
+ "interpretation": (
+ "dry-run: fill the profile fields, then pass --run; physical validation "
+ "is a separate supervised step and is never done by this tool"
+ ),
+ }
+
+
+# ---------------------------------------------------------------------------
+# steps
+# ---------------------------------------------------------------------------
+
+
+def _run_step(name: str, fn, *args, **kwargs) -> dict:
+ try:
+ detail = fn(*args, **kwargs)
+ except Exception as exc: # a failed step must not hide the others
+ return {"name": name, "status": "error", "reason": f"{type(exc).__name__}: {exc}"}
+ entry = {"name": name}
+ entry.update(detail)
+ return entry
+
+
+def step_probe(loaded, db: SymbolDB, options: DeviceOptions, out_dir: str, qemu_ok: bool) -> dict:
+ vd = VersionDiffOptions(
+ old=loaded.path,
+ new=loaded.path,
+ out=out_dir,
+ probe_instructions=options.probe_instructions,
+ probe_duration=options.probe_duration,
+ log_bytes=options.log_bytes,
+ approval_token=options.approval_token,
+ )
+ result = run_probe(loaded, db, vd, out_dir, qemu_ok)
+ return {
+ "status": result["status"],
+ "reason": result.get("reason"),
+ "session": result.get("session"),
+ "stopped_pc": result.get("stopped_pc"),
+ "stopped_symbol": result.get("stopped_symbol"),
+ "fault_count": result.get("fault_count"),
+ "first_fault": result.get("first_fault"),
+ }
+
+
+def step_symbols(loaded, db: SymbolDB, sources: list[str], notes: list[str],
+ options: DeviceOptions, out_dir: str) -> dict:
+ records = [
+ {"name": symbol.name, "addr": symbol.addr, "size": symbol.size, "source": symbol.source}
+ for symbol in db
+ ]
+ rel = None
+ if records:
+ rel = os.path.join("symbols", f"{options.device}.json")
+ path = os.path.join(out_dir, rel)
+ ensure_dir(os.path.dirname(path))
+ db.save_json(path, meta={
+ "device": options.device,
+ "firmware": os.path.basename(loaded.path),
+ "bias": loaded.bias,
+ "sources": sources,
+ "notes": notes,
+ })
+ return {
+ "status": "ok" if records else "skipped",
+ "reason": None if records else "no symbols: no ELF .symtab and no Ghidra/JSON sources supplied",
+ "count": len(records),
+ "sources": sources,
+ "notes": notes,
+ "db": rel,
+ }
+
+
+def step_matrix(firmware: str, reference: str | None, options: DeviceOptions, out_dir: str) -> dict:
+ specs = []
+ if reference and os.path.isfile(reference):
+ specs.append(TargetSpec(path=os.path.abspath(reference), name="reference"))
+ specs.append(TargetSpec(path=os.path.abspath(firmware), name=options.device))
+ matrix = run_matrix(
+ specs,
+ MatrixOptions(
+ out=out_dir,
+ probe_instructions=options.probe_instructions,
+ probe_duration=options.probe_duration,
+ scenario_duration=options.scenario_duration,
+ max_stops=options.max_stops,
+ max_rows=options.max_rows,
+ log_bytes=options.log_bytes,
+ ),
+ )
+ write_json(os.path.join(out_dir, "portability-report.json"), matrix)
+ write_text(
+ os.path.join(out_dir, "portability-report.md"),
+ render_matrix_markdown(matrix, MatrixOptions(out=out_dir, max_rows=options.max_rows)),
+ )
+ statuses = {row["name"]: row["status"] for row in matrix["targets"]}
+ return {
+ "status": "error" if "error" in statuses.values() else "ok",
+ "report": "portability-matrix/portability-report.json",
+ "targets": statuses,
+ "comparisons": [
+ {"a": row["a"], "b": row["b"], "status": row["status"], "all_zero": row["all_zero"]}
+ for row in matrix["comparisons"]
+ ],
+ }
+
+
+def build_checklist(profile: dict, firmware: str, reference: str | None, results: dict) -> dict:
+ """docs/07 evidence checklist + docs/22 §3.1 profile items."""
+ missing = profile_missing_fields(profile)
+ pairing_ok = bool(str(profile.get("firmware_family", "")).strip()
+ and str(profile.get("firmware_version", "")).strip())
+ physical = profile.get("physical_validation") or {}
+ physical_status = str(physical.get("status", "pending")).lower()
+ probe = results.get("probe") or {}
+ symbols = results.get("symbols") or {}
+ matrix = results.get("matrix") or {}
+ items = [
+ {
+ "id": "profile",
+ "item": "device profile fields recorded (PCI IDs, BARs, FB, engines, interrupts, ROM)",
+ "status": "ok" if not missing else "pending",
+ "detail": "missing: " + ", ".join(missing) if missing else "all profile sections populated",
+ },
+ {
+ "id": "pairing",
+ "item": "firmware hash + family/version pairing recorded before debugging (docs/07 P2.2)",
+ "status": "ok" if pairing_ok else "pending",
+ "detail": f"family={profile.get('firmware_family') or '-'} version={profile.get('firmware_version') or '-'}",
+ },
+ {
+ "id": "hash",
+ "item": "firmware sha256 recorded",
+ "status": "ok" if firmware and os.path.isfile(firmware) else "missing",
+ "detail": sha256_file(firmware) if firmware and os.path.isfile(firmware) else "firmware not found",
+ },
+ {
+ "id": "probe",
+ "item": "probe first-touch captured and explained as profile vs logic",
+ "status": "ok" if probe.get("status") == "ok" else "pending",
+ "detail": f"stopped_pc={probe.get('stopped_pc')} faults={probe.get('fault_count')}",
+ },
+ {
+ "id": "symbols",
+ "item": "symbol inventory built with provenance",
+ "status": "ok" if symbols.get("status") == "ok" else "pending",
+ "detail": f"count={symbols.get('count')} sources={symbols.get('sources')}",
+ },
+ {
+ "id": "matrix",
+ "item": "portability matrix run vs the reference firmware",
+ "status": "ok" if matrix.get("status") == "ok" and reference else "pending",
+ "detail": f"targets={matrix.get('targets')}",
+ },
+ {
+ "id": "docs07",
+ "item": "docs/07 evidence checklist attachable (hash, probe, symbols, sessions, compare, M1-M6)",
+ "status": "ok" if firmware else "missing",
+ "detail": "bundle-manifest.json hashes every artifact",
+ },
+ {
+ "id": "physical",
+ "item": "physical/supervised validation done (outside this simulator-only tool)",
+ "status": "ok" if physical_status in ("passed", "ok", "done") else "pending",
+ "detail": f"status={physical_status}; docs/22 §3.5: simulator-only evidence never justifies the support claim",
+ },
+ ]
+ required_pending = [row["id"] for row in items
+ if row["status"] != "ok" and row["id"] != "physical"]
+ if required_pending:
+ recommendation = "hold: " + ", ".join(required_pending) + " not complete"
+ elif physical_status in ("passed", "ok", "done"):
+ recommendation = "go: simulator evidence complete and physical validation recorded"
+ else:
+ recommendation = "simulator-evidence-complete; physical validation pending (supervised step)"
+ return {
+ "format": "greeneye-device-checklist/1",
+ "tool": "harness/port_new_device.py",
+ "device": profile.get("name") or "-",
+ "firmware": firmware,
+ "reference": reference,
+ "profile_missing_fields": missing,
+ "items": items,
+ "recommendation": recommendation,
+ "safety": "simulator-only: no /dev, no kernel module, no mdev, no guest, no physical GPU",
+ }
+
+
+def render_checklist_markdown(checklist: dict) -> str:
+ lines = [
+ f"# green-eye port checklist: {checklist['device']}",
+ "",
+ "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_new_device.py`.",
+ "",
+ f"* firmware: `{checklist['firmware']}`",
+ f"* reference: `{checklist['reference']}`",
+ f"* recommendation: **{checklist['recommendation']}**",
+ f"* safety: {checklist['safety']}",
+ "",
+ "| id | item | status | detail |",
+ "|---|---|---|---|",
+ ]
+ for row in checklist["items"]:
+ lines.append(f"| {row['id']} | {row['item']} | {row['status']} | {row['detail']} |")
+ lines += ["", "Evidence: `port-checklist.json`, `bundle-manifest.json`, `portability-matrix/`.", ""]
+ return "\n".join(lines)
+
+
+def build_bundle(out_dir: str) -> dict:
+ files: list[dict] = []
+ for root, dirs, names in os.walk(out_dir):
+ dirs.sort()
+ for name in sorted(names):
+ path = os.path.join(root, name)
+ rel = os.path.relpath(path, out_dir)
+ if rel == "bundle-manifest.json":
+ continue
+ files.append({"path": rel, "bytes": os.path.getsize(path), "sha256": sha256_file(path)})
+ files.sort(key=lambda row: row["path"])
+ return {"format": BUNDLE_FORMAT, "tool": "harness/port_new_device.py", "files": files}
+
+
+def run_device_port(options: DeviceOptions) -> dict:
+ if options.out is None:
+ raise ValueError("--out is required unless --dry-run is used")
+ out_root = os.path.abspath(options.out)
+ device_dir = os.path.join(out_root, options.device)
+ profile_path = options.profile or os.path.join(out_root, "profiles", f"{options.device}.json")
+ profile, created = load_profile(profile_path)
+ created_now = not created
+ if not created:
+ profile = scaffold_profile(options.device)
+ profile = merge_profile(profile, options)
+ labels = profile_labels(profile, options.device, profile_path)
+ firmware = profile.get("firmware") or ""
+ reference = profile.get("reference") or ""
+ check_rm610_gate(labels + [os.path.basename(firmware), os.path.basename(reference)],
+ approval_token=options.approval_token)
+ if created_now:
+ ensure_dir(os.path.dirname(profile_path))
+ write_json(profile_path, profile)
+ if not firmware:
+ raise ValueError(
+ "no device firmware: pass --firmware PATH or set \"firmware\" in the profile "
+ f"({profile_path}); see docs/07 P2"
+ )
+ loaded = load_any_target(firmware)
+ if not loaded.ok:
+ raise ValueError(f"firmware not usable: {loaded.path}: {loaded.reason or loaded.kind}")
+ if created_now:
+ profile["firmware_sha256"] = loaded.sha256
+ write_json(profile_path, profile)
+
+ ensure_dir(device_dir)
+ db, sources, notes = load_symbol_sources(
+ loaded,
+ symbols_json=options.symbols,
+ ghidra_exports=options.ghidra_exports,
+ ghidra_bias=options.ghidra_bias,
+ )
+ qemu_ok = qemu_available()
+ results: dict = {}
+ steps = [
+ _run_step("probe_first_touch", step_probe, loaded, db, options, device_dir, qemu_ok),
+ _run_step("symbol_inventory", step_symbols, loaded, db, sources, notes, options, device_dir),
+ _run_step("portability_matrix", step_matrix, firmware, reference, options,
+ os.path.join(device_dir, "portability-matrix")),
+ ]
+ results["probe"] = steps[0]
+ results["symbols"] = steps[1]
+ results["matrix"] = steps[2]
+ checklist = build_checklist(profile, firmware, reference, results)
+ write_json(os.path.join(device_dir, "port-checklist.json"), checklist)
+ write_text(os.path.join(device_dir, "port-checklist.md"), render_checklist_markdown(checklist))
+ checklist_step = {
+ "name": "port_checklist",
+ "status": "ok",
+ "report": "port-checklist.json",
+ "recommendation": checklist["recommendation"],
+ }
+ steps.append(checklist_step)
+ failed = [step["name"] for step in steps if step["status"] == "error"]
+ report = {
+ "format": FORMAT,
+ "tool": "harness/port_new_device.py",
+ "dry_run": False,
+ "rm610_gate": {"status": "passed", "labels": sorted(set(labels))},
+ "device": options.device,
+ "profile": {
+ "path": os.path.relpath(profile_path, out_root),
+ "created": created_now,
+ "missing_fields": profile_missing_fields(profile),
+ },
+ "firmware": {"path": firmware, "kind": loaded.kind, "sha256": loaded.sha256},
+ "reference": {"path": reference, "found": bool(reference) and os.path.isfile(reference)},
+ "qemu_available": qemu_ok,
+ "steps": steps,
+ "checklist": checklist,
+ "failed": failed,
+ "manifest": "bundle-manifest.json",
+ }
+ write_json(os.path.join(device_dir, "device-port.json"), report)
+ write_text(os.path.join(device_dir, "device-port.md"), render_report_markdown(report))
+ write_json(os.path.join(device_dir, "bundle-manifest.json"), build_bundle(device_dir))
+ return report
+
+
+def render_report_markdown(report: dict) -> str:
+ lines = [
+ f"# green-eye device port: {report['device']}",
+ "",
+ "> TOOL IN DEVELOPMENT - NOT READY YET! Generated by `harness/port_new_device.py`.",
+ "",
+ f"* format: `{report['format']}`",
+ f"* firmware: `{report['firmware']['path']}` ({report['firmware']['kind']}, "
+ f"sha256 `{report['firmware']['sha256'][:16]}`)",
+ f"* reference: `{report['reference']['path'] or '-'}`",
+ f"* qemu available: {report['qemu_available']}",
+ "",
+ "| step | status | report | reason |",
+ "|---|---|---|---|",
+ ]
+ for step in report["steps"]:
+ lines.append(f"| {step['name']} | {step['status']} | {step.get('report') or '-'} "
+ f"| {step.get('reason') or '-'} |")
+ lines += [
+ "",
+ f"* checklist recommendation: **{report['checklist']['recommendation']}**",
+ f"* manifest: `{report['manifest']}`",
+ "",
+ ]
+ return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------------
+# CLI
+# ---------------------------------------------------------------------------
+
+
+_HELP_EPILOG = """\
+invocation:
+ dry-run first (default; nothing is created or touched):
+ python3 harness/port_new_device.py --device rtx4090 --profile profiles/rtx4090.json
+ execute (scaffolds the profile if absent; simulator only):
+ python3 harness/port_new_device.py --run --device rtx4090 \\
+ --firmware ../../firmware/57X-GSP/570.144/gsp-570.144.fwimage.bin \\
+ --reference firmware/uproc_demo/build/uproc_demo.elf --out dumps/port-rtx4090
+
+agent (MCP, docs/20): read the dry-run plan, fill the profile, then --run;
+read device-port.json and port-checklist.json from the bundle.
+RM 610 (docs/23): rm610-labelled device/firmware labels are refused without
+--approval-token RM610-APPROVED or GREENEYE_RM610_APPROVED=1.
+Safety: no /dev, no kernel module, no mdev, no guest, no physical GPU.
+"""
+
+
+def build_parser() -> argparse.ArgumentParser:
+ parser = argparse.ArgumentParser(
+ prog="port_new_device",
+ description=__doc__.splitlines()[0],
+ epilog=_HELP_EPILOG,
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ )
+ parser.add_argument("--device", required=True, help="device name (profile key, e.g. rtx4090)")
+ parser.add_argument("--profile", help="profile JSON path (default: <out>/profiles/<device>.json)")
+ parser.add_argument("--firmware", help="device GSP firmware image or re-hosted ELF (overrides profile)")
+ parser.add_argument("--reference", help="reference firmware for the portability matrix")
+ parser.add_argument("--symbols", help="green-eye symbol DB JSON for the device firmware")
+ parser.add_argument("--ghidra-exports", help="Ghidra exports TSV for the device firmware")
+ parser.add_argument("--ghidra-bias", type=lambda value: int(value, 0), default=0)
+ parser.add_argument("--out", help="evidence bundle directory (required for --run)")
+ parser.add_argument("--run", dest="run", action="store_true",
+ help="execute (default is the dry-run plan)")
+ parser.add_argument("--dry-run", dest="run", action="store_false",
+ help="print the plan and exit 0 (default)")
+ parser.set_defaults(run=False)
+ parser.add_argument("--approval-token", help=f"explicit RM610 token ({RM610_TOKEN})")
+ parser.add_argument("--probe-instructions", type=int, default=2000)
+ parser.add_argument("--probe-duration", type=float, default=0.5)
+ parser.add_argument("--scenario-duration", type=float, default=0.4)
+ parser.add_argument("--max-stops", type=int, default=5000)
+ parser.add_argument("--max-rows", type=int, default=40)
+ parser.add_argument("--step-timeout", type=float, default=600.0)
+ parser.add_argument("--log-bytes", type=int, default=MAX_LOG_BYTES)
+ return parser
+
+
+def options_from_args(args: argparse.Namespace) -> DeviceOptions:
+ if not DEVICE_RE.match(args.device):
+ raise ValueError(f"invalid device name: {args.device!r}")
+ return DeviceOptions(
+ device=args.device,
+ profile=args.profile,
+ firmware=args.firmware,
+ reference=args.reference,
+ symbols=args.symbols,
+ ghidra_exports=args.ghidra_exports,
+ ghidra_bias=args.ghidra_bias,
+ out=args.out,
+ run=args.run,
+ approval_token=args.approval_token,
+ probe_instructions=args.probe_instructions,
+ probe_duration=args.probe_duration,
+ scenario_duration=args.scenario_duration,
+ max_stops=args.max_stops,
+ max_rows=args.max_rows,
+ step_timeout=args.step_timeout,
+ log_bytes=args.log_bytes,
+ )
+
+
+def main(argv: list[str] | None = None) -> int:
+ parser = build_parser()
+ args = parser.parse_args(argv)
+ try:
+ options = options_from_args(args)
+ except ValueError as exc:
+ parser.error(str(exc))
+ try:
+ check_rm610_gate(
+ [options.device, os.path.basename(options.profile or ""),
+ os.path.basename(options.firmware or ""), os.path.basename(options.reference or "")],
+ approval_token=options.approval_token,
+ )
+ if options.dry_run:
+ plan = build_plan(options)
+ print(json.dumps(plan, indent=2, sort_keys=True))
+ print(f"[device] dry-run: profile {plan['profile']['action']}, "
+ f"firmware found={plan['firmware']['found']}; nothing created or touched")
+ return 0
+ report = run_device_port(options)
+ except Rm610GateError as exc:
+ print(f"[device] RM610 gate refused: {exc}", file=sys.stderr)
+ return 3
+ except (ValueError, FileNotFoundError, json.JSONDecodeError) as exc:
+ print(f"[device] error: {exc}", file=sys.stderr)
+ return 2
+ for step in report["steps"]:
+ reason = f" reason={step['reason']}" if step.get("reason") else ""
+ print(f"[device] {step['name']}: status={step['status']}{reason}")
+ print(f"[device] checklist: {report['checklist']['recommendation']}")
+ print(f"[device] wrote {os.path.join(os.path.abspath(options.out), options.device, 'device-port.json')}")
+ print(f"[device] wrote {os.path.join(os.path.abspath(options.out), options.device, 'bundle-manifest.json')}")
+ return 1 if report["failed"] else 0
+
+
+if __name__ == "__main__": # pragma: no cover
+ raise SystemExit(main())
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()
generated by cgit v1.2.3 (git 2.39.1) at 2026-10-12 07:36:09 +0000