chore: merge the per-run probe lock

tests/__init__.py

@@ -16,6 +16,11 @@ import os
 from pathlib import Path
 
 os.environ.setdefault("RETROBIOS_OFFLINE", "1")
+# A test never waits on, nor blocks, the measurements running on the host.
+os.environ.setdefault(
+    "RETROBIOS_PROBE_LOCK",
+    str(Path(__file__).resolve().parent.parent / "tmp" / "tests" / "probe.lock"),
+)
 
 (Path(__file__).resolve().parent.parent / "tmp" / "tests").mkdir(
     parents=True, exist_ok=True

tests/campaign/oracle.yml

@@ -14,6 +14,9 @@
 #               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.
+# bios_patterns per core, log lines by which it says it loaded the console's
+#               own BIOS: they stand for the bios-stage screens of a core
+#               that hides them by default.
 #
 # 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
@@ -47,7 +50,11 @@ cores:
       Without a BIOS, or with one of the wrong size, it logs "Using HLE BIOS"
       and runs its own (frontend/libretro.c:3893-3899 at c8816799). The BIOS
       boot logo is off by default (pcsx_rearmed_show_bios_bootlogo), so its
-      runs reach no bios-stage screen.
+      runs reach no bios-stage screen: the line it logs once a file passes
+      its signature check and is kept stands for them.
+    bios_patterns:
+      - pattern: '^found (US|EU|JP) BIOS file, crc32 '
+        source_ref: frontend/libretro.c:3646-3669 at c8816799
     error_patterns: []
     error_screens: []
   "PCSX1":

tests/test_campaign.py

@@ -155,6 +155,14 @@ class RuleTable(unittest.TestCase):
         optional = rules.rule_need(NONE_BOOTS, PRESENT, {"name": "a", "required": False})
         self.assertEqual([v.field for v in optional], ["hle_fallback"])
 
+    def test_a_file_the_content_never_uses_settles_nothing(self) -> None:
+        unused = run(rules.BOOTS, "bios.bin", ("game",))
+        self.assertFalse(rules.read(NONE_BOOTS, unused, unused, unused))
+        self.assertEqual(rules.rule_need(NONE_BOOTS, unused, self.ENTRY, used=False), [])
+        # A wrong file that hangs where no file boots: the core reads it.
+        self.assertTrue(rules.read(NONE_BOOTS, unused, run(rules.HANG), unused))
+        self.assertTrue(rules.read(NONE_BOOTS, PRESENT, None, None))
+
     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)
@@ -397,6 +405,19 @@ 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_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)
+        data = yaml.safe_load(self.oracle.read_text())
+        data["content"][0]["references"]["Other"] = {"recorded_from": "2026-10-27-000000000000",
+                                                     "frames": [{"sha1": "a" * 40, "stage": "game",
+                                                                 "shows": "x"}]}
+        self.oracle.write_text(yaml.safe_dump(data), encoding="utf-8")
+        self.assertEqual(self.make().oracle_digest(key), before)
+        data["cores"] = {"Fake": {"error_patterns": [{"pattern": "BIOS", "reason": "r"}]}}
+        self.oracle.write_text(yaml.safe_dump(data), encoding="utf-8")
+        self.assertNotEqual(self.make().oracle_digest(key), before)
+
     def test_evidence_does_not_change_the_digest(self) -> None:
         profile = yaml.safe_load(PROFILE)
         before = campaign.profile_digest(profile)
@@ -1105,6 +1126,52 @@ class FakeCorePasses(Workspace):
         self.assertEqual({rep["verdict_evidence"]["rule"] for r in records for v in r["variants"]
                           for rep in v["repetitions"]}, {"timeout"})
 
+    def test_two_passes_interleave_their_runs(self) -> None:
+        """Each core_probe takes the host lock for its own run: a pass never
+        holds it between runs, so a second pass runs between them."""
+        import threading
+        import time
+
+        spans: dict[str, list[tuple[float, float]]] = {"a": [], "b": []}
+        other = Workspace(methodName="setUp")
+        other.setUp()
+        self.addCleanup(other.tearDown)
+        passes = []
+        for name, workspace in (("a", self), ("b", other)):
+            c = FakeCorePasses.passing(workspace, "hle") if workspace is self else self.passing_in(workspace)
+            real = c.host.run
+
+            def timed(argv: list[str], *args: object, _name: str = name, _real=real, **kwargs: object):
+                start = time.monotonic()
+                try:
+                    return _real(argv, *args, **kwargs)
+                finally:
+                    spans[_name].append((start, time.monotonic()))
+            c.host.run = timed
+            passes.append(c)
+        threads = [threading.Thread(target=c.run, args=(set(), None)) for c in passes]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        ends = {name: sorted(end for _, end in runs) for name, runs in spans.items()}
+        self.assertGreater(len(ends["a"]), 2)
+        self.assertGreater(len(ends["b"]), 2)
+        # Neither pass finished all its runs before the other started finishing its own.
+        self.assertLess(ends["b"][0], ends["a"][-1])
+        self.assertLess(ends["a"][0], ends["b"][-1])
+
+    def passing_in(self, workspace: Workspace) -> campaign.Campaign:
+        workspace.put_game()
+        host = campaign.Host(loadavg=lambda: (0.0, 0.0, 0.0), free_gb=lambda path: 500.0,
+                             cpu_count=1_000_000)
+        c = campaign.Campaign(workspace.paths, campaign.Settings(
+            check_repeat=2, probe_timeout=30, delays=(1,), echo=False), host,
+            lock_path=workspace.tmp / "auto.lock")
+        so = self.cores["hle"]
+        c.unit_cores = lambda *a, **k: [campaign.Core("fake", so, campaign.file_sha256(so))]
+        return c
+
     def test_missing_content_leaves_the_checks_pending(self) -> None:
         c = self.passing(romfetch=str(self.tmp / "no-romfetch"))
         shutil.rmtree(self.game)

tests/test_core_probe.py

@@ -211,6 +211,13 @@ class OracleTests(unittest.TestCase):
         self.assertEqual((evidence["rule"], evidence["frame"], evidence["stages"]),
                          ("reference_frame", 20, ["bios", "game"]))
 
+    def test_a_bios_load_line_stands_for_hidden_bios_screens(self) -> None:
+        oracle = {**self.ORACLE, "bios_patterns": [r"^found US BIOS file"]}
+        verdict, evidence = core_probe.judge(
+            self.rep(frames((0, self.TITLE)), logs=["found US BIOS file, crc32 1: x"]), oracle)
+        self.assertEqual((verdict, evidence["stages"]), ("boots", ["bios", "game"]))
+        self.assertEqual(core_probe.judge(self.rep(frames((0, self.TITLE))), oracle)[1]["stages"], ["game"])
+
     def test_the_bios_screens_alone_are_not_a_boot(self) -> None:
         # A console that shows its logo and then refuses the disc never reached the game.
         verdict, evidence = core_probe.judge(
@@ -275,6 +282,7 @@ class OracleTests(unittest.TestCase):
         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(picked["bios_patterns"], [])
         self.assertEqual(core_probe.oracle_for(data, "b" * 40, "Fake")["references"], [])
         self.assertEqual(core_probe.oracle_for(data, "a" * 40, "Nobody")["error_patterns"], ["shared"])
 
@@ -283,7 +291,7 @@ class OracleTests(unittest.TestCase):
         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 []
+            patterns += (core.get("error_patterns") or []) + (core.get("bios_patterns") or [])
             for screen in core.get("error_screens") or ():
                 self.assertRegex(screen["sha1"], r"^[0-9a-f]{40}$")
         for pattern in patterns:
@@ -496,6 +504,32 @@ class FakeCoreTests(unittest.TestCase):
         sealed = measurement_records.seal({**record, "emulator": "fake", "subject": "s"})
         self.assertEqual([e.message for e in Draft202012Validator(schema).iter_errors(sealed)], [])
 
+    def test_a_run_waits_for_the_host_lock_and_frees_it(self) -> None:
+        import fcntl
+        import time
+
+        lock = self.tmp / "host.lock"
+        argv = [sys.executable, str(REPO_ROOT / "scripts" / "core_probe.py"), "run", "--core",
+                str(self.core), "--frames", "1", "--no-trace", "--work-dir", str(self.tmp / "work")]
+        env = {**os.environ, "RETROBIOS_PROBE_LOCK": str(lock)}
+        with lock.open("a") as held:
+            fcntl.flock(held, fcntl.LOCK_EX)
+            waiting = subprocess.Popen(argv, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
+                                       text=True)
+            time.sleep(1.5)
+            self.assertIsNone(waiting.poll(), "the run went ahead while another held the lock")
+            # A caller that holds the lock for this run says so.
+            held_run = subprocess.run(argv, env={**env, "RETROBIOS_PROBE_LOCK_HELD": "1"},
+                                      capture_output=True, text=True, timeout=60, check=False)
+            self.assertEqual(held_run.returncode, 0, held_run.stderr)
+            fcntl.flock(held, fcntl.LOCK_UN)
+        _, err = waiting.communicate(timeout=60)
+        self.assertEqual(waiting.returncode, 0, err)
+        self.assertIn("another measurement is running", err)
+        # Released at the end of the run: the lock is free again.
+        with lock.open("a") as again:
+            fcntl.flock(again, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
     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",

tests/test_measurements.py

@@ -220,7 +220,8 @@ class ReadAndIndex(Workspace):
         keys = {"emulator", "emulator_name", "profiled", "systems", "system_names", "subject", "kind",
                 "claim", "method", "observable", "variable", "status", "statement", "content",
                 "variants", "repetitions", "flags", "date", "host", "core", "frontend_commit", "id",
-                "runs", "record_url", "sqlite", "verdicts"}
+                "runs", "record_url", "sqlite", "verdicts", "title", "host_short", "groups",
+                "proves", "timing", "setup"}
         for item in index["items"]:
             self.assertEqual(set(item), keys)
         alpha = {i["subject"]: i for i in index["items"] if i["emulator"] == "alpha"}
@@ -247,11 +248,9 @@ class ReadAndIndex(Workspace):
         self.assertNotIn("differ", statement)
         # A check on another observable keeps its comparison.
         self.assertEqual(sm.result_of(full_record("alpha", "c", "2026-10-06T10:00:00+00:00"))[0], "differs")
-        item = {"claim": "c", "runs": 1, "core": {"library": "Fake 1", "commit": None}, "status": "mixed",
-                "statement": "", "verdicts": sm.verdicts_of(record), "method": "m", "host": "h",
-                "date": "2026-10-06", "variants": 2, "repetitions": 2, "flags": 0, "record_url": "u"}
-        text = "\n".join(sm.tested_item(item))
-        self.assertIn("v1: `hang`", text)
+        groups = sm.variant_groups(record)
+        self.assertEqual([(g["variant"], g["result"], g["group"]) for g in groups],
+                         [("v0", "works", 0), ("v1", "hang", 1)])
 
     def test_links_name_the_last_commit_that_touched_the_records(self) -> None:
         first = self.repository()
@@ -431,11 +430,118 @@ class SitePages(unittest.TestCase):
         self.assertEqual(sm.tested_section("beetle_saturn", []), [])
         text = "\n".join(sm.tested_section("beetle_saturn", [self.ITEM]))
         self.assertTrue(text.startswith("## Tested\n"))
-        for expected in ("**Result:** `same`", "**Method:** Twice per card. Every input fixed.", "**Ran on:** Test CPU, 12 CPUs",
-                         self.ITEM["record_url"], "1 flag raised", "2 runs",
-                         "../downloads/measurements/beetle_saturn.sqlite", "../wiki/measurements.md",
-                         "<image>"):
+        for expected in ("same with every file", "<strong>Method:</strong> Twice per card. Every input fixed.",
+                         "<strong>Ran on:</strong> Test CPU, 12 CPUs", self.ITEM["record_url"],
+                         "1 flag raised", "2 runs", "downloads/measurements/beetle_saturn.sqlite",
+                         "wiki/measurements/", "&lt;image&gt;", "1 test · last run 2026-10-27",
+                         '<p class="rb-tested-file"><code>mpeg_card.bin</code></p>',
+                         'class="rb-tested-legend"', "No file staged."):
             self.assertIn(expected, text)
+        self.assertNotIn("confirmed", text)
+        self.assertNotIn(" open", text)
+
+    def item(self, **values: object) -> dict:
+        return {**self.ITEM, "host_short": "Test CPU, Linux", **values}
+
+    def test_summary_counts(self) -> None:
+        items = [self.item(proves=["hle_fallback of a.bin"], open_findings=2),
+                 self.item(id="b", subject="b", date="2026-10-28T00:00:00+00:00", open_findings=2),
+                 self.item(id="c", kind="bench", status="measured", subject="c",
+                           timing={"ms_per_frame": 1.5, "fps": 660.0, "stdev": 0.01})]
+        self.assertEqual(sm.tested_counts(items), {"tests": 3, "confirmed": 1, "open": 2, "benchmarks": 1,
+                                                   "last": "2026-10-28"})
+        text = "\n".join(sm.tested_section("beetle_saturn", items))
+        self.assertIn("3 tests · 1 confirmed · 2 open · last run 2026-10-28 on Test CPU, Linux.", text)
+        self.assertIn("<td>1.500</td><td>660.0</td>", text)
+
+    def test_short_titles(self) -> None:
+        cases = [
+            ({"subject": "mpeg-card-lunar-cr2-build", "variable": "mpeg_card.bin",
+              "content": "Redump: Lunar - Silver Star Story Complete (Japan), T-27904G V1.00"},
+             "`mpeg_card.bin`, Lunar (CR2 build)"),
+            ({"subject": "negative-scph5501-bin-mednafen-psx-bust-a-move-2-arcade-edition-usa",
+              "variable": "scph5501.bin", "content": "Bust-A-Move 2 - Arcade Edition (USA)",
+              "core": {"file": "mednafen_psx_libretro.so"}},
+             "`scph5501.bin`, Bust-A-Move 2 (wrong files)"),
+            ({"subject": "issue85-lunar-opening-movie", "variable": None,
+              "content": "Redump: Lunar - Silver Star Story Complete (Japan), T-27904G V1.00"},
+             "Lunar (issue 85 opening movie)"),
+            ({"subject": "with-command-auth-device-and-command-get-9a211715", "variable": "mpeg_card.bin",
+              "content": "Redump: Lunar"},
+             "`mpeg_card.bin`, Lunar (with command auth device and get)"),
+        ]
+        for item, title in cases:
+            self.assertEqual(sm.short_title(item), title)
+        self.assertEqual(sm.short_host({"cpu_model": "13th Gen Intel(R) Core(TM) i9-13900K",
+                                        "os": "Linux-6.18.54.1-microsoft-standard-WSL2-x86_64"}),
+                         "Intel Core i9-13900K, Linux (WSL2)")
+
+    def test_rows_of_one_file_never_read_alike(self) -> None:
+        def item(rid: str, build: str, files: int, observable: str, clock: bool = True) -> dict:
+            record = {"core": {"build_command": build}, "subject": "mpeg-card-lunar", "kind": "check",
+                      "check": {"observable": observable}, "variants": [{}] * files,
+                      "inputs": {"options": {"beetle_saturn_mpeg_card": "enabled"},
+                                 "env": {"FAKETIME": "x"} if clock else {}}}
+            return self.item(id=rid, content="Redump: Lunar - Silver Star Story", setup=sm.setup_of(record))
+        items = [item("a", "make -j8", 11, "video_sha1"),
+                 item("b", "make -j8 after editing cdb.c", 11, "video_sha1"),
+                 item("c", "make -j10, with the CR2 patch to cdb.c", 3, "state:MPEG.AuthState"),
+                 item("d", "make -j8", 11, "video_sha1", clock=False)]
+        titles = sm.distinct_titles(items)
+        self.assertEqual(titles, {
+            "a": "Lunar",
+            "b": "Lunar, patched build",
+            "c": "Lunar, CR2 patch, 3 files, MPEG.AuthState",
+            "d": "Lunar, running clock"})
+        self.assertEqual(len(set(titles.values())), len(items))
+        self.assertNotIn("...", " ".join(titles.values()))
+        # Alone in its file, a test is named by its game.
+        self.assertEqual(sm.distinct_titles(items[:1]), {"a": "Lunar"})
+
+    def test_no_empty_sentence(self) -> None:
+        text = "\n".join(sm.tested_section("beetle_saturn", [self.item(claim=None, method=None, statement="",
+                                                                         kind="run", status="ran")]))
+        self.assertNotRegex(text, r"</strong>\s*\.|:\s*\.\s|`ran`\. \.")
+        self.assertNotIn("Claim:", text)
+        self.assertNotIn("Method:", text)
+
+    def test_dots_follow_the_result_groups(self) -> None:
+        record = full_record("alpha", "boot", "2026-10-06T10:00:00+00:00", values=("1" * 40, "2" * 40, "1" * 40))
+        record["check"]["variable"] = "card.bin"
+        record["variants"][2]["system_files"] = {"card.bin": None}
+        groups = sm.variant_groups(record)
+        self.assertEqual([g["group"] for g in groups],
+                         [i for i, group in enumerate(record["check"]["result"]["groups"])
+                          for _ in group["variants"]][:1] + [1, 0])
+        self.assertEqual([g["absent"] for g in groups], [False, False, True])
+        text = "\n".join(sm.tested_section("alpha", [self.item(groups=groups)]))
+        strip = text.split('class="rb-dots"', 1)[1].split("</span></span>", 1)[0]
+        self.assertEqual(re.findall(r'rb-dot rb-dot-(\d)( rb-dot-empty)?"', strip),
+                         [("0", ""), ("1", ""), ("0", " rb-dot-empty")])
+        self.assertIn('aria-label="3 variants. v0: result A; v1: result B; v2: result A"', text)
+
+    def test_the_details_hold_the_method_and_the_host(self) -> None:
+        text = "\n".join(sm.tested_section("beetle_saturn", [self.item(runs=1)]))
+        details = text.split('<div class="rb-tested-more">', 1)[1].split("</div>", 1)[0]
+        self.assertIn("Twice per card", details)
+        self.assertIn("<strong>Ran on:</strong> Test CPU, Linux.", details)
+        self.assertIn(f'<a href="{self.ITEM["record_url"]}">Full record</a>.', details)
+        self.assertNotIn("1 run", details)
+
+    def test_the_beetle_saturn_block_stays_short(self) -> None:
+        stored = mr.read_records(REPO_ROOT)
+        profiles = sm.load_profiles(REPO_ROOT / "emulators")
+        index = sm.build_index(stored, {"repo": REPO_URL, "commit": COMMIT}, profiles)
+        items = sm.by_emulator(index).get("beetle_saturn")
+        if not items:
+            self.skipTest("no beetle_saturn record")
+        block = "\n".join(sm.tested_section("beetle_saturn", items))
+        # What a reader sees before opening anything: the closed details hide their body.
+        shown = re.sub(r'<div class="rb-tested-more">.*?</div>', " ", block, flags=re.S)
+        shown = re.sub(r'<details class="rb-tested-earlier">.*?</details>\s*</details>', " ", shown,
+                       flags=re.S)
+        shown = re.sub(r"<[^>]+>", " ", shown)
+        self.assertLess(len(re.findall(r"[A-Za-z0-9][\w.'-]*", shown)), 120, shown)
 
     def test_hosts_table_on_the_method_page(self) -> None:
         page = sm.with_hosts(f"x\n\n{sm.HOSTS_MARKER}\n",