Something went wrong. Try again.
Hardware-in-the-loop development platform for the Commodore 64.
Something went wrong. Try again.
26 kB · 618 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619#!/usr/bin/env python3"""Automated VICE debugging-and-checkpoint acceptance scenario.
This scenario exercises one reviewed fixture end to end through the existingCLI only. It proves the frozen memory-checkpoint capture and the checkpointcomparison path on VICE 3.8, and it proves that a deliberately wrong build ofthe same source still fails with resolved crash evidence.
What the scenario commands, in order:
1. Build both checkpoint-demo variants through the CLI ``test`` command. The CLI runs each manifest's declared build command (``make build`` / ``make build-buggy``), so the artifacts are produced by a real source rebuild and not supplied by this script.2. Run the corrected manifest twice from one build (``--repeat 2``). Both iterations must pass with identical artifact bytes and one valid frozen memory-checkpoint capture each.3. Run the buggy manifest once. It must exit ``4`` with an ``outcome`` of ``failed`` and a ``failure.code`` of ``crash``, and its crash report must resolve the instrumented site to the single ``crash_here!`` call in ``src/main.rs``.4. Compare the two corrected VICE runs with ``runs compare-checkpoints``. A ``matched`` state proves run-to-run repeatability of the frozen ledger; it is not a physical-device gate.5. Re-run the repository's known-good ``smoke`` test on VICE from the root manifest and confirm the asset still produces the accepted artifact hash.
Verification reads only registered run files and their ``SHA256SUMS``. Everyledger byte is checked against the fixture's independently computed goldenimage (``fixtures/checkpoint-demo/tools/expected_ledger.py``), which recomputesthe expected values from the reviewed inputs alone. A run with no capture, atorn capture, a digest that does not match a registered file, or a byte thatdiffers from the golden image fails the scenario.
Scope and limits:
- This is emulator-only acceptance. It never claims a physical-device gate.- It performs no C64 Ultimate access, no device reset, and no configuration, firmware, deployment, or DMA operation. It opens no network target.- It must not be treated as proof for arbitrary programs. The deterministic fake target, if used elsewhere, manufactures observations and does not execute the PRG.- The host must provide the CLI binary, ``make``, the pinned Rust-MOS Docker image already present locally (the scenario never pulls), and VICE 3.8.
Usage: python3 tools/vice_debugging_and_checkpoint_acceptance.py python3 tools/vice_debugging_and_checkpoint_acceptance.py --binary target/release/ultimate-tool python3 tools/vice_debugging_and_checkpoint_acceptance.py --skip-health-smoke"""
from __future__ import annotations
import argparseimport hashlibimport importlib.utilimport jsonimport osfrom pathlib import Pathimport shutilimport subprocessimport sysimport tomllibimport uuidfrom datetime import datetime, timezone
SCHEMA_VERSION = 1
REPOSITORY_ROOT = Path(__file__).resolve().parent.parentFIXTURE_ROOT = REPOSITORY_ROOT / "fixtures" / "checkpoint-demo"DEFAULT_CHECKPOINT_MANIFEST = FIXTURE_ROOT / "c64dev.toml"DEFAULT_BUGGY_MANIFEST = FIXTURE_ROOT / "c64dev-buggy.toml"DEFAULT_HEALTH_MANIFEST = REPOSITORY_ROOT / "c64dev.toml"EXPECTED_LEDGER_MODULE = FIXTURE_ROOT / "tools" / "expected_ledger.py"
REQUIRED_VICE_VERSION = "3.8"EXPECTED_HEALTH_ARTIFACT = ( "dac6abb2ea01752e94c4318463efa6c54942713ffb021b67f1a4e1d66fa02560")# The instrumented crash_here! call in the reviewed fixture source.EXPECTED_CRASH_SITE_HEX = "$0EF1"EXPECTED_CRASH_SYMBOL = "main"EXPECTED_CRASH_LINE = 107EXPECTED_CRASH_SOURCE_SUFFIX = "fixtures/checkpoint-demo/src/main.rs"
CLI_TIMEOUT_SECONDS = 180
class Failure(RuntimeError): """A scenario step failed; the message is the acceptance conclusion."""
def sha256_hex(data: bytes) -> str: return hashlib.sha256(data).hexdigest()
def load_toml(path: Path) -> dict: with open(path, "rb") as handle: return tomllib.load(handle)
def load_expected_ledger(): """Import the fixture's independent golden-value builder module.""" spec = importlib.util.spec_from_file_location("checkpoint_demo_expected", EXPECTED_LEDGER_MODULE) if spec is None or spec.loader is None: raise Failure(f"cannot import golden-value builder {EXPECTED_LEDGER_MODULE}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module
# ---------------------------------------------------------------------------# Host requirements# ---------------------------------------------------------------------------
def resolve_vice_executable(manifest: dict, override: str | None) -> str: if override: candidate = override else: candidate = manifest.get("vice", {}).get("executable", "x64sc") resolved = shutil.which(candidate) if resolved is None: raise Failure(f"VICE executable {candidate!r} is not on PATH (host requirement)") return resolved
def probe_vice_version(executable: str) -> str: """Run an explicit ``--version`` probe; the version is never substituted.""" result = subprocess.run( [executable, "--version"], capture_output=True, text=True, timeout=30, check=False, ) banner = (result.stdout + result.stderr).strip() if result.returncode != 0: raise Failure(f"{executable} --version exited {result.returncode}: {banner}") first_line = banner.splitlines()[0] if banner else "" if "VICE" not in banner: raise Failure(f"{executable} --version did not report a VICE banner: {first_line!r}") # Require the exact required version token, no wildcard substitution. token = REQUIRED_VICE_VERSION if not any( line.split()[0] == token or f" {token})" in line or f"VICE {token}" in line for line in banner.splitlines() ): raise Failure( f"VICE {REQUIRED_VICE_VERSION} is required but {executable} reported: {first_line!r}" ) return first_line
def require_docker_image(image: str) -> None: if shutil.which("docker") is None: raise Failure("docker is required to build the fixture but is not on PATH") result = subprocess.run( ["docker", "image", "inspect", image], capture_output=True, text=True, timeout=60, check=False, ) if result.returncode != 0: raise Failure( f"pinned build image {image} is not present locally (the scenario never pulls)" )
# ---------------------------------------------------------------------------# CLI invocation# ---------------------------------------------------------------------------
def invoke_cli(binary: Path, arguments: list[str], expected_exit: int) -> dict: command = [str(binary), "--format", "json", *arguments] try: result = subprocess.run( command, capture_output=True, text=True, timeout=CLI_TIMEOUT_SECONDS, check=False, ) except subprocess.TimeoutExpired as error: raise Failure( f"CLI {' '.join(arguments)} exceeded the {CLI_TIMEOUT_SECONDS}s bound" ) from error if result.returncode != expected_exit: raise Failure( f"CLI {' '.join(arguments)} exited {result.returncode}, expected " f"{expected_exit}: {result.stderr.strip()}" ) try: envelope = json.loads(result.stdout) except json.JSONDecodeError as error: raise Failure(f"CLI {' '.join(arguments)} did not emit a machine envelope: {error}") from error if envelope.get("record_kind") != "result" or envelope.get("error") is not None: raise Failure( f"CLI {' '.join(arguments)} returned a non-result envelope: " f"{json.dumps(envelope.get('error'))}" ) return envelope["data"]
def cli_version(binary: Path) -> str: result = subprocess.run( [str(binary), "--version"], capture_output=True, text=True, timeout=30, check=False ) return (result.stdout or result.stderr).strip()
# ---------------------------------------------------------------------------# Registered-evidence verification# ---------------------------------------------------------------------------
def verify_registered_evidence(run_directory: Path) -> int: """Verify every file named in SHA256SUMS; return the verified count.""" sums_path = run_directory / "SHA256SUMS" if not sums_path.is_file(): raise Failure(f"{run_directory} has no SHA256SUMS manifest") verified = 0 for line in sums_path.read_text().splitlines(): if not line.strip(): continue digest, name = line.split(" ", 1) registered = run_directory / name if not registered.is_file(): raise Failure(f"registered evidence file {name} is missing in {run_directory}") actual = sha256_hex(registered.read_bytes()) if actual != digest: raise Failure(f"registered evidence {name} digest mismatch in {run_directory}") verified += 1 return verified
def read_registered_file(run_directory: Path, name: str | None, run_label: str) -> bytes: if not name: raise Failure(f"{run_label} capture is missing a registered read file") path = run_directory / name if not path.is_file(): raise Failure(f"{run_label} registered read {name} is missing") return path.read_bytes()
def verify_capture(run_directory: Path, run_json: dict, expected_ledger, run_label: str) -> dict: """Validate one run's frozen memory-checkpoint capture against the golden image.
Reads only registered run files. Rejects a missing capture, a torn or invalid capture, a per-read digest that does not match its file, and any byte that differs from the independently computed golden ledger. """ metadata = run_json.get("memory_checkpoint_capture") if not metadata: raise Failure(f"{run_label} recorded no memory checkpoint capture")
evidence_file = metadata.get("evidence_file") if not evidence_file: raise Failure(f"{run_label} capture metadata names no evidence file") capture_bytes = (run_directory / evidence_file).read_bytes() if sha256_hex(capture_bytes) != metadata.get("sha256"): raise Failure(f"{run_label} capture document digest does not match its metadata")
capture = json.loads(capture_bytes) if capture.get("schema_version") != 1: raise Failure(f"{run_label} capture has unsupported schema {capture.get('schema_version')}") if not capture.get("valid"): raise Failure(f"{run_label} capture is invalid: {capture.get('invalid_reason')}") if not capture.get("stable"): raise Failure(f"{run_label} capture was not stable across reads")
rows = capture.get("rows") or [] if len(rows) != 3: raise Failure(f"{run_label} capture has {len(rows)} rows, expected 3")
expected_order = list(expected_ledger.ROW_ADDRESSES.items()) image_parts: list[bytes] = [] row_report = [] for index, row in enumerate(rows): expected_name, expected_address = expected_order[index] if row.get("name") != expected_name or row.get("address") != expected_address: raise Failure( f"{run_label} row {index} is {row.get('name')}@${row.get('address'):04X}, " f"expected {expected_name}@${expected_address:04X}" ) first = read_registered_file(run_directory, row.get("first_read_file"), run_label) second = read_registered_file(run_directory, row.get("second_read_file"), run_label) if first != second: raise Failure(f"{run_label} row {expected_name} reads differed between passes") if sha256_hex(first) != row.get("first_read_sha256"): raise Failure(f"{run_label} row {expected_name} first read digest mismatch") if sha256_hex(second) != row.get("second_read_sha256"): raise Failure(f"{run_label} row {expected_name} second read digest mismatch") if len(first) != row.get("length"): raise Failure(f"{run_label} row {expected_name} length differs from declaration") image_parts.append(first) row_report.append({"name": expected_name, "address": expected_address, "hex": first.hex()})
marker = capture.get("marker") or {} marker_before = read_registered_file(run_directory, marker.get("before_file"), run_label) marker_after = read_registered_file(run_directory, marker.get("after_file"), run_label) golden_marker = bytes(expected_ledger.SAFE_MARKER) if marker_before != golden_marker or marker_after != golden_marker: raise Failure(f"{run_label} marker reads do not equal the golden safe-state marker") expected_marker_hex = bytes(marker.get("expected_hex", "".encode()), "ascii") if marker.get("before_hex") != expected_marker_hex.decode() or marker.get( "after_hex" ) != expected_marker_hex.decode(): raise Failure(f"{run_label} marker hex claims do not match the expected marker")
image = b"".join(image_parts) + marker_after golden = expected_ledger.corrected_ledger() if image != golden: raise Failure(f"{run_label} captured ledger differs from the golden image") if sha256_hex(image) != sha256_hex(golden): raise Failure(f"{run_label} captured ledger digest differs from the golden image")
return { "evidence_file": evidence_file, "capture_sha256": metadata.get("sha256"), "semantics": capture.get("semantics"), "marker_address": marker.get("address"), "marker_hex": marker_after.hex(), "rows": row_report, "ledger_image_hex": image.hex(), "ledger_sha256": sha256_hex(image), "golden_sha256": sha256_hex(golden), "verified_files": None, }
# ---------------------------------------------------------------------------# Scenario steps# ---------------------------------------------------------------------------
def run_checkpoint_pair(binary: Path, manifest: Path) -> dict: """Build once through the CLI and run the corrected manifest twice.""" data = invoke_cli( binary, ["test", "checkpoint-pass", "--target", "vice", "--repeat", "2", "--manifest", str(manifest)], 0, ) runs = data.get("runs") or [] if len(runs) != 2: raise Failure(f"corrected repeat produced {len(runs)} runs, expected 2") if data.get("outcome") != "passed": raise Failure(f"corrected outcome was {data.get('outcome')!r}, expected 'passed'")
hashes = {run.get("artifact_sha256") for run in runs} if len(hashes) != 1 or None in hashes: raise Failure(f"corrected iterations disagreed on artifact hash: {hashes}") artifact_sha256 = hashes.pop()
captures = [] for index, run in enumerate(runs): if run.get("outcome") != "passed": raise Failure(f"corrected iteration {index + 1} outcome {run.get('outcome')!r}") directory = Path(run["directory"]) verified = verify_registered_evidence(directory) run_json = json.loads((directory / "run.json").read_text()) capture = verify_capture( directory, run_json, GOLDEN, f"corrected iteration {index + 1}" ) capture["verified_files"] = verified capture["run_id"] = run["run_id"] capture["directory"] = str(directory) captures.append(capture)
return { "manifest": str(manifest), "test": "checkpoint-pass", "repeat": 2, "artifact_sha256": artifact_sha256, "run_ids": [run["run_id"] for run in runs], "build_command": run_json.get("build", {}).get("command"), "captures": captures, }
def run_buggy_crash(binary: Path, manifest: Path) -> dict: """Run the buggy manifest and require resolved crash evidence.""" data = invoke_cli(binary, ["test", "checkpoint-wrapped", "--target", "vice", "--manifest", str(manifest)], 4) runs = data.get("runs") or [] if len(runs) != 1: raise Failure(f"buggy run produced {len(runs)} runs, expected 1") run = runs[0] if data.get("outcome") != "failed": raise Failure(f"buggy outcome was {data.get('outcome')!r}, expected 'failed'") failure = run.get("failure") or {} if failure.get("code") != "crash": raise Failure(f"buggy failure code was {failure.get('code')!r}, expected 'crash'") if run.get("memory_checkpoint_capture"): raise Failure("buggy run requested a checkpoint capture it must not have")
directory = Path(run["directory"]) verified = verify_registered_evidence(directory) run_id = run["run_id"]
report = invoke_cli( binary, ["runs", "crash-report", run_id, "--manifest", str(manifest)], 0 ) if report.get("state") != "resolved": raise Failure( f"crash report state was {report.get('state')!r} " f"({report.get('unavailable_reason')})" ) if report.get("site_address_hex") != EXPECTED_CRASH_SITE_HEX: raise Failure( f"crash site was {report.get('site_address_hex')!r}, " f"expected {EXPECTED_CRASH_SITE_HEX}" ) matches = report.get("matches") or [] if not matches: raise Failure("crash report resolved no site") resolved = matches[0] location = resolved.get("location") or {} if resolved.get("symbol") != EXPECTED_CRASH_SYMBOL: raise Failure(f"crash symbol was {resolved.get('symbol')!r}, expected {EXPECTED_CRASH_SYMBOL!r}") if not str(location.get("file", "")).endswith(EXPECTED_CRASH_SOURCE_SUFFIX): raise Failure(f"crash source file was {location.get('file')!r}") if location.get("line") != EXPECTED_CRASH_LINE: raise Failure(f"crash source line was {location.get('line')!r}, expected {EXPECTED_CRASH_LINE}")
return { "manifest": str(manifest), "test": "checkpoint-wrapped", "run_id": run_id, "exit_code": data.get("exit_code"), "outcome": "failed", "failure_code": "crash", "crash_site_hex": report.get("site_address_hex"), "crash_symbol": resolved.get("symbol"), "crash_source": {"file": location.get("file"), "line": location.get("line"), "column": location.get("column")}, "verified_files": verified, }
def compare_pair(binary: Path, manifest: Path, baseline: str, candidate: str) -> dict: data = invoke_cli( binary, ["runs", "compare-checkpoints", baseline, candidate, "--manifest", str(manifest)], 0, ) if data.get("state") != "matched": raise Failure(f"checkpoint comparison state was {data.get('state')!r}, expected 'matched'") return { "baseline_run_id": baseline, "candidate_run_id": candidate, "state": data.get("state"), "baseline_artifact_sha256": data.get("baseline_artifact_sha256"), "candidate_artifact_sha256": data.get("candidate_artifact_sha256"), "comparison_id": data.get("comparison_id"), }
def run_health_smoke(binary: Path, manifest: Path) -> dict: data = invoke_cli(binary, ["test", "smoke", "--target", "vice", "--manifest", str(manifest)], 0) runs = data.get("runs") or [] if len(runs) != 1: raise Failure(f"health smoke produced {len(runs)} runs, expected 1") run = runs[0] if data.get("outcome") != "passed" or run.get("outcome") != "passed": raise Failure(f"health smoke outcome was {run.get('outcome')!r}, expected 'passed'") if run.get("artifact_sha256") != EXPECTED_HEALTH_ARTIFACT: raise Failure( f"health smoke artifact hash was {run.get('artifact_sha256')!r}, " f"expected the accepted root {EXPECTED_HEALTH_ARTIFACT}" ) directory = Path(run["directory"]) verified = verify_registered_evidence(directory) return { "manifest": str(manifest), "test": "smoke", "run_id": run["run_id"], "artifact_sha256": run.get("artifact_sha256"), "expected_artifact_sha256": EXPECTED_HEALTH_ARTIFACT, "verified_files": verified, }
# ---------------------------------------------------------------------------# Entry point# ---------------------------------------------------------------------------
def write_acceptance(report: dict, acceptance_dir: Path) -> Path: acceptance_dir.mkdir(parents=True, exist_ok=True) path = acceptance_dir / f"{report['acceptance_id']}.json" path.write_text(json.dumps(report, indent=2, sort_keys=False) + "\n") return path
def main(argv: list[str]) -> int: global GOLDEN
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument( "--binary", type=Path, default=Path( os.environ.get("ULTIMATE_TOOL_BINARY", str(REPOSITORY_ROOT / "target" / "debug" / "ultimate-tool")) ), ) parser.add_argument("--checkpoint-manifest", type=Path, default=DEFAULT_CHECKPOINT_MANIFEST) parser.add_argument("--buggy-manifest", type=Path, default=DEFAULT_BUGGY_MANIFEST) parser.add_argument("--health-manifest", type=Path, default=DEFAULT_HEALTH_MANIFEST) parser.add_argument("--vice-executable", default=None) parser.add_argument("--acceptance-dir", type=Path, default=REPOSITORY_ROOT / ".ultimate-tool" / "acceptance") parser.add_argument( "--skip-health-smoke", action="store_true", help="Skip the final repository smoke re-check (emulator-only scenario otherwise).", ) options = parser.parse_args(argv)
binary = options.binary.resolve() if not binary.is_file() or not os.access(binary, os.X_OK): raise Failure(f"CLI binary {binary} is missing or not executable (host requirement)")
checkpoint_manifest = options.checkpoint_manifest.resolve() buggy_manifest = options.buggy_manifest.resolve() for manifest in (checkpoint_manifest, buggy_manifest): if not manifest.is_file(): raise Failure(f"manifest {manifest} is missing")
corrected_manifest = load_toml(checkpoint_manifest) buggy_toml = load_toml(buggy_manifest)
vice_executable = resolve_vice_executable(corrected_manifest, options.vice_executable) vice_version = probe_vice_version(vice_executable) require_docker_image(corrected_manifest["build"]["container_image"]) require_docker_image(buggy_toml["build"]["container_image"]) if shutil.which("make") is None: raise Failure("make is required to build the fixture but is not on PATH")
GOLDEN = load_expected_ledger()
checks: list[dict] = []
def record(name: str, detail: str) -> None: checks.append({"name": name, "passed": True, "detail": detail})
checkpoint = run_checkpoint_pair(binary, checkpoint_manifest) record( "corrected_repeat_matched", f"{len(checkpoint['run_ids'])} VICE runs share artifact {checkpoint['artifact_sha256']}", ) record("frozen_ledger_golden", "captured ledger bytes equal the independently computed golden image")
buggy = run_buggy_crash(binary, buggy_manifest) record("buggy_crash_resolved", f"{buggy['crash_site_hex']} -> {buggy['crash_symbol']}:{buggy['crash_source']['line']}")
comparison = compare_pair( binary, checkpoint_manifest, checkpoint["run_ids"][0], checkpoint["run_ids"][1] ) record("checkpoint_comparison_matched", "run-to-run repeatability of the frozen ledger")
health = None if not options.skip_health_smoke: health_manifest = options.health_manifest.resolve() if not health_manifest.is_file(): raise Failure(f"health manifest {health_manifest} is missing") health = run_health_smoke(binary, health_manifest) record("health_smoke_regression", f"root smoke artifact {health['artifact_sha256']}")
report = { "schema_version": SCHEMA_VERSION, "acceptance_id": str(uuid.uuid4()), "generated_at": datetime.now(timezone.utc).isoformat(), "scenario": "automated_vice_debugging_and_checkpoint_acceptance", "tool": {"binary": str(binary), "version": cli_version(binary)}, "host": { "vice_executable": vice_executable, "vice_version": vice_version, "vice_version_required": REQUIRED_VICE_VERSION, "docker_image": corrected_manifest["build"]["container_image"], "make": shutil.which("make"), }, "scope": { "physical_gate": False, "ultimate_access": False, "device_reset": False, "config_or_firmware_write": False, "network_target_access": False, "note": "Emulator-only VICE acceptance; no physical C64 Ultimate gate is claimed.", }, "checkpoint_run": checkpoint, "buggy_run": buggy, "checkpoint_comparison": comparison, "health_smoke": health, "checks": checks, }
path = write_acceptance(report, options.acceptance_dir.resolve()) print(f"acceptance {report['acceptance_id']} written to {path}") for check in checks: print(f" PASS {check['name']}: {check['detail']}") print("This is a VICE-only acceptance; no physical-device gate is claimed.") return 0
GOLDEN = None
if __name__ == "__main__": try: raise SystemExit(main(sys.argv[1:])) except Failure as error: print(f"acceptance failed: {error}", file=sys.stderr) raise SystemExit(1) except (OSError, ValueError, KeyError, subprocess.TimeoutExpired) as error: print(f"acceptance error: {error}", file=sys.stderr) raise SystemExit(2)