#!/usr/bin/env python3 """Automated VICE debugging-and-checkpoint acceptance scenario. This scenario exercises one reviewed fixture end to end through the existing CLI only. It proves the frozen memory-checkpoint capture and the checkpoint comparison path on VICE 3.8, and it proves that a deliberately wrong build of the 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``. Every ledger byte is checked against the fixture's independently computed golden image (``fixtures/checkpoint-demo/tools/expected_ledger.py``), which recomputes the expected values from the reviewed inputs alone. A run with no capture, a torn capture, a digest that does not match a registered file, or a byte that differs 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 argparse import hashlib import importlib.util import json import os from pathlib import Path import shutil import subprocess import sys import tomllib import uuid from datetime import datetime, timezone SCHEMA_VERSION = 1 REPOSITORY_ROOT = Path(__file__).resolve().parent.parent FIXTURE_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 = 107 EXPECTED_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)