bench: record whether another lane held the NPU while the arm ran (box.npu, #1197)

tests/hyperloom_benchmark_test.py

@@ -187,6 +187,41 @@ def main():
     assert got["gtt_used_gib"] == 1.0, got  # start GTT, not the last one
     assert bench.box_summary([]) is None
 
+    # And the NPU is a condition of the score, not just of the die: the 2026-11-05 policy entry for this
+    # box says the engines coexist safely but that a GPU arm beside a running NPU *matrix* corrupts both
+    # lanes' numbers. This lane's arm was refused 27 times on that gate, and a number measured beside a
+    # job has to be able to say so -- the owner's resident `npu-serve-keeper` is baseline, not a job.
+    real_home = os.environ.pop("HL_REAL_HOME", None)
+    try:
+        with tempfile.TemporaryDirectory() as tmp:
+            home = pathlib.Path(tmp)
+            os.environ["HL_REAL_HOME"] = tmp
+            assert bench.npu_conditions() == {"npu_holder": None, "npu_job": False,
+                                              "npu_queued": 0}, bench.npu_conditions()
+            (home / "NPU-TOKEN").write_text("npu-serve-keeper pid 4675\n", encoding="utf-8")
+            assert bench.npu_conditions()["npu_job"] is False   # the resident serve is baseline
+            (home / "NPU-QUEUE").mkdir()
+            (home / "NPU-QUEUE/.log").write_text("noise\n", encoding="utf-8")
+            (home / "NPU-QUEUE/0179392.mosaic-regate9").write_text("pid=1\n", encoding="utf-8")
+            (home / "NPU-TOKEN").write_text("gpu-hold mosaic-regate9 pid 1584738 2026-11-05\n",
+                                             encoding="utf-8")
+            got = bench.npu_conditions()
+            assert got["npu_job"] is True and got["npu_queued"] == 1, got  # dotfiles are not tickets
+            assert got["npu_holder"].startswith("gpu-hold mosaic-regate9"), got
+            # A half-written or unreadable claim file is not a quiet NPU.
+            (home / "NPU-TOKEN").write_text("\n", encoding="utf-8")
+            assert bench.npu_conditions()["npu_holder"] is None, bench.npu_conditions()
+            # And the arm's own summary counts how many of its samples saw a job.
+            got = bench.box_summary([{"tctl_c": 60.0, "npu_job": False},
+                                     {"tctl_c": 61.0, "npu_job": True},
+                                     {"tctl_c": 62.0, "npu_job": True}])
+            assert (got["npu_job_samples"], got["npu_samples"]) == (2, 3), got
+            assert got["npu_job"] is False, got  # the start snapshot, same shape as gtt_used_gib
+    finally:
+        os.environ.pop("HL_REAL_HOME", None)
+        if real_home is not None:
+            os.environ["HL_REAL_HOME"] = real_home
+
     # And the verdict has to name what it was relative to: `HL_STATE` is per-box and the base logits are
     # regenerated when missing, so a KLD number reads PASS against a reference the JSON never named.
     with tempfile.TemporaryDirectory() as tmp:

tools/hyperloom/custom/benchmark.py

@@ -398,6 +398,37 @@ def die_temp_c():
     return None
 
 
+def npu_conditions():
+    """Whether another lane holds the NPU, read from the claim files (device-free).
+
+    Same class as the die temperature: a condition of the score, recorded rather than refused. The
+    fleet's policy for this box (okf policies/strixhalo-thermal-and-gpu-claims.md, 2026-11-05) is that
+    the two engines coexist safely -- the real cross-engine hazard is a shared dma-buf range, not
+    simultaneity -- but a GPU *arm* beside a running NPU *matrix* corrupts both lanes' numbers. This
+    lane's long-context arm was refused 27 times on 2026-11-05 for exactly this and produced no
+    number, and a result that did measure beside an NPU job could not have said so.
+
+    `npu-serve-keeper` is the owner's resident serve: baseline box state, not a job. Every field is
+    null/0/false where the box does not expose it, so an unreadable claim file cannot be read back as
+    a quiet NPU.
+    """
+    home = user_home()
+    holder = None
+    try:
+        lines = (home / "NPU-TOKEN").read_text(encoding="utf-8", errors="replace").splitlines()
+        holder = lines[0].strip() if lines and lines[0].strip() else None
+    except OSError:
+        holder = None
+    queued = 0
+    try:
+        queued = sum(1 for path in (home / "NPU-QUEUE").iterdir() if not path.name.startswith("."))
+    except OSError:
+        queued = 0
+    return {"npu_holder": holder,
+            "npu_job": bool(holder) and not holder.startswith("npu-serve-keeper"),
+            "npu_queued": queued}
+
+
 def box_conditions():
     """The conditions the measurement is running in, device-free (sysfs + /proc).
 
@@ -418,9 +449,11 @@ def box_conditions():
                 break
     except (OSError, ValueError, IndexError):
         available = None
-    return {"tctl_c": die_temp_c(),
-            "gtt_used_gib": gtt_used, "gtt_total_gib": gtt_total,
-            "mem_available_gib": available}
+    conditions = {"tctl_c": die_temp_c(),
+                  "gtt_used_gib": gtt_used, "gtt_total_gib": gtt_total,
+                  "mem_available_gib": available}
+    conditions.update(npu_conditions())
+    return conditions
 
 
 def box_summary(samples):
@@ -437,6 +470,10 @@ def box_summary(samples):
     summary["tctl_c_min"] = min(die) if die else None
     summary["tctl_c_max"] = max(die) if die else None
     summary["tctl_samples"] = len(die)
+    # Same reason as the die's spread: an NPU job can start after the arm is claimed, so the start
+    # snapshot alone does not say whether the number was measured beside one.
+    summary["npu_job_samples"] = sum(1 for s in samples if s.get("npu_job"))
+    summary["npu_samples"] = len(samples)
     return summary