feat: judge recorded runs again offline

core_probe rejudge appends, for each live record the oracle now judges otherwise, a record with the same runs judged again: verdicts, bios_used and inputs.oracle recomputed, supersedes the earlier one, judged_at and judged_by beside recorded_at. Records of another probe are left to it. The campaign reads a superseded record as its successor, and a file the log shows refused or loaded settles validation with no BIOS screen.

scripts/campaign.py

@@ -1281,15 +1281,21 @@ class Campaign:
         self.settle(unit.emulator, content, core, index, entry, [verdict])
 
     def read_record(self, emulator: str, rid: str | None) -> dict | None:
+        """A record by its id, or the record that supersedes it: a run
+        judged again offline is read as judged last."""
         if not rid:
             return None
         path = self.paths.root / mr.records_path(emulator)
         if not path.is_file():
             return None
-        for line in path.read_text(encoding="utf-8").splitlines():
-            if f'"id":"{rid}"' in line:
-                return json.loads(line)
-        return None
+        records = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
+        by_id = {r["id"]: r for r in records}
+        successor = {r["supersedes"]: r["id"] for r in records if r.get("supersedes")}
+        seen = set()
+        while rid in successor and rid not in seen:
+            seen.add(rid)
+            rid = successor[rid]
+        return by_id.get(rid)
 
     def judged(self, emulator: str, rid: str) -> bool:
         """True for a record the boot oracle judged: only those prove a field."""

scripts/campaign_rules.py

@@ -25,8 +25,9 @@ name (zeros of its size, its first half, another region's dump).
   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
+core reports a BIOS error, when its log says it ran its built-in BIOS
+(bios_used builtin), or when it boots without the BIOS screens the right
+file shows. It is used when the log names the file it loaded. 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
@@ -64,6 +65,7 @@ class Run:
     record_id: str
     label: str
     stages: tuple[str, ...] = ()
+    bios_used: str = "unknown"
 
     @property
     def shows_bios(self) -> bool:
@@ -101,7 +103,8 @@ def run_of(record: dict, label: str) -> Run:
     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 ()))
+    return Run(item.get("verdict", UNKNOWN), record["id"], label, tuple(evidence.get("stages") or ()),
+               item.get("bios_used", "unknown"))
 
 
 def ids(*runs: Run) -> tuple[str, ...]:
@@ -138,8 +141,9 @@ def rule_need(none: Run, present: Run, entry: dict, used: bool = True) -> list[V
     if none.verdict == BOOTS and not used:
         return []
     if none.verdict == BOOTS:
+        on = " on its built-in BIOS" if none.bios_used == "builtin" else ""
         out = [Verdict("boots_without_file", "hle_fallback", True,
-                       f"the core boots the content with no BIOS at all, {name} included",
+                       f"the core boots the content{on} with no BIOS at all, {name} included",
                        ids(none), True)]
         if entry.get("required") is True:
             out.append(Verdict(
@@ -167,6 +171,11 @@ def file_use(wrong: Run, none: Run, present: Run) -> str | None:
         return None
     if wrong.verdict == BIOS_ERROR:
         return REFUSED
+    # The log names the BIOS the core ran: that settles it before any screen.
+    if wrong.verdict == BOOTS and wrong.bios_used == "builtin":
+        return REFUSED
+    if wrong.verdict == BOOTS and wrong.bios_used.startswith("staged:"):
+        return USED
     if wrong.verdict == BOOTS:
         if wrong.shows_bios:
             return USED

scripts/core_probe.py

@@ -1110,6 +1110,16 @@ def build_parser() -> argparse.ArgumentParser:
     check.add_argument("--claim", required=True, help="the statement under test")
     check.add_argument("--method", required=True, help="how the run tests it")
     check.add_argument("--repeat", type=int, default=2, help="runs of each variant, each in a fresh process")
+    rejudge = sub.add_parser("rejudge", help="judge recorded runs again against the current oracle, "
+                                               "offline: no core runs")
+    rejudge.add_argument("--record", metavar="DIR", default=str(ROOT),
+                         help="the checkout whose measurements/ are judged again (default: this one)")
+    rejudge.add_argument("--emulator", action="append", default=[],
+                         help="only these emulators' records (default: every one)")
+    rejudge.add_argument("--oracle", default=None, metavar="FILE",
+                         help="the oracle to judge against (default: tests/campaign/oracle.yml)")
+    rejudge.add_argument("--dry-run", action="store_true",
+                         help="print what would be appended, write nothing")
     worker = sub.add_parser("_worker", help="internal: one run in this process, started by the others")
     worker.add_argument("spec", help="JSON file describing the run")
     worker.add_argument("out", help="JSON file the result is written to")
@@ -1582,8 +1592,89 @@ def _ensure(path: Path) -> Path:
     return path
 
 
+def rejudged(record: dict, data: dict, identity: dict | None, now: str) -> dict | None:
+    """The record judged again from its own runs, superseding it, or None
+    when the judgement it carries is the one the oracle gives now.
+
+    Nothing of the runs changes: logs, frames, files and timings are the
+    record's. The verdicts, their evidence, bios_used and inputs.oracle are
+    computed again, and judged_at and judged_by say when and by what.
+    """
+    import copy  # noqa: PLC0415
+
+    # Another probe (standalone_probe, pack_probe) judges with its own rules.
+    if record.get("record_version", 0) < 4 or record["frontend"].get("script") != "scripts/core_probe.py":
+        return None
+    out = copy.deepcopy(record)
+    content_sha1 = ((out["inputs"].get("content") or {}).get("sha1"))
+    for variant in out["variants"]:
+        for rep in variant["repetitions"]:
+            judge_repetition(rep, oracle_for(data, content_sha1, rep.get("library_name")))
+        judge_variant(variant)
+    first = out["variants"][0]["repetitions"][0]
+    out["inputs"]["oracle"] = ({**identity, **oracle_for(data, content_sha1, first.get("library_name"))}
+                               if identity else None)
+    if out.get("check"):
+        out["check"]["result"]["verdicts"] = {v["label"]: v["verdict"] for v in out["variants"]}
+
+    def judgement(item: dict) -> list:
+        return [[v.get("verdict"), v.get("verdict_evidence"), v.get("bios_used"),
+                 [[r.get("verdict"), r.get("verdict_evidence"), r.get("bios_used")] for r in v["repetitions"]]]
+                for v in item["variants"]]
+    if record.get("record_version") == RECORD_VERSION_NOW() and judgement(out) == judgement(record):
+        return None
+    for key in ("id", "migrated_from_version"):
+        out.pop(key, None)
+    out["record_version"] = RECORD_VERSION_NOW()
+    out["supersedes"] = record["id"]
+    out["judged_at"] = now
+    ident = frontend_identity()
+    out["judged_by"] = {"script": ident["script"], "commit": ident["commit"],
+                        "script_sha256": ident["script_sha256"]}
+    return out
+
+
+def RECORD_VERSION_NOW() -> int:  # noqa: N802
+    from measurement_records import RECORD_VERSION  # noqa: PLC0415
+
+    return RECORD_VERSION
+
+
+def rejudge_records(args: argparse.Namespace) -> int:
+    """Append, for each live record whose judgement the oracle now gives
+    otherwise, the record judged again, superseding it."""
+    from measurement_records import line_text, live, read_records, records_path, seal  # noqa: PLC0415
+
+    root = Path(args.record)
+    data, identity = load_oracle(Path(args.oracle) if args.oracle else DEFAULT_ORACLE)
+    now = dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()
+    stored = read_records(root)
+    alive = {r["id"] for r in live(s.record for s in stored)}
+    written = 0
+    for entry in stored:
+        if entry.record["id"] not in alive or (args.emulator and entry.emulator not in args.emulator):
+            continue
+        new = rejudged(entry.record, data, identity, now)
+        if new is None:
+            continue
+        sealed = seal(new)
+        before = {v["label"]: v.get("verdict") for v in entry.record["variants"]}
+        after = {v["label"]: f"{v['verdict']} ({v['bios_used']})" for v in sealed["variants"]}
+        print(json.dumps({"emulator": entry.emulator, "supersedes": entry.record["id"], "id": sealed["id"],
+                          "before": before, "after": after}))
+        written += 1
+        if not args.dry_run:
+            with (root / records_path(entry.emulator)).open("a", encoding="utf-8") as handle:
+                handle.write(line_text(sealed) + "\n")
+    print(f"{written} records judged again" + (" (dry run, nothing written)" if args.dry_run else ""),
+          file=sys.stderr)
+    return 0
+
+
 def main(argv: Sequence[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
+    if args.mode == "rejudge":
+        return rejudge_records(args)
     if args.mode == "_worker":
         spec = json.loads(Path(args.spec).read_text(encoding="utf-8"))
         Path(args.out).write_text(json.dumps(run_worker(spec)), encoding="utf-8")

tests/test_campaign.py

@@ -192,6 +192,20 @@ class RuleTable(unittest.TestCase):
         no_logo = run(rules.BOOTS, stages=("game",))
         self.assertIsNone(rules.file_use(run(rules.BOOTS, stages=("game",)), NONE_BOOTS, no_logo))
 
+    def test_the_bios_the_log_names_settles_the_file_use(self) -> None:
+        present = run(rules.BOOTS, stages=("game",))  # no BIOS screen to compare with
+        fallback = rules.Run(rules.BOOTS, "r", "zeros", ("game",), "builtin")
+        loaded = rules.Run(rules.BOOTS, "r", "other", ("game",), "staged:bios.bin")
+        self.assertEqual(rules.file_use(fallback, NONE_BOOTS, present), rules.REFUSED)
+        self.assertEqual(rules.file_use(loaded, NONE_BOOTS, present), rules.USED)
+        self.assertIsNone(rules.file_use(run(rules.BOOTS, stages=("game",)), NONE_BOOTS, present))
+        record = {"id": "x", "variants": [{"label": "a", "verdict": "boots", "bios_used": "builtin",
+                                           "verdict_evidence": {"rule": "r", "detail": "d"}}]}
+        self.assertEqual(rules.run_of(record, "a").bios_used, "builtin")
+        builtin_none = rules.Run(rules.BOOTS, "n", "none", ("game",), "builtin")
+        (hle,) = rules.rule_need(builtin_none, PRESENT, {"name": "a"})
+        self.assertIn("on its built-in BIOS", hle.statement)
+
     def test_validation_agrees_or_adds_size(self) -> None:
         refused = run(rules.BOOTS, stages=("game",))
         used = run(rules.HANG)
@@ -411,6 +425,15 @@ class StateAndResume(Workspace):
         self.assertTrue(campaign.should_run({**done, "status": "pending"}, "a", "d"))
         self.assertFalse(campaign.should_run({**done, "status": "failed"}, "a", "d"))
 
+    def test_a_record_judged_again_is_read_in_its_place(self) -> None:
+        lines = self.root / mr.records_path("fake")
+        lines.parent.mkdir(parents=True, exist_ok=True)
+        old = {"id": "2026-10-27-aaaaaaaaaaaa", "supersedes": None, "verdict": "old"}
+        new = {"id": "2026-10-28-bbbbbbbbbbbb", "supersedes": old["id"], "verdict": "new"}
+        lines.write_text("\n".join(json.dumps(r) for r in (old, new)) + "\n", encoding="utf-8")
+        self.assertEqual(self.make().read_record("fake", old["id"])["verdict"], "new")
+        self.assertIsNone(self.make().read_record("fake", "2026-10-27-cccccccccccc"))
+
     def test_only_its_own_oracle_entries_make_a_check_run_again(self) -> None:
         key = [hashlib.sha1(b"GAME").hexdigest(), "Fake"]
         before = self.make().oracle_digest(key)

tests/test_core_probe.py

@@ -565,6 +565,35 @@ class FakeCoreTests(unittest.TestCase):
         self.assertEqual({v["bios_used"] for v in record["variants"]}, {"unknown"})
         self.assert_schema(record)
 
+    def test_recorded_runs_are_judged_again_offline(self) -> None:
+        oracle = self.oracle_file()
+        record_dir = checkout(self.tmp / "rejudge")
+        code, first = self.probe(["run", "--core", str(self.core), "--frames", "4", "--frame-hash-every", "1",
+                                  "--content", str(self.tmp / "game.bin"), "--oracle", str(oracle),
+                                  "--no-trace", "--record", str(record_dir), "--max-load", "1000"])
+        self.assertEqual((code, first["variants"][0]["bios_used"]), (0, "unknown"))
+        # The core's fallback line becomes known: the same runs now name the BIOS.
+        oracle.write_text(oracle.read_text().replace(
+            "    error_patterns: []\n", "    error_patterns: []\n    fallback_patterns:\n"
+            "      - pattern: '^card absent'\n        reason: test\n", 1), encoding="utf-8")
+        argv = ["rejudge", "--record", str(record_dir), "--oracle", str(oracle)]
+        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+            self.assertEqual(core_probe.main(argv), 0)
+        lines = [json.loads(line) for line in (record_dir / "measurements" / "fake.jsonl").read_text().splitlines()]
+        self.assertEqual(len(lines), 2)
+        again = lines[1]
+        self.assertEqual((again["supersedes"], again["record_version"]), (first["id"], 5))
+        self.assertEqual(again["variants"][0]["bios_used"], "builtin")
+        self.assertEqual(again["variants"][0]["repetitions"][0]["logs"], first["variants"][0]["repetitions"][0]["logs"])
+        self.assertEqual(again["recorded_at"], first["recorded_at"])
+        self.assertIn("judged_at", again)
+        self.assertEqual(measurement_records.check_line(again), [])
+        self.assert_schema(again)
+        # Judged once: a second pass has nothing to add.
+        with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+            core_probe.main(argv)
+        self.assertEqual(len((record_dir / "measurements" / "fake.jsonl").read_text().splitlines()), 2)
+
     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",