lab: no measurement on a shared GPU (refused / contaminated)

CLAUDE.md

@@ -38,7 +38,9 @@ pytest/ruff config). Each folder with commands has its own `python -m`.
   `proftools.py` (trace = nsys, kernel = ncu, hostprof = py-spy, each wrapping lab.profile's workload in the jail
   with validated args) + `evaltools.py` (bench → `regimes` on
   the seen split; equiv → `correctness` candidate vs the target's reference; submit → held-out full tier vs the
-  base commit, one aggregate per regime through the Thresholdout guard; all three serve the pristine tree via
+  base commit, one aggregate per regime through the Thresholdout guard; nothing is measured while another process holds
+  the GPU (`refused`), and a process other than the engine on the GPU mid-run makes it `contaminated` (never evidence,
+  never the cached base; `serve.Served.exclusive()`); all three serve the pristine tree via
   `serve.py`, jailed (on Linux, port bridged out of a network-less container); every serve also keeps its passive data, `artifacts.py`: device
   samples via `gpu.DeviceSampler` (DCGM if `dcgmi`, else nvidia-smi), engine telemetry files from `{dir}`, all
   client rows, the serve log; bench/equiv as ledger blobs under `result.artifacts`, submit only in

lab/LEDGER.md

@@ -21,7 +21,10 @@ on write. Every record: `id`, `at`, `schema` (set by the writer), `kind`, `run`,
 | `note` | `note`, `restore`, refusals | free text, or `{verdict: refused, reason}` |
 
 A tool that could not run writes `{verdict: refused | error, reason}` and the agent sees that text; it never
-crashes the session. The integrity verdict (`feedback/lab_verdict.py`) reads only these records: a run is valid
+crashes the session. `bench`, `equiv`, `submit` and `baseline` also write `refused` when a process already held
+the GPU (no engine launched) and `contaminated` when a process other than the served engine held it during the
+measurement; both carry `gpu_processes: [{pid, used_mib, name}]`. A contaminated run is never evidence and never
+cached as the base. The integrity verdict (`feedback/lab_verdict.py`) reads only these records: a run is valid
 iff a completed `submit` exists whose snapshot also has a completed seen-split `bench`, a passing `test`, a
 passing `equiv` (tier full) and no failing `equiv`, and no record reports a write-surface violation.
 

lab/RUNTIME.md

@@ -130,8 +130,13 @@ Transport now: SSH from the controller, one worker. Transport later: a queue. Th
 ## Leftover processes, today
 
 On Linux, srt already isolates them (own PID and network namespace; killed when the jail exits). During a
-session they still share the GPU and CPU with the engine being measured, and nothing stops them before a
-measurement: `serve.py` tears down only the engine's own process group. Rule 4 closes it.
+session they could still share the GPU with the engine being measured. Since 2026-10-09 `serve.py` refuses to
+launch an engine while any process holds the GPU (`GpuBusy`) and polls the GPU's processes every second while it
+serves; anything outside the engine's own processes marks the measurement `contaminated` (`Served.exclusive()`):
+never evidence for submit, never cached as the base, and caught before the holdout guard spends a query. The
+facts (pid, MiB, name) go to the agent. It refuses rather than kills because the lab cannot yet tell the agent's
+processes apart; rule 4's freeze-and-kill comes with the agent container. Not covered: CPU contention, and a GPU
+burst shorter than the poll interval.
 
 ## At scale (not built)
 

lab/evaltools.py

@@ -121,6 +121,19 @@ def equiv_rows(out: Path) -> list[dict]:
     return [{k: v for k, v in json.loads(x).items() if k != "text"} for x in p.read_text().splitlines() if x.strip()]
 
 
+def _failed(e: BaseException) -> dict:
+    """The record of a measurement that did not complete; GPU processes that stopped it are kept as facts."""
+    if isinstance(e, serve.GpuBusy):
+        return {"verdict": "refused", "reason": str(e), "gpu_processes": e.apps}
+    if isinstance(e, serve.Contaminated):
+        return {"verdict": "contaminated", "reason": str(e), "gpu_processes": e.apps}
+    return {"verdict": "error", "reason": _why(e)}
+
+
+def _said(tool: str, result: dict) -> str:
+    return f"{tool} {'failed' if result['verdict'] == 'error' else result['verdict']}: {result['reason']}"
+
+
 def _why(e: BaseException) -> str:
     if isinstance(e, serve.NotReady):
         return f"engine did not start: {e}"
@@ -147,11 +160,11 @@ def bench(self, args: dict) -> str:
             with serve.Served(tree, t.engine, log=self.s.run_dir / "serve-bench.log",
                               jailed=self.serve_jailed, artifacts=work) as srv:
                 results = run_regimes(srv.url, t, names, "seen", tier, seed, rows)
+                srv.exclusive()
         except Exception as e:                      # a tool never crashes the session: the failure is the record
-            result = {"verdict": "error", "reason": _why(e), "seconds": time.monotonic() - t0,
-                      "artifacts": self._store(work, rows)}
+            result = {**_failed(e), "seconds": time.monotonic() - t0, "artifacts": self._store(work, rows)}
             self._record("bench", "bench", args, result, snap, config={"split": "seen", "tier": tier, "seed": seed})
-            return f"bench failed: {result['reason']}"
+            return _said("bench", result)
         result = {"verdict": "ok", "metrics": headline(results), "regimes": results, "tier": tier,
                   "seconds": time.monotonic() - t0, "ready_s": srv.ready_s, "artifacts": self._store(work, rows)}
         self._record("bench", "bench", args, result, snap, config={"split": "seen", "tier": tier, "seed": seed})
@@ -179,11 +192,11 @@ def equiv(self, args: dict) -> str:
                                      Thresholds(min_score_ratio=t.min_score_ratio, min_length_ratio=t.min_length_ratio),
                                      concurrency=self.equiv_concurrency, api=client.Api(t.engine.api, t.chat_kwargs),
                                      config={"tier": tier})
+                srv.exclusive()
         except Exception as e:
-            self._record("equiv", "equiv", args, {"verdict": "error", "reason": _why(e),
-                                                  "artifacts": self._store(work, equiv_rows(out))},
-                         snap, config={"split": "seen", "tier": tier})
-            return f"equiv failed: {_why(e)}"
+            result = {**_failed(e), "artifacts": self._store(work, equiv_rows(out))}
+            self._record("equiv", "equiv", args, result, snap, config={"split": "seen", "tier": tier})
+            return _said("equiv", result)
         passed = {"pass": True, "fail": False}.get(res["verdict"])     # inconclusive is None: neither passing nor failing
         record = {"verdict": res["verdict"], "passed": passed, "reasons": res["reasons"], "gates": res["gates"],
                   "metrics": res["metrics"], "thresholds": res["thresholds"], "tier": tier,
@@ -222,6 +235,7 @@ def submit(self, args: dict) -> str:
                 with serve.Served(tree, t.engine, log=self.s.run_dir / "serve-submit.log", jailed=self.serve_jailed,
                                   artifacts=private) as srv:
                     results = run_regimes(srv.url, t, names, "heldout", "full", self.seed, rows)
+                    srv.exclusive()             # before the holdout guard: a contaminated run spends no query
             finally:
                 artifacts.write_rows(private, rows)
             new = headline(results)
@@ -237,8 +251,9 @@ def submit(self, args: dict) -> str:
             self._record("submit", "submit", args, {"verdict": "refused", "reason": str(e)}, snap)
             return f"submit refused: {e}"
         except Exception as e:
-            self._record("submit", "submit", args, {"verdict": "error", "reason": _why(e)}, snap)
-            return f"submit failed: {_why(e)}"
+            result = _failed(e)
+            self._record("submit", "submit", args, result, snap)
+            return _said("submit", result)
         rec = self._record_heldout("submit", "submit", args, {"tier": "full"}, metrics, snap)
         out = {"snapshot": snap.id, "split": "heldout", "tier": "full", "metrics": metrics, "record": rec["id"]}
         task = getattr(self.s, "task", None)
@@ -286,6 +301,7 @@ def _base_measure(self, t: target.Target, names: list[str], split: str, tier: st
             grader.export(self.s.workspace.repo, self.s.workspace.base, tree)
             with serve.Served(tree, t.engine, log=self.s.run_dir / f"serve-base-{split}.log", jailed=self.serve_jailed) as srv:
                 have.update(headline(run_regimes(srv.url, t, missing, split, tier, self.seed)))
+                srv.exclusive()                 # a contaminated base is never cached: every later submit reads it
             cache.write_text(json.dumps(have, indent=1))
         return have
 
@@ -300,7 +316,7 @@ def baseline(self, commit: str) -> dict:
             seen = self.base_seen(target.load(), names)
             result = {"verdict": "ok", "metrics": {n: seen[n] for n in names}}
         except Exception as e:                      # no GPU or a base that will not serve is itself the fact
-            result = {"verdict": "error", "reason": _why(e)}
+            result = _failed(e)
         return ledger.append({"kind": "baseline", "run": self.s.run_id, "session": self.s.session_id,
                               "config": {"split": "seen", "tier": BENCH_TIER, "commit": commit},
                               "result": result, "cost": self._cost("baseline")}, self.s.ledger_root)

lab/gpu.py

@@ -161,3 +161,37 @@ def stop(self) -> None:
                 self._file.close()
                 self._proc = None
         (self.path.parent / "meta.json").write_text(json.dumps(meta, indent=1))
+
+
+def compute_apps() -> list[dict] | None:
+    """Every process holding a GPU context (host pids): pid, used MiB, name. None where there is no nvidia-smi."""
+    if not available():
+        return None
+    out = subprocess.run(["nvidia-smi", "--query-compute-apps=pid,used_memory,process_name",
+                          "--format=csv,noheader,nounits"], capture_output=True, text=True)
+    if out.returncode != 0:
+        return None
+    apps = []
+    for line in out.stdout.splitlines():
+        pid, mib, name = ([p.strip() for p in line.split(",", 2)] + ["", ""])[:3]
+        if pid.isdigit():
+            apps.append({"pid": int(pid), "used_mib": int(mib) if mib.isdigit() else None, "name": name})
+    return apps
+
+
+def descendants(pid: int) -> set[int]:
+    """`pid` and every process below it, from /proc (Linux); `{pid}` where there is no /proc."""
+    children: dict[int, list[int]] = {}
+    for stat in Path("/proc").glob("[0-9]*/stat"):
+        try:
+            ppid = int(stat.read_text().rsplit(")", 1)[1].split()[1])
+        except (OSError, ValueError, IndexError):
+            continue
+        children.setdefault(ppid, []).append(int(stat.parent.name))
+    out, todo = set(), [pid]
+    while todo:
+        p = todo.pop()
+        if p not in out:
+            out.add(p)
+            todo += children.get(p, [])
+    return out

lab/safety/container.py

@@ -96,6 +96,12 @@ def new_name() -> str:
     return f"lab-jail-{uuid.uuid4().hex[:12]}"
 
 
+def pids(name: str) -> set[int]:
+    """The container's processes as host pids; empty once it is gone."""
+    out = subprocess.run([_docker(), "top", name, "-eo", "pid"], capture_output=True, text=True)
+    return {int(x) for x in out.stdout.split() if x.isdigit()} if out.returncode == 0 else set()
+
+
 def remove(name: str) -> None:
     """Kill and remove the container; a no-op if it is already gone."""
     subprocess.run([_docker(), "rm", "-f", name], capture_output=True)

lab/safety/grader.py

@@ -14,7 +14,7 @@
 from fnmatch import fnmatch
 from pathlib import Path
 
-from lab import engine, target
+from lab import engine, gpu, target
 from lab.safety import container, jail
 from lab.safety.surfaces import ALWAYS_DENY, HIDDEN, may_write
 
@@ -220,18 +220,19 @@ def _jail_argv(tree: Path, argv: list[str], env_extra: dict | None, domains, loc
 def jailed_popen(tree: Path, argv: list[str], *, env_extra: dict | None = None, stdout=None, stderr=None,
                  local_binding: bool = True, port: int | None = None):
     """A long-running jailed process (a served engine) in its own session, reachable on the host's 127.0.0.1:`port`.
-    Returns (proc, tmp dir to remove, stop): `stop` removes what lives outside proc's process group."""
+    Returns (proc, tmp dir to remove, stop, pids): `stop` removes what lives outside proc's process group; `pids()`
+    is every host pid the jail runs."""
     wrapped, env, tmp, name = _jail_argv(tree, argv, env_extra, (), local_binding, tag=".serve", port=port)
     proc = subprocess.Popen(wrapped, cwd=tree, env=env, stdout=stdout, stderr=stderr, start_new_session=True)
     if name is None:
-        return proc, tmp, lambda: None
+        return proc, tmp, lambda: None, lambda: gpu.descendants(proc.pid)
     bridge = container.Bridge(port, tmp / container.BRIDGE) if port else None
 
     def stop() -> None:
         container.remove(name)
         if bridge:
             bridge.stop()
-    return proc, tmp, stop
+    return proc, tmp, stop, lambda: container.pids(name)
 
 
 def jailed(tree: Path, argv: list[str], *, timeout_s: float, env_extra: dict | None = None,

lab/serve.py

@@ -2,7 +2,8 @@
 free localhost port, with the target's env; ready when its health path answers 200; torn down with its whole
 process group. `with Served(tree, spec) as s: ... s.url`. With `artifacts`, the run's passive data lands in that dir:
 `device/` (samples for the whole served window), `telemetry/` (files the engine wrote, when the target asks for them)
-and `serve.log` (this run's part of the log)."""
+and `serve.log` (this run's part of the log). The GPU must be free when the engine starts (else `GpuBusy`, nothing
+launched); while it is served, any other process holding the GPU is recorded and `exclusive()` raises `Contaminated`."""
 
 from __future__ import annotations
 
@@ -12,6 +13,7 @@
 import shutil
 import socket
 import subprocess
+import threading
 import time
 import urllib.error
 import urllib.request
@@ -21,6 +23,7 @@
 from lab.safety import grader
 
 TELEMETRY_SUBDIR = "lab-telemetry"        # inside the served tree, the one place a jailed engine may write
+POLL_S = 1.0                              # how often the GPU's processes are checked while an engine is served
 
 
 def free_port() -> int:
@@ -40,6 +43,26 @@ class NotReady(RuntimeError):
     pass
 
 
+def _listed(apps: list[dict]) -> str:
+    return "; ".join(f"pid {a['pid']} ({a['name']}, {a['used_mib']} MiB)" for a in apps)
+
+
+class GpuBusy(RuntimeError):
+    """Something already held the GPU, so no engine was launched."""
+
+    def __init__(self, apps: list[dict]):
+        super().__init__(f"the GPU was in use before the engine started: {_listed(apps)}")
+        self.apps = apps
+
+
+class Contaminated(RuntimeError):
+    """A process outside the served engine held the GPU during the measurement."""
+
+    def __init__(self, apps: list[dict]):
+        super().__init__(f"another process held the GPU during the measurement: {_listed(apps)}")
+        self.apps = apps
+
+
 class Served:
     def __init__(self, tree: Path, server: target.Server, log: Path | None = None, port: int | None = None,
                  timeout_s: float | None = None, env_extra: dict | None = None, jailed: bool = True,
@@ -60,10 +83,17 @@ def __init__(self, tree: Path, server: target.Server, log: Path | None = None, p
         self.proc: subprocess.Popen | None = None
         self.tmp: Path | None = None
         self.stop_jail = None
+        self.pids = None                            # every host pid of the served engine
+        self.foreign: dict[int, dict] = {}          # GPU processes seen while serving that were not the engine's
+        self._stop = threading.Event()
+        self._watcher: threading.Thread | None = None
         self.started_at: float | None = None
         self.ready_s: float | None = None
 
     def __enter__(self) -> "Served":
+        busy = gpu.compute_apps()
+        if busy:
+            raise GpuBusy(busy)
         argv = argv_for(self.server, self.port)
         self.log_offset = self.log.stat().st_size if self.log and self.log.exists() else 0
         if self.artifacts:
@@ -75,11 +105,15 @@ def __enter__(self) -> "Served":
         out = open(self.log, "ab") if self.log else subprocess.DEVNULL
         self.started_at = time.monotonic()
         if self.jailed:
-            self.proc, self.tmp, self.stop_jail = grader.jailed_popen(self.tree, argv, env_extra=self.env_extra,
-                                                                      stdout=out, stderr=out, port=self.port)
+            self.proc, self.tmp, self.stop_jail, self.pids = grader.jailed_popen(
+                self.tree, argv, env_extra=self.env_extra, stdout=out, stderr=out, port=self.port)
         else:
             self.proc = subprocess.Popen(argv, cwd=self.tree, env={**os.environ, **self.env_extra}, stdout=out,
                                          stderr=out, start_new_session=True)
+            self.pids = lambda: gpu.descendants(self.proc.pid)
+        if busy is not None:                        # None: no nvidia-smi here, nothing to watch
+            self._watcher = threading.Thread(target=self._watch, daemon=True)
+            self._watcher.start()
         try:
             self._wait_ready()
         except BaseException:
@@ -105,7 +139,31 @@ def _wait_ready(self) -> None:
                 raise NotReady(f"{health} not ready after {self.timeout_s:.0f}s" + (f"; log: {self.log}" if self.log else ""))
             time.sleep(0.5)
 
+    def _watch(self) -> None:
+        while True:
+            before = self.pids()
+            apps = gpu.compute_apps() or []
+            mine = before | (self.pids() if apps else set())     # a process the engine starts meanwhile is its own
+            for a in apps:
+                if a["pid"] not in mine:
+                    self.foreign.setdefault(a["pid"], a)
+            if self._stop.wait(POLL_S):
+                return
+
+    def _stop_watching(self) -> None:
+        self._stop.set()
+        if self._watcher is not None:
+            self._watcher.join()
+            self._watcher = None
+
+    def exclusive(self) -> None:
+        """Call once the measurement is done: raises `Contaminated` if anything else held the GPU since launch."""
+        self._stop_watching()
+        if self.foreign:
+            raise Contaminated(list(self.foreign.values()))
+
     def __exit__(self, *exc) -> None:
+        self._stop_watching()           # before teardown: an engine going away must not read as someone else
         if self.proc is not None and self.proc.poll() is None:
             try:
                 os.killpg(self.proc.pid, signal.SIGTERM)

lab/validate.py

@@ -90,6 +90,7 @@ def validate(t: target.Target, tier: str = "dev", noise_runs: int = 3, regimes:
                 for n in names:
                     r = suite.REGIMES[n](ctx)
                     values[n].append(r["value"] if r["valid"] else None)
+            srv.exclusive()                     # a band from a shared GPU would skew every submit
         for n in names:
             report["noise"][n] = write_noise(n, values[n], t, {"tier": "short", "split": "seen"})
     report["bands_measured"] = all(d["band_pct"] is not None for d in report["noise"].values()) if names and noise_runs else None