Stop two harnesses from measuring each other's store

A scratch copy is one run's working store. The video harness defaults to
bench/runs/video_audio and copy_store() copies straight over it, so a smoke
run started while the full run was mid-ingest replaced the database under it:
the full run's after-arm would have scored a store whose speech atoms had been
swapped out, and the smoke itself died on "database is locked" inside
text_index. Six harnesses share that copy path, so the claim lives in
bench/scratch.py: a live pid in held_by refuses the path, a stale one is taken
over, and the same process may re-copy per arm.

The reviewer's smoke knob also measured nothing instead of saying so. With
--limit the probe sample narrows to video only, and a small --probes strides
past that modality entirely, so the run crashed in aggregate() after the
ingest. probe_sample() now switches the arm off at --probes 0 and fails in
half a second otherwise.

The frozen-report guard had two holes found by running it: it compared
cwd-relative parents, so an absolute path to the frozen file passed, and it
refused the harness's own default destination once that file existed, which
would have stopped the shipped run. It now compares resolved paths and refuses
only committed evidence.

Record fixes from review round 3: the round-2 line credited a reproduction to
the wrong round and quoted a git log -S hit for a string that never existed in
code; the suite counts now name the commit each was measured on (401 at
83dfc14, 406 at eab6bd7, 410 at cf2de31, 411 at 92237c8).

Gates: pytest -q 418 passed; ruff clean; bench.filesize_check OK; mypy cold
cache and pyright clean on the four touched files.

bench/scratch.py

@@ -3,10 +3,40 @@ migrates whatever database it opens."""
 
 from __future__ import annotations
 
+import atexit
+import os
 import shutil
 import time
 from pathlib import Path
 
+HELD = "held_by"
+
+
+def _alive(pid: str) -> bool:
+    """A live process holds the copy. Without /proc there is nothing to check, so a stale lock cannot block a run."""
+    return pid.isdigit() and Path(f"/proc/{pid}").exists()
+
+
+def claim(scratch: Path) -> None:
+    """Refuse a copy another run is already using: two harnesses on one scratch overwrite each other's store mid-measurement."""
+    lock = scratch / HELD
+    if lock.exists():
+        holder = lock.read_text(encoding="utf-8").strip()
+        if _alive(holder) and holder != str(os.getpid()):
+            raise SystemExit(
+                f"{scratch} is held by pid {holder}; wait for that run or pass a different scratch path"
+            )
+    scratch.mkdir(parents=True, exist_ok=True)
+    lock.write_text(str(os.getpid()), encoding="utf-8")
+    atexit.register(_release, scratch)
+
+
+def _release(scratch: Path) -> None:
+    """Drop the claim, but only if it is still ours."""
+    lock = scratch / HELD
+    if lock.exists() and lock.read_text(encoding="utf-8").strip() == str(os.getpid()):
+        lock.unlink(missing_ok=True)
+
 
 def copy_store(home: Path, scratch: Path) -> tuple[Path, float]:
     """Work on a copy: the baseline store is frozen evidence and must not be migrated in place."""
@@ -15,6 +45,7 @@ def copy_store(home: Path, scratch: Path) -> tuple[Path, float]:
         raise SystemExit(
             f"{source}: no product database, build it with python -m bench.ingest_baseline"
         )
+    claim(scratch)
     data = scratch / "data"
     data.mkdir(parents=True, exist_ok=True)
     began = time.perf_counter()

bench/tests/test_scratch.py

@@ -0,0 +1,31 @@
+"""A scratch copy is one run's working store; two runs on one path silently measure each other's data."""
+
+import os
+from pathlib import Path
+
+import pytest
+
+from bench import scratch
+
+
+def test_a_live_holder_refuses_the_same_scratch(tmp_path: Path) -> None:
+    """The collision I hit: a second run re-copied the store under a run that was ingesting into it."""
+    if not Path("/proc/1").exists():
+        pytest.skip("liveness is read from /proc")
+    (tmp_path / scratch.HELD).write_text("1", encoding="utf-8")
+    with pytest.raises(SystemExit, match="is held by pid 1"):
+        scratch.claim(tmp_path)
+
+
+def test_a_stale_claim_is_taken_over(tmp_path: Path) -> None:
+    """A crashed run must not lock the scratch away forever."""
+    (tmp_path / scratch.HELD).write_text("999999", encoding="utf-8")
+    scratch.claim(tmp_path)
+    assert (tmp_path / scratch.HELD).read_text(encoding="utf-8") == str(os.getpid())
+
+
+def test_one_run_may_copy_more_than_once(tmp_path: Path) -> None:
+    """Harnesses that score several arms re-copy on purpose; that is not a collision."""
+    scratch.claim(tmp_path)
+    scratch.claim(tmp_path)
+    assert (tmp_path / scratch.HELD).read_text(encoding="utf-8") == str(os.getpid())

bench/tests/test_video_audio_check.py

@@ -11,8 +11,9 @@ from bench import video_audio_check
 def test_a_frozen_report_is_not_overwritten_by_a_harness(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    """A reviewer runs harnesses in a worktree of its own; the committed reports stay put."""
+    """A reviewer runs harnesses in a worktree of its own; the committed evidence stays put."""
     monkeypatch.setattr(video_audio_check, "PAYLOAD", tmp_path / "video_audio.json")
+    monkeypatch.setattr(video_audio_check, "_tracked", lambda path: True)
     frozen = tmp_path / "VIDEO_AUDIO.md"
     frozen.write_text("committed evidence", encoding="utf-8")
     with pytest.raises(SystemExit, match="is a frozen report"):
@@ -23,6 +24,47 @@ def test_a_frozen_report_is_not_overwritten_by_a_harness(
     video_audio_check.refuse_frozen_overwrite(elsewhere)
 
 
+def test_a_harness_may_write_its_own_untracked_output(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The default destination is the reports directory; refusing it would stop the shipped run."""
+    monkeypatch.setattr(video_audio_check, "PAYLOAD", tmp_path / "video_audio.json")
+    monkeypatch.setattr(video_audio_check, "_tracked", lambda path: False)
+    fresh = tmp_path / "VIDEO_AUDIO.md"
+    fresh.write_text("not yet evidence", encoding="utf-8")
+    video_audio_check.refuse_frozen_overwrite(fresh)
+
+
+def test_an_absolute_path_to_the_frozen_directory_is_refused_too(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The guard compares where the file really lands, not how the path was spelled."""
+    monkeypatch.setattr(video_audio_check, "PAYLOAD", tmp_path / "video_audio.json")
+    monkeypatch.setattr(video_audio_check, "_tracked", lambda path: True)
+    frozen = tmp_path / "VIDEO_AUDIO.md"
+    frozen.write_text("committed evidence", encoding="utf-8")
+    with pytest.raises(SystemExit, match="is a frozen report"):
+        video_audio_check.refuse_frozen_overwrite(Path(frozen.resolve()))
+
+
+def test_the_probe_arm_switches_off_rather_than_ranking_nothing() -> None:
+    """--probes 0 means "skip that arm"; a small count with the smoke knob means "you measured nothing"."""
+    assert video_audio_check.probe_sample(0, 0) == []
+    # The shipped sample keeps every modality; the smoke knob narrows it to video, where a
+    # small count can miss the modality entirely (the reviewer's --limit 1 --probes 1 crash).
+    assert any(query.modality == video_audio_check.VIDEO_MODALITY for query in video_audio_check.probe_sample(300, 0))
+    for probes in (1, 5):
+        with pytest.raises(SystemExit, match=f"--probes {probes} picked no answerable video query"):
+            video_audio_check.probe_sample(probes, probes)
+
+
+def test_an_empty_video_arm_is_a_lie_not_a_zero() -> None:
+    """The video arm is the point of the harness; a row of zeros there must stop the run."""
+    with pytest.raises(SystemExit, match="nothing to rank"):
+        video_audio_check._required(None, "every video question")
+    assert video_audio_check._score_of(None) == "skipped"
+
+
 def test_a_missing_speech_snapshot_says_so_before_ranking(tmp_path: Path) -> None:
     """Without a snapshot every video defers, and the after row would equal the before row."""
     with pytest.raises(SystemExit, match="no speech snapshot"):

bench/video_audio_check.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 import argparse
 import json
 import statistics
+import subprocess
 import sys
 import time
 from pathlib import Path
@@ -75,12 +76,14 @@ def speech_facts(store: Path) -> dict[str, Any]:
     return {"atoms": int(atoms), "characters": int(characters)}
 
 
-def rank_arm(home: Path, probes: list[Any], top_k: int) -> dict[str, Any]:
-    """Rank the probes through the shipped words-only path and score them."""
+def rank_arm(home: Path, probes: list[Any], top_k: int) -> dict[str, Any] | None:
+    """Rank the probes through the shipped words-only path and score them; None skips the arm."""
     from ely_eye.config import Settings
     from ely_eye.db import Database
     from ely_eye.retrieval import RetrievalService
 
+    if not probes:
+        return None
     settings = Settings(
         home=home,
         dense_embeddings_enabled=False,
@@ -140,15 +143,33 @@ def add_speech(home: Path, files: list[Path], asr: Path, limit: int) -> dict[str
     return {"files": batches, "seconds": time.perf_counter() - began}
 
 
+def probe_sample(probes: int, limit: int) -> list[Any]:
+    """The standard stratified sample, asked about video only when the smoke knob is on.
+
+    A small `--probes` is not "measure a little": the sample strides the whole suite, so a
+    count under a few dozen can miss this modality entirely and the arm would rank nothing.
+    """
+    if probes <= 0:
+        return []
+    picked = [query for query in scale_probes(load_queries(), probes) if query.gold]
+    if limit:
+        picked = [query for query in picked if query.modality == VIDEO_MODALITY][:limit]
+    if not picked:
+        raise SystemExit(
+            f"--probes {probes} picked no answerable video query: the sample strides the whole "
+            "suite, so a small count can skip this modality and the arm would rank nothing. "
+            "Pass --probes 0 to skip the probe arm, or the shipped 300 to keep every modality."
+        )
+    return picked
+
+
 def measure(home: Path, scratch: Path, videos: Path, asr: Path, probes: int, limit: int) -> dict[str, Any]:
     """Score the video questions and the standard probe sample before and after the speech pass."""
     if not (home / "data" / "ely_eye.sqlite").exists():
         raise SystemExit(f"{home}: no store to copy; run `python -m bench.dense_scale` first")
     files = guard_videos(videos, home)
     whole = video_queries() if not limit else video_queries()[:limit]
-    sample = [query for query in scale_probes(load_queries(), probes) if query.gold]
-    if limit:
-        sample = [query for query in sample if query.modality == VIDEO_MODALITY][:limit]
+    sample = probe_sample(probes, limit)
     print(f"{len(whole)} answerable video queries, {len(sample)} stratified probes, {len(files)} files", flush=True)
 
     copy_home, copied = copy_store(home, scratch)
@@ -156,11 +177,11 @@ def measure(home: Path, scratch: Path, videos: Path, asr: Path, probes: int, lim
     before_speech = speech_facts(copy_home)
     if before_speech["atoms"]:
         print(f"note: the copy already carries {before_speech['atoms']} speech atoms", flush=True)
-    before_video = rank_arm(copy_home, whole, SHIPPED_TOP_K)
+    before_video = _required(rank_arm(copy_home, whole, SHIPPED_TOP_K), "every video question")
     before_sample = rank_arm(copy_home, sample, SHIPPED_TOP_K)
     print(
         f"  before: video recall@10 {before_video['recall_at_10']:.3f}, "
-        f"probes {before_sample['recall_at_10']:.3f}",
+        f"probes {_score_of(before_sample)}",
         flush=True,
     )
 
@@ -171,11 +192,11 @@ def measure(home: Path, scratch: Path, videos: Path, asr: Path, probes: int, lim
             f"re-ingesting {len(files)} files added no speech atoms; the snapshot at {asr} is "
             "available but produced nothing, so the after row would be a measurement of nothing"
         )
-    after_video = rank_arm(copy_home, whole, SHIPPED_TOP_K)
+    after_video = _required(rank_arm(copy_home, whole, SHIPPED_TOP_K), "every video question")
     after_sample = rank_arm(copy_home, sample, SHIPPED_TOP_K)
     print(
         f"  after:  video recall@10 {after_video['recall_at_10']:.3f}, "
-        f"probes {after_sample['recall_at_10']:.3f}, speech atoms {after_speech['atoms']}",
+        f"probes {_score_of(after_sample)}, speech atoms {after_speech['atoms']}",
         flush=True,
     )
     return {
@@ -193,6 +214,18 @@ def measure(home: Path, scratch: Path, videos: Path, asr: Path, probes: int, lim
     }
 
 
+def _required(row: dict[str, Any] | None, label: str) -> dict[str, Any]:
+    """The video arm is never empty; a row of zeros there would be a silent lie."""
+    if row is None:
+        raise SystemExit(f"{label}: nothing to rank, so this run would print a row of zeros")
+    return row
+
+
+def _score_of(row: dict[str, Any] | None) -> str:
+    """One number for the progress line, or the reason there is none."""
+    return "skipped" if row is None else f"{row['recall_at_10']:.3f}"
+
+
 def write_markdown(report: dict[str, Any], destination: Path) -> None:
     """Print the two rows beside each other, with the load they were measured under."""
     load = report["machine"].get("load_average_1_5_15")
@@ -217,15 +250,19 @@ def write_markdown(report: dict[str, Any], destination: Path) -> None:
     for stage in ("before", "after"):
         for arm, label in (("video", "every video question"), ("probes", "the stratified probe sample")):
             row = report[stage][arm]
+            if row is None:
+                lines.append(f"| {label} ({stage}) | 0 | skipped (--probes 0) | — | — | — | — |")
+                continue
             lines.append(
                 f"| {label} ({stage}) | {row['queries']} | {row['recall_at_10']:.3f} | {row['mrr']:.3f} | "
                 f"{row['hit_at_10']:.3f} | {row['wilson_lower']:.3f} | {row['p50_seconds']:.3f} |"
             )
     gained = report["after"]["video"]["recall_at_10"] - report["before"]["video"]["recall_at_10"]
-    overall = report["after"]["probes"]["recall_at_10"] - report["before"]["probes"]["recall_at_10"]
+    probe_row = report["after"]["probes"]
+    overall = "not measured (--probes 0)" if probe_row is None else f"{probe_row['recall_at_10'] - report['before']['probes']['recall_at_10']:+.3f}"
     lines += [
         "",
-        f"Speech moved video recall@10 by {gained:+.3f} and the whole probe sample by {overall:+.3f}.",
+        f"Speech moved video recall@10 by {gained:+.3f} and the whole probe sample by {overall}.",
         "",
         "Latency is read from the same pass and moves with machine load; the recall columns do not.",
         "",
@@ -234,9 +271,21 @@ def write_markdown(report: dict[str, Any], destination: Path) -> None:
     destination.write_text("\n".join(lines), encoding="utf-8")
 
 
+def _tracked(path: Path) -> bool:
+    """Only committed evidence is frozen; a harness's own untracked output is fair game."""
+    return (
+        subprocess.run(
+            ["git", "ls-files", "--error-unmatch", str(path)], capture_output=True, check=False
+        ).returncode
+        == 0
+    )
+
+
 def refuse_frozen_overwrite(destination: Path) -> None:
     """A reviewer running this in a worktree of its own must not overwrite the committed evidence."""
-    if destination.parent == PAYLOAD.parent and destination.exists():
+    if destination.exists() and destination.resolve().parent == PAYLOAD.resolve().parent and _tracked(
+        destination
+    ):
         raise SystemExit(
             f"{destination} is a frozen report; point --report and --payload somewhere else, or "
             "delete the frozen copy on purpose rather than under a harness"