"""Admission sample at window open. A SAMPLE, not a lease. Two populations, because one cannot see the other (IR-100 third limit, todlando): builders carry cargo/rustc/link names; a RUNNING TEST BINARY carries none of them, so a clean builder census says nothing about executions in flight. Each arm gets a positive control, because a filter that cannot express its hunt returns a clean zero. """ import json, os, subprocess, sys, time import psutil sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import lane_run as L DEPS = os.path.join(L.POOL, "debug", "deps") TREE = os.path.dirname(L.POOL) def record_name(): """Name the record after the sha it DESCRIBES, plus the minute it was taken. This used to be the literal "admission-61bfd85c.json" - a tag from an earlier window - so every later admission wrote its census under a sha it did not describe AND overwrote the previous one. A file named for a sha it does not describe is exactly how a stale read passes for a fresh one, which is the same class as the reused .raw paths the window drivers already guard by refusing to read them before the report carries end_utc. """ r = subprocess.run(["git", "rev-parse", "--short", "HEAD"], cwd=TREE, capture_output=True, text=True) tag = r.stdout.strip() or "unknown" return "admission-%s-%s.json" % (tag, time.strftime("%H%M%SZ", time.gmtime())) def running_from_pool(): """Processes whose IMAGE lives in this lane's deps dir - the executions the builder census is structurally blind to.""" hits, unreadable = [], [] for p in psutil.process_iter(["pid", "name", "create_time"]): try: exe = p.exe() or "" except (psutil.Error, OSError) as err: try: unreadable.append({"pid": p.info["pid"], "name": p.info["name"], "error": type(err).__name__}) except Exception: unreadable.append({"pid": None, "name": None, "error": type(err).__name__}) continue if not exe: # An EMPTY path is a BLIND SPOT, not a miss. psutil returns "" for a # few processes instead of raising, and the old filter skipped those # silently - they were counted nowhere and read exactly like "not in # the pool". Measured 2026-09-11: 556 processes, 553 readable, 1 # raised, 2 empty. The hole was 3, and it was reported as 1. unreadable.append({"pid": p.info["pid"], "name": p.info["name"], "error": "EmptyPath"}) continue if os.path.normcase(exe).startswith(os.path.normcase(DEPS)): hits.append({"pid": p.info["pid"], "exe": exe, "birth": p.info["create_time"]}) return {"running": hits, "unreadable": unreadable} SLEEPER = os.path.join(os.environ.get("SystemRoot", r"C:\Windows"), "System32", "ping.exe") LIVE_ARGS = ["-n", "8", "127.0.0.1"] # ~7 s of life, no network egress def _sleeper_named(directory, name): """A process that LIVES, wearing the name/path the filter under test keys on. First attempt used `cargo --version` and a real test binary's `--list`: both are correct subjects and both EXITED before either filter could sample them, so both controls read false and the sample's clean zero proved nothing. A control has to outlive the poll, so the sleeper is copied under the name (builder census keys on NAME) or into the directory (pool filter keys on PATH). """ os.makedirs(directory, exist_ok=True) dst = os.path.join(directory, name) import shutil shutil.copyfile(SLEEPER, dst) return dst def control_builder(): """A process NAMED cargo.exe that lives; the census keys on the name.""" ctl = _sleeper_named(os.path.join(L.EVID, "temp", "control"), "cargo.exe") p = subprocess.Popen([ctl] + LIVE_ARGS, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) seen = False for _ in range(200): if any(b["pid"] == p.pid for b in L.builder_census()["builders"]): seen = True break if p.poll() is not None: break time.sleep(0.01) p.kill(); p.wait() try: os.remove(ctl) except OSError: pass return {"pid": p.pid, "control": ctl, "seen_by_census": seen} def control_pool_exe(): """A process whose IMAGE sits in deps and lives; the filter keys on path.""" if not os.path.isdir(DEPS): return {"skipped": "deps dir absent", "seen_by_filter": None} exe = _sleeper_named(DEPS, "zz-admission-control.exe") p = subprocess.Popen([exe] + LIVE_ARGS, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) seen = False for _ in range(300): if any(h["pid"] == p.pid for h in running_from_pool()["running"]): seen = True break if p.poll() is not None: break time.sleep(0.01) p.kill(); p.wait() try: os.remove(exe) except OSError: pass return {"exe": exe, "pid": p.pid, "seen_by_filter": seen} if __name__ == "__main__": out = { "utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "free_bytes": L.free_bytes(), "pool": L.pool_bytes(), "builders": L.builder_census(), "pool_executions": running_from_pool(), "control_builder": control_builder(), "control_pool_exe": control_pool_exe(), } # Exclusive create, never truncate: naming the record after the resolved sha # makes a collision unlikely, but "unlikely" is not a guarantee, and the fault # this fix exists for was a SILENT overwrite (window 5 and 6 records are gone # from disk; those ENDs stand on the transcript alone). Refusing loudly is the # guarantee; a second admission in the same second is a name to fix, not bytes # to discard. path = os.path.join(L.EVID, record_name()) try: fh = open(path, "x") except FileExistsError: sys.exit("admission record already exists, refusing to overwrite: %s" % path) with fh: json.dump(out, fh, indent=2) print(json.dumps({k: out[k] for k in ("utc", "free_bytes", "control_builder", "control_pool_exe")}, indent=2)) print("by_owner:", out["builders"]["by_owner"]) print("pool_executions:", len(out["pool_executions"]["running"]), "unreadable:", len(out["pool_executions"]["unreadable"]), out["pool_executions"]["unreadable"]) for b in out["builders"]["builders"]: print(" builder", b["pid"], b["name"], b["owner"], b["manifest"]) for h in out["pool_executions"]["running"]: print(" exec", h["pid"], h["exe"])