kernels/p6_harness_probe.py

"""Phase 6 pre-flight on free CPU, second pass: how do we get a working lm-eval without moving the image?

The first pass (12:13Z) answered a question nobody had asked and killed two assumptions at once:

* `pip install --user --no-deps lm-eval==0.4.13` **cannot be imported at all** -- `lm_eval/api/metrics.py:11`
  does `import sacrebleu` at module scope, so `--tasks list` exits 1 with `ModuleNotFoundError`. The
  `--no-deps` flag was E-045/12's answer to "a resolving install might drop a different transformers into
  ~/.local"; it bought that safety at the price of a harness that does not start. Review-found, CPU-proved.
* The rest of this file's own first draft then died on `%`-formatting a template that contains `%s` specs.

So the real question is which install keeps the image's `transformers 5.0.0` and `torch` *and* gets the
metric libraries. lm-eval declares `transformers>=4.1`, which 5.0.0 already satisfies -- pip leaves a
satisfied requirement alone -- so a resolving install may be safe here. This measures it rather than
arguing: it snapshots the versions and file locations of the packages Phase 6 depends on, installs with
deps, snapshots again, and only then parses `--tasks list` and dumps the nine task configs.

**No eval data is touched and none can be:** installing, listing task ids and reading YAML configs never
calls `datasets.load_dataset`; that happens only when a task is requested against a model, which is
Phase 6's own step (§3.3).
"""
import json
import os
import re
import subprocess
import sys

os.chdir("/kaggle/working")
PIN = "0.4.13"
TASKS = ["arc_challenge", "arc_easy", "hellaswag", "mmlu", "piqa", "truthfulqa_mc1",
         "truthfulqa_mc2", "winogrande", "gsm8k"]
PRIMARY_METRIC = {"arc_challenge": "acc,none", "arc_easy": "acc,none", "hellaswag": "acc,none",
                  "mmlu": "acc,none", "piqa": "acc,none", "truthfulqa_mc1": "acc,none",
                  "truthfulqa_mc2": "acc,none", "winogrande": "acc,none",
                  "gsm8k": "exact_match,flexible-extract"}
WATCH = ["transformers", "torch", "numpy", "pandas", "datasets", "evaluate", "accelerate", "scipy",
         "sympy", "sacrebleu", "lm_eval"]

SNAPSHOT = """
import importlib.metadata as md, json, os, sys
out = {}
for name in %s:
    try:
        mod = __import__(name)
        where = getattr(mod, "__file__", None) or "?"
    except Exception:
        where = None
    try:
        ver = md.version(name.replace("_", "-"))
    except Exception:
        ver = getattr(__import__(name), "__version__", None) if where else None
    out[name] = {"version": str(ver) if ver else None,
                 "file": where, "user_dir": bool(where and "/.local/" in where)}
print("SNAPSHOT " + json.dumps(out, default=str))
""" % (WATCH,)


def sh(argv, label, timeout=2400):
    print("=== " + label, flush=True)
    p = subprocess.run(argv, capture_output=True, text=True, timeout=timeout)
    out = (p.stdout or "") + (p.stderr or "")
    print(label, "rc", p.returncode, "chars", len(out), flush=True)
    return p.returncode, out


def snapshot(label):
    open("snap.py", "w").write(SNAPSHOT)
    rc, out = sh([sys.executable, "snap.py"], "snapshot_" + label, 600)
    line = next((l for l in out.splitlines() if l.startswith("SNAPSHOT ")), "")
    d = json.loads(line[len("SNAPSHOT "):]) if line else {}
    for k, v in sorted(d.items()):
        print("   %-12s %-10s %s" % (k, v["version"], ("<USER>" if v["user_dir"] else v["file"])),
              flush=True)
    return d


def parse_ids(text):
    return set(re.findall(r"^\s*(?:-\s*)?([a-z][a-z0-9_]{2,})\s*$", text, re.M))


before = snapshot("before")

rc, out = sh([sys.executable, "-m", "pip", "install", "--user", "--quiet",
              "lm-eval==" + PIN], "pip_install_with_deps", 2400)
print("   ", out.strip()[-400:], flush=True)
after = snapshot("after")
moved = {k: (before.get(k, {}).get("version"), v["version"])
         for k, v in after.items() if before.get(k, {}).get("version") != v["version"]
         and k != "lm_eval"}
print("PIP_MOVED", json.dumps(moved, default=str), flush=True)
assert after["transformers"]["version"] == "5.0.0", "transformers moved off 5.0.0"
assert after["transformers"]["file"] == before["transformers"]["file"], "transformers moved location"

rc, listed = sh([sys.executable, "-m", "lm_eval", "--tasks", "list"], "tasks_list", 1200)
ids = parse_ids(listed)
print("PARSED_IDS", len(ids), "all nine present:", all(t in ids for t in TASKS), flush=True)
if not ids:
    print("RAW_TAIL", repr(listed[-1500:]), flush=True)

open("taskconfig.py", "w").write("""
import json
from lm_eval.tasks import TaskManager
tm = TaskManager()
try:
    out = {"index_type": str(type(next(iter(tm.task_index.values()))))}
except Exception as e:
    out = {"index_type": "empty or unreachable: " + type(e).__name__}
recs = {}
for t in %s:
    r = {"registered": t in tm.task_index}
    try:
        item = tm.task_index[t]
        cfg = item if isinstance(item, dict) else getattr(item, "config", {})
        r["keys"] = sorted(cfg)[:14]
        r["num_fewshot"] = cfg.get("num_fewshot")
        r["metric_list"] = [(m.get("metric_uri") if isinstance(m, dict) else m)
                            for m in (cfg.get("metric_list") or [])]
        r["dataset_path"] = cfg.get("dataset_path")
        r["dataset_name"] = cfg.get("dataset_name")
        r["test_split"] = cfg.get("test_split")
        r["validation_split"] = cfg.get("validation_split")
        r["fewshot_split"] = cfg.get("fewshot_split")
    except Exception as e:
        r["error"] = "%%s: %%s" %% (type(e).__name__, str(e)[:220])
    recs[t] = r
print("TASKCONFIG " + json.dumps({"meta": out, "tasks": recs}, default=str))
""" % (TASKS,))
rc, tc = sh([sys.executable, "taskconfig.py"], "task_configs", 1200)
line = next((l for l in tc.splitlines() if l.startswith("TASKCONFIG ")), "")
info = json.loads(line[len("TASKCONFIG "):]) if line else {"meta": {}, "tasks": {}}
cfgs = info.get("tasks", {})
print("TASKCONFIG_META", json.dumps(info.get("meta"), default=str), flush=True)
for t in TASKS:
    c = cfgs.get(t, {})
    m = PRIMARY_METRIC[t]
    print("TASK %-16s fewshot=%-6s test=%-10s val=%-8s metric %-34s declared=%s" % (
        t, c.get("num_fewshot"), c.get("test_split"), c.get("validation_split"), m,
        m in (c.get("metric_list") or [])), flush=True)

print("HARNESS_PROBE_JSON", json.dumps({
    "pip_moved": moved, "tasks_listed": len(ids), "all_present": all(t in ids for t in TASKS),
    "primary_declared": {t: PRIMARY_METRIC[t] in (cfgs.get(t) or {}).get("metric_list", [])
                         for t in TASKS}, "configs": cfgs}, default=str)[:6000], flush=True)
ok = bool(ids) and all(t in ids for t in TASKS) and bool(cfgs)
print("VERDICT P6_HARNESS_PROBE", "OK" if ok else "PROBLEM", flush=True)
raise SystemExit(0 if ok else 6)