summaryrefslogtreecommitdiffstats
path: root/debug/risc-v-gsp-rm-vm/harness/value_demos/s10_full_session_evidence.py
blob: 75d7b1e6d4c7a53de61755bce71ea42240d80fbc (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
#!/usr/bin/env python3
"""s10 - full session evidence: one run produces the whole evidence pack.

Real scenario (docs/14-ADVERSARIAL-INTEROP-GUIDE.md 6; Handoff-vGPU-IO.md
PART 91 ADDENDA 6-7 and PART 92 ADDENDA 6-11, 2026-10-01..10-02): a finding
is only a finding with firmware hashes, baseline/candidate sessions, an
analyze report, a compare report, the M1-M6 table, ``greeneye_hits.tsv`` and
an unexplained-delta list.  The physical milestone cycles each gathered this
by hand across separate boots and log windows (cycle evidence dirs
``backups/evidence-20261001-cycle{11,12,13}``, P6 evidence dirs).

Analog: one recorded session with hooks/watchpoints/notes/snapshots, a second
identical run for determinism, an analyze report, a session compare, an
exported hits TSV and a manifest with hashes - plus measured tool overhead
(startup, session size, trace slowdown).
"""
from __future__ import annotations

import os

try:  # package import (tests) or direct script execution (run_all)
    from . import _common as C
except ImportError:
    import sys as _sys

    _sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    import _common as C


def _stimulus(ge, record, backend: str) -> dict:
    session = ge.launch(kernel=C.DEMO_ELF, backend=backend, record=record, name=os.path.basename(record))
    try:
        ctrl = session.symbols.lookup("gsp_ctrl")
        stats = {"dispatches": [], "sum_hits": 0, "result_stores": 0}
        session.hook(
            "rpc_dispatch",
            on_enter=lambda hit: stats["dispatches"].append(hit.arg(0)),
        )
        session.hook("rpc_sum", on_enter=lambda hit: stats.__setitem__("sum_hits", stats["sum_hits"] + 1))
        session.watch(ctrl + C.CTRL_RESULT, size=8, kind="w", callback=lambda hit: stats.__setitem__("result_stores", stats["result_stores"] + 1))
        session.run(duration=0.4)
        session.snapshot(ctrl, C.CTRL_SIZE, "selftest-done")
        session.memory.write_u64(ctrl + C.CTRL_ARG0, 0x1234)
        session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 1)
        session.run(duration=0.2)
        session.note("stimulus", cmd=1, arg0=0x1234, expected=0x1234 ^ C.PING_XOR)
        session.snapshot(ctrl, C.CTRL_SIZE, "after-ping")
        stats["ping_reply"] = session.memory.read_u64(ctrl + C.CTRL_RESULT)
        session.memory.write_u64(ctrl + C.CTRL_MAILBOX, 5)
        session.run(duration=0.2)
        stats["state_reply"] = session.memory.read_u64(ctrl + C.CTRL_RESULT)
        stats["rpc_count"] = session.memory.read_u64(ctrl + C.CTRL_RPC_COUNT)
        return stats
    finally:
        session.close()


def _dir_size(path: str) -> int:
    total = 0
    for root, _dirs, files in os.walk(path):
        for name in files:
            total += os.path.getsize(os.path.join(root, name))
    return total


def _trace_overhead(ge, out: str) -> dict:
    import time

    def timed(trace):
        record = C.session_dir(out, "trace-on" if trace else "trace-off")
        start = time.monotonic()
        session = ge.launch(kernel=C.DEMO_ELF, backend="native", record=record, trace=trace, name="vd-s10-trace")
        run_start = time.monotonic()
        try:
            session.run(duration=0.25)
        finally:
            run_s = time.monotonic() - run_start
            session.close()
        return run_s, time.monotonic() - start

    off_run, off_total = timed(None)
    on_run, on_total = timed(True)
    return {
        "untraced_run_s": round(off_run, 4),
        "traced_run_s": round(on_run, 4),
        "run_slowdown_x": round(on_run / off_run, 2) if off_run else None,
        "untraced_total_s": round(off_total, 4),
        "traced_total_s": round(on_total, 4),
        "total_slowdown_x": round(on_total / off_total, 2) if off_total else None,
        "note": "total includes writing blobs/trace.jsonl (~200k exec lines) at close",
    }


def run() -> dict:
    import time

    import greeneye as ge
    from greeneye.analyze import compare_sessions, render_summary
    from greeneye.ghidra import export_hits

    backend = C.choose_backend("qemu")
    out = C.artifact_dir("s10_full_session_evidence")
    commands = [
        'ge.launch(backend="%s", record="session-A") + hooks/watch/notes/snapshots' % backend,
        "repeat the identical stimulus into session-B",
        "render_summary(A); compare_sessions(A, B, --symbols); export_hits(A)",
        "hash firmware + artifacts into manifest.json with the M1-M6 skeleton",
    ]
    scenario = C.scenario(
        ids=[
            "docs/14-ADVERSARIAL-INTEROP-GUIDE.md 6: evidence-pack checklist and report template",
            "Handoff-vGPU-IO.md PART 91 ADDENDUM 6-7 (2026-10-01): cycle 11-13 evidence dirs",
            "PART 92 ADDENDUM 6-11 (2026-10-02): P6 evidence dirs and per-cycle release transcripts",
            "docs/06-ACCURACY.md 5: M1-M6 measurement contract",
        ],
        baseline=(
            "every physical milestone gathered hash + session + logs + compare "
            "by hand across separate boots and log windows; cycles 11-13 and "
            "the P6 runs each have their own evidence directory"
        ),
        projection=(
            "on real firmware the same one-session pack is produced once H2.2 "
            "executes the image; the manifest/M1-M6 format is backend-agnostic"
        ),
        blocked=True,
        device="H2.2 device model for real .fwimage session evidence",
    )
    C.render_banner(
        "s10_full_session_evidence",
        "Can one run produce the complete evidence pack (sessions, analyze, compare, hits, hashes, M1-M6)?",
        "milestone evidence packs gathered manually across boots",
        scenario["static_baseline"],
        commands,
    )

    start = time.monotonic()
    session_a_dir = C.session_dir(out, "session-A")
    session_b_dir = C.session_dir(out, "session-B")
    session_a = _stimulus(ge, session_a_dir, backend)
    session_b = _stimulus(ge, session_b_dir, backend)

    symbols = ge.SymbolDB()
    symbols.load_elf(ge.load_elf(C.DEMO_ELF))
    report = render_summary(session_a_dir, symbols=symbols, max_rows=30)
    C.write_text(os.path.join(out, "analyze.md"), report)
    compare = compare_sessions(session_a_dir, session_b_dir, symbols=symbols, max_rows=30)
    C.write_text(os.path.join(out, "compare.md"), compare)
    hits = export_hits(session_a_dir, os.path.join(out, "greeneye_hits.tsv"), symbols=symbols)
    trace = _trace_overhead(ge, out)
    size_a = _dir_size(session_a_dir)
    size_b = _dir_size(session_b_dir)
    deltas_match = session_a["dispatches"] == session_b["dispatches"] and session_a["ping_reply"] == session_b["ping_reply"]
    manifest = {
        "format": "greeneye-value-evidence/1",
        "firmware": {
            "path": os.path.relpath(C.DEMO_ELF, C.TOOL_ROOT),
            "sha256": C.sha256_file(C.DEMO_ELF),
        },
        "sessions": {
            "a": {"path": os.path.relpath(session_a_dir, C.TOOL_ROOT), "bytes": size_a, "sha256_manifest": C.sha256_file(os.path.join(session_a_dir, "events.jsonl"))},
            "b": {"path": os.path.relpath(session_b_dir, C.TOOL_ROOT), "bytes": size_b, "sha256_manifest": C.sha256_file(os.path.join(session_b_dir, "events.jsonl"))},
        },
        "stimulus": {"commands": [1, 5], "arg0": 0x1234, "note": "PING then STATE"},
        "observed": {"a": session_a, "b": session_b, "delta_match": deltas_match},
        "metrics": [
            {"id": "M1", "name": "inventory coverage", "status": "observed", "value": "dispatch inventory [1,2,3,4,5,5,5] self-test + [1,5] stimulus"},
            {"id": "M2", "name": "reply agreement", "status": "not measured", "reason": "no physical reference capture (needs H2/production trace)"},
            {"id": "M3", "name": "ordering agreement", "status": "observed", "value": "identical dispatch sequence across A/B"},
            {"id": "M4", "name": "state agreement", "status": "observed", "value": "rpc_state stage cycling on the control block"},
            {"id": "M5", "name": "error-path agreement", "status": "not measured", "reason": "error path demoed in s08; no reference to agree against"},
            {"id": "M6", "name": "memory side-effects", "status": "observed", "value": "2 named region snapshots (selftest-done, after-ping) per session"},
        ],
        "artifacts": {
            "analyze": "analyze.md",
            "compare": "compare.md",
            "hits": os.path.relpath(hits, out),
            "manifest": "manifest.json",
        },
        "overhead": {
            "session_a_bytes": size_a,
            "session_b_bytes": size_b,
            "trace": trace,
            "doc16_trace_slowdown": "~2.4x exec trace on the demo workload (docs/16 measured 0.54 MIPS traced vs 1.5 MIPS untraced)",
        },
        "unexplained_deltas": [] if deltas_match else ["session A/B observed values differ"],
    }
    payload = {
        "success": deltas_match and session_a["ping_reply"] == (0x1234 ^ C.PING_XOR) and size_a > 0,
        "backend": backend,
        "elapsed_within_demo_s": round(time.monotonic() - start, 3),
        "manifest": manifest,
        "question": "one run -> hash + sessions + analyze + compare + hits + M1-M6, plus tool overhead",
        "answer": (
            f"one {backend} stimulus pair produced {session_a['dispatches']} dispatches, "
            f"PING reply {session_a['ping_reply']:#x}, STATE reply {session_a['state_reply']:#x}, "
            f"{session_a['result_stores']} publication stores; session {size_a} B; "
            f"A/B deterministic={deltas_match}; analyze+compare+hits+manifest written"
        ),
        "success_criteria": {
            "pass": "reply matches the ABI, both sessions agree, all artifacts exist, deltas list is empty",
            "fail": "nondeterminism, missing artifact, or an unexplained delta",
        },
        "real_case": scenario,
    }
    C.write_json(os.path.join(out, "manifest.json"), manifest)
    C.write_json(os.path.join(out, "evidence.json"), payload)
    return payload


def main() -> int:
    return 0 if C.timed_demo(run, C.artifact_dir("s10_full_session_evidence"))["success"] else 1


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