diff --git a/scripts/campaign.py b/scripts/campaign.py index 7ab5efcb9..79b0f21dc 100644 --- a/scripts/campaign.py +++ b/scripts/campaign.py @@ -1,41 +1,51 @@ #!/usr/bin/env python3 -"""Run the real-condition campaign unattended: cores, content, checks, rules. +"""Run the real-condition campaign end to end: cores, content, checks, rules. -For each emulator profile in scope, the campaign takes the core (the libretro -x86_64 Linux nightly, else a build of the profile's pinned commit), the test -content of each of its systems (tests/campaign/content.yml, fetched through -romfetch when it is not on disk), and runs core_probe.py: +A maintainer starts each pass. For each emulator profile in scope, the +campaign takes the core (the libretro x86_64 Linux nightly, else a build of +the profile's pinned commit), the test content of each of its systems +(tests/campaign/content.yml, fetched through romfetch when it is not on +disk), and runs core_probe.py. The boot oracle of core_probe judges every +run: boots, bios_error, crash, hang or unknown, against the reference frames +and error patterns of tests/campaign/oracle.yml. - files every declared file the collection holds, staged, and which ones - the core opens - boot the content with no file against the content with the declared - file, compared on the frames the core shows + files every declared file the collection holds, staged: the full set, + and which files the core opens + none no BIOS at all + negative per declared file of the content's region: the file alone, then + zeros of its size, its first half and other regions' dumps + staged under its name path a declared file staged where the core looked for it alias a declared file staged under a name the core looked for variants each .variants/ copy of a file the core opened - bench the content timed, warm-up excluded, 5 repetitions + bench with --bench only: the content timed, warm-up excluded Every run is a record in measurements/.jsonl. The fixed rule table -(campaign_rules.py) turns records into verdicts. A verdict the table settles -is written into the profile with the record ids as `evidence:`. Any other -goes to measurements/findings.jsonl, where the site lists it as open. Each -pass adds one line to measurements/campaign-runs.jsonl. +(campaign_rules.py) reads the verdicts, never a comparison of frames. A +verdict the table settles is written into the profile with the record ids +as `evidence:`. Any other goes to measurements/findings.jsonl, where the +site lists it as open. Each pass adds one line to +measurements/campaign-runs.jsonl. + +`references` runs the full set once per core and content with its frames +saved as PNG, and lists the screens that stay on for several samples: the +candidates a reviewer looks at before writing them into oracle.yml. State is .cache/campaign/state.json of the main checkout: per check done, -failed or pending with the reason, and the core sha256 and profile digest it -ran against. A pass skips what is done unless the core build or the profile -changed. The campaign holds the host-wide automation lock, waits for an idle -host, stops cleanly on a busy host or a full disk, and resumes on the next -pass. Under autorun (`campaign` job) the lock is the job's. +failed or pending with the reason, and the core sha256, profile digest, +oracle sha256 and record version it ran against. A pass skips what is done +unless one of them changed. The campaign holds the host-wide automation +lock, waits for an idle host, stops cleanly on a busy host or a full disk, +and the next pass a maintainer starts resumes there. Usage: - python scripts/campaign.py run [--only CORE,...] [--system ID] [--budget 6h] + python scripts/campaign.py run [--only CORE,...] [--system ID] [--budget 6h] [--bench] python scripts/campaign.py run --dry-run python scripts/campaign.py run --commit [--commit-every N] + python scripts/campaign.py references --only CORE,... [--system ID] python scripts/campaign.py status [--json] python scripts/campaign.py findings [--all] [--json] python scripts/campaign.py gate - python scripts/campaign.py install-timer [--enable] """ from __future__ import annotations @@ -86,6 +96,7 @@ PENDING = "pending" CRASH_ATTEMPTS = 3 NETWORK_DELAYS = (10, 60, 300) MAX_ALIAS_CANDIDATES = 6 +MAX_REGION_DUMPS = 2 BENCH_WARMUP = 600 BENCH_FRAMES = 1800 BENCH_REPEAT = 5 @@ -175,6 +186,11 @@ def region_matches(entry: dict, region: str | None) -> bool: return region in regions or "world" in regions +def regions_of(entry: dict) -> list[str]: + regions = entry.get("region") or [] + return [regions] if isinstance(regions, str) else list(regions) + + def is_bios(entry: dict) -> bool: return entry.get("category", "bios") == "bios" and entry.get("type") != "directory" @@ -235,6 +251,8 @@ class Settings: bench_frames: int = BENCH_FRAMES romfetch: str = "" network: bool = True + # Benchmarks only choose between variants: never part of a default pass. + bench: bool = False delays: tuple[float, ...] = NETWORK_DELAYS echo: bool = True # Third-party code (a nightly core, a project's Makefile) runs only under @@ -252,6 +270,7 @@ class Paths: content: Path db: Path content_file: Path = CONTENT_FILE + oracle: Path = core_probe.DEFAULT_ORACLE @property def state(self) -> Path: @@ -470,14 +489,17 @@ def save_state(path: Path, state: dict) -> None: autorun.save_state(path, state) -def should_run(previous: dict | None, core_sha: str, digest: str) -> bool: +def should_run(previous: dict | None, core_sha: str, digest: str, oracle_sha: str = "") -> bool: """A check runs again when it never ran, is pending, or ran against - another core build or another version of the profile.""" + another core build, another version of the profile, another oracle, or + before records carried a verdict.""" if not previous: return True if previous.get("status") == PENDING: return True - return previous.get("core_sha256") != core_sha or previous.get("profile_digest") != digest + return (previous.get("core_sha256") != core_sha or previous.get("profile_digest") != digest + or previous.get("record_version") != mr.RECORD_VERSION + or previous.get("oracle_sha256", "") != oracle_sha) # -------------------------------------------------------------------------- @@ -875,6 +897,7 @@ class Campaign: self.cores = Cores(self) self.store = ContentStore(self) self.content = load_content(paths.content_file) + self.oracle_sha = file_sha256(paths.oracle) if paths.oracle.is_file() else "" self.deadline = 0.0 self.current: Pass | None = None self.dirty_at_start: set[str] = set() @@ -1004,6 +1027,8 @@ class Campaign: "--emulator", emulator, "--subject", subject, "--max-load", str(self.host.max_load), "--timeout", str(self.settings.probe_timeout), "--work-dir", str(self.paths.root / "tmp" / "core_probe")] + if self.paths.oracle.is_file(): + argv += ["--oracle", str(self.paths.oracle)] if not self.settings.allow_unsandboxed: argv.append("--require-sandbox") if core.artifact: @@ -1026,7 +1051,8 @@ class Campaign: def remember(self, emulator: str, key: str, result: ProbeResult, core: Core | None, digest: str) -> None: entry = {"status": result.status, "at": now_iso(), "profile_digest": digest, - "core_sha256": core.sha256 if core else None} + "core_sha256": core.sha256 if core else None, + "record_version": mr.RECORD_VERSION, "oracle_sha256": self.oracle_sha} if result.reason: entry["reason"] = result.reason if result.log: @@ -1042,6 +1068,20 @@ class Campaign: self.state["checks"].setdefault(emulator, {})[key] = { "status": PENDING, "reason": reason, "at": now_iso(), "profile_digest": digest} + def needed(self, emulator: str, key: str, core: Core, digest: str) -> bool: + return should_run(self.state["checks"].get(emulator, {}).get(key), core.sha256, digest, + self.oracle_sha) + + def check(self, emulator: str, key: str, core: Core, digest: str, argv: list[str], + log_name: str) -> dict | None: + """The record of a check: run now when needed, else the one kept.""" + if not self.needed(emulator, key, core, digest): + previous = self.state["checks"].get(emulator, {}).get(key) or {} + return self.read_record(emulator, previous.get("record")) + result = self.probe(argv, log_name) + self.remember(emulator, key, result, core, digest) + return result.record + def system_entries(self, profile: dict, system: str) -> list[tuple[int, dict]]: from validation import filter_files_by_mode @@ -1060,84 +1100,148 @@ class Campaign: held.sort(key=lambda item: (item[1].get("priority", 1 << 30), item[0])) return held[0] if held else None + @staticmethod + def other_regions(entry: dict, staged: dict[str, tuple[int, dict, Path]], local: Path) -> list[Path]: + """Held dumps of the same system for regions this entry does not + serve, one per distinct file: the wrong-region variants.""" + mine = set(regions_of(entry)) + if not mine or "world" in mine: + return [] + out: dict[str, Path] = {} + for _, other, path in staged.values(): + theirs = set(regions_of(other)) + if theirs and "world" not in theirs and not theirs & mine and path.resolve() != local.resolve(): + out.setdefault(file_sha1(path), path) + return sorted(out.values(), key=lambda p: p.name)[:MAX_REGION_DUMPS] + def run_unit_content(self, unit: Unit, core: Core, content: Content, digest: str) -> None: emulator, profile = unit.emulator, unit.profile - checks = self.state["checks"].setdefault(emulator, {}) entries = self.system_entries(profile, content.system) - boot = self.boot_entry(entries, content) - keys = {name: self.check_key(core, content, name) for name in ("files", "bench")} - if boot: - keys["boot"] = self.check_key(core, content, f"boot:{boot[1]['name']}") - todo = [k for k in keys.values() if should_run(checks.get(k), core.sha256, digest)] - if not todo: - return - try: - folder = self.store.get(content) - load = load_file(folder, content.load) - except Unavailable as exc: - for key in todo: - self.pending(emulator, key, str(exc), digest) - return - frames = ["--frames", str(content.frames), "--frame-hash-every", str(content.frame_hash_every)] - stem = f"{core.name}-{slug(content.name)}" - - # files: every declared file the collection holds, at its place. staged: dict[str, tuple[int, dict, Path]] = {} for index, entry in entries: local = self.local_file(entry) if local and entry_dest(entry) not in staged: staged[entry_dest(entry)] = (index, entry, local) - files_record = None - if keys["files"] in todo: - argv = self.probe_args("run", core, emulator, slug(f"files-{stem}"), content, load) + frames - for dest, (_, _, local) in sorted(staged.items()): - argv += ["--system-file", f"{dest}={local}"] - result = self.probe(argv, f"{emulator}-{stem}-files") - self.remember(emulator, keys["files"], result, core, digest) - files_record = result.record - elif checks.get(keys["files"], {}).get("status") == DONE: - files_record = self.read_record(emulator, checks[keys["files"]].get("record")) + tested = {dest: item for dest, item in staged.items() if region_matches(item[1], content.region)} + keys = {"files": self.check_key(core, content, "files"), + "none": self.check_key(core, content, "none")} + for dest in tested: + keys[f"negative:{dest}"] = self.check_key(core, content, f"negative:{dest}") + if self.settings.bench: + keys["bench"] = self.check_key(core, content, "bench") + if not any(self.needed(emulator, key, core, digest) for key in keys.values()): + return + try: + folder = self.store.get(content) + load = load_file(folder, content.load) + except Unavailable as exc: + for key in keys.values(): + if self.needed(emulator, key, core, digest): + self.pending(emulator, key, str(exc), digest) + return + frames = ["--frames", str(content.frames), "--frame-hash-every", str(content.frame_hash_every)] + repeat = ["--repeat", str(self.settings.check_repeat)] + stem = f"{core.name}-{slug(content.name)}" - boot_record = None - if boot and keys["boot"] in todo: - index, entry = boot - local = self.local_file(entry) - dest = entry_dest(entry) - argv = self.probe_args("check", core, emulator, slug(f"boot-{entry['name']}-{stem}"), - content, load) + frames - argv += ["--vary", f"{dest}=", "--vary", f"{dest}={local}", "--observable", "frame_hashes", - "--repeat", str(self.settings.check_repeat), - "--claim", f"{core.name} reaches the game on {content.name} without {entry['name']}", - "--method", (f"Two variants, {entry['name']} absent and {entry['name']} as the " - f"collection holds it, each run {self.settings.check_repeat} times " - f"from a fresh process for {content.frames} frames with no button " - f"pressed. Every {content.frame_hash_every}th frame is hashed. The " - f"runs without the file are compared with the runs with it on the " - f"frames that are not one colour.")] - result = self.probe(argv, f"{emulator}-{stem}-boot") - self.remember(emulator, keys["boot"], result, core, digest) - boot_record = result.record - if boot_record: - absent, present = (v["label"] for v in boot_record["variants"]) - verdicts = rules.rule_bios_need(rules.boot_outcome(boot_record, absent, present), - entry, boot_record["id"]) + # files: every declared file the collection holds, at its place. + argv = self.probe_args("run", core, emulator, slug(f"files-{stem}"), content, load) + frames + repeat + for dest, (_, _, local) in sorted(staged.items()): + argv += ["--system-file", f"{dest}={local}"] + files_record = self.check(emulator, keys["files"], core, digest, argv, f"{emulator}-{stem}-files") + # none: no BIOS at all. + argv = self.probe_args("run", core, emulator, slug(f"none-{stem}"), content, load) + frames + repeat + none_record = self.check(emulator, keys["none"], core, digest, argv, f"{emulator}-{stem}-none") + if not files_record or not none_record: + return + full = rules.run_of(files_record, files_record["variants"][0]["label"]) + none = rules.run_of(none_record, none_record["variants"][0]["label"]) + self.log(f"{emulator}: {core.name} on {content.name}: full set {full.verdict}, " + f"no BIOS {none.verdict}") + + for dest, (index, entry, local) in sorted(tested.items()): + record = self.negative(unit, core, content, load, frames, digest, keys[f"negative:{dest}"], + dest, entry, local, staged) + if not record: + continue + labels = [v["label"] for v in record["variants"]] + present = rules.run_of(record, labels[0]) + zeros, half = (rules.run_of(record, label) for label in labels[1:3]) + for label in labels[3:]: + self.log(f"{emulator}: {rules.rule_region(rules.run_of(record, label), present, entry, content.region)}") + if dest in self.opened(files_record, 0) or dest in self.opened(record, 0): + verdicts = rules.rule_need(none, present, entry) + validation = rules.rule_validation(zeros, half, none, present, entry) + if validation: + verdicts.append(validation) self.settle(emulator, content, core, index, entry, verdicts) - self.follow_aliases(unit, core, content, load, frames, digest, index, entry, - local, boot_record, absent) + variants = self.variant_files(entry, local) + if variants: + self.run_variants(unit, core, content, load, frames, digest, index, entry, dest, + local, variants) + else: + self.moved(unit, core, content, load, frames, digest, index, entry, dest, local, + files_record, record) - if files_record and keys["files"] in todo: - self.read_files(unit, core, content, load, frames, digest, staged, files_record) + boot = self.boot_entry(entries, content) + if boot and entry_dest(boot[1]) in tested: + index, entry = boot + self.follow_aliases(unit, core, content, load, frames, digest, index, entry, + tested[entry_dest(entry)][2], none_record) - if keys["bench"] in todo: + if self.settings.bench and full.verdict == rules.BOOTS: argv = self.probe_args("bench", core, emulator, slug(f"bench-{stem}"), content, load) argv += ["--warmup", str(self.settings.bench_warmup), "--frames", str(self.settings.bench_frames), "--repeat", str(self.settings.bench_repeat), "--no-trace"] if boot: - local = self.local_file(boot[1]) - argv += ["--system-file", f"{entry_dest(boot[1])}={local}"] - result = self.probe(argv, f"{emulator}-{stem}-bench") - self.remember(emulator, keys["bench"], result, core, digest) + argv += ["--system-file", f"{entry_dest(boot[1])}={self.local_file(boot[1])}"] + self.check(emulator, keys["bench"], core, digest, argv, f"{emulator}-{stem}-bench") + + def negative(self, unit: Unit, core: Core, content: Content, load: Path, frames: list[str], + digest: str, key: str, dest: str, entry: dict, local: Path, + staged: dict[str, tuple[int, dict, Path]]) -> dict | None: + """The file alone, then wrong files under its name: zeros of its + size, its first half, and other regions' dumps.""" + stem = f"{core.name}-{slug(content.name)}" + size = local.stat().st_size + others = self.other_regions(entry, staged, local) + argv = self.probe_args("check", core, unit.emulator, slug(f"negative-{entry['name']}-{stem}"), + content, load) + frames + argv += ["--vary", f"{dest}={local}", "--vary", f"{dest}=@zeros:{size}", + "--vary", f"{dest}=@head:{size // 2}:{local}"] + for other in others: + argv += ["--vary", f"{dest}={other}"] + regions = f", then {', '.join(p.name for p in others)} staged under its name" if others else "" + argv += ["--observable", "frame_hashes", "--repeat", str(self.settings.check_repeat), + "--claim", (f"{core.name} boots {content.name} with {entry['name']} alone and " + f"refuses a wrong file under its name"), + "--method", (f"{entry['name']} alone as {dest}, then {size} zero bytes and its " + f"first {size // 2} bytes{regions}, nothing else staged. Each run " + f"{self.settings.check_repeat} times from a fresh process for " + f"{content.frames} frames with no button pressed, every " + f"{content.frame_hash_every}th frame hashed. The boot oracle judges " + f"each variant against the reference frames of {content.name}.")] + return self.check(unit.emulator, key, core, digest, argv, f"{unit.emulator}-{stem}-negative") + + def moved(self, unit: Unit, core: Core, content: Content, load: Path, frames: list[str], + digest: str, index: int, entry: dict, dest: str, local: Path, files_record: dict, + alone: dict) -> None: + """A file the core opened neither in the full set nor alone: where + else it looked for a file of that name, else flagged.""" + probed = [p for record in (files_record, alone) for r in record["variants"][0]["repetitions"] + for p in r.get("files_probed_absent") or ()] + others = [p for p in dict.fromkeys(probed) if p != dest + and p.rsplit("/", 1)[-1].casefold() == dest.rsplit("/", 1)[-1].casefold()] + for candidate in others[:MAX_ALIAS_CANDIDATES]: + found = self.follow_up(unit, core, content, load, frames, digest, entry, local, candidate, + f"path:{entry['name']}:{candidate}") + if found and candidate in self.opened(found, 0): + verdict = rules.rule_path(entry, dest, rules.Opened(candidate, found["id"])) + if verdict: + self.settle(unit.emulator, content, core, index, entry, [verdict]) + return + verdict = rules.rule_never_opened(entry, [files_record["id"], alone["id"]], content.name) + self.settle(unit.emulator, content, core, index, entry, [verdict]) def read_record(self, emulator: str, rid: str | None) -> dict | None: if not rid: @@ -1150,49 +1254,10 @@ class Campaign: return json.loads(line) return None - def read_files(self, unit: Unit, core: Core, content: Content, load: Path, frames: list[str], - digest: str, staged: dict[str, tuple[int, dict, Path]], record: dict) -> None: - """Which staged files the core opened: path follow-ups, variants, flags.""" - emulator = unit.emulator - rep = record["variants"][0]["repetitions"][0] - if rep.get("files_opened") is None: - return - opened = {item["path"] for item in rep.get("files_opened") or ()} - probed = list(rep.get("files_probed_absent") or ()) - for dest, (index, entry, local) in sorted(staged.items()): - if dest in opened: - variants = self.variant_files(entry, local) - if variants: - self.run_variants(unit, core, content, load, frames, digest, index, entry, dest, - local, variants) - continue - if not region_matches(entry, content.region): - continue - # A fallback is read only when the files before it are absent: - # staged alone, it shows whether the core reads it at all. - alone = self.follow_up(unit, core, content, load, frames, digest, entry, local, dest, - f"alone:{entry['name']}") - if alone is None: - continue - if dest in self.opened(alone, 1): - continue - probed_alone = [p for r in alone["variants"][0]["repetitions"] - for p in r.get("files_probed_absent") or ()] - others = [p for p in dict.fromkeys(probed + probed_alone) if p != dest - and p.rsplit("/", 1)[-1].casefold() == dest.rsplit("/", 1)[-1].casefold()] - moved = False - for candidate in others[:MAX_ALIAS_CANDIDATES]: - record_c = self.follow_up(unit, core, content, load, frames, digest, entry, local, - candidate, f"path:{entry['name']}:{candidate}") - if record_c and candidate in self.opened(record_c, 1): - verdict = rules.rule_path(entry, dest, rules.Opened(candidate, record_c["id"])) - if verdict: - self.settle(emulator, content, core, index, entry, [verdict]) - moved = True - break - if not moved: - verdict = rules.rule_never_opened(entry, [record["id"], alone["id"]], content.name) - self.settle(emulator, content, core, index, entry, [verdict]) + def judged(self, emulator: str, rid: str) -> bool: + """True for a record the boot oracle judged: only those prove a field.""" + record = self.read_record(emulator, rid) + return bool(record) and record.get("record_version", 0) >= 4 @staticmethod def opened(record: dict, position: int) -> set[str]: @@ -1203,34 +1268,23 @@ class Campaign: def follow_up(self, unit: Unit, core: Core, content: Content, load: Path, frames: list[str], digest: str, entry: dict, local: Path, dest: str, check: str) -> dict | None: - """The file staged at dest alone, against nothing there.""" - key = self.check_key(core, content, check) - previous = self.state["checks"].get(unit.emulator, {}).get(key) - if not should_run(previous, core.sha256, digest): - return self.read_record(unit.emulator, (previous or {}).get("record")) + """The file staged at dest alone: does the core open it there.""" stem = f"{core.name}-{slug(content.name)}" - argv = self.probe_args("check", core, unit.emulator, slug(f"{check}-{stem}"), content, load) - argv += frames + ["--vary", f"{dest}=", "--vary", f"{dest}={local}", - "--observable", "frame_hashes", "--repeat", "1", - "--claim", f"{core.name} opens {entry['name']} as {dest}", - "--method", (f"{entry['name']} staged as {dest}, against nothing there, " - f"{content.frames} frames of {content.name}. The files the " - f"core opened are read from strace.")] - result = self.probe(argv, f"{unit.emulator}-{stem}-{slug(check)}") - self.remember(unit.emulator, key, result, core, digest) - return result.record + argv = self.probe_args("run", core, unit.emulator, slug(f"{check}-{stem}"), content, load) + argv += frames + ["--system-file", f"{dest}={local}"] + return self.check(unit.emulator, self.check_key(core, content, check), core, digest, argv, + f"{unit.emulator}-{stem}-{slug(check)}") def follow_aliases(self, unit: Unit, core: Core, content: Content, load: Path, frames: list[str], - digest: str, index: int, entry: dict, local: Path, record: dict, - absent_label: str) -> None: - """Names the core looked for, with the file absent, that no entry declares.""" + digest: str, index: int, entry: dict, local: Path, none_record: dict) -> None: + """Names the core looked for, with no BIOS, that no entry declares.""" declared = {e["name"] for e in unit.profile.get("files") or ()} declared |= {a for e in unit.profile.get("files") or () for a in e.get("aliases") or ()} suffix = Path(entry["name"]).suffix.casefold() dest = entry_dest(entry) folder = dest.rsplit("/", 1)[0] if "/" in dest else "" probed: list[str] = [] - for rep in rules.variant(record, absent_label)["repetitions"]: + for rep in none_record["variants"][0]["repetitions"]: for path in rep.get("files_probed_absent") or (): name = path.rsplit("/", 1)[-1] where = path.rsplit("/", 1)[0] if "/" in path else "" @@ -1240,7 +1294,7 @@ class Campaign: for candidate in probed[:MAX_ALIAS_CANDIDATES]: found = self.follow_up(unit, core, content, load, frames, digest, entry, local, candidate, f"alias:{entry['name']}:{candidate.rsplit('/', 1)[-1]}") - if found and candidate in self.opened(found, 1): + if found and candidate in self.opened(found, 0): verdict = rules.rule_alias(self.reload_entry(unit, index), rules.Opened(candidate, found["id"])) if verdict: self.settle(unit.emulator, content, core, index, self.reload_entry(unit, index), [verdict]) @@ -1249,7 +1303,7 @@ class Campaign: digest: str, index: int, entry: dict, dest: str, primary: Path, variants: list[Path]) -> None: key = self.check_key(core, content, f"variants:{entry['name']}") - if not should_run(self.state["checks"].get(unit.emulator, {}).get(key), core.sha256, digest): + if not self.needed(unit.emulator, key, core, digest): return stem = f"{core.name}-{slug(content.name)}" argv = self.probe_args("check", core, unit.emulator, slug(f"variants-{entry['name']}-{stem}"), @@ -1257,16 +1311,14 @@ class Campaign: for path in [primary, *variants]: argv += ["--vary", f"{dest}={path}"] argv += ["--observable", "frame_hashes", "--repeat", str(self.settings.check_repeat), - "--claim", f"the copies of {entry['name']} change what {core.name} shows on {content.name}", - "--method", (f"Each copy the collection holds staged as {dest}, each run " - f"{self.settings.check_repeat} times for {content.frames} frames, " - f"compared on the sampled frame hashes.")] - result = self.probe(argv, f"{unit.emulator}-{stem}-variants") - self.remember(unit.emulator, key, result, core, digest) - if result.record: - status = result.record["check"]["result"]["status"] - verdict = rules.rule_variants(status, [v["label"] for v in result.record["variants"]], - entry, result.record["id"]) + "--claim", f"every copy of {entry['name']} boots {content.name} on {core.name}", + "--method", (f"Each copy the collection holds staged alone as {dest}, each run " + f"{self.settings.check_repeat} times for {content.frames} frames. The " + f"boot oracle judges each copy.")] + record = self.check(unit.emulator, key, core, digest, argv, f"{unit.emulator}-{stem}-variants") + if record: + verdict = rules.rule_variants([rules.run_of(record, v["label"]) for v in record["variants"]], + entry) if verdict: self.settle(unit.emulator, content, core, index, entry, [verdict]) @@ -1291,9 +1343,10 @@ class Campaign: agrees = verdict.field in rules.SETTLED_FIELDS and rules.agrees(fresh, verdict) if verdict.apply: updates: dict[str, object] = {} - if not agrees or verdict.field not in fresh: + if not agrees or (verdict.field not in fresh and verdict.field != "validation"): updates[verdict.field] = verdict.value - evidence = rules.merge_evidence(fresh.get("evidence"), verdict) + evidence = rules.merge_evidence(fresh.get("evidence"), verdict, + keep=lambda rid: self.judged(emulator, rid)) if evidence != fresh.get("evidence"): updates["evidence"] = evidence if updates: @@ -1364,6 +1417,60 @@ class Campaign: return record["core"].get("library_version") return None + # -- reference frames ----------------------------------------------------------- + + def references(self, only: set[str], system: str | None, png_dir: Path) -> list[dict]: + """The full set run once per core and content, frames written as PNG, + and the screens that stay on long enough to serve as references.""" + self.deadline = self.host.clock() + self.settings.budget + listed = [] + for unit in self.units(only, system): + contents = self.unit_contents(unit, system) + if unit.pending or not contents: + continue + try: + cores = self.unit_cores(unit.emulator, unit.profile, unit.buildbot_only) + except Unavailable as exc: + self.log(f"{unit.emulator}: {exc}") + continue + for core in cores: + for content in contents: + self.guard() + self.store.hold(content, "references") + try: + load = load_file(self.store.get(content), content.load) + except Unavailable as exc: + self.log(f"{unit.emulator}: {exc}") + continue + stem = f"{core.name}-{slug(content.name)}" + target = png_dir / slug(core.name) / slug(content.name) + shutil.rmtree(target, ignore_errors=True) + argv = self.probe_args("run", core, unit.emulator, slug(f"references-{stem}"), + content, load) + argv += ["--frames", str(content.frames), "--frame-hash-every", + str(content.frame_hash_every), "--repeat", str(self.settings.check_repeat), + "--png-dir", str(target)] + for index, entry in self.system_entries(unit.profile, content.system): + local = self.local_file(entry) + if local and f"{entry_dest(entry)}=" not in " ".join(argv): + argv += ["--system-file", f"{entry_dest(entry)}={local}"] + result = self.probe(argv, f"{unit.emulator}-{stem}-references") + self.store.release(content, "references") + if not result.record: + self.log(f"{unit.emulator}: references of {content.name}: {result.reason}") + continue + record = result.record + listed.append({ + "emulator": unit.emulator, "core": core.name, + "library_name": record["core"]["library_name"], "content": content.name, + "content_sha1": record["inputs"]["content"]["sha1"], "record": record["id"], + "verdict": record["variants"][0]["verdict"], + "png_dir": str(target / record["variants"][0]["label"]), + "candidates": reference_candidates(record), + }) + save_state(self.paths.state, self.state) + return listed + # -- the pass ----------------------------------------------------------------- def units(self, only: set[str], system: str | None) -> list[Unit]: @@ -1567,6 +1674,32 @@ class Campaign: return made +def reference_candidates(record: dict, min_samples: int = 3) -> list[dict]: + """Screens of a record's first variant that are not one colour and stay + on for at least min_samples consecutive samples in every repetition: + what a reviewer looks at to choose reference frames.""" + found: list[dict[str, tuple[int, int]]] = [] + for rep in record["variants"][0]["repetitions"]: + observables = rep.get("observables") or {} + uniform = set(observables.get("uniform_frames") or ()) + runs: dict[str, tuple[int, int]] = {} + previous, start, length = None, 0, 0 + for frame, digest in [*(observables.get("frame_hashes") or ()), (None, None)]: + if digest == previous: + length += 1 + continue + if previous and previous not in uniform and length >= min_samples: + first, longest = runs.get(previous, (start, 0)) + runs[previous] = (min(first, start), max(longest, length)) + previous, start, length = digest, frame, 1 + found.append(runs) + common = set(found[0]) if found else set() + for runs in found[1:]: + common &= set(runs) + return sorted(({"sha1": digest, "first_frame": found[0][digest][0], "samples": found[0][digest][1]} + for digest in common), key=lambda item: item["first_frame"]) + + def reason_class(reason: str) -> str: """A pending reason without its detail, to count them.""" head = reason.split(":", 1)[0].strip() if ":" in reason else reason @@ -1661,9 +1794,15 @@ def main(argv: Sequence[str] | None = None) -> int: parser = argparse.ArgumentParser(allow_abbrev=False, description=__doc__.split("\n\n")[0], formatter_class=argparse.RawDescriptionHelpFormatter) sub = parser.add_subparsers(dest="command", required=True) - run_p = sub.add_parser("run", help="one pass: cores, content, checks, rules") - run_p.add_argument("--only", default="", help="emulator profiles or core names, comma-separated") - run_p.add_argument("--system", help="one system id") + scope = argparse.ArgumentParser(add_help=False, allow_abbrev=False) + scope.add_argument("--only", default="", help="emulator profiles or core names, comma-separated") + scope.add_argument("--system", help="one system id") + scope.add_argument("--romfetch", default="", + help=f"romfetch command (default: ${ROMFETCH_ENV}, else romfetch on PATH)") + scope.add_argument("--offline", action="store_true", help="no buildbot, no romfetch, no build") + scope.add_argument("--allow-unsandboxed", action="store_true", + help="run cores and builds without bubblewrap when it is missing (refused by default)") + run_p = sub.add_parser("run", parents=[scope], help="one pass: cores, content, checks, rules") run_p.add_argument("--budget", type=parse_budget, default=6 * 3600.0, help="stop starting checks after this long (6h, 90m, 3600s)") run_p.add_argument("--max-load-wait", type=float, default=3600, @@ -1671,34 +1810,28 @@ def main(argv: Sequence[str] | None = None) -> int: run_p.add_argument("--min-free-gb", type=float, default=30, help="pause below this free space") run_p.add_argument("--cache-cap-gb", type=float, default=20, help="content the campaign fetched is evicted past this size, oldest use first") - run_p.add_argument("--romfetch", default="", - help=f"romfetch command (default: ${ROMFETCH_ENV}, else romfetch on PATH)") - run_p.add_argument("--offline", action="store_true", help="no buildbot, no romfetch, no build") - run_p.add_argument("--allow-unsandboxed", action="store_true", - help="run cores and builds without bubblewrap when it is missing (refused by default)") + run_p.add_argument("--bench", action="store_true", + help="also time each content that boots (benchmarks only choose between variants)") run_p.add_argument("--dry-run", action="store_true", help="print the plan, run nothing") run_p.add_argument("--commit", action="store_true", help="commit records, findings and each changed profile behind the gate") run_p.add_argument("--commit-every", type=int, default=0, metavar="N", help="with --commit, also commit after every N emulators") + refs_p = sub.add_parser("references", parents=[scope], + help="run the full set with PNG frames and list reference candidates") + refs_p.add_argument("--png-dir", default=str(ROOT / "tmp" / "campaign" / "references"), + help="where the frames are written, one folder per core and content") status_p = sub.add_parser("status", help="done, failed and pending checks, and why") status_p.add_argument("--json", action="store_true", help="the report as JSON") findings_p = sub.add_parser("findings", help="profile claims the runs contradict, open for review") findings_p.add_argument("--all", action="store_true", help="applied and resolved too") findings_p.add_argument("--json", action="store_true", help="the findings as JSON") sub.add_parser("gate", help="tests, schemas, verify and pack of the changed profiles") - timer_p = sub.add_parser("install-timer", - help="write the autorun units, the campaign job's timer among them") - timer_p.add_argument("--enable", action="store_true", - help="also reload systemd and enable the timers") args = parser.parse_args(argv) paths = Paths.default() if args.command == "gate": return gate(paths.root, changed_profiles(paths.root)) - if args.command == "install-timer": - command = [sys.executable, str(ROOT / "scripts" / "autorun.py"), "install-timers"] - return subprocess.run(command + (["--enable"] if args.enable else []), check=False).returncode if args.command == "findings": lines = open_findings(paths.root, args.all) if args.json: @@ -1713,6 +1846,9 @@ def main(argv: Sequence[str] | None = None) -> int: settings = Settings(budget=args.budget, max_load_wait=args.max_load_wait, min_free_gb=args.min_free_gb, cache_cap_gb=args.cache_cap_gb, romfetch=args.romfetch, network=not args.offline, + allow_unsandboxed=args.allow_unsandboxed, bench=args.bench) + elif args.command == "references": + settings = Settings(romfetch=args.romfetch, network=not args.offline, allow_unsandboxed=args.allow_unsandboxed) campaign = Campaign(paths, settings) if args.command == "status": @@ -1720,13 +1856,22 @@ def main(argv: Sequence[str] | None = None) -> int: print(json.dumps(report, indent=2) if args.json else render_status(report)) return 0 only = {item.strip() for item in args.only.split(",") if item.strip()} + if args.command == "references": + try: + with host_lock(campaign.lock_path): + listed = campaign.references(only, args.system, Path(args.png_dir)) + except autorun.LockBusy as exc: + print(f"campaign: deferred, {exc}", file=sys.stderr) + return 75 + print(json.dumps(listed, indent=2)) + return 0 if args.dry_run: campaign.run(only, args.system, dry_run=True) return 0 campaign.dirty_at_start = set(autorun.changed_files(paths.root)) if args.commit else set() def stop(signum: int, frame: object) -> None: - # A timer's TimeoutStartSec ends the pass like Ctrl-C: state saved, pass line written. + # SIGTERM ends the pass like Ctrl-C: state saved, pass line written. raise KeyboardInterrupt signal.signal(signal.SIGTERM, stop) try: diff --git a/scripts/campaign_profiles.py b/scripts/campaign_profiles.py index 0df90652a..4efee99fc 100644 --- a/scripts/campaign_profiles.py +++ b/scripts/campaign_profiles.py @@ -23,6 +23,7 @@ ANCHORS = { "hle_fallback": ("required", "name"), "path": ("name",), "aliases": ("path", "name"), + "validation": ("sha1", "md5", "crc32", "size", "name"), } diff --git a/scripts/campaign_rules.py b/scripts/campaign_rules.py index e9da2543e..ff1ec0d4b 100644 --- a/scripts/campaign_rules.py +++ b/scripts/campaign_rules.py @@ -1,25 +1,36 @@ -"""The campaign's fixed rule table: which observable settles which profile field. +"""The campaign's fixed rule table: which verdicts settle which profile field. -Each rule is a pure function of a core probe record (or what was read from -one) and a profile entry. It returns verdicts. A verdict with `apply` set is -settled by the table and written into the profile with the record ids as -`evidence:`. Any other verdict goes to the review queue -(measurements/findings.jsonl). +Each rule is a pure function of verdicts read from core probe records (the +boot oracle of core_probe.py: boots, bios_error, crash, hang, unknown) and +a profile entry. It returns verdicts on profile fields. One with `apply` set +is settled by the table and written into the profile with the record ids as +`evidence:`. Any other goes to the review queue (measurements/findings.jsonl). - observable field settled - boots to the game with no file hle_fallback true - no game screen without it, the game with it required true - hle_fallback false - opens the staged file under another path path that path - opens the staged file under another name aliases name added - a staged file the content should use, never read - flagged only +Three runs are compared for a file, all on the same content: `none` with no +BIOS at all, `present` with the file alone, and wrong files staged under its +name (zeros of its size, its first half, another region's dump). -"Boots to the game" is read from the sampled frames: the run without the -file shows at least MIN_SHARED_SCREENS distinct screens that the run with -the file also shows, single-colour frames excluded. "No game screen" is a -run that did not load, or whose every sampled frame is a single colour. A -run that shows screens of its own that the reference never shows settles -nothing. + runs field settled + none boots hle_fallback true + none fails, present boots required true + hle_fallback false + zeros refused, present boots validation a content check + zeros used, present boots validation no content check + half refused, zeros used validation size + half used validation no size check + opens the staged file under another path path that path + opens the staged file under another name aliases name added + a staged file the content should use, never - flagged only + read + copies of one file that boot and fail - queued + +"Fails" is bios_error, crash or hang. A wrong file is refused when the +core reports a BIOS error, or when it boots without the BIOS screens the +right file shows (it fell back to a built-in BIOS). It is used when it +turns the run into a hang or a crash that the run without any file does not +show. Anything else settles nothing. Of validation, the table writes only a +missing `size`. A contradicted content check goes to review: which hash a +core computes is read in its code. Standard library only. """ @@ -29,23 +40,34 @@ from __future__ import annotations from collections.abc import Iterable, Sequence from dataclasses import dataclass, field -MIN_SHARED_SCREENS = 2 BOOTS = "boots" -FAILS = "fails" -INCONCLUSIVE = "inconclusive" +BIOS_ERROR = "bios_error" +CRASH = "crash" +HANG = "hang" +UNKNOWN = "unknown" +FAILS = (BIOS_ERROR, CRASH, HANG) +REFUSED = "refused" +USED = "used" +CONTENT_CHECKS = ("md5", "sha1", "sha256", "crc32", "adler32") # A field an entry leaves out reads as this value. -FIELD_DEFAULTS: dict[str, object] = {"hle_fallback": False, "path": None, "aliases": []} -SETTLED_FIELDS = ("required", "hle_fallback", "path", "aliases") +FIELD_DEFAULTS: dict[str, object] = {"hle_fallback": False, "path": None, "aliases": [], + "validation": []} +SETTLED_FIELDS = ("required", "hle_fallback", "path", "aliases", "validation") @dataclass(frozen=True) -class Boot: - """What the runs without the file showed, against the runs with it.""" +class Run: + """The verdict of one variant of a record: what a rule reads.""" - status: str - detail: str - shared: int = 0 + verdict: str + record_id: str + label: str + stages: tuple[str, ...] = () + + @property + def shows_bios(self) -> bool: + return "bios" in self.stages @dataclass(frozen=True) @@ -67,13 +89,6 @@ class Opened: extra: dict = field(default_factory=dict) -def screens(repetition: dict) -> set[str]: - """The distinct sampled frames of a repetition that are not one colour.""" - observables = repetition.get("observables") or {} - uniform = set(observables.get("uniform_frames") or ()) - return {digest for _, digest in observables.get("frame_hashes") or () if digest not in uniform} - - def variant(record: dict, label: str) -> dict: for item in record.get("variants") or (): if item.get("label") == label: @@ -81,43 +96,16 @@ def variant(record: dict, label: str) -> dict: raise KeyError(f"record {record.get('id')}: no variant {label!r}") -def boot_outcome(record: dict, absent_label: str, present_label: str) -> Boot: - """Whether the core reached the game without the file. +def run_of(record: dict, label: str) -> Run: + """A variant's verdict as a rule reads it. A record from before the + oracle has none, and reads as unknown.""" + item = variant(record, label) + evidence = item.get("verdict_evidence") or {} + return Run(item.get("verdict", UNKNOWN), record["id"], label, tuple(evidence.get("stages") or ())) - Every repetition must agree: one that boots and one that does not leave - the question open. - """ - present = variant(record, present_label)["repetitions"] - if any(not rep.get("loaded") for rep in present): - return Boot(INCONCLUSIVE, "with the file, the core did not load the content") - reference: set[str] = set() - for rep in present: - shown = screens(rep) - if not shown: - return Boot(INCONCLUSIVE, "with the file, the core showed no game screen") - reference |= shown - outcomes: list[str] = [] - shared_counts: list[int] = [] - for rep in variant(record, absent_label)["repetitions"]: - if not rep.get("loaded"): - outcomes.append(FAILS) - continue - shown = screens(rep) - if not shown: - outcomes.append(FAILS) - continue - shared = len(shown & reference) - shared_counts.append(shared) - outcomes.append(BOOTS if shared >= MIN_SHARED_SCREENS else INCONCLUSIVE) - shared = min(shared_counts) if shared_counts else 0 - if all(o == BOOTS for o in outcomes): - return Boot(BOOTS, f"without the file, every run showed {shared} or more screens the " - f"run with the file shows", shared) - if all(o == FAILS for o in outcomes): - return Boot(FAILS, "without the file, no run loaded the content or showed more than " - "one colour, while the run with the file reached the game") - return Boot(INCONCLUSIVE, "without the file, the runs showed screens the run with the file " - "never shows, or disagreed with each other", shared) + +def ids(*runs: Run) -> tuple[str, ...]: + return tuple(dict.fromkeys(run.record_id for run in runs)) def current(entry: dict, name: str) -> object: @@ -130,32 +118,110 @@ def agrees(entry: dict, verdict: Verdict) -> bool: """True when the profile already says what the run showed.""" if verdict.field == "aliases": return set(verdict.value or ()) <= set(entry.get("aliases") or ()) + if verdict.field == "validation": + return verdict.rule == "validation_agrees" return current(entry, verdict.field) == verdict.value -def rule_bios_need(boot: Boot, entry: dict, record_id: str) -> list[Verdict]: - """Boot with no file against boot with the declared file.""" +def rule_need(none: Run, present: Run, entry: dict) -> list[Verdict]: + """No BIOS at all against the file alone: required and hle_fallback.""" name = entry["name"] - evidence = (record_id,) - if boot.status == BOOTS: + if none.verdict == BOOTS: out = [Verdict("boots_without_file", "hle_fallback", True, - f"the core reaches the game without {name}: {boot.detail}", evidence, True)] + f"the core boots the content with no BIOS at all, {name} included", + ids(none), True)] if entry.get("required") is True: out.append(Verdict( "boots_without_file", "required", False, - f"the profile says {name} is required, and the core reaches the game " - f"without it. The table settles hle_fallback from this run, not required", - evidence)) + f"the profile says {name} is required, and the core boots without it. The table " + f"settles hle_fallback from this run, not required", ids(none))) return out - if boot.status == FAILS: + if none.verdict in FAILS and present.verdict == BOOTS: return [ Verdict("fails_without_file", "required", True, - f"the core reaches the game only with {name}: {boot.detail}", evidence, True), + f"with no BIOS the run ends in {none.verdict}, with {name} alone it boots", + ids(none, present), True), Verdict("fails_without_file", "hle_fallback", False, - f"no stand-in for {name}: {boot.detail}", evidence, True), + f"no stand-in for {name}: with no BIOS the run ends in {none.verdict}", + ids(none, present), True), ] - return [Verdict("inconclusive", "hle_fallback", None, - f"the runs with and without {name} settle nothing: {boot.detail}", evidence)] + return [Verdict("unsettled", "hle_fallback", None, + f"with no BIOS the run is {none.verdict} and with {name} alone {present.verdict}: " + f"no rule settles a field from that", ids(none, present))] + + +def file_use(wrong: Run, none: Run, present: Run) -> str | None: + """Whether the core refused a wrong file or ran it, or None.""" + if present.verdict != BOOTS: + return None + if wrong.verdict == BIOS_ERROR: + return REFUSED + if wrong.verdict == BOOTS: + if wrong.shows_bios: + return USED + return REFUSED if present.shows_bios else None + if wrong.verdict in (HANG, CRASH) and none.verdict in (BOOTS, BIOS_ERROR): + return USED + return None + + +def declared_checks(entry: dict) -> tuple[set[str], bool]: + """The checks the entry's validation names, and whether it is one list + (a per-build dict is never edited by the table).""" + value = entry.get("validation") or [] + if isinstance(value, dict): + return {str(v) for items in value.values() for v in items or ()}, False + return {str(v) for v in value}, True + + +def rule_validation(zeros: Run | None, half: Run | None, none: Run, present: Run, + entry: dict) -> Verdict | None: + """Same-size and short wrong files against the right one: validation.""" + name = entry["name"] + content = {REFUSED: True, USED: False}.get(file_use(zeros, none, present)) if zeros else None + half_use = file_use(half, none, present) if half else None + size: bool | None = None + if half_use == USED: + size, content = False, False if content is None else content + elif half_use == REFUSED and content is False: + size = True + if content is None and size is None: + return None + declared, editable = declared_checks(entry) + evidence = ids(*(run for run in (none, present, zeros, half) if run)) + said = sorted(declared) + shown = [] + problems = [] + if content is not None: + shown.append("checks the content" if content else "does not check the content") + if content != bool(declared & set(CONTENT_CHECKS)): + problems.append("a same-size file of zeros is " + ("refused" if content else "run") + + f", the profile declares validation {said}") + missing_size = size is True and "size" not in declared + if size is not None: + shown.append("checks the size" if size else "does not check the size") + if size is False and "size" in declared: + problems.append(f"the first half of {name} is run, the profile declares a size check") + statement = f"the core {' and '.join(shown)} of {name}" + if problems: + return Verdict("validation_contradicted", "validation", said, + f"{statement}: {'; '.join(problems)}", evidence) + if missing_size: + if not editable: + return Verdict("validation_contradicted", "validation", said, + f"{statement}: the per-build validation has no size check", evidence) + value = [*(entry.get("validation") or []), "size"] + return Verdict("validation_size_added", "validation", value, + f"{statement}: half of the file is refused while zeros of its size are run", + evidence, True) + return Verdict("validation_agrees", "validation", entry.get("validation") or [], statement, + evidence, editable) + + +def rule_region(dump: Run, present: Run, entry: dict, game_region: str | None) -> str: + """What the core did with another region's dump under this file's name.""" + return (f"{dump.label} staged as {entry['name']} with a {game_region or 'region-less'} game: " + f"{dump.verdict}, {entry['name']} itself: {present.verdict}") def rule_path(entry: dict, declared: str, opened: Opened) -> Verdict | None: @@ -184,20 +250,25 @@ def rule_never_opened(entry: dict, record_ids: Sequence[str], content: str) -> V f"{content}", tuple(record_ids)) -def rule_variants(status: str, labels: Iterable[str], entry: dict, record_id: str) -> Verdict | None: - """Copies of one file that change what the core shows: a pack choice to review.""" - if status != "differs": +def rule_variants(runs: Iterable[Run], entry: dict) -> Verdict | None: + """Copies of one file whose verdicts differ: a pack choice to review.""" + runs = list(runs) + if len({run.verdict for run in runs}) < 2: return None - return Verdict("variant_changes_output", "variants", sorted(labels), - f"the copies of {entry['name']} change what the core shows", - (record_id,)) + shown = ", ".join(f"{run.label} {run.verdict}" for run in runs) + return Verdict("variant_changes_verdict", "variants", sorted(run.label for run in runs), + f"the copies of {entry['name']} do not boot alike: {shown}", ids(*runs)) -def merge_evidence(existing: dict | None, verdict: Verdict) -> dict: - """The entry's evidence: map with the verdict's record ids added under its field.""" +def merge_evidence(existing: dict | None, verdict: Verdict, + keep: object = lambda rid: True) -> dict: + """The entry's evidence: map with the verdict's record ids added under its + field. An id `keep` refuses is dropped from that field: a record from + before the oracle no longer proves anything.""" out = {key: list(value) for key, value in (existing or {}).items()} - ids = out.setdefault(verdict.field, []) + ids_ = [rid for rid in out.get(verdict.field, []) if keep(rid)] for rid in verdict.evidence: - if rid not in ids: - ids.append(rid) - return {key: out[key] for key in sorted(out)} + if rid not in ids_: + ids_.append(rid) + out[verdict.field] = ids_ + return {key: out[key] for key in sorted(out) if out[key]} diff --git a/scripts/core_probe.py b/scripts/core_probe.py index 1582459d0..2ce9e137e 100644 --- a/scripts/core_probe.py +++ b/scripts/core_probe.py @@ -375,15 +375,25 @@ class Frontend: self.unanswered: dict[int, int] = {} self.input_queries = 0 self.input_pressed = 0 + # What a run showed so far, one JSON line per event, flushed at once: + # a core that kills the worker leaves its logs and frames behind. + events = spec.get("events") + self.events = open(events, "a", encoding="utf-8", buffering=1) if events else None # noqa: SIM115 self.callbacks = ( EnvFn(self.environment), VideoFn(self.video_refresh), AudioFn(self.audio_sample), AudioBatchFn(self.audio_batch), PollFn(lambda: None), InputFn(self.input_state), ) self.log_callback = RetroLogCallback(LogFn(self.log)) + def event(self, kind: str, **values: object) -> None: + if self.events: + self.events.write(json.dumps({"event": kind, **values}) + "\n") + def log(self, level: int, fmt: bytes, *args: int) -> None: name = LOG_LEVELS[level] if 0 <= level < len(LOG_LEVELS) else str(level) - self.logs.append({"frame": self.frame, "level": name, "text": render_printf(fmt or b"", args)}) + line = {"frame": self.frame, "level": name, "text": render_printf(fmt or b"", args)} + self.logs.append(line) + self.event("log", **line) def environment(self, cmd: int, data: int) -> bool: cmd &= ~ENV_EXPERIMENTAL @@ -446,8 +456,10 @@ class Frontend: if every and self.frame % every == 0: digest = hashlib.sha1(self.last_frame).hexdigest() self.frame_hashes.append([self.frame, digest]) - if digest not in self.uniform and uniform_frame(self.last_frame, PIXEL_BYTES[self.pixel_format]): + flat = digest in self.uniform or uniform_frame(self.last_frame, PIXEL_BYTES[self.pixel_format]) + if flat: self.uniform.add(digest) + self.event("frame", frame=self.frame, sha1=digest, uniform=flat) png_dir = self.spec.get("png_dir") if png_dir and self.last_frame: rgb = to_rgb(self.pixel_format, *self.last_size, self.last_frame) @@ -483,6 +495,7 @@ def run_worker(spec: dict) -> dict: getattr(core, name)(fn) info = RetroSystemInfo() core.retro_get_system_info(ctypes.byref(info)) + front.event("info", library_name=info.library_name.decode(), library_version=info.library_version.decode()) core.retro_init() game = None if spec.get("content"): @@ -494,7 +507,9 @@ def run_worker(spec: dict) -> dict: front.option_buffers.append(holder) game.data, game.size = ctypes.cast(holder, ctypes.c_void_p), len(blob) core.retro_load_game.restype = ctypes.c_bool - if not core.retro_load_game(ctypes.byref(game) if game else None): + loaded = bool(core.retro_load_game(ctypes.byref(game) if game else None)) + front.event("loaded", value=loaded) + if not loaded: return {"loaded": False, "logs": front.logs, "library_name": info.library_name.decode()} # A frontend plugs a joypad into each port after loading, as RetroArch does. for port in range(spec["joypad_ports"]): @@ -619,10 +634,41 @@ def core_identity(args: argparse.Namespace) -> dict: # Runs (parent side) # -------------------------------------------------------------------------- +@dataclass(frozen=True) +class Synthetic: + """A wrong file made for a negative test: `@zeros:SIZE` is SIZE zero + bytes, `@head:SIZE:PATH` the first SIZE bytes of PATH.""" + + kind: str + size: int + source: Path | None = None + + @classmethod + def parse(cls, text: str) -> Synthetic: + kind, _, rest = text[1:].partition(":") + size_text, _, source = rest.partition(":") + if kind not in ("zeros", "head") or not size_text.isdigit() or (kind == "head") != bool(source): + raise argparse.ArgumentTypeError( + f"{text!r}: a made file is @zeros:SIZE or @head:SIZE:PATH") + return cls(kind, int(size_text), Path(source) if source else None) + + def label(self, dest: str) -> str: + if self.kind == "zeros": + return f"{dest}-zeros-{self.size}" + return f"{dest}-head-{self.size}-of-{self.source.name}" + + def write(self, out: Path) -> None: + if self.kind == "zeros": + out.write_bytes(bytes(self.size)) + return + with self.source.open("rb") as handle: + out.write_bytes(handle.read(self.size)) + + @dataclass class Variant: label: str - files: dict[str, Path | None] = field(default_factory=dict) + files: dict[str, Path | Synthetic | None] = field(default_factory=dict) def stage_system_dir(base: dict[str, Path], variant: Variant, target: Path) -> dict[str, str | None]: @@ -632,10 +678,14 @@ def stage_system_dir(base: dict[str, Path], variant: Variant, target: Path) -> d for dest, src in {**base, **variant.files}.items(): out = target / dest if src is None: + out.unlink(missing_ok=True) staged[dest] = None continue out.parent.mkdir(parents=True, exist_ok=True) - shutil.copyfile(src, out) + if isinstance(src, Synthetic): + src.write(out) + else: + shutil.copyfile(src, out) staged[dest] = file_digest(out) return staged @@ -729,6 +779,45 @@ def run_group(command: list[str], timeout: float, env: dict[str, str]) -> subpro return subprocess.CompletedProcess(command, proc.returncode, out, err) +WORKER_TRACEBACK = "Traceback (most recent call last)" + + +def exit_description(returncode: int) -> dict: + """The status of a worker that died. A signal reaches the probe as a + negative code, or as 128 + its number through bubblewrap's init.""" + number = -returncode if returncode < 0 else returncode - 128 if 128 < returncode < 160 else 0 + name = None + if number: + with contextlib.suppress(ValueError): + name = signal.Signals(number).name + return {"status": returncode, "signal": name} + + +def partial_result(events: Path, spec: dict) -> dict: + """A repetition rebuilt from the events a worker wrote before it died.""" + lines = [] + if events.is_file(): + for line in events.read_text(encoding="utf-8", errors="replace").splitlines(): + with contextlib.suppress(ValueError): + lines.append(json.loads(line)) + info = next((e for e in lines if e.get("event") == "info"), {}) + loaded = next((e["value"] for e in lines if e.get("event") == "loaded"), False) + frames = [e for e in lines if e.get("event") == "frame"] + result: dict = { + "loaded": bool(loaded), + "library_name": info.get("library_name", ""), + "logs": [{k: e[k] for k in ("frame", "level", "text")} for e in lines if e.get("event") == "log"], + } + if info.get("library_version"): + result["library_version"] = info["library_version"] + if spec["frame_hash_every"]: + result["observables"] = { + "frame_hashes": [[e["frame"], e["sha1"]] for e in frames], + "uniform_frames": sorted({e["sha1"] for e in frames if e.get("uniform")}), + } + return result + + def run_once(args: argparse.Namespace, system_dir: Path, scratch: Path, state_out: Path | None, png_dir: Path | None) -> dict: save_dir = scratch / "save" @@ -748,6 +837,7 @@ def run_once(args: argparse.Namespace, system_dir: Path, scratch: Path, state_ou "png_dir": str(png_dir) if png_dir else None, "state_out": str(state_out) if state_out else None, "env": dict(args.env), + "events": str((scratch / "events.jsonl").resolve()), } spec_path, out_path, trace_path = scratch / "spec.json", scratch / "result.json", scratch / "strace.txt" spec_path.write_text(json.dumps(spec), encoding="utf-8") @@ -764,10 +854,20 @@ def run_once(args: argparse.Namespace, system_dir: Path, scratch: Path, state_ou if args.require_sandbox and isolation != SANDBOX: raise ProbeError(f"refused to run the core outside a sandbox: {isolation}") # The core sees the variables it was given and nothing of the host's. - proc = run_group(prefix + command, args.timeout, dict(args.env)) - if proc.returncode != 0 or not out_path.exists(): - raise ProbeError(f"worker exited {proc.returncode}: {proc.stderr.strip()[-2000:]}") - result = json.loads(out_path.read_text(encoding="utf-8")) + try: + proc = run_group(prefix + command, args.timeout, dict(args.env)) + except subprocess.TimeoutExpired: + result = partial_result(scratch / "events.jsonl", spec) + result["exit"] = {"status": None, "signal": "SIGKILL", "timeout_s": args.timeout} + else: + if proc.returncode == 0 and out_path.exists(): + result = json.loads(out_path.read_text(encoding="utf-8")) + elif WORKER_TRACEBACK in proc.stderr: + # The probe's own code failed: no verdict on the core is drawn. + raise ProbeError(f"worker exited {proc.returncode}: {proc.stderr.strip()[-2000:]}") + else: + result = partial_result(scratch / "events.jsonl", spec) + result["exit"] = exit_description(proc.returncode) places = [(str(system_dir.resolve()), ""), (str(save_dir.resolve()), "")] if args.content: places.append((str(Path(args.content).resolve().parent), "")) @@ -820,6 +920,10 @@ def run_variant(args: argparse.Namespace, base: dict[str, Path], variant: Varian if state_out: state_out.parent.mkdir(parents=True, exist_ok=True) rep = run_once(args, system_dir, scratch, state_out, png_dir) + oracle = oracle_for(args.oracle_data, args.content_sha1, rep.get("library_name")) + rep["verdict"], rep["verdict_evidence"] = judge(rep, oracle) + if rep.get("exit"): + flags.append(f"repetition {index}: {rep['verdict_evidence']['detail']}") if not rep.get("loaded"): flags.append(f"repetition {index}: retro_load_game returned false") for entry in rep.get("files_opened") or []: @@ -833,11 +937,14 @@ def run_variant(args: argparse.Namespace, base: dict[str, Path], variant: Varian if rep.get("observables") != reference: differing = sorted(k for k in (reference or {}) if (rep.get("observables") or {}).get(k) != reference[k]) flags.append(f"repetition {index} disagrees with repetition 0 on {', '.join(differing)}") + verdict, evidence = variant_verdict(reps) return { "label": variant.label, "system_files": {dest: sha for dest, sha in sorted(staged.items())}, "repetitions": reps, "flags": flags, + "verdict": verdict, + "verdict_evidence": evidence, } @@ -894,12 +1001,18 @@ def build_parser() -> argparse.ArgumentParser: help="refuse to record when the 1-minute load average exceeds this " "(default: a quarter of the CPU count)") common.add_argument("--work-dir", default=str(DEFAULT_WORK_DIR), help="scratch directory") - common.add_argument("--timeout", type=int, default=1800, help="seconds per repetition") + common.add_argument("--timeout", type=int, default=1800, + help="seconds per repetition, after which the run is killed and judged a hang") + common.add_argument("--oracle", default=None, metavar="FILE", + help="reference frames and error patterns the verdicts are judged against " + "(default: tests/campaign/oracle.yml)") common.add_argument("--no-trace", action="store_true", help="do not run the worker under strace") common.add_argument("--require-sandbox", action="store_true", help="refuse to run the core unless bubblewrap can isolate it") - sub.add_parser("run", parents=[common], help="one execution") + run = sub.add_parser("run", parents=[common], help="one execution") + run.add_argument("--repeat", type=int, default=1, + help="runs, each in a fresh process: the verdict is the one they all agree on") bench = sub.add_parser("bench", parents=[common], help="timed repetitions") bench.add_argument("--repeat", type=int, default=MIN_BENCH_REPETITIONS, help=f"timed repetitions, each in a fresh process (at least {MIN_BENCH_REPETITIONS} to record)") @@ -941,12 +1054,16 @@ def validate_args(args: argparse.Namespace) -> list[str]: for dest, src in args.system_file: if not Path(src).is_file(): problems.append(f"--system-file {dest}={src}: not a file") + if args.repeat < 1: + problems.append("--repeat must be >= 1") if args.frames < 1 or args.warmup < 0: problems.append("--frames must be >= 1 and --warmup >= 0") if args.require_sandbox: isolation = sandbox_argv([])[1] if isolation != SANDBOX: problems.append(f"--require-sandbox: {isolation}") + if args.oracle and not Path(args.oracle).is_file(): + problems.append(f"--oracle {args.oracle}: not a file") if args.png_dir and not args.frame_hash_every: problems.append("--png-dir needs --frame-hash-every") if args.record: @@ -970,7 +1087,15 @@ def validate_args(args: argparse.Namespace) -> list[str]: if len(dests) != 1: problems.append("--vary must name the same DEST in every occurrence") for dest, src in args.vary: - if src and not Path(src).is_file(): + if src.startswith("@"): + try: + made = Synthetic.parse(src) + except argparse.ArgumentTypeError as exc: + problems.append(f"--vary {dest}: {exc}") + continue + if made.source and not made.source.is_file(): + problems.append(f"--vary {dest}={src}: {made.source} is not a file") + elif src and not Path(src).is_file(): problems.append(f"--vary {dest}={src}: not a file") names = set(BASE_OBSERVABLES) | {f"state:{v}" for v in args.state_var} if args.frame_hash_every: @@ -986,11 +1111,12 @@ def variants_for(args: argparse.Namespace) -> list[Variant]: out: list[Variant] = [] seen: set[str] = set() for dest, src in args.vary: - label = Path(src).name if src else f"{dest}-absent" + made = Synthetic.parse(src) if src.startswith("@") else None + label = made.label(dest) if made else Path(src).name if src else f"{dest}-absent" while label in seen: label += "+" seen.add(label) - out.append(Variant(label, {dest: Path(src) if src else None})) + out.append(Variant(label, {dest: made or (Path(src) if src else None)})) return out @@ -1036,6 +1162,125 @@ def check_result(variants: list[dict], observable: str) -> dict: } +# -------------------------------------------------------------------------- +# Boot oracle: what a run reached, from positive evidence +# -------------------------------------------------------------------------- + +BOOTS, BIOS_ERROR, CRASH, HANG, UNKNOWN = "boots", "bios_error", "crash", "hang", "unknown" +VERDICTS = (BOOTS, BIOS_ERROR, CRASH, HANG, UNKNOWN) +DEFAULT_ORACLE = ROOT / "tests" / "campaign" / "oracle.yml" + + +def oracle_for(data: dict, content_sha1: str | None, library_name: str | None) -> dict: + """The oracle entries that judge one content on one core: its reference + frames, the error patterns every core shares and the core's own, and the + core's known error screens.""" + core = (data.get("cores") or {}).get(library_name or "") or {} + references: list[dict] = [] + for item in data.get("content") or (): + if content_sha1 and item.get("sha1") == content_sha1: + block = (item.get("references") or {}).get(library_name or "") or {} + references = [{"sha1": f["sha1"], "stage": f["stage"], "shows": f["shows"]} + for f in block.get("frames") or ()] + patterns = [p["pattern"] for p in (data.get("error_patterns") or ())] + patterns += [p["pattern"] for p in core.get("error_patterns") or ()] + screens = [{"sha1": s["sha1"], "shows": s["shows"]} for s in core.get("error_screens") or ()] + return {"references": references, "error_patterns": patterns, "error_screens": screens} + + +def sampled(rep: dict) -> list[tuple[int, str]]: + observables = rep.get("observables") or {} + return [(int(frame), digest) for frame, digest in observables.get("frame_hashes") or ()] + + +def judge(rep: dict, oracle: dict) -> tuple[str, dict]: + """The verdict of one repetition and the evidence that decided it. + + In order: a worker killed after the timeout is a hang and one that died + a crash, from the process status. A reference frame reached is a boot. + Without one, an error pattern of the core log or a known error screen + is a BIOS error. Every sampled frame of the second half one colour is a + hang. Anything else is unknown: a run is never called a boot because + its frames differ from another run's. + """ + ended = rep.get("exit") + if ended and ended.get("timeout_s") is not None: + return HANG, {"rule": "timeout", "detail": f"killed after {ended['timeout_s']} s", + "exit_status": ended.get("status"), "signal": ended.get("signal")} + if ended: + how = f"signal {ended['signal']}" if ended.get("signal") else f"exit status {ended.get('status')}" + return CRASH, {"rule": "exit_status", "detail": f"the worker died: {how}", + "exit_status": ended.get("status"), "signal": ended.get("signal")} + frames = sampled(rep) + final = (rep.get("observables") or {}).get("frame_sha1") + if final: + frames.append((int(rep.get("frames", 0)) + int(rep.get("warmup", 0)), final)) + references = {r["sha1"]: r for r in oracle.get("references") or ()} + reached = [(frame, digest) for frame, digest in frames if digest in references] + if reached: + frame, digest = reached[0] + stages = list(dict.fromkeys(references[d]["stage"] for _, d in reached)) + return BOOTS, {"rule": "reference_frame", + "detail": f"frame {frame} shows {references[digest]['shows']}", + "frame": frame, "sha1": digest, "stages": stages} + for pattern in oracle.get("error_patterns") or (): + for line in rep.get("logs") or (): + if re.search(pattern, line["text"]): + return BIOS_ERROR, {"rule": "error_pattern", + "detail": f"the core logged an error at frame {line['frame']}", + "pattern": pattern, "log": line["text"], "frame": line["frame"]} + screens = {s["sha1"]: s for s in oracle.get("error_screens") or ()} + for frame, digest in frames: + if digest in screens: + return BIOS_ERROR, {"rule": "error_screen", + "detail": f"frame {frame} shows {screens[digest]['shows']}", + "frame": frame, "sha1": digest} + if not rep.get("loaded"): + return UNKNOWN, {"rule": "no_evidence", + "detail": "retro_load_game returned false and no error pattern matched"} + uniform = set((rep.get("observables") or {}).get("uniform_frames") or ()) + tail = [digest for frame, digest in sampled(rep)][len(sampled(rep)) // 2:] + if tail and all(digest in uniform for digest in tail): + return HANG, {"rule": "uniform_tail", + "detail": f"the last {len(tail)} sampled frames are one colour each"} + if not oracle.get("references"): + return UNKNOWN, {"rule": "no_evidence", "detail": "no reference frame is known for this " + "content on this core"} + return UNKNOWN, {"rule": "no_evidence", "detail": "no reference frame reached, no error pattern " + "and no error screen"} + + +def variant_verdict(reps: Sequence[dict]) -> tuple[str, dict]: + """The verdict all repetitions agree on, else unknown.""" + verdicts = [rep["verdict"] for rep in reps] + if len(set(verdicts)) == 1: + first = reps[0]["verdict_evidence"] + evidence = {"rule": "repetitions_agree", + "detail": f"{len(reps)} of {len(reps)} repetitions: {first['detail']}", + "repetitions": verdicts} + for key in ("frame", "sha1", "stages", "pattern", "log", "exit_status", "signal"): + if key in first: + evidence[key] = first[key] + return verdicts[0], evidence + return UNKNOWN, {"rule": "repetitions_disagree", + "detail": f"the repetitions disagree: {', '.join(verdicts)}", + "repetitions": verdicts} + + +def load_oracle(path: Path | None) -> tuple[dict, dict | None]: + """The oracle file's data and its identity, or nothing without a file.""" + if path is None or not path.is_file(): + return {}, None + import yaml # noqa: PLC0415 + + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + try: + shown = path.resolve().relative_to(ROOT).as_posix() + except ValueError: + shown = path.name + return data, {"file": shown, "sha256": file_digest(path, "sha256")} + + def default_emulator(core_file: Path, library_name: str | None) -> str: """The profile whose cores: names the .so, else the library name as a slug.""" stem = core_file.name.split(".")[0].removesuffix("_libretro") @@ -1097,8 +1342,10 @@ def probe(args: argparse.Namespace) -> int: print(f"ERROR: load average {load_before[0]} exceeds {max_load}: the machine is busy, " f"nothing recorded (raise --max-load to override)", file=sys.stderr) return 3 - repetitions = 1 if args.mode == "run" else args.repeat + repetitions = args.repeat base = {dest: Path(src) for dest, src in args.system_file} + args.oracle_data, oracle_identity = load_oracle(Path(args.oracle) if args.oracle else DEFAULT_ORACLE) + args.content_sha1 = file_digest(Path(args.content)) if args.content else None work = Path(tempfile.mkdtemp(prefix=f"{args.mode}-", dir=_ensure(Path(args.work_dir)))) try: variants = [run_variant(args, base, variant, work / f"v{i}", repetitions) @@ -1136,6 +1383,9 @@ def probe(args: argparse.Namespace) -> int: + (["frame_hashes", "uniform_frames"] if args.frame_hash_every else []), "frame_hash_every": args.frame_hash_every, "joypad_ports": args.joypad_ports, + "oracle": ({**oracle_identity, **oracle_for(args.oracle_data, args.content_sha1, + first.get("library_name"))} + if oracle_identity else None), }, "variants": variants, "flags": [f"{v['label']}: {flag}" for v in variants for flag in v["flags"]], @@ -1150,7 +1400,8 @@ def probe(args: argparse.Namespace) -> int: if args.mode == "check": record["check"] = {"claim": args.claim, "method": args.method, "observable": args.observable, "variable": args.vary[0][0], - "result": check_result(variants, args.observable)} + "result": {**check_result(variants, args.observable), + "verdicts": {v["label"]: v["verdict"] for v in variants}}} try: record["emulator"] = args.emulator or default_emulator(Path(args.core), first.get("library_name")) except ProbeError as exc: diff --git a/scripts/measurement_records.py b/scripts/measurement_records.py index 1dbf5bdd5..f61185331 100644 --- a/scripts/measurement_records.py +++ b/scripts/measurement_records.py @@ -27,7 +27,7 @@ from collections.abc import Iterable from dataclasses import dataclass from pathlib import Path -RECORD_VERSION = 3 +RECORD_VERSION = 4 # 3: before the boot oracle. 4: every repetition and variant carries a verdict. RECORD_VERSIONS = (3, 4) RECORDS_DIR = "measurements" diff --git a/tests/campaign/oracle.yml b/tests/campaign/oracle.yml new file mode 100644 index 000000000..fa567f1d6 --- /dev/null +++ b/tests/campaign/oracle.yml @@ -0,0 +1,26 @@ +# The boot oracle of scripts/core_probe.py: what a run must reach to boot, +# and what tells a BIOS error apart. +# +# content per piece of test content (by the sha1 of the file handed to +# the core, the .cue for a disc), per core (by the library name +# retro_get_system_info gives): the reference frames a run with +# the full BIOS set reaches. Each was recorded once +# (`recorded_from`, the record id) and reviewed by looking at its +# PNG. `stage: bios` is a screen only the console's own BIOS +# draws, `stage: game` a screen of the content. A run that reaches +# one of them boots. +# error_patterns log lines after which a core stops, read from its source +# (`source_ref`). Shared ones at the top level, a core's own under +# cores:. A line a core logs and then carries on (a fallback to a +# built-in BIOS) is not an error pattern. +# error_screens per core, the sha1 of a frame that shows its BIOS error message. +# +# A run that reaches no reference frame, matches no pattern or screen and +# does not end on one colour is `unknown`: frames that differ from another +# run's never make a boot. + +error_patterns: [] + +cores: {} + +content: [] diff --git a/tests/test_campaign.py b/tests/test_campaign.py index 52da1cf43..6bfabbb83 100644 --- a/tests/test_campaign.py +++ b/tests/test_campaign.py @@ -16,6 +16,7 @@ import io import json import os import shutil +import struct import subprocess import sys import tempfile @@ -92,15 +93,20 @@ void retro_init(void) {} void retro_deinit(void) {} static FILE *try_open(const char *dir, const char *name) { char path[4096]; snprintf(path, sizeof path, "%s/%s", dir, name); return fopen(path, "rb"); } +/* booted: 0 a black screen to the end, 1 the BIOS logo then the title, + 2 the title alone (the built-in stand-in of FAKE_HLE). */ bool retro_load_game(const void *g) { - const char *dir = NULL; FILE *f; (void)g; env(9, &dir); + const char *dir = NULL; FILE *f; unsigned char buf[8]; size_t n = 0; (void)g; env(9, &dir); f = try_open(dir, "bios.bin"); if (!f) f = try_open(dir, "BIOS.BIN"); + if (f) { n = fread(buf, 1, sizeof buf, f); fclose(f); } + int valid = f && n == 4 && buf[0] == 'B'; #ifdef FAKE_HLE - if (f) fclose(f); - booted = 1; + /* Checks the size only: a file of the wrong size is refused for the + stand-in, one of the right size is run, whatever it holds. */ + booted = valid ? 1 : (f && n == 4) ? 0 : 2; #else - if (f) { fclose(f); booted = 1; } else booted = 0; + booted = valid ? 1 : 0; #endif return true; } void retro_unload_game(void) {} @@ -109,7 +115,8 @@ void retro_run(void) { #ifdef FAKE_HANG for (;;) {} #endif - for (int i = 0; i < 8; i++) pixels[i] = booted ? (uint16_t)(frame / 2 + i) : 0; + int shown = frame < 4 ? 0 : booted == 1 && frame < 8 ? 1 : booted ? 2 : 0; + for (int i = 0; i < 8; i++) pixels[i] = shown == 0 ? 0 : (uint16_t)((shown == 1 ? 1 : 20) + i); frame++; video(pixels, 4, 2, 8); batch(s, 2); } size_t retro_serialize_size(void) { return 0; } @@ -117,78 +124,94 @@ bool retro_serialize(void *d, size_t n) { (void)d; (void)n; return false; } """ -def rep(frames: list[str], uniform: list[str] | None = None, loaded: bool = True) -> dict: - return {"loaded": loaded, "observables": { - "frame_hashes": [[i * 10, h] for i, h in enumerate(frames)], - "uniform_frames": sorted(set(uniform or []))}} +def run(verdict: str, label: str = "v", stages: tuple[str, ...] = (), + rid: str = "2026-10-27-aaaaaaaaaaaa") -> rules.Run: + return rules.Run(verdict, rid, label, stages) -def boot_record(absent: list[dict], present: list[dict], rid: str = "2026-10-27-aaaaaaaaaaaa") -> dict: - return {"id": rid, "variants": [{"label": "bios.bin-absent", "repetitions": absent}, - {"label": "bios.bin", "repetitions": present}]} - - -class BootOutcome(unittest.TestCase): - GAME = ("black", "logo", "title", "title2") - - def test_boots_when_screens_are_shared(self) -> None: - record = boot_record([rep(["black", "hle", "logo", "title"], ["black"])] * 2, - [rep(self.GAME, ["black"])] * 2) - boot = rules.boot_outcome(record, "bios.bin-absent", "bios.bin") - self.assertEqual((boot.status, boot.shared), (rules.BOOTS, 2)) - - def test_one_colour_and_no_load_fail(self) -> None: - black = boot_record([rep(["black"], ["black"]), rep([], [], loaded=False)], - [rep(self.GAME, ["black"])]) - self.assertEqual(rules.boot_outcome(black, "bios.bin-absent", "bios.bin").status, rules.FAILS) - - def test_a_shared_blank_screen_proves_nothing(self) -> None: - # The black frame both runs start on is one colour: it is no game screen. - record = boot_record([rep(["black", "error"], ["black"])], [rep(self.GAME, ["black"])]) - self.assertEqual(rules.boot_outcome(record, "bios.bin-absent", "bios.bin").status, - rules.INCONCLUSIVE) - - def test_one_shared_screen_is_not_enough(self) -> None: - record = boot_record([rep(["logo", "error"])], [rep(self.GAME, ["black"])]) - self.assertEqual(rules.boot_outcome(record, "bios.bin-absent", "bios.bin").status, - rules.INCONCLUSIVE) - - def test_repetitions_that_disagree_settle_nothing(self) -> None: - record = boot_record([rep(["logo", "title"]), rep(["black"], ["black"])], - [rep(self.GAME, ["black"])]) - self.assertEqual(rules.boot_outcome(record, "bios.bin-absent", "bios.bin").status, - rules.INCONCLUSIVE) - - def test_a_reference_that_shows_nothing_settles_nothing(self) -> None: - record = boot_record([rep(["logo", "title"])], [rep(["black"], ["black"])]) - self.assertEqual(rules.boot_outcome(record, "bios.bin-absent", "bios.bin").status, - rules.INCONCLUSIVE) - record = boot_record([rep(["logo", "title"])], [rep(self.GAME, loaded=False)]) - self.assertEqual(rules.boot_outcome(record, "bios.bin-absent", "bios.bin").status, - rules.INCONCLUSIVE) +NONE_BOOTS = run(rules.BOOTS, "none", ("game",), "2026-10-27-000000000001") +NONE_HANGS = run(rules.HANG, "none", (), "2026-10-27-000000000001") +PRESENT = run(rules.BOOTS, "bios.bin", ("bios", "game"), "2026-10-27-000000000002") class RuleTable(unittest.TestCase): ENTRY = MappingProxyType({"name": "bios.bin", "required": True, "aliases": ["Bios.bin"]}) RID = "2026-10-27-bbbbbbbbbbbb" - def test_boots_without_settles_hle_and_queues_required(self) -> None: - verdicts = rules.rule_bios_need(rules.Boot(rules.BOOTS, "x", 3), self.ENTRY, self.RID) + def test_a_verdict_is_read_from_the_record(self) -> None: + record = {"id": self.RID, "variants": [ + {"label": "a", "verdict": "boots", "verdict_evidence": {"rule": "repetitions_agree", + "detail": "d", "stages": ["bios"]}}, + {"label": "old"}]} + self.assertEqual(rules.run_of(record, "a"), rules.Run("boots", self.RID, "a", ("bios",))) + # A record from before the oracle proves nothing. + self.assertEqual(rules.run_of(record, "old").verdict, rules.UNKNOWN) + + def test_boots_with_no_bios_settles_hle_and_queues_required(self) -> None: + verdicts = rules.rule_need(NONE_BOOTS, PRESENT, self.ENTRY) self.assertEqual([(v.field, v.value, v.apply) for v in verdicts], [("hle_fallback", True, True), ("required", False, False)]) - self.assertEqual(verdicts[0].evidence, (self.RID,)) - optional = rules.rule_bios_need(rules.Boot(rules.BOOTS, "x", 3), - {"name": "a", "required": False}, self.RID) + self.assertEqual(verdicts[0].evidence, (NONE_BOOTS.record_id,)) + optional = rules.rule_need(NONE_BOOTS, PRESENT, {"name": "a", "required": False}) self.assertEqual([v.field for v in optional], ["hle_fallback"]) - def test_fails_without_settles_required_and_no_hle(self) -> None: - verdicts = rules.rule_bios_need(rules.Boot(rules.FAILS, "x"), self.ENTRY, self.RID) - self.assertEqual([(v.field, v.value, v.apply) for v in verdicts], - [("required", True, True), ("hle_fallback", False, True)]) + def test_fails_without_and_boots_alone_settles_required(self) -> None: + for failure in rules.FAILS: + verdicts = rules.rule_need(run(failure, "none"), PRESENT, self.ENTRY) + self.assertEqual([(v.field, v.value, v.apply) for v in verdicts], + [("required", True, True), ("hle_fallback", False, True)]) - def test_inconclusive_is_queued(self) -> None: - (verdict,) = rules.rule_bios_need(rules.Boot(rules.INCONCLUSIVE, "x"), self.ENTRY, self.RID) - self.assertFalse(verdict.apply) + def test_unknown_or_no_boot_alone_settles_nothing(self) -> None: + for none, present in ((run(rules.UNKNOWN), PRESENT), (NONE_HANGS, run(rules.HANG)), + (NONE_HANGS, run(rules.UNKNOWN))): + (verdict,) = rules.rule_need(none, present, self.ENTRY) + self.assertFalse(verdict.apply) + self.assertEqual(verdict.rule, "unsettled") + + def test_file_use(self) -> None: + self.assertEqual(rules.file_use(run(rules.BIOS_ERROR), NONE_HANGS, PRESENT), rules.REFUSED) + # It boots without the BIOS screens the right file shows: a fallback. + self.assertEqual(rules.file_use(run(rules.BOOTS, stages=("game",)), NONE_BOOTS, PRESENT), + rules.REFUSED) + self.assertEqual(rules.file_use(run(rules.BOOTS, stages=("bios", "game")), NONE_BOOTS, PRESENT), + rules.USED) + self.assertEqual(rules.file_use(run(rules.HANG), NONE_BOOTS, PRESENT), rules.USED) + self.assertEqual(rules.file_use(run(rules.CRASH), NONE_BOOTS, PRESENT), rules.USED) + # A hang the run with no file shows too says nothing about the file. + self.assertIsNone(rules.file_use(run(rules.HANG), NONE_HANGS, PRESENT)) + self.assertIsNone(rules.file_use(run(rules.UNKNOWN), NONE_BOOTS, PRESENT)) + self.assertIsNone(rules.file_use(run(rules.BIOS_ERROR), NONE_BOOTS, run(rules.HANG))) + no_logo = run(rules.BOOTS, stages=("game",)) + self.assertIsNone(rules.file_use(run(rules.BOOTS, stages=("game",)), NONE_BOOTS, no_logo)) + + def test_validation_agrees_or_adds_size(self) -> None: + refused = run(rules.BOOTS, stages=("game",)) + used = run(rules.HANG) + entry = {"name": "bios.bin", "validation": ["size", "md5"]} + verdict = rules.rule_validation(refused, refused, NONE_BOOTS, PRESENT, entry) + self.assertEqual((verdict.rule, verdict.apply), ("validation_agrees", True)) + self.assertTrue(rules.agrees(entry, verdict)) + added = rules.rule_validation(used, refused, NONE_BOOTS, PRESENT, {"name": "bios.bin"}) + self.assertEqual((added.rule, added.value, added.apply), ("validation_size_added", ["size"], True)) + self.assertFalse(rules.agrees({"name": "bios.bin"}, added)) + self.assertEqual(rules.rule_validation(used, used, NONE_BOOTS, PRESENT, {"name": "b"}).rule, + "validation_agrees") + + def test_validation_contradicted_is_queued(self) -> None: + refused = run(rules.BIOS_ERROR) + used = run(rules.HANG) + cases = [ + (refused, None, {"name": "b", "validation": ["size"]}), # content check undeclared + (used, None, {"name": "b", "validation": ["md5"]}), # declared, not done + (None, used, {"name": "b", "validation": ["size"]}), # half of it runs + (used, refused, {"name": "b", "validation": {"core": ["md5"]}}), # per-build form + ] + for zeros, half, entry in cases: + verdict = rules.rule_validation(zeros, half, NONE_BOOTS, PRESENT, entry) + self.assertEqual((verdict.rule, verdict.apply), ("validation_contradicted", False), entry) + self.assertFalse(rules.agrees(entry, verdict)) + self.assertIsNone(rules.rule_validation(run(rules.UNKNOWN), None, NONE_BOOTS, PRESENT, + {"name": "b"})) def test_path_and_alias(self) -> None: self.assertIsNone(rules.rule_path(self.ENTRY, "bios.bin", rules.Opened("bios.bin", self.RID))) @@ -204,8 +227,9 @@ class RuleTable(unittest.TestCase): self.assertEqual(verdict.rule, "never_opened") def test_variants(self) -> None: - self.assertIsNone(rules.rule_variants("same", ["a", "b"], self.ENTRY, self.RID)) - self.assertFalse(rules.rule_variants("differs", ["b", "a"], self.ENTRY, self.RID).apply) + self.assertIsNone(rules.rule_variants([run(rules.BOOTS, "a"), run(rules.BOOTS, "b")], self.ENTRY)) + verdict = rules.rule_variants([run(rules.BOOTS, "b"), run(rules.HANG, "a")], self.ENTRY) + self.assertEqual((verdict.value, verdict.apply), (["a", "b"], False)) def test_agreement_and_evidence(self) -> None: self.assertTrue(rules.agrees({"name": "a"}, rules.Verdict("r", "hle_fallback", False, "", ()))) @@ -213,6 +237,22 @@ class RuleTable(unittest.TestCase): self.assertTrue(rules.agrees(self.ENTRY, rules.Verdict("r", "aliases", ["Bios.bin"], "", ()))) merged = rules.merge_evidence({"required": ["x"]}, rules.Verdict("r", "required", True, "", ("y", "x"))) self.assertEqual(merged, {"required": ["x", "y"]}) + # An id a frame comparison wrote no longer proves the field. + merged = rules.merge_evidence({"required": ["old"], "path": ["p"]}, + rules.Verdict("r", "required", True, "", ("new",)), + keep=lambda rid: rid != "old") + self.assertEqual(merged, {"path": ["p"], "required": ["new"]}) + + def test_reference_candidates(self) -> None: + def rep(hashes: list[str]) -> dict: + return {"observables": {"frame_hashes": [[i * 10, h] for i, h in enumerate(hashes)], + "uniform_frames": ["black"]}} + record = {"variants": [{"repetitions": [ + rep(["black"] * 4 + ["logo"] * 3 + ["flash"] + ["title"] * 5), + rep(["black"] * 5 + ["logo"] * 3 + ["title"] * 3)]}]} + self.assertEqual(campaign.reference_candidates(record), + [{"sha1": "logo", "first_frame": 40, "samples": 3}, + {"sha1": "title", "first_frame": 80, "samples": 5}]) class ProfileEditor(unittest.TestCase): @@ -286,6 +326,11 @@ class FakeHost(campaign.Host): return result +def screen(base: int) -> str: + """The sha1 of a frame of FAKE_CORE: eight 16-bit pixels from base up.""" + return hashlib.sha1(struct.pack("<8H", *range(base, base + 8))).hexdigest() + + class Workspace(unittest.TestCase): def setUp(self) -> None: TMP.mkdir(parents=True, exist_ok=True) @@ -309,9 +354,17 @@ class Workspace(unittest.TestCase): "name": "Fake Game (World)", "catalog": "no-intro", "reason": "a test", "frames": 12, "frame_hash_every": 2}]}}}), encoding="utf-8") self.game = self.tmp / "content" / "fakesys" / "Fake Game (World)" + self.oracle = self.tmp / "oracle.yml" + self.oracle.write_text(yaml.safe_dump({ + "error_patterns": [], "cores": {}, "content": [{ + "name": "Fake Game (World)", "sha1": hashlib.sha1(b"GAME").hexdigest(), + "references": {"Fake": {"recorded_from": "2026-10-27-000000000000", "frames": [ + {"sha1": screen(1), "stage": "bios", "shows": "the fake logo"}, + {"sha1": screen(20), "stage": "game", "shows": "the fake title"}]}}}]}), + encoding="utf-8") self.paths = campaign.Paths(root=self.root, home=self.tmp / "home", content=self.tmp / "content", db=self.root / "database.json", - content_file=self.content_file) + content_file=self.content_file, oracle=self.oracle) def tearDown(self) -> None: shutil.rmtree(self.tmp, ignore_errors=True) @@ -333,8 +386,12 @@ class Workspace(unittest.TestCase): class StateAndResume(Workspace): def test_should_run(self) -> None: self.assertTrue(campaign.should_run(None, "a", "d")) - done = {"status": "done", "core_sha256": "a", "profile_digest": "d"} + done = {"status": "done", "core_sha256": "a", "profile_digest": "d", + "record_version": mr.RECORD_VERSION, "oracle_sha256": ""} self.assertFalse(campaign.should_run(done, "a", "d")) + # A record from before the verdicts, or judged against another oracle, runs again. + self.assertTrue(campaign.should_run({**done, "record_version": 3}, "a", "d")) + self.assertTrue(campaign.should_run(done, "a", "d", "f" * 64)) self.assertTrue(campaign.should_run(done, "b", "d")) self.assertTrue(campaign.should_run(done, "a", "e")) self.assertTrue(campaign.should_run({**done, "status": "pending"}, "a", "d")) @@ -752,8 +809,14 @@ class Findings(Workspace): def entry(self, index: int = 0) -> dict: return yaml.safe_load(self.profile.read_text())["files"][index] + def none(self, verdict: str) -> rules.Run: + return rules.Run(verdict, self.RID, "none") + + def present(self) -> rules.Run: + return rules.Run(rules.BOOTS, self.RID, "bios.bin", ("game",)) + def test_a_settled_field_is_written_with_its_evidence(self) -> None: - verdicts = rules.rule_bios_need(rules.Boot(rules.BOOTS, "shown", 4), self.entry(), self.RID) + verdicts = rules.rule_need(self.none(rules.BOOTS), self.present(), self.entry()) self.c.settle("fake", self.content, self.core, 0, self.entry(), verdicts) entry = self.entry() self.assertEqual(entry["hle_fallback"], True) @@ -767,7 +830,7 @@ class Findings(Workspace): self.assertEqual(len(self.lines()), 2) def test_an_agreeing_run_adds_evidence_only(self) -> None: - verdicts = rules.rule_bios_need(rules.Boot(rules.FAILS, "nothing"), self.entry(), self.RID) + verdicts = rules.rule_need(self.none(rules.HANG), self.present(), self.entry()) self.c.settle("fake", self.content, self.core, 0, self.entry(), verdicts) entry = self.entry() self.assertEqual((entry["required"], entry["hle_fallback"]), (True, False)) @@ -775,7 +838,7 @@ class Findings(Workspace): self.assertEqual(self.lines(), []) def test_a_later_agreement_resolves_the_open_finding(self) -> None: - (verdict,) = rules.rule_bios_need(rules.Boot(rules.INCONCLUSIVE, "?"), self.entry(), self.RID) + (verdict,) = rules.rule_need(self.none(rules.UNKNOWN), self.present(), self.entry()) self.c.settle("fake", self.content, self.core, 0, self.entry(), [verdict]) self.assertEqual(self.lines()[-1]["status"], "open") later = rules.Verdict("inconclusive", "hle_fallback", False, "agrees now", ("2026-10-28-000000000000",)) @@ -789,7 +852,7 @@ class Findings(Workspace): self.assertNotIn("evidence", self.entry(1)) def test_finding_lines_validate(self) -> None: - verdicts = rules.rule_bios_need(rules.Boot(rules.BOOTS, "shown", 4), self.entry(), self.RID) + verdicts = rules.rule_need(self.none(rules.BOOTS), self.present(), self.entry()) self.c.settle("fake", self.content, self.core, 0, self.entry(), verdicts) self.c.write_run_line(self.c.current) try: @@ -802,7 +865,7 @@ class Findings(Workspace): self.assertEqual(list(Draft202012Validator(schema).iter_errors(line)), [], name) def test_the_batch_commit_plan(self) -> None: - verdicts = rules.rule_bios_need(rules.Boot(rules.BOOTS, "shown", 4), self.entry(), self.RID) + verdicts = rules.rule_need(self.none(rules.BOOTS), self.present(), self.entry()) self.c.settle("fake", self.content, self.core, 0, self.entry(), verdicts) self.c.current.touched.add(mr.records_path("fake")) plan = self.c.plan_commits(set()) @@ -937,12 +1000,19 @@ class FakeCorePasses(Workspace): isolations = {rep["isolation"] for r in records for v in r["variants"] for rep in v["repetitions"]} self.assertEqual(isolations, {campaign.core_probe.SANDBOX}) kinds = sorted(r["subject"].split("-")[0] for r in records) - self.assertEqual(kinds, ["alias", "bench", "boot", "files"]) - boot = next(r for r in records if r["subject"].startswith("boot")) - alias = next(r for r in records if r["subject"].startswith("alias")) + self.assertEqual(kinds, ["alias", "files", "negative", "none"]) + by_kind = {r["subject"].split("-")[0]: r for r in records} + self.assertEqual(by_kind["files"]["variants"][0]["verdict"], "boots") + self.assertEqual(by_kind["none"]["variants"][0]["verdict_evidence"]["stages"], ["game"]) + self.assertEqual(by_kind["negative"]["check"]["result"]["verdicts"], + {"bios.bin": "boots", "bios.bin-zeros-4": "hang", + "bios.bin-head-2-of-bios.bin": "boots"}) entry = yaml.safe_load((self.root / "emulators" / "fake.yml").read_text())["files"][0] self.assertEqual(entry["hle_fallback"], True) - self.assertEqual(entry["evidence"], {"aliases": [alias["id"]], "hle_fallback": [boot["id"]]}) + self.assertEqual(entry["validation"], ["size"]) + self.assertEqual(entry["evidence"], { + "aliases": [by_kind["alias"]["id"]], "hle_fallback": [by_kind["none"]["id"]], + "validation": [by_kind["none"]["id"], by_kind["negative"]["id"]]}) statuses = {(f["field"], f["status"]) for f in mr.read_campaign(self.root, mr.FINDINGS)} self.assertEqual(statuses, {("hle_fallback", "applied"), ("required", "open"), ("aliases", "applied")}) @@ -961,9 +1031,23 @@ class FakeCorePasses(Workspace): with rebuilt.open("ab") as handle: handle.write(b"\0") third = self.passing("hle", core=rebuilt) - # BIOS.BIN is declared now, so no alias run follows the boot test. + # BIOS.BIN is declared now, so no alias run follows. self.assertEqual(len(third.run(set(), None).records), 3) + def test_a_size_check_the_profile_lacks_is_written(self) -> None: + profile = self.root / "emulators" / "fake.yml" + profile.write_text(PROFILE.replace(" validation: [size]\n", ""), encoding="utf-8") + self.passing("hle").run(set(), None) + entry = yaml.safe_load(profile.read_text())["files"][0] + self.assertEqual(entry["validation"], ["size"]) + applied = {f["field"] for f in mr.read_campaign(self.root, mr.FINDINGS) if f["status"] == "applied"} + self.assertIn("validation", applied) + + def test_bench_runs_only_when_asked(self) -> None: + self.passing("plain", bench=True).run(set(), None) + subjects = [s.record["subject"] for s in mr.read_records(self.root)] + self.assertEqual(sum(s.startswith("bench") for s in subjects), 1) + def test_no_stand_in_settles_required_and_finds_an_alias(self) -> None: c = self.passing() c.run(set(), None) @@ -982,9 +1066,13 @@ class FakeCorePasses(Workspace): self.assertEqual(left.stdout, b"") checks = c.state["checks"]["fake"] self.assertTrue(checks) - self.assertEqual({entry["status"] for entry in checks.values()}, {"failed"}) - self.assertEqual({entry["reason"] for entry in checks.values()}, {"timeout"}) - self.assertEqual(mr.read_records(self.root), []) + # A run killed after its timeout is a record whose verdict is a hang. + self.assertEqual({entry["status"] for entry in checks.values()}, {"done"}) + records = [s.record for s in mr.read_records(self.root)] + self.assertTrue(records) + self.assertEqual({v["verdict"] for r in records for v in r["variants"]}, {"hang"}) + self.assertEqual({rep["verdict_evidence"]["rule"] for r in records for v in r["variants"] + for rep in v["repetitions"]}, {"timeout"}) def test_missing_content_leaves_the_checks_pending(self) -> None: c = self.passing(romfetch=str(self.tmp / "no-romfetch")) diff --git a/tests/test_core_probe.py b/tests/test_core_probe.py index 0d3c7d358..bf6b00fb4 100644 --- a/tests/test_core_probe.py +++ b/tests/test_core_probe.py @@ -9,6 +9,7 @@ from __future__ import annotations import argparse import contextlib +import hashlib import io import json import os @@ -20,6 +21,7 @@ import tempfile import unittest import unittest.mock from pathlib import Path +from typing import ClassVar REPO_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(REPO_ROOT / "scripts")) @@ -42,6 +44,7 @@ FAKE_CORE = r""" #include #include #include +#include struct retro_system_info { const char *n, *v, *e; bool full, block; }; struct retro_game_geometry { unsigned bw, bh, mw, mh; float a; }; struct retro_system_timing { double fps, rate; }; @@ -78,11 +81,15 @@ bool retro_load_game(const void *g) { if (env(27, &cb)) { logger = cb; cb.log(1, "card %s byte %d\n", f ? "present" : "absent", (int)card); - cb.log(1, "tag %s\n", tag ? tag : "unset"); } + cb.log(1, "tag %s\n", tag ? tag : "unset"); + if (card == 0xEE) cb.log(3, "BIOS not found: card.bin\n"); } return true; } void retro_unload_game(void) {} void retro_run(void) { int16_t s[4] = {1, 2, 3, 4}; + const char *end = getenv("CORE_PROBE_TEST_END"); + if (end && frames == 1 && !strcmp(end, "crash")) { if (logger.log) logger.log(3, "about to fall\n"); raise(SIGSEGV); } + if (end && frames == 1 && !strcmp(end, "hang")) for (;;) {} if (frames++ == 0 && logger.log) logger.log(1, "ports %u\n", ports); input(0, 1, 0, 8); for (int i = 0; i < 8; i++) pixels[i] = card; @@ -177,6 +184,127 @@ class ParsingTests(unittest.TestCase): self.assertEqual([p.name for p in core_probe.cue_tracks(cue)], ["a b.bin", "c.bin"]) +def frames(*pairs: tuple[int, str]) -> list[list]: + return [[frame, digest] for frame, digest in pairs] + + +class OracleTests(unittest.TestCase): + """judge() turns a repetition into a verdict from positive evidence only.""" + + TITLE, LOGO, ERROR, BLACK, OTHER = ("1" * 40, "2" * 40, "3" * 40, "4" * 40, "5" * 40) + ORACLE: ClassVar[dict] = {"references": [{"sha1": "2" * 40, "stage": "bios", "shows": "the console logo"}, + {"sha1": "1" * 40, "stage": "game", "shows": "the title"}], + "error_patterns": [r"BIOS not found"], + "error_screens": [{"sha1": "3" * 40, "shows": "the core's missing-BIOS message"}]} + + def rep(self, hashes: list, logs: list[str] = (), uniform: list[str] = (), loaded: bool = True, + **extra: object) -> dict: + return {"loaded": loaded, "library_name": "Fake", "frames": 100, "warmup": 0, + "logs": [{"frame": 0, "level": "error", "text": text} for text in logs], + "observables": {"frame_hashes": hashes, "uniform_frames": list(uniform)}, **extra} + + def test_a_reference_frame_is_a_boot_with_its_stages(self) -> None: + verdict, evidence = core_probe.judge( + self.rep(frames((0, self.BLACK), (10, self.LOGO), (20, self.TITLE)), + logs=["BIOS not found"]), self.ORACLE) + self.assertEqual(verdict, "boots") + self.assertEqual((evidence["rule"], evidence["frame"], evidence["stages"]), + ("reference_frame", 10, ["bios", "game"])) + + def test_an_error_pattern_or_screen_without_a_reference_is_a_bios_error(self) -> None: + verdict, evidence = core_probe.judge( + self.rep(frames((0, self.OTHER)), logs=["x", "BIOS not found: scph5501.bin"]), self.ORACLE) + self.assertEqual((verdict, evidence["rule"], evidence["log"]), + ("bios_error", "error_pattern", "BIOS not found: scph5501.bin")) + verdict, evidence = core_probe.judge(self.rep(frames((0, self.ERROR))), self.ORACLE) + self.assertEqual((verdict, evidence["rule"], evidence["frame"]), ("bios_error", "error_screen", 0)) + self.assertEqual(core_probe.judge(self.rep([], logs=["BIOS not found"], loaded=False), + self.ORACLE)[0], "bios_error") + + def test_process_status_decides_crash_and_hang_first(self) -> None: + crashed = self.rep(frames((0, self.TITLE)), exit={"status": -11, "signal": "SIGSEGV"}) + self.assertEqual(core_probe.judge(crashed, self.ORACLE)[0], "crash") + killed = self.rep([], exit={"status": None, "signal": "SIGKILL", "timeout_s": 5}) + self.assertEqual(core_probe.judge(killed, self.ORACLE), ( + "hang", {"rule": "timeout", "detail": "killed after 5 s", "exit_status": None, + "signal": "SIGKILL"})) + + def test_a_one_colour_second_half_is_a_hang(self) -> None: + run = self.rep(frames((0, self.OTHER), (10, self.OTHER), (20, self.BLACK), (30, self.BLACK)), + uniform=[self.BLACK]) + self.assertEqual(core_probe.judge(run, self.ORACLE)[1]["rule"], "uniform_tail") + lit = self.rep(frames((0, self.BLACK), (10, self.BLACK), (20, self.OTHER), (30, self.BLACK)), + uniform=[self.BLACK]) + self.assertEqual(core_probe.judge(lit, self.ORACLE)[0], "unknown") + + def test_frames_that_differ_are_never_a_boot(self) -> None: + verdict, evidence = core_probe.judge(self.rep(frames((0, self.OTHER), (10, "6" * 40))), + self.ORACLE) + self.assertEqual((verdict, evidence["rule"]), ("unknown", "no_evidence")) + verdict, evidence = core_probe.judge(self.rep(frames((0, self.TITLE))), + {**self.ORACLE, "references": []}) + self.assertEqual(verdict, "unknown") + self.assertIn("no reference frame is known", evidence["detail"]) + self.assertEqual(core_probe.judge(self.rep([], loaded=False), self.ORACLE)[0], "unknown") + + def test_a_variant_needs_every_repetition_to_agree(self) -> None: + boot = {"verdict": "boots", "verdict_evidence": {"rule": "reference_frame", "detail": "d", + "frame": 3, "stages": ["game"]}} + verdict, evidence = core_probe.variant_verdict([boot, boot]) + self.assertEqual((verdict, evidence["rule"], evidence["stages"]), + ("boots", "repetitions_agree", ["game"])) + hang = {"verdict": "hang", "verdict_evidence": {"rule": "uniform_tail", "detail": "d"}} + verdict, evidence = core_probe.variant_verdict([boot, hang]) + self.assertEqual((verdict, evidence["repetitions"]), ("unknown", ["boots", "hang"])) + + def test_oracle_entries_are_picked_by_content_and_core(self) -> None: + data = {"error_patterns": [{"pattern": "shared"}], + "cores": {"Fake": {"error_patterns": [{"pattern": "own"}], + "error_screens": [{"sha1": self.ERROR, "shows": "s"}]}}, + "content": [{"name": "disc", "sha1": "a" * 40, "references": { + "Fake": {"frames": [{"sha1": self.TITLE, "stage": "game", "shows": "t"}]}, + "Other": {"frames": [{"sha1": self.LOGO, "stage": "bios", "shows": "l"}]}}}]} + picked = core_probe.oracle_for(data, "a" * 40, "Fake") + self.assertEqual(picked["references"], [{"sha1": self.TITLE, "stage": "game", "shows": "t"}]) + self.assertEqual(picked["error_patterns"], ["shared", "own"]) + self.assertEqual(core_probe.oracle_for(data, "b" * 40, "Fake")["references"], []) + self.assertEqual(core_probe.oracle_for(data, "a" * 40, "Nobody")["error_patterns"], ["shared"]) + + def test_the_repository_oracle_is_well_formed(self) -> None: + import yaml + data = yaml.safe_load(core_probe.DEFAULT_ORACLE.read_text(encoding="utf-8")) + patterns = list(data.get("error_patterns") or ()) + for core in (data.get("cores") or {}).values(): + patterns += core.get("error_patterns") or [] + for screen in core.get("error_screens") or (): + self.assertRegex(screen["sha1"], r"^[0-9a-f]{40}$") + for pattern in patterns: + self.assertTrue(pattern.get("source_ref") or pattern.get("reason"), pattern) + core_probe.re.compile(pattern["pattern"]) + for item in data.get("content") or (): + self.assertRegex(item["sha1"], r"^[0-9a-f]{40}$") + for block in item["references"].values(): + self.assertRegex(block["recorded_from"], r"^\d{4}-\d{2}-\d{2}-[0-9a-f]{12}$") + stages = {frame["stage"] for frame in block["frames"]} + self.assertIn("game", stages, item["name"]) + for frame in block["frames"]: + self.assertRegex(frame["sha1"], r"^[0-9a-f]{40}$") + self.assertIn(frame["stage"], ("bios", "game")) + self.assertTrue(frame["shows"]) + + def test_made_files(self) -> None: + zeros = core_probe.Synthetic.parse("@zeros:16") + self.assertEqual((zeros.label("bios.bin"), zeros.source), ("bios.bin-zeros-16", None)) + head = core_probe.Synthetic.parse("@head:8:/x/scph5501.bin") + self.assertEqual(head.label("bios.bin"), "bios.bin-head-8-of-scph5501.bin") + for bad in ("@zeros:", "@zeros:8:/x", "@head:8", "@ones:8", "@head:x:/y"): + with self.assertRaises(argparse.ArgumentTypeError): + core_probe.Synthetic.parse(bad) + self.assertEqual(core_probe.exit_description(-11), {"status": -11, "signal": "SIGSEGV"}) + self.assertEqual(core_probe.exit_description(139), {"status": 139, "signal": "SIGSEGV"}) + self.assertEqual(core_probe.exit_description(1), {"status": 1, "signal": None}) + + class ResultTests(unittest.TestCase): @staticmethod def variant(label: str, value: str, loaded: bool = True) -> dict: @@ -290,6 +418,76 @@ class FakeCoreTests(unittest.TestCase): schema = json.loads((REPO_ROOT / "schemas" / "core-probe-record.schema.json").read_text()) self.assertEqual(list(Draft202012Validator(schema).iter_errors(line)), []) + def oracle_file(self) -> Path: + title = hashlib.sha1(struct.pack("<8H", *[1] * 8)).hexdigest() + content = self.tmp / "game.bin" + content.write_bytes(b"game") + path = self.tmp / "oracle.yml" + path.write_text( + "error_patterns:\n - pattern: 'BIOS not found'\n reason: test\n" + "cores:\n Fake:\n error_patterns: []\n error_screens: []\n" + f"content:\n - name: game\n sha1: {core_probe.file_digest(content)}\n" + " references:\n Fake:\n recorded_from: 2026-10-27-000000000000\n" + f" frames:\n - sha1: {title}\n stage: game\n" + " shows: card one\n", encoding="utf-8") + return path + + def test_verdicts_of_a_negative_check(self) -> None: + oracle = self.oracle_file() + error = self.tmp / "error.bin" + error.write_bytes(b"\xee") + code, record = self.probe(["check", "--core", str(self.core), "--frames", "4", + "--frame-hash-every", "1", "--content", str(self.tmp / "game.bin"), + "--oracle", str(oracle), "--vary", "card.bin=", + "--vary", f"card.bin={self.cards[0]}", "--vary", f"card.bin={error}", + "--vary", "card.bin=@zeros:1", + "--vary", f"card.bin=@head:0:{self.cards[0]}", + "--observable", "frame_hashes", "--claim", "c", "--method", "m", + "--no-trace", "--repeat", "2"]) + self.assertEqual(code, 0) + verdicts = record["check"]["result"]["verdicts"] + self.assertEqual(verdicts, {"card.bin-absent": "hang", "card0.bin": "boots", + "error.bin": "bios_error", "card.bin-zeros-1": "hang", + "card.bin-head-0-of-card0.bin": "hang"}) + boot = record["variants"][1] + self.assertEqual((boot["verdict_evidence"]["rule"], boot["verdict_evidence"]["stages"]), + ("repetitions_agree", ["game"])) + self.assertEqual(boot["repetitions"][0]["verdict_evidence"]["rule"], "reference_frame") + self.assertEqual(record["variants"][3]["system_files"], + {"card.bin": hashlib.sha1(b"\x00").hexdigest()}) + self.assertEqual(record["inputs"]["oracle"]["error_patterns"], ["BIOS not found"]) + self.assertEqual(record["record_version"], 4) + self.assert_schema(record) + + def test_a_crash_and_a_hang_are_verdicts(self) -> None: + oracle = self.oracle_file() + code, record = self.probe(["run", "--core", str(self.core), "--frames", "4", + "--frame-hash-every", "1", "--oracle", str(oracle), + "--env", "CORE_PROBE_TEST_END=crash"]) + self.assertEqual(code, 0) + crashed = record["variants"][0] + self.assertEqual((crashed["verdict"], crashed["repetitions"][0]["exit"]["signal"]), + ("crash", "SIGSEGV")) + self.assertEqual(crashed["repetitions"][0]["logs"][-1]["text"], "about to fall") + self.assertEqual(crashed["repetitions"][0]["observables"]["frame_hashes"][0][0], 0) + self.assert_schema(record) + code, record = self.probe(["run", "--core", str(self.core), "--frames", "4", + "--frame-hash-every", "1", "--oracle", str(oracle), "--timeout", "3", + "--env", "CORE_PROBE_TEST_END=hang", "--no-trace"]) + self.assertEqual(code, 0) + self.assertEqual((record["variants"][0]["verdict"], + record["variants"][0]["repetitions"][0]["verdict_evidence"]["rule"]), + ("hang", "timeout")) + + def assert_schema(self, record: dict) -> None: + try: + from jsonschema import Draft202012Validator + except ImportError: + return + schema = json.loads((REPO_ROOT / "schemas" / "core-probe-record.schema.json").read_text()) + sealed = measurement_records.seal({**record, "emulator": "fake", "subject": "s"}) + self.assertEqual([e.message for e in Draft202012Validator(schema).iter_errors(sealed)], []) + def test_second_run_is_appended_beside_the_first(self) -> None: record_dir = checkout(self.tmp / "appended") argv = ["run", "--core", str(self.core), "--frames", "1", "--no-trace", "--max-load", "1000",