"""Lane producer runner for the #304 Rust allocation. Samples volume capacity and a scoped builder census around one producer, enforces a hard deadline, and retains the real exit code, raw stderr, stdout and the observed descendant identities. No retry, no kill-by-image, no PID-only kill. """ import ctypes, json, os, subprocess, sys, tempfile, time import psutil # Producer output is nextest's, which draws box characters. A cp1252 stdout # raises UnicodeEncodeError on them and kills the driver AFTER the producer has # already written its evidence — losing every later producer in the sequence. for _stream in (sys.stdout, sys.stderr): try: _stream.reconfigure(encoding="utf-8", errors="replace") except (AttributeError, ValueError): pass EVID = os.path.dirname(os.path.abspath(__file__)) VOLUME = "C:\\" BUILDER_NAMES = { "cargo.exe", "rustc.exe", "link.exe", "cargo-nextest.exe", "rustdoc.exe", "lld-link.exe", "runner.worker.exe", } # Allocation gates (304-RUST-ALLOCATION.txt) STOP_FREE = 34_359_738_368 # stop if free <= 32 GiB MAX_GROWTH = 68_719_476_736 # stop if start_free - free >= 64 GiB def free_bytes(path=VOLUME): f = ctypes.c_ulonglong(0) ok = ctypes.windll.kernel32.GetDiskFreeSpaceExW( ctypes.c_wchar_p(path), ctypes.byref(f), None, None) if not ok: raise OSError("GetDiskFreeSpaceExW failed on %s" % path) return f.value # THE POOL IS THE LANE'S, NAMED BY THE LANE, NEVER INHERITED. This was a literal # path to the hertz-304 worktree's target. A copy of this rig driving a DIFFERENT # lane would then have censused, and reported growth for, a pool it was not # running in -- the same class as the admission record named after a sha it did # not describe, and as the two cells this lane guards. There is no default: an # unset var REFUSES, because a rig that silently measures the wrong subject is # worse than one that will not start. POOL = os.environ.get("HERTZ_LANE_POOL") if not POOL: raise SystemExit( "HERTZ_LANE_POOL is unset. Set it to THIS lane's target dir, e.g.\n" " HERTZ_LANE_POOL=/target\n" "The rig refuses to guess: censusing another lane's pool reports growth " "and executions that are not this lane's.") if not os.path.isdir(POOL): raise SystemExit("HERTZ_LANE_POOL is not a directory: %s" % POOL) def pool_bytes(root=POOL): """Logical bytes and file count of THIS lane's pool subtree. A volume free-space meter is unattributable whenever any other lane moves bytes (doyle, 2026-09-11); this number is the lane's own regardless of what the volume does. Unreadable entries are counted, never silently skipped. """ # A missing root would walk to a clean zero and read as "no growth". if not os.path.isdir(root): raise OSError("pool subtree is absent, not empty: %s" % root) total, files, unreadable = 0, 0, 0 for base, _dirs, names in os.walk(root, onerror=lambda _e: None): for name in names: try: total += os.lstat(os.path.join(base, name)).st_size files += 1 except OSError: unreadable += 1 return {"bytes": total, "files": files, "unreadable": unreadable} def _attribute(proc): """Who owns this builder: CI, an editor's analyzer, this session, or unknown. A bare count is not readable (todlando, 2026-09-11): an editor-driven `cargo check --workspace --all-targets` counts exactly like a CI leg, so a nonzero census reads as "CI or a peer" when it may be a language server, and a zero leases nothing. Attribute, and carry the manifest when it is legible. """ chain, manifest = [], None try: manifest = next((a for a in proc.cmdline() if a.endswith("Cargo.toml")), None) except (psutil.Error, OSError): pass q = proc try: for _ in range(8): q = q.parent() if q is None: break chain.append(q.name().lower()) except (psutil.Error, OSError): chain.append("") if any("runner." in c for c in chain): owner = "ci" elif any("rust-analyzer" in c for c in chain): owner = "analyzer" elif "" in chain: owner = "unknown" else: owner = "agent" return owner, chain, manifest def builder_census(): seen, unreadable = [], [] for p in psutil.process_iter(["pid", "name", "create_time"]): try: name = (p.info["name"] or "").lower() except Exception as exc: unreadable.append(str(exc)) continue if name in BUILDER_NAMES: owner, chain, manifest = _attribute(p) seen.append({"pid": p.info["pid"], "name": name, "birth": p.info["create_time"], "owner": owner, "manifest": manifest, "ancestry": chain}) return {"builders": seen, "unreadable": unreadable, "by_owner": {o: sum(1 for b in seen if b["owner"] == o) for o in ("ci", "analyzer", "agent", "unknown")}} def priv_for(label): """The private TMP base for one producer. ONE definition, two callers - the runner that exports it and the window driver that records which base an arm actually got. Window 6's arm B was a non-measurement because the driver drove the switch through env_extra (child-only) while this branch reads the PARENT env; a second copy of this rule is how that stays possible. """ if os.environ.get("HERTZ_RIG_TMP_IN_REPO") == "1": return os.path.join(EVID, "temp", label) base = (os.environ.get("TEMP") or os.environ.get("TMP") or tempfile.gettempdir()) # Named after THIS rig dir, not a literal lane tag: two lanes running from # two copies of this rig must not share one TMP base, or a stale fixture # from the other lane is reachable under a name this lane believes is its own. return os.path.join(base, os.path.basename(EVID) + "-rig", label) def run(label, argv, cwd, deadline_s, env_extra=None, stdout_path=None): env = dict(os.environ) for k in ("SPT_ENDPOINT", "SPT_AGENT", "SPT_SESSION_ID", "SPT_HOME", "CARGO_TARGET_DIR", "RUSTFLAGS"): env.pop(k, None) # THE RIG'S OWN TMP REDIRECT WAS A DEFECT, measured 2026-09-11 after # w5-lib-full: this base used to be EVID/temp/