feat: take the host probe lock per run

core_probe takes probe.lock itself around one execution, load check included, and releases it when the run ends. A campaign pass never holds it between runs, so two passes interleave. RETROBIOS_PROBE_LOCK names another file, RETROBIOS_PROBE_LOCK_HELD says the caller holds it. Tests use their own lock file.

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/test_campaign.py

@@ -1126,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

@@ -504,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",