"""T1's shared pins, paths and small helpers (stdlib only). Every other t1-tune module imports these, so a pin lives in exactly one place (PLAN-deem-T1-tune-v2.md, appendix B). Nothing here touches a model, a GPU or the network. """ from __future__ import annotations import hashlib import json import os import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) # .../bench/deem-2026-09-27/t1-tune DEEM_KIT = os.path.dirname(HERE) # .../bench/deem-2026-09-27 BENCH = os.path.dirname(DEEM_KIT) # .../bench # --- the sources and the held-out sets, each pinned by sha256 ------------------------------------- POC_DIR = os.path.join(BENCH, "dinner-persona-baseline-2026-09-12", "out", "dinners") POC_TABLES = ("ancients", "cross", "operator", "span") # fold n = POC_TABLES[n-1] POC_ASSIGNMENTS = ("GGG", "GMG", "MGM", "MMM") POC_MOVES = ("M1", "M2") POC_JUDGE = os.path.join(POC_DIR, "judgments-chairs.jsonl") S0_PATH = os.path.join(BENCH, "jev-2026-09-21", "kit", "task_c.json") S0_SHA = "61a0b1d6e7cf66fd91992c0c3a4572f6fbae90e6471048fd7dc1eb41ff2d7f39" H48_PATH = os.path.join(DEEM_KIT, "kit", "h48.json") H48_SHA = "" VOICE_PATH = os.path.join(DEEM_KIT, "d1-voice", "kit", "voice-items.json") VOICE_SHA = "cea9e12d7983a3c52a204c1ed3e2d6c709e7504066f7ff37c1f86153bcf95d3c" #: every wall read on master: D0's pinned read and the COI series' three 2026-09-22 receipts WALL_PATHS = (os.path.join(DEEM_KIT, "kit", "wall-5f4a4e1c.json"), os.path.join(BENCH, "cost-of-intelligence-2026-09-23", "KIT.lock.receipts", "wall-6429e0c8.json"), os.path.join(BENCH, "cost-of-intelligence-2026-09-23", "KIT.lock.receipts", "wall-8dd38e12.json"), os.path.join(BENCH, "cost-of-intelligence-2026-09-23", "KIT.lock.receipts", "wall-bedf6e9a.json")) BUILD_KIT_DIR = os.path.join(BENCH, "jev-2026-09-21", "kit") # S0's builder: NAME_TOKENS, imported HOSTOS_DIR = os.path.join(BENCH, "render-grid-2026-09-24") # --- the model, the runtime, D0's instrument ----------------------------------------------------- BENCH_DIR = os.environ.get("DEEM_BENCH_DIR", os.path.expanduser("/workshop/bench-deem-2026-09-27")) BASE_CKPT = os.path.join(BENCH_DIR, "weights", "deem-0.8-v1") BASE_SHA = "80438125177c0681e855158d46720cbbd4ff07ea620cd55b0f456aafbfcf778d" BASE_REVISION = "8cbabbb2c4a7ef13c6b43f0ef3ae4157983c6d21" DEEM_COMMIT = "6755b30bf6bbd9a81f8db6ba42cc0fd62c9f4719" CKPT_ROOT = os.path.join(BENCH_DIR, "t1", "ckpt") # disk, never git, never /dev/shm VENV_PY = os.path.join(BENCH_DIR, ".venv", "bin", "python") RUN_DEEM_D0_SHA = "78fe9562a7de88ed9a30fa03bb3b2c769b99d496301ae6d8972a2e6edc09b2a6" HOSTOS_SHA = "a6a74bba57250be1936825466c6614876d757cf81373fcf6fa0b48587cf0e3ad" GUESTS = ("darwin", "einstein", "hypatia", "ibn_sina", "sagan", "socrates") # S0's option order LETTERS = "ABCDEF" LETTER_IDS = (32, 33, 34, 35, 36, 37) # D0 receipts/letter-ids.json SEED_HEADLINE = 20260927 # --- guards (plan 6.1 and the lane brief) -------------------------------------------------------- TEMP_PAUSE_C = 100.0 TEMP_RESUME_C = 90.0 MEM_FLOOR_BYTES = 6 * 1024 ** 3 DISK_FLOOR_BYTES = 6 * 1024 ** 3 sys.path.insert(0, BUILD_KIT_DIR) from build_kit import NAME_TOKENS # noqa: E402 (S0's own rule, imported, never copied) def sha(data) -> str: return hashlib.sha256(data if isinstance(data, bytes) else data.encode("utf-8")).hexdigest() def file_sha(path: str) -> str: h = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(1 << 22), b""): h.update(chunk) return h.hexdigest() def utc() -> str: return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def load_pinned(path: str, want: str): raw = open(path, "rb").read() if sha(raw) != want: raise SystemExit("%s is not the pinned file (sha256 %s, want %s)" % (path, sha(raw), want)) return json.loads(raw) def set_fingerprint(texts) -> str: """S0's convention: sha256 over the sorted distinct text sha256 values, first 16 hex.""" return sha("".join(sorted({sha(t) for t in texts})))[:16] def labelled_fingerprint(rows) -> str: """The (text, label) fingerprint: sha256 over sorted 'text_sha256:label' strings, first 16 hex.""" return sha("\n".join(sorted("%s:%s" % (sha(r["text"]), r["label"]) for r in rows)))[:16] def names_in(text: str, exclude: str | None = None) -> list[str]: """Guests whose NAME_TOKENS occur in the text (S0's substring rule), optionally excluding one.""" low = text.lower() return [g for g in GUESTS if g != exclude and any(tok in low for tok in NAME_TOKENS[g])] def pkg_temp_c() -> float | None: """The CPU package temperature (x86_pkg_temp), None if unreadable.""" for zone in sorted(os.listdir("/sys/class/thermal")): if not zone.startswith("thermal_zone"): continue try: if open("/sys/class/thermal/%s/type" % zone).read().strip() == "x86_pkg_temp": return int(open("/sys/class/thermal/%s/temp" % zone).read()) / 1000 except OSError: continue return None def mem_available_bytes() -> int: for line in open("/proc/meminfo"): if line.startswith("MemAvailable:"): return int(line.split()[1]) * 1024 raise RuntimeError("MemAvailable not in /proc/meminfo") def vm_rss_bytes(pid: int | None = None) -> int | None: try: for line in open("/proc/%s/status" % (pid or "self")): if line.startswith("VmRSS:"): return int(line.split()[1]) * 1024 except OSError: return None return None def disk_free_bytes(path: str = "/") -> int: st = os.statvfs(path) return st.f_bavail * st.f_frsize def rapl() -> dict: """/run/rapl/energy (rapl-sampler.service, 1 Hz): {domain: [uj, max_range, sample_ts]}; {} if absent.""" try: out = {} with open("/run/rapl/energy") as fh: for line in fh: ts, dom, uj, mx = line.split() out[dom] = [int(uj), int(mx), float(ts)] return out except (OSError, ValueError): return {} def read_os() -> dict: sys.path.insert(0, HOSTOS_DIR) import hostos # the canonical copy, by path return hostos.read_os() def write_json(path: str, obj, indent=1) -> None: os.makedirs(os.path.dirname(path), exist_ok=True) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: json.dump(obj, fh, ensure_ascii=False, indent=indent, sort_keys=True) fh.write("\n") os.replace(tmp, path)