Neural pathways: the store's directories as a network¶

Open in Colab

The p4 store grows as a directory tree: a p4 holds .claude directories, and a .claude may hold a p4, never a p4 directly in a p4. Read that tree as a network's pathways: one small model per directory node, each trained on its own era of the store's text, a parent built from its children. nested-store.md measured that tree against one large model. This notebook runs the controls that settle what the tree adds: is it the φ structure, or only the splitting into small models?

  1. Trees. The same nodes, data shards and budget, three ways:

    • Zeckendorf tree: nodes at the store's Zeckendorf addresses, the p4 nesting rule.
    • Binary tree: the same number of nodes at power-of-2 heap addresses, no adjacency rule.
    • Flat: independent nodes, no tree.

    All three are compared against one large model.

  2. Masks as routes. The Zeckendorf nodes, pruned four ways (φ's adjacency mask, 2:4, a random half of the rows, the heaviest half). Each is used as an ensemble and as a route that sends the task to one node. If the heaviest-rows mask routes as well as φ, the adjacency rule adds nothing to routing.

  3. Early stopping. Every pretraining checks its ranking on held-back dev pools every CHECK_S seconds and keeps its best weights. Each model's curve shows whether longer pretraining helps or hurts; the T4 runs suggested it hurts. The dev pools are kept out of every score.

Everything is saved to Google Drive as it is produced, and Run all after a disconnect resumes where Drive says it stopped (see the Q-matrix notebook for how). Secrets: GITHUB_TOKEN (read access to the store repository; write access to ezexe/phi only with PUSH_RESULTS).

Runtime: the T4 GPU if you can get one (the T4 copy is pathways_colab_t4.ipynb); the CPU works, slower. Each seed pretrains four model families for MINUTES each, then fine-tunes about 120 small models: roughly an hour a seed on a T4.

0. Settings¶

Edit, then Runtime → Run all. A new RUN name starts a fresh run; the same name resumes.

In [ ]:
DRIVE_DIR = "/content/drive/MyDrive/phi-colab"   # everything this notebook keeps lives under DRIVE_DIR/RUN
RUN = "pathways-1"

STORE_REPO = "ezexe/AudioPlugin-p4"   # the p4 store measured
STORE_BRANCH = "master"
STORE_PATH = ""                       # or a store already on disk/Drive; then STORE_REPO is not cloned
PHI_REF = "main"
P4_REF = "main"

K_LIST = [4]                          # nodes per tree; k = 4 is where the Zeckendorf ensemble won before
SEEDS = [0, 1, 2]                     # three seeds: the seed-to-seed noise was as large as most differences
MINUTES = 6.0                         # pretraining budget per family (each node gets MINUTES / k)
EPOCHS = 10
FOLDS = 4
CHECK_S = 30                          # early stopping: seconds between dev checks; 0 turns it off

DEVICE = "auto"                       # auto: the GPU when the runtime has one; "cuda" requires it; "cpu" forces the CPU

PUSH_RESULTS = False                  # True: commit the results into ezexe/phi on PUSH_BRANCH (a PR is yours to open)
PUSH_BRANCH = "colab/pathways-results"

1. Drive, state and logging¶

Mounts Drive and creates the run's folders. state.json records each finished step; every command's output is appended to a log on Drive as it runs.

In [ ]:
import datetime, json, os, pathlib, shutil, subprocess, sys, time

try:
    from google.colab import drive
    drive.mount("/content/drive")
    IN_COLAB = True
except ImportError:                    # run elsewhere: DRIVE_DIR is an ordinary folder
    IN_COLAB = False

ROOT = pathlib.Path(DRIVE_DIR) / RUN
for d in ("results", "cache", "logs", "models", "pairs"):
    (ROOT / d).mkdir(parents=True, exist_ok=True)
STATE = ROOT / "state.json"
WORK = pathlib.Path("/content/work" if IN_COLAB else ROOT.parent / (RUN + "-work"))
WORK.mkdir(parents=True, exist_ok=True)


def now():
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M")


def save_json(path, obj):
    """Write through a temporary file and rename, so a disconnect mid-write never leaves half a file on Drive."""
    path = pathlib.Path(path)
    tmp = path.with_name(path.name + ".tmp")
    tmp.write_text(json.dumps(obj, indent=1), encoding="utf-8")
    os.replace(tmp, path)


def load_state():
    try:
        return json.loads(STATE.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {"created": now(), "done": {}}


def mark(step, **info):
    st = load_state()
    st.setdefault("done", {})[step] = {"at": now(), **info}
    save_json(STATE, st)


def done(step):
    return step in load_state().get("done", {})


HEARTBEAT_S = 120                      # a line when a command has printed nothing for this long


def run(cmd, log, cwd=None, env=None, check=True):
    """Run a command, echo its output here with the time and append it to logs/<log>.log on Drive line by line;
    after HEARTBEAT_S of silence, say it is still running and what it last said."""
    import threading
    path = ROOT / "logs" / f"{log}.log"
    started = time.time()
    last = {"t": started, "line": "(started)"}
    with open(path, "a", encoding="utf-8") as f:
        f.write(f"\n# {now()} $ {' '.join(map(str, cmd))}\n")
        f.flush()
        p = subprocess.Popen([str(c) for c in cmd], cwd=cwd,
                             env={**os.environ, "PYTHONUNBUFFERED": "1", **(env or {})},
                             stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
        stop = threading.Event()

        def beat():
            while not stop.wait(15):
                if time.time() - last["t"] >= HEARTBEAT_S:
                    print(f"{time.strftime('%H:%M:%S')} ... {log} still running ({(time.time() - started) / 60:.0f} min); "
                          f"last: {last['line'][:100]}", flush=True)
                    last["t"] = time.time()
        threading.Thread(target=beat, daemon=True).start()
        for line in p.stdout:
            last["t"], last["line"] = time.time(), line.strip()
            print(f"{time.strftime('%H:%M:%S')} {line}", end="", flush=True)
            f.write(line)
            f.flush()
        code = p.wait()
        stop.set()
        f.write(f"# exit {code} after {(time.time() - started) / 60:.1f} min\n")
    if check and code != 0:
        raise RuntimeError(f"{cmd[0]} exited {code}; see {path}")
    return code


print("run folder:", ROOT)
print("done so far:", ", ".join(load_state().get("done", {})) or "nothing")

2. Repositories, pinned¶

Clones ezexe/phi, ezexe/p4 and the store into the runtime's disk. The first session records each commit in state.json; later sessions check out those same commits, so a resumed run measures the same code on the same data. Delete pins from state.json to move to newer commits.

In [ ]:
def token():
    if not IN_COLAB:
        return os.environ.get("GITHUB_TOKEN", "")
    from google.colab import userdata
    try:
        return userdata.get("GITHUB_TOKEN") or ""
    except Exception:                  # no such secret, or access not granted to this notebook
        return ""


def url(repo):
    t = token()
    return f"https://x-access-token:{t}@github.com/{repo}.git" if t else f"https://github.com/{repo}.git"


def checkout(repo, ref, dest, pin):
    """Clone with full history (the shards are cut by when git first added each file) but no LFS objects, then
    check out the pinned commit, or `ref` on the first session."""
    env = {"GIT_LFS_SKIP_SMUDGE": "1", "GIT_TERMINAL_PROMPT": "0"}
    if not (dest / ".git").exists():
        r = subprocess.run(["git", "clone", "-q", "--filter=blob:none", "--no-checkout", url(repo), str(dest)],
                           env={**os.environ, **env}, capture_output=True, text=True)
        if r.returncode != 0:          # the error may hold the URL, so the token is stripped before printing
            raise RuntimeError(f"cloning {repo} failed: " + r.stderr.replace(token() or "\0", "***"))
    target = pin or f"origin/{ref}"
    subprocess.run(["git", "-C", str(dest), "checkout", "-q", "--force", target], env={**os.environ, **env}, check=True)
    return subprocess.run(["git", "-C", str(dest), "rev-parse", "HEAD"], capture_output=True, text=True).stdout.strip()


st = load_state()
pins = st.get("pins", {})
PHI = WORK / "phi"
P4 = WORK / "p4"
pins["phi"] = checkout("ezexe/phi", PHI_REF, PHI, pins.get("phi"))
pins["p4"] = checkout("ezexe/p4", P4_REF, P4, pins.get("p4"))
if STORE_PATH:
    STORE = pathlib.Path(STORE_PATH)
else:
    STORE = WORK / "store"
    pins["store"] = checkout(STORE_REPO, STORE_BRANCH, STORE, pins.get("store"))
st = load_state()
st["pins"] = pins
save_json(STATE, st)

PLUGIN = P4 / "lbiza" / "plugins" / "vlds"
PICK = PLUGIN / "scripts" / "pick_model.py"
SUITE = PHI / "scripts" / "session" / "nested_suite.py"
run([sys.executable, "-m", "pip", "install", "-q", "--no-deps", "-e", str(PHI)], "setup")
sys.path.insert(0, str(PHI))

# the training device, checked once per runtime: a short pretrain and fine-tune on it, the CPU if that fails
import torch
import phi_rank as pr
want = "cpu" if DEVICE == "cpu" else ("cuda" if torch.cuda.is_available() else "cpu")
if DEVICE == "cuda" and want != "cuda":
    raise RuntimeError("DEVICE = 'cuda' but this runtime has no GPU: Runtime -> Change runtime type -> T4 GPU, then Run all")
if want == "cuda":
    try:
        os.environ["PHI_DEVICE"] = "cuda"
        m = pr.Ranker(1 << 13, 16)
        texts = [f"the ring buffer drain rule {i}" for i in range(64)]
        assert pr.pretrain(m, list(zip(texts, texts[1:])), 0.05, log=lambda *_: None) > 0
        pr.finetune(m, [{"task": "the ring buffer", "rows": [{"text": t, "label": int(i < 3)}
                                                               for i, t in enumerate(texts)]}], epochs=1)
        assert next(m.parameters()).device.type == "cpu"
    except Exception as e:                 # under auto a broken CUDA path must not stop the run: the CPU still measures
        if DEVICE == "cuda":
            raise RuntimeError(f"GPU self-check failed: {type(e).__name__}: {e}") from e
        print(f"GPU self-check failed ({type(e).__name__}: {e}); training on the CPU")
        want = "cpu"
os.environ["PHI_DEVICE"] = want
gpu = torch.cuda.get_device_name(0) if want == "cuda" else ""
print(f"training on: {want} {gpu}".strip(), f"| CPU cores: {os.cpu_count()}")
st = load_state()
devices = st.setdefault("devices", [])
if devices and devices[-1]["device"] != (gpu or want):
    print(f"note: earlier sessions of this run trained on {devices[-1]['device']}. Runs already on Drive keep "
          "their models; new runs will differ in how many steps fit the time budget.")
devices.append({"at": now(), "device": gpu or want, "cpus": os.cpu_count()})
save_json(STATE, st)

SESSION = next(p for p in sorted((STORE / "sessions").iterdir()) if (p / "vlds").is_dir()) / "vlds"
print("pins:", json.dumps(pins))
print("store:", STORE, "| a session store for the plugin:", SESSION.relative_to(STORE))

4. The store's labelled pairs¶

The vlds plugin exports every labelled pool of the store (pick_model.py pairs). The file is kept on Drive.

In [ ]:
PAIRS = ROOT / "pairs" / "pairs.jsonl"
if not (done("pairs") and PAIRS.exists()):
    run([sys.executable, str(PICK), "--store", str(SESSION), "pairs", "--jsonl", str(PAIRS)], "pairs")
    n = sum(1 for _ in open(PAIRS, encoding="utf-8"))
    mark("pairs", rows=n)
pools = len({json.loads(l)["pool_id"] for l in open(PAIRS, encoding="utf-8")})
print(f"{pools} labelled pools in {PAIRS}")
assert pools >= FOLDS, "too few labelled pools to fold"

5. The pathways suite¶

One pathways_suite.py run per (k, seed). Its --cache folder on Drive keeps every pretrained family and every model's out-of-fold scores as they finish, so a run cut short continues from there. A finished run's JSON lands in results/pathways-k<k>-s<seed>.json and is never redone. Each stage prints a timestamped line, and a heartbeat follows any 2 minutes of silence.

In [ ]:
PATHWAYS = PHI / "scripts" / "session" / "pathways_suite.py"


def pw_done(k, seed):
    p = ROOT / "results" / f"pathways-k{k}-s{seed}.json"
    try:
        return len(json.loads(p.read_text())["runs"]) == 1
    except (OSError, ValueError, KeyError):
        return False


plan = [(k, s) for k in K_LIST for s in SEEDS]
for i, (k, seed) in enumerate(plan, 1):
    if pw_done(k, seed):
        print(f"[{i}/{len(plan)}] k={k} seed={seed}: done")
        continue
    print(f"[{i}/{len(plan)}] k={k} seed={seed}: running on {os.environ.get('PHI_DEVICE', 'cpu')} (resumes from Drive's cache)")
    t0 = time.time()
    run([sys.executable, str(PATHWAYS), "--p4", str(STORE), "--pairs", str(PAIRS), "--k", k, "--minutes", MINUTES,
         "--epochs", EPOCHS, "--folds", FOLDS, "--check-s", CHECK_S, "--seeds", seed, "--cache", ROOT / "cache",
         "--out", ROOT / "results" / f"pathways-k{k}-s{seed}.json"], f"pathways-k{k}-s{seed}")
    mark(f"pathways-k{k}-s{seed}", seconds=round(time.time() - t0))
print("pathways: every planned run is on Drive")

6. Summary and verdicts¶

Reads every pathways result on Drive and writes summary.json, summary.md and summary.png. A claim needs every seed of a k to agree before it is called, and one with no finished run reads not measured.

In [ ]:
import statistics
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
from IPython.display import Markdown, Image, display

runs = {}
for p in sorted((ROOT / "results").glob("pathways-k*-s*.json")):
    d = json.loads(p.read_text())
    for r in d["runs"]:
        runs.setdefault(d["k"], []).append(r)


def every(k, f):
    return all(f(r["quality"]) for r in runs[k])


def verdict(ok, measured):
    return "not measured" if not measured else ("holds" if ok else "does not hold")


ks = sorted(runs)
done_k = [k for k in ks if len(runs[k]) >= 2]
claims = {
    "the Zeckendorf directory tree beats the binary directory tree (ensemble, every seed)":
        verdict(all(every(k, lambda q: q["zeck_ens"] > q["btree_ens"]) for k in done_k), bool(done_k)),
    "the Zeckendorf tree's root beats the binary tree's root (every seed)":
        verdict(all(every(k, lambda q: q["zeck_root"] > q["btree_root"]) for k in done_k), bool(done_k)),
    "the Zeckendorf tree beats flat independent nodes (ensemble, every seed)":
        verdict(all(every(k, lambda q: q["zeck_ens"] > q["flat_ens"]) for k in done_k), bool(done_k)),
    "the Zeckendorf tree beats one large model (ensemble, every seed)":
        verdict(all(every(k, lambda q: q["zeck_ens"] > q["big"]) for k in done_k), bool(done_k)),
    "the φ mask routes to the best node more often than the heaviest-rows mask (every seed)":
        verdict(all(every(k, lambda q: q["phi_route_hit"] > q["heavy_route_hit"]) for k in done_k), bool(done_k)),
    "φ routing ranks better than routing by a random half of the rows (every seed)":
        verdict(all(every(k, lambda q: q["phi_routed"] > q["rand_routed"]) for k in done_k), bool(done_k)),
    "φ routing ranks better than 2:4 routing (every seed)":
        verdict(all(every(k, lambda q: q["phi_routed"] > q["24_routed"]) for k in done_k), bool(done_k)),
    "the one large model's best dev ranking comes before its pretraining budget ends (every seed)":
        verdict(all(r["early_stopping"]["big"].get("best_steps", 0) < r["early_stopping"]["big"].get("steps", 0)
                    for k in done_k for r in runs[k]),
                bool(done_k) and all("best_steps" in r["early_stopping"].get("big", {}) for k in done_k for r in runs[k])),
}
cols = ["big", "zeck_ens", "btree_ens", "flat_ens", "zeck_root", "btree_root", "phi_routed", "24_routed",
        "rand_routed", "heavy_routed", "phi_route_hit", "24_route_hit", "rand_route_hit", "heavy_route_hit"]
rows = [{"k": k, "seeds": len(runs[k]), **{c: round(statistics.mean(r["quality"][c] for r in runs[k]), 4) for c in cols}}
        for k in ks]
summary = {"run": RUN, "at": now(), "pins": load_state().get("pins"), "devices": load_state().get("devices"),
           "plan": {"k": K_LIST, "seeds": SEEDS, "minutes": MINUTES, "epochs": EPOCHS, "check_s": CHECK_S},
           "by_k": rows, "early_stopping": {k: [r["early_stopping"] for r in runs[k]] for k in ks}, "claims": claims}
save_json(ROOT / "summary.json", summary)

lines = [f"# Neural pathways on Colab: {RUN}", "", f"Pinned commits: `{json.dumps(summary['pins'])}`", "",
         "Held-out r-precision, mean over seeds (dev pools excluded):", "",
         "| k | seeds | one large | Zeckendorf tree | binary tree | flat | Zeck root | binary root |",
         "|---|---|---|---|---|---|---|---|"]
for r in rows:
    lines.append(f"| {r['k']} | {r['seeds']} | {r['big']} | {r['zeck_ens']} | {r['btree_ens']} | {r['flat_ens']} "
                 f"| {r['zeck_root']} | {r['btree_root']} |")
lines += ["", "Routing by each mask (routed r-precision / how often the routed node is the best node):", "",
          "| k | φ | 2:4 | random half | heaviest half |", "|---|---|---|---|---|"]
for r in rows:
    lines.append(f"| {r['k']} | {r['phi_routed']} / {r['phi_route_hit']} | {r['24_routed']} / {r['24_route_hit']} "
                 f"| {r['rand_routed']} / {r['rand_route_hit']} | {r['heavy_routed']} / {r['heavy_route_hit']} |")
lines += ["", "Early stopping, the one large model (best dev score at step, of steps run):", ""]
for k in ks:
    for r in runs[k]:
        e = r["early_stopping"].get("big", {})
        lines.append(f"- k = {k}, seed {r['seed']}: " + (f"best {e['best_score']} at step {e['best_steps']:,} of {e['steps']:,}"
                                                            if "best_steps" in e else "early stopping was off"))
lines += ["", "## Verdicts", ""] + [f"- {c}: **{v}**" for c, v in claims.items()]
(ROOT / "summary.md").write_text("\n".join(lines) + "\n", encoding="utf-8")

curves = [(k, r["seed"], r["curves"].get("big")) for k in ks for r in runs[k] if r.get("curves", {}).get("big")]
if curves:
    fig, ax = plt.subplots(figsize=(7, 4))
    for k, seed, c in curves:
        ax.plot([x for x, _ in c], [y for _, y in c], marker="o", ms=3, label=f"k={k} seed {seed}")
    ax.set(xlabel="pretraining steps", ylabel="dev r-precision (zero-shot)",
           title="one large model: does longer pretraining help?")
    ax.legend()
    fig.tight_layout()
    fig.savefig(ROOT / "summary.png", dpi=140)
display(Markdown((ROOT / "summary.md").read_text()))
if curves:
    display(Image(str(ROOT / "summary.png")))

7. Optional: commit the results into ezexe/phi¶

With PUSH_RESULTS = True and a GITHUB_TOKEN that can write to ezexe/phi, this cell:

  • copies the results into .docs/p4/nested-store-results/colab-<RUN>/
  • commits them on PUSH_BRANCH
  • pushes the branch

Opening a pull request and updating nested-store.md stay yours. This cell runs once per run; delete push from state.json to push again.

In [ ]:
if PUSH_RESULTS and not done("push"):
    if not token():
        raise RuntimeError("PUSH_RESULTS needs a GITHUB_TOKEN secret that can write to ezexe/phi")
    dest = WORK / "phi-push"
    if dest.exists():
        shutil.rmtree(dest)
    subprocess.run(["git", "clone", "-q", "--depth", "1", url("ezexe/phi"), str(dest)], check=True,
                   capture_output=True)
    subprocess.run(["git", "-C", str(dest), "checkout", "-q", "-B", PUSH_BRANCH], check=True)
    out = dest / ".docs" / "p4" / "nested-store-results" / f"colab-{RUN}"
    out.mkdir(parents=True, exist_ok=True)
    for p in list((ROOT / "results").glob("*.json")) + [ROOT / "summary.json", ROOT / "summary.md", ROOT / "summary.png",
              ROOT / "report.md"]:
        if p.exists():
            shutil.copy2(p, out / p.name)
    git = ["git", "-C", str(dest), "-c", "user.name=phi colab", "-c", "user.email=phi-colab@users.noreply.github.com"]
    subprocess.run(git + ["add", "-A"], check=True)
    subprocess.run(git + ["commit", "-q", "-m", f"nested-store results from Colab run {RUN}"], check=True)
    r = subprocess.run(["git", "-C", str(dest), "push", "-q", "-u", "origin", PUSH_BRANCH], capture_output=True, text=True)
    if r.returncode != 0:
        raise RuntimeError("push failed: " + r.stderr.replace(token(), "***"))
    mark("push", branch=PUSH_BRANCH)
    print(f"pushed to {PUSH_BRANCH}: open a pull request on https://github.com/ezexe/phi/compare/{PUSH_BRANCH}")
else:
    print("push:", "done earlier" if done("push") else "off (PUSH_RESULTS = False)")