day35/scripts/gen_loop.py

#!/usr/bin/env python3
"""gen_loop.py — the automated GROW -> FILL -> PROBE -> (MERGE+REGROW) generation loop.

DAY-34 PROVEN RESULT this automates
-----------------------------------
Layer-duplication GROWTH flips fine-tuning from destructive to constructive. On the
FROZEN 48B a rank32/160-step SFT DAMAGES the sealed ceiling (13 -> 8.5). On a GROWN
model the SAME dose IMPROVES it (15 -> 16). So the compounding loop is, each GENERATION:
  grow 8 identity-twin layers (uniform placement — A/B/C showed placement null)
    -> harvest the model's OWN best-of-8 failures as oracle-verified GT training data
    -> SFT-fill at the sweet spot (rank32, 160 steps, 3 epochs)
    -> DOUBLE-PROBE the sealed ceiling (salt-decorrelated bo8 x2, averaged)
    -> multiplicative gate (keep = ceiling_avg >= best*1.01)
    -> MERGE the fill into the weights, then REGROW = next generation.
Gen-1 measured 13 -> 16 (+23%). GOAL: does it STACK across generations
(16 -> ~19 -> ...) = sustained multiplicative >=1%/gen compounding.

GROWTH-STACKING DECISION — WEIGHT-LEVEL COMPOUNDING VIA MERGE (the crux)
-----------------------------------------------------------------------
Each generation's fill is a LoRA ADAPTER. To make gen{g+1} build ON TOP of gen{g}'s
learning we CONSOLIDATE: merge gen{g}'s adapter into gen{g}'s grown weights, THEN grow the
+8 twins on the merged base. Two facts make this the correct path (not accumulate-corpus):
  * Our model is BORN 4bit. `merge_and_unload` on a bnb-4bit base merges the LoRA delta
    straight into the 4bit weights and the model STAYS 4bit — there is nothing to
    requantize (foom_expand's proven day-15 path; [[project_brain_expansion_works]]). So
    the "4bit merge is lossy" caveat does NOT bite a born-4bit model.
  * DO NOT run a bf16->nf4 requantize pass after merge — it errors on the uint8 storage.
    Detect 4bit by CLASS NAME ("4bit" in Linear4bit/Params4bit), NOT by .dtype (bnb masks
    uint8 as bf16). We just save_pretrained directly.
Why merge beats a per-layer adapter carry-over: a regrown model has MORE layers at NEW
indices, so the previous adapter's per-layer tensors would not map onto it (TS_RESUME would
mismatch). Merging bakes the delta into the base BEFORE the layer count changes, so the
adapter's rank no longer caps cumulative learning ([[reference_lora_base_binding_merge]] pt2)
— this is TRUE weight-level compounding, the reason the ceiling can keep climbing.
The gate just TRACKS the best ceiling and whether we are compounding; growth proceeds every
gen regardless (a single non-keep gen must not stall the layer ladder).

ARCHITECTURE (why subprocess phases)
------------------------------------
The orchestrator holds NO GPU. Each GPU phase is a clean subprocess that fully releases the
GPU on exit — the discipline train_sft_sub.py uses to dodge the bnb-4bit vLLM<->HF unload
leak ([[HF unload leak]]). Per gen:
  [grow]    self-dispatch  PHASE=grow    (torch load bf16 -> MERGE prev adapter -> +8 twins -> save 4bit)
  [harvest] self-dispatch  PHASE=harvest (vLLM best-of-8 on train, verified GT for failures)
  [fill]    scripts/train_sft_sub.py     (fresh LoRA on this gen's corpus)
  [probe]   scripts/probe_once.py x2     (salt 0 + salt 1, averaged = decision-grade)
Every candidate/GT grade runs through ceiling_ratchet's killpg+timeout sandbox grader.

Usage (pod, GPU free): scripts/run_gl.sh  (or set envs and: python3 scripts/gen_loop.py)
"""
import json
import os
import subprocess
import sys

sys.path.insert(0, "/workspace/RSI")
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

SCRIPTS = os.path.dirname(os.path.abspath(__file__))

# ---- env knobs -------------------------------------------------------------
START_MODEL = os.environ.get("START_MODEL", "/workspace/RSI/expanded_models/gen2_B_grown")
N_GENS = int(os.environ.get("N_GENS", "4"))
GROW_LAYERS = int(os.environ.get("GROW_LAYERS", "8"))
FILL_STEPS = os.environ.get("FILL_STEPS", "160")
FILL_RANK = os.environ.get("FILL_RANK", "32")
FILL_LR = os.environ.get("FILL_LR", "1e-4")
FILL_EPOCHS = os.environ.get("FILL_EPOCHS", "3")
HOLD_SET = os.environ.get("HOLD_SET", "hard_holdout")
HOLD_N = int(os.environ.get("HOLD_N", "60"))
CEIL_K = int(os.environ.get("CEIL_K", "8"))
HARVEST_K = int(os.environ.get("HARVEST_K", "8"))
TRAIN_N = int(os.environ.get("TRAIN_N", "200"))
WORKDIR = os.environ.get("WORKDIR", "/workspace/RSI/expanded_models/gen_loop")
OUT = os.environ.get("OUT", "/workspace/RSI/outputs/gen_loop.jsonl")
PREREG = os.environ.get("PREREG", "/workspace/RSI/outputs/prereg.json")
CR_TEMP = os.environ.get("CR_TEMP", "0.8")
MAXTOK = int(os.environ.get("MAXTOK", "700"))
CFG_SEQ = int(os.environ.get("MAX_MODEL_LEN", "4096"))
GEN_CHUNK = int(os.environ.get("GEN_CHUNK", "240"))
SEED = int(os.environ.get("CR_SEED", "34"))
# Per-gen grown-base probe: OFF by default. Function-preservation of identity-twin growth is
# verified twice (day-34 A/B/C placement-null + gen1_base 14.5 vs gen0 15.0) — re-measuring it
# every gen costs ~80 min for zero decision value. Flip on only to re-audit growth itself.
PROBE_BASE = os.environ.get("GL_PROBE_BASE", "0") == "1"
# Adaptive probing: probe salt=0 first; only run the salt=1 confirm when p1 lands within
# GL_PROBE_MARGIN of the gate boundary (ruler noise band +-2-3). A clear pass/fail decides on
# ONE probe (~40 min saved); only boundary calls pay for the decorrelated average.
ADAPTIVE_PROBE = os.environ.get("GL_ADAPTIVE_PROBE", "1") == "1"
PROBE_MARGIN = float(os.environ.get("GL_PROBE_MARGIN", "2.0"))
# RATE knobs — the 1%/20min bar is a RATE (>=+3%/h sustained), so the cycle must be ~1h:
#  * FILLS_PER_GROW: growth (~20 min load+merge+save) is amortized over several fill-cycles;
#    within a growth epoch the layer count is fixed, so fills ACCUMULATE into one adapter via
#    TS_RESUME and only the epoch's best adapter is merged at the next grow.
#  * GPU_UTIL_SOLO: 0.45 was the CORESIDENT train+vLLM envelope; our phases are serialized
#    subprocesses holding the GPU exclusively -> harvest generates at 0.80 (~2x throughput).
#  * PROBE_UTIL: the sealed ruler stays at 0.45 until the switchover calibration EMPIRICALLY
#    proves 0.80 reproduces a known probe exactly (VLLM_DETERMINISTIC batch-invariance check);
#    gl_flags.env then flips it — proof before speed on anything that touches the metric.
FILLS_PER_GROW = int(os.environ.get("GL_FILLS_PER_GROW", "3"))
GPU_UTIL_SOLO = os.environ.get("GL_GPU_UTIL_SOLO", "0.80")
PROBE_UTIL = os.environ.get("GL_PROBE_UTIL", os.environ.get("GPU_UTIL", "0.45"))
MAX_WEAK = int(os.environ.get("GL_MAX_WEAK", "2"))   # weak cycles before regrow (saturation)

# VRAM ceiling — beyond this, consolidation/merge is required (out of scope, stop cleanly)
MAX_SAFET_GB = float(os.environ.get("GL_MAX_SAFET_GB", "40"))
MAX_LAYERS = int(os.environ.get("GL_MAX_LAYERS", "140"))

BNB = {"load_in_4bit": True, "bnb_4bit_compute_dtype": "bfloat16"}


# ============================================================================
# helpers (orchestrator side — no torch/vLLM import here)
# ============================================================================
def _safetensors_gb(model_dir):
    """Sum of all *.safetensors shards in a model dir, in GB."""
    tot = 0
    try:
        for fn in os.listdir(model_dir):
            if fn.endswith(".safetensors"):
                tot += os.path.getsize(os.path.join(model_dir, fn))
    except Exception:
        return 0.0
    return tot / 1e9


def _n_layers(model_dir):
    try:
        cfg = json.load(open(os.path.join(model_dir, "config.json")))
        return int(cfg.get("num_hidden_layers", 0))
    except Exception:
        return 0


def _vram_ceiling_hit(model_dir, tag):
    gb = _safetensors_gb(model_dir)
    nl = _n_layers(model_dir)
    if gb > MAX_SAFET_GB or nl > MAX_LAYERS:
        print(f"[gl] VRAM CEILING ({tag}): safetensors={gb:.1f}GB (cap {MAX_SAFET_GB}) "
              f"layers={nl} (cap {MAX_LAYERS}) — consolidation/merge needed, out of scope. STOP.",
              flush=True)
        return True
    return False


def _run_phase(phase, extra_env, desc):
    """Self-dispatch one GPU phase as a clean subprocess (frees ALL GPU on exit)."""
    env = dict(os.environ, GL_PHASE=phase)
    env.update({k: str(v) for k, v in extra_env.items()})
    print(f"[gl] --> subprocess phase={phase} ({desc})", flush=True)
    r = subprocess.run([sys.executable, os.path.abspath(__file__)], env=env)
    ok = (r.returncode == 0)
    print(f"[gl] <-- phase={phase} rc={r.returncode} ({'ok' if ok else 'FAIL'})", flush=True)
    return ok


def _probe_cache_load():
    p = os.path.join(WORKDIR, "probe_cache.json")
    try:
        return json.load(open(p))
    except Exception:
        return {}


def _probe_cache_save(cache):
    p = os.path.join(WORKDIR, "probe_cache.json")
    tmp = p + ".tmp"
    json.dump(cache, open(tmp, "w"), indent=1)
    os.replace(tmp, p)


def _double_probe(model_dir, adapter, tag, gate=None):
    """Decision-grade probe via scripts/probe_once.py — adaptive double (salt 0 [+ salt 1]).

    Two salted bo8 probes take DIFFERENT sample paths under VLLM_DETERMINISTIC=1, so the
    average has ~1pp spread instead of a single-probe +-2-3 (day-33 finding). ADAPTIVE mode:
    when `gate` is given and p1 lands >PROBE_MARGIN away from it, the call is already decided
    — skip the salt=1 confirm (~40 min). Boundary calls still get the decorrelated average.
    Results are CACHED in WORKDIR/probe_cache.json so a restarted loop re-pays nothing.
    Each probe is its own subprocess -> GPU fully freed before the next GPU phase."""
    cache = _probe_cache_load()
    key = f"{tag}|{model_dir}|{adapter or ''}"
    if key in cache:
        e = cache[key]
        print(f"[gl] probe cache HIT [{tag}] avg={e['avg']:.1f}/{HOLD_N} probes={e['probes']}", flush=True)
        return e["avg"], e["probes"]
    solved = []
    for salt in (0, 1):
        po_out = os.path.join(WORKDIR, f"probe_{tag}_s{salt}.jsonl")
        try:
            os.remove(po_out)
        except Exception:
            pass
        env = dict(os.environ)
        env.pop("GL_PHASE", None)
        env.update({k: str(v) for k, v in dict(
            MODEL=model_dir, ADAPTER=(adapter or ""), SALT=salt, CEIL_K=CEIL_K, HOLD_N=HOLD_N,
            CR_TEMP=CR_TEMP, MAXTOK=MAXTOK, MAX_MODEL_LEN=CFG_SEQ, GEN_CHUNK=GEN_CHUNK,
            PO_OUT=po_out, GPU_UTIL=PROBE_UTIL).items()})
        print(f"[gl] --> probe {tag} salt={salt} model={model_dir} adapter={adapter}", flush=True)
        r = subprocess.run([sys.executable, os.path.join(SCRIPTS, "probe_once.py")], env=env)
        if r.returncode != 0 or not os.path.exists(po_out):
            print(f"[gl] probe {tag} salt={salt} FAILED (rc={r.returncode})", flush=True)
            return None, []
        row = json.loads(open(po_out).read().strip().splitlines()[-1])
        solved.append(int(row["solved"]))
        if salt == 0 and ADAPTIVE_PROBE and gate is not None and abs(solved[0] - gate) > PROBE_MARGIN:
            print(f"[gl] probe [{tag}] p1={solved[0]} is {abs(solved[0]-gate):.1f} from gate "
                  f"{gate:.1f} (> margin {PROBE_MARGIN}) — decided on ONE probe, skipping confirm",
                  flush=True)
            break
    avg = sum(solved) / len(solved)
    ps = " ".join(f"p{i+1}={s}" for i, s in enumerate(solved))
    print(f"[gl] DECISION probe [{tag}] avg={avg:.1f}/{HOLD_N} ({ps})", flush=True)
    cache[key] = {"avg": avg, "probes": solved}
    _probe_cache_save(cache)
    return avg, solved


# ============================================================================
# PHASE: grow  (subprocess — MERGE prev adapter, then grow variant B / uniform)
# ============================================================================
def phase_grow():
    import copy
    import gc
    import torch
    import foom_expand as fx

    src = os.environ["GL_SRC"]
    out_dir = os.environ["GL_OUT"]
    merge_adapter = os.environ.get("GL_ADAPTER", "").strip()   # prev-gen fill to consolidate

    # transformers 5.9 save fix — revert_weight_conversion raises on a structurally-modified
    # bnb-4bit model; passthrough is safe (Qwen load-time conversions are layout-identity).
    try:
        import transformers.modeling_utils as _mu
        if hasattr(_mu, "revert_weight_conversion"):
            _mu.revert_weight_conversion = lambda model_to_save, state_dict, *a, **k: state_dict
            print("[grow] patched revert_weight_conversion -> passthrough", flush=True)
    except Exception as _pe:
        print(f"[grow] revert_weight_conversion patch warn: {_pe}", flush=True)

    from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
    print(f"[grow] loading {src} ...", flush=True)
    model = AutoModelForCausalLM.from_pretrained(
        src, torch_dtype=torch.bfloat16, device_map="cuda", trust_remote_code=True)
    tok = AutoTokenizer.from_pretrained(src, trust_remote_code=True)

    # ---- MERGE the previous generation's fill adapter into the weights (foom_expand path).
    # merge_and_unload on a born-4bit base keeps the model 4bit — nothing to requantize;
    # this BAKES gen{g}'s learning into the base so gen{g+1} builds on top (weight-level
    # compounding, adapter rank no longer caps cumulative learning).
    if merge_adapter and os.path.exists(os.path.join(merge_adapter, "adapter_model.safetensors")):
        from peft import PeftModel
        print(f"[grow] MERGING prev-gen adapter {merge_adapter} (bake fill into base, stays 4bit) ...", flush=True)
        model = PeftModel.from_pretrained(model, merge_adapter)
        model = model.merge_and_unload()
    elif merge_adapter:
        print(f"[grow] WARN: merge adapter {merge_adapter} missing adapter_model.safetensors — skipping merge", flush=True)
    gc.collect(); torch.cuda.empty_cache()
    print(f"[grow] VRAM after load+merge: {torch.cuda.memory_allocated()/1e9:.1f} GB", flush=True)

    layers = fx.find_layer_list(model)
    n = len(layers)
    # variant B / uniform placement — A/B/C showed placement null; uniform = day-15 recipe.
    idxs = sorted(fx.pick_indices(n, GROW_LAYERS, "interleave"))
    assert len(idxs) == GROW_LAYERS, f"got {len(idxs)} dup indices != GROW_LAYERS {GROW_LAYERS}"
    print(f"[grow] variant B/uniform: {n} -> {n+GROW_LAYERS} layers, dup at {idxs}", flush=True)

    dev = next(model.parameters()).device
    for idx in sorted(idxs, reverse=True):        # high->low so earlier indices stay valid
        twin = copy.deepcopy(layers[idx]).to(dev)
        fx.make_identity(twin, dev)               # zero o_proj/down_proj -> function-preserving
        layers.insert(idx + 1, twin)
    new_n = len(layers)
    assert new_n == n + GROW_LAYERS, f"layer count mismatch {new_n} != {n}+{GROW_LAYERS}"
    if hasattr(model.config, "num_hidden_layers"):
        model.config.num_hidden_layers = new_n
    # Qwen2.5+ per-layer layer_types must mirror the SAME duplications (foom_expand fix)
    _lt = getattr(model.config, "layer_types", None)
    if isinstance(_lt, list) and len(_lt) == n:
        _lt = list(_lt)
        for idx in sorted(idxs, reverse=True):
            _lt.insert(idx + 1, _lt[idx])
        model.config.layer_types = _lt
        print(f"[grow] updated layer_types: {n} -> {len(_lt)} entries", flush=True)
    gc.collect(); torch.cuda.empty_cache()

    # Detect 4bit by CLASS NAME (bnb masks uint8 as bf16 via .dtype). A merged born-4bit
    # model stays 4bit => save directly, NEVER requantize (the recurring uint8 crash).
    _lin_types = sorted({type(m).__name__ for m in model.modules()
                         if isinstance(m, torch.nn.Linear)})
    already_4bit = any("4bit" in t.lower() for t in _lin_types) or any(
        "4bit" in type(getattr(m, "weight", None)).__name__.lower()
        for m in model.modules() if isinstance(m, torch.nn.Linear))
    print(f"[grow] Linear types={_lin_types}; already_4bit={already_4bit} (save directly, no requantize)", flush=True)
    if not getattr(model.config, "quantization_config", None):
        model.config.quantization_config = BitsAndBytesConfig(
            load_in_4bit=True, bnb_4bit_compute_dtype=torch.bfloat16,
            bnb_4bit_quant_type="nf4", bnb_4bit_use_double_quant=True).to_dict()

    os.makedirs(out_dir, exist_ok=True)
    print(f"[grow] saving 4bit checkpoint -> {out_dir} ...", flush=True)
    model.save_pretrained(out_dir, safe_serialization=True)
    tok.save_pretrained(out_dir)
    params = sum(p.numel() for p in model.parameters())
    json.dump({"src": src, "merged_adapter": merge_adapter, "strategy": "gen_loop/uniform-B",
               "layers_added": GROW_LAYERS, "dup_indices": idxs, "original_layers": n,
               "new_layer_count": new_n, "param_count": params},
              open(os.path.join(out_dir, "expansion_metadata.json"), "w"), indent=2)
    print(f"[grow] param count: {params} ({params/1e9:.2f}B)", flush=True)

    del model
    gc.collect(); torch.cuda.empty_cache()

    # self-verify: reload the saved 4bit + probe-generate (foom_expand step [4])
    print("[grow] verify: reloading saved 4bit + probe generate ...", flush=True)
    chk = AutoModelForCausalLM.from_pretrained(out_dir, device_map="cuda", trust_remote_code=True)
    ids = tok("def add(a, b):\n    return", return_tensors="pt").to(chk.device)
    g = chk.generate(**ids, max_new_tokens=16, do_sample=False)
    txt = tok.decode(g[0][ids["input_ids"].shape[1]:], skip_special_tokens=True)
    nlay = len(fx.find_layer_list(chk))
    assert nlay == new_n, f"reload layer count {nlay} != {new_n}"
    print(f"[grow] RELOAD OK: {nlay} layers, probe completion: {txt!r}", flush=True)
    sys.exit(0)


# ============================================================================
# PHASE: harvest  (subprocess — external_gt_ceiling harvest, fresh corpus this gen)
# ============================================================================
def phase_harvest():
    import random
    import ceiling_ratchet as cr
    from src.utils.external_benchmarks import _try_load_from_datasets, _extract_code
    from src.utils.vllm_backend import VLLMModelLoader

    model_dir = os.environ["GL_MODEL"]
    corpus_path = os.environ["GL_CORPUS_PATH"]
    gen = int(os.environ.get("GL_GEN", "1"))

    loader = VLLMModelLoader(model_path=model_dir, dtype="bfloat16", max_model_len=CFG_SEQ,
                             gpu_memory_utilization=float(os.environ.get("GPU_UTIL", "0.45")),
                             allow_remote_code=True, quantization_config=BNB, max_lora_rank=128,
                             enable_chunked_prefill=False, enable_lora=True, enforce_eager=True)
    loader.load()
    print(f"[harvest] loaded {model_dir}", flush=True)

    def gen_batch(ps, temp, mt=MAXTOK):
        outs = []
        for cs in range(0, len(ps), GEN_CHUNK):
            outs.extend(loader.generate_batch(ps[cs:cs + GEN_CHUNK], max_new_tokens=mt,
                                              temperature=temp, top_p=(1.0 if temp == 0 else 0.95)))
        return outs

    PR = json.load(open(PREREG))
    HOLD_IDS = set(PR[HOLD_SET])
    apps = [it for it in (_try_load_from_datasets("apps") or [])
            if not (it.meta.get("fn_name") or "").strip()
            and it.meta.get("inputs") and it.meta.get("outputs")]
    # train = disjoint from the sealed ceiling AND carrying usable ground-truth (it.answer)
    train = [it for it in apps if cr._thash(it) not in HOLD_IDS
             and getattr(it, "answer", "") and str(it.answer).strip()]
    # vary the sampled subset per gen so successive gens harvest different failures
    random.seed(SEED + gen)
    train = random.sample(train, min(TRAIN_N, len(train)))

    # cross-gen SOLVED CACHE: a problem proven solved by an earlier gen never contributes
    # corpus (only failures do) — and merge preserves capability — so re-running best-of-K on
    # it is pure waste. Skip known-solved; the harvest frontier SHRINKS as gens compound.
    sc_path = os.path.join(WORKDIR, "solved_cache.json")
    tc_path = os.path.join(WORKDIR, "trained_cache.json")
    try:
        solved_ids = set(json.load(open(sc_path)))
    except Exception:
        solved_ids = set()
    try:
        trained_ids = set(json.load(open(tc_path)))
    except Exception:
        trained_ids = set()
    n_pre = len(train)
    # also skip problems whose GT is ALREADY in a previous fill corpus — re-training the same
    # rows is repetition, not new frontier; each cycle's corpus must be genuinely new support
    train = [it for it in train
             if cr._thash(it) not in solved_ids and cr._thash(it) not in trained_ids]
    print(f"[harvest] gen{gen}: train(with-GT)={n_pre} -> {len(train)} after solved+trained "
          f"caches ({n_pre - len(train)} skipped) HARVEST_K={HARVEST_K}", flush=True)

    # best-of-K on TRAIN to find the problems the model FAILS (their GT = genuinely-new support)
    prompts = [it.prompt for it in train for _ in range(HARVEST_K)]
    souts = gen_batch(prompts, float(CR_TEMP))
    solved_train = [False] * len(train)
    cands = [(i, (_extract_code(s) or s)) for i in range(len(train))
             for s in souts[i * HARVEST_K:(i + 1) * HARVEST_K]]
    okres = list(cr._POOL.map(lambda ic: cr.solves_all(ic[1], train[ic[0]]), cands))
    for (i, _), ok in zip(cands, okres):
        if ok:
            solved_train[i] = True
    n_failed = sum(1 for x in solved_train if not x)
    print(f"[harvest] model solves {sum(solved_train)}/{len(train)} | {n_failed} FAILED "
          f"(their oracle-verified GT = new support)", flush=True)
    solved_ids |= {cr._thash(train[i]) for i, ok in enumerate(solved_train) if ok}
    json.dump(sorted(solved_ids), open(sc_path + ".tmp", "w"))
    os.replace(sc_path + ".tmp", sc_path)

    # verified GT for FAILED problems only (max new-support)
    corpus, gt_bad, new_trained = [], 0, []
    for i, it in enumerate(train):
        if solved_train[i]:
            continue
        gt = str(it.answer).strip()
        if not gt:
            continue
        if not cr.solves_all(gt, it):            # verify GT passes the oracle on our sampled TCs
            gt_bad += 1
            continue
        corpus.append({"prompt": it.prompt, "response": "```python\n" + gt + "\n```"})
        new_trained.append(cr._thash(it))
    print(f"[harvest] verified GT corpus={len(corpus)} ({gt_bad} GT rejected by oracle)", flush=True)
    trained_ids |= set(new_trained)
    json.dump(sorted(trained_ids), open(tc_path + ".tmp", "w"))
    os.replace(tc_path + ".tmp", tc_path)
    if not corpus:
        print("[harvest] EMPTY corpus — nothing to fill", flush=True)
        sys.exit(1)
    os.makedirs(os.path.dirname(corpus_path) or ".", exist_ok=True)
    open(corpus_path, "w").write("\n".join(json.dumps(x) for x in corpus) + "\n")
    print(f"[harvest] wrote {len(corpus)} rows -> {corpus_path}", flush=True)
    sys.exit(0)


# ============================================================================
# orchestrator
# ============================================================================
def orchestrate():
    os.makedirs(WORKDIR, exist_ok=True)
    os.makedirs(os.path.dirname(OUT) or ".", exist_ok=True)
    print(f"[gl] START gen_loop: START_MODEL={START_MODEL} N_GENS={N_GENS} GROW_LAYERS={GROW_LAYERS} "
          f"fill(rank{FILL_RANK}/steps{FILL_STEPS}/ep{FILL_EPOCHS})", flush=True)

    def log(row):
        open(OUT, "a").write(json.dumps(row) + "\n")

    traj = []

    # ---- gen 0: DOUBLE-probe START_MODEL as-is (the grown base, no fill) = gen-0 ceiling ----
    avg0, probes0 = _double_probe(START_MODEL, adapter=None, tag="gen0")
    if avg0 is None:
        print("[gl] gen-0 probe FAILED — cannot establish baseline ceiling. STOP.", flush=True)
        log({"gen": 0, "error": "gen0_probe_failed"})
        return
    best_ceiling = avg0
    n0 = _n_layers(START_MODEL)
    log({"gen": 0, "model": START_MODEL, "layers": n0, "ceiling_avg": avg0, "probes": probes0,
         "best_ceiling": best_ceiling, "kept": True, "corpus_size": 0, "delta_pct": 0.0})
    traj.append((0, avg0, best_ceiling, True))
    print(f"[gl] gen0 ceiling={avg0:.1f}/{HOLD_N} (layers={n0}) = baseline best", flush=True)

    prev_grown = START_MODEL     # gen g grows on this; gen1 grows on START (no merge, START unfilled)
    prev_adapter = ""            # gen g merges the (g-1) epoch's BEST fill into the base before growing
    for g in range(1, N_GENS + 1):
        grown_dir = os.path.join(WORKDIR, f"gen{g}_grown")
        grow_row = {"gen": g, "src": prev_grown, "merged_adapter": prev_adapter,
                    "grown_dir": grown_dir, "layers_added": GROW_LAYERS}

        # [1] GROW = MERGE prev epoch's best fill into prev_grown, then +8 twins (weight-level
        # compounding). ~20 min — amortized over FILLS_PER_GROW fill-cycles below.
        meta_p = os.path.join(grown_dir, "expansion_metadata.json")
        if os.path.exists(meta_p) and os.path.exists(os.path.join(grown_dir, "config.json")):
            print(f"[gl] gen{g}: grown dir exists, reusing {grown_dir}", flush=True)
        elif not _run_phase("grow",
                            {"GL_SRC": prev_grown, "GL_OUT": grown_dir, "GL_ADAPTER": prev_adapter},
                            f"gen{g} merge+grow +{GROW_LAYERS}"):
            grow_row["error"] = "grow_failed"
            log(grow_row)
            print(f"[gl] gen{g}: GROW failed — loop stops cleanly.", flush=True)
            return
        meta = json.load(open(meta_p))
        grow_row.update({"layers": meta.get("new_layer_count"), "dup_indices": meta.get("dup_indices"),
                         "params": meta.get("param_count")})

        # [1b] optional re-audit: double-probe the grown base (function-preservation verified 2x)
        if PROBE_BASE:
            b_avg, b_probes = _double_probe(grown_dir, adapter=None, tag=f"gen{g}_base")
            grow_row.update({"grown_base_ceiling": b_avg, "grown_base_probes": b_probes})

        # ---- FILL-CYCLES within this growth epoch -------------------------------------
        # Layer count is FIXED inside the epoch, so fills ACCUMULATE into one adapter via
        # TS_RESUME (fresh corpus each cycle; trained-cache guarantees no repetition). The
        # epoch ends after FILLS_PER_GROW cycles or MAX_WEAK consecutive non-keeps
        # (= the twins are saturated -> grow new capacity). This is the ~1h RATE cycle:
        # harvest (frontier, solo-util) + fill (160 steps) + adaptive probe.
        resume_adapter = ""            # accumulating adapter chain within the epoch
        epoch_best_adapter = ""        # what the NEXT grow merges (best kept, weight-level compounding)
        weak = 0
        for c in range(1, FILLS_PER_GROW + 1):
            # c==1 keeps the legacy gen{g} naming so pre-restructure banked artifacts
            # (gen1_corpus/gen1_adapter/probe tag "gen1") are reused, not re-paid.
            suffix = f"gen{g}" if c == 1 else f"gen{g}c{c}"
            corpus_path = os.path.join(WORKDIR, f"{suffix}_corpus.jsonl")
            adapt_dir = os.path.join(WORKDIR, f"{suffix}_adapter")
            row = dict(grow_row, cycle=c, tag=suffix)

            # [2] HARVEST fresh frontier corpus (exclusive GPU -> solo util; reused on restart)
            if os.path.exists(corpus_path) and sum(1 for l in open(corpus_path) if l.strip()) > 0:
                print(f"[gl] {suffix}: corpus exists, reusing {corpus_path}", flush=True)
            elif not _run_phase("harvest",
                                {"GL_MODEL": grown_dir, "GL_CORPUS_PATH": corpus_path,
                                 "GL_GEN": g * 100 + c, "GPU_UTIL": GPU_UTIL_SOLO},
                                f"{suffix} harvest"):
                row["error"] = "harvest_failed"
                log(row)
                print(f"[gl] {suffix}: HARVEST failed (empty frontier or crash) — end epoch.", flush=True)
                break
            corpus_size = sum(1 for l in open(corpus_path) if l.strip()) if os.path.exists(corpus_path) else 0
            row["corpus_size"] = corpus_size

            # [3] FILL: continue the epoch's adapter on the new corpus (TS_RESUME accumulation)
            fill_env = dict(os.environ, TS_BASE=grown_dir, TS_POOL=corpus_path, TS_OUT=adapt_dir,
                            TS_CYCLE="1", TS_RANK=FILL_RANK, TS_LR=FILL_LR, TS_EPOCHS=FILL_EPOCHS,
                            TS_STEPS=FILL_STEPS)
            if resume_adapter:
                fill_env["TS_RESUME"] = resume_adapter
            else:
                fill_env.pop("TS_RESUME", None)
            fill_env.pop("GL_PHASE", None)
            os.makedirs(adapt_dir, exist_ok=True)
            ckpt = os.path.join(adapt_dir, "lora_cycle_1")
            if os.path.exists(os.path.join(ckpt, "adapter_model.safetensors")):
                print(f"[gl] {suffix}: fill adapter exists, reusing {ckpt}", flush=True)
            else:
                print(f"[gl] {suffix}: FILL on {corpus_size} rows "
                      f"(rank{FILL_RANK}/steps{FILL_STEPS}/ep{FILL_EPOCHS}"
                      f"{' resume=' + resume_adapter if resume_adapter else ' fresh'})", flush=True)
                fr = subprocess.run([sys.executable, os.path.join(SCRIPTS, "train_sft_sub.py")],
                                    env=fill_env)
                if fr.returncode != 0 or not os.path.exists(os.path.join(ckpt, "adapter_model.safetensors")):
                    row["error"] = "fill_failed"
                    log(row)
                    print(f"[gl] {suffix}: FILL failed (rc={fr.returncode}) — end epoch.", flush=True)
                    break
            row["fill_adapter"] = ckpt

            # [4] PROBE: adaptive decision-grade probe vs the multiplicative gate
            avg, probes = _double_probe(grown_dir, adapter=ckpt, tag=suffix, gate=best_ceiling * 1.01)
            if avg is None:
                row["error"] = "probe_failed"
                log(row)
                print(f"[gl] {suffix}: PROBE failed — end epoch.", flush=True)
                break
            row.update({"ceiling_avg": avg, "probes": probes})

            # [5] GATE (multiplicative): keep = ceiling_avg >= best*1.01
            kept = avg >= best_ceiling * 1.01
            damaged = avg < best_ceiling - PROBE_MARGIN   # beyond ruler noise = real regression
            delta_pct = (avg / best_ceiling - 1.0) * 100.0 if best_ceiling > 0 else 0.0
            if kept:
                best_ceiling = avg
                epoch_best_adapter = ckpt
                resume_adapter = ckpt
                weak = 0
            elif damaged:
                # do NOT carry a damaging adapter forward — revert the chain to the last good one
                print(f"[gl] {suffix}: DAMAGE ({avg:.1f} < best {best_ceiling:.1f} - {PROBE_MARGIN}) — "
                      f"reverting chain to {epoch_best_adapter or 'fresh'}", flush=True)
                resume_adapter = epoch_best_adapter
                weak += 1
            else:
                resume_adapter = ckpt     # neutral: keep the learning, it may compound next cycle
                weak += 1
            row.update({"best_ceiling": best_ceiling, "kept": kept, "damaged": damaged,
                        "delta_pct": round(delta_pct, 2), "weak_streak": weak})
            log(row)
            traj.append((suffix, avg, best_ceiling, kept))
            print(f"[gl] {suffix}: ceiling_avg={avg:.1f}/{HOLD_N} vs best={best_ceiling:.1f} "
                  f"delta={delta_pct:+.1f}% kept={kept} weak={weak} layers={row.get('layers')} "
                  f"corpus={corpus_size}", flush=True)

            if weak >= MAX_WEAK:
                print(f"[gl] gen{g}: {weak} weak cycles — twins saturated, REGROW.", flush=True)
                break

        # [6] VRAM guard — stop cleanly before regrowing past what a single GPU can hold
        if _vram_ceiling_hit(grown_dir, f"after gen{g}"):
            break

        # advance: gen{g+1} grows on gen{g}_grown after MERGING this epoch's best fill in
        prev_grown = grown_dir
        prev_adapter = epoch_best_adapter

    # ---- trajectory summary ----
    print("[gl] DONE", flush=True)
    print("[gl] trajectory (cycle: ceiling_avg | best | kept):", flush=True)
    for gi, av, bc, kp in traj:
        print(f"[gl]   {gi}: {av:.1f}/{HOLD_N}  best={bc:.1f}  kept={kp}", flush=True)
    if len(traj) >= 2:
        first = traj[0][1]
        last = traj[-1][2]   # best_ceiling at end = the banked capability
        n_kept = sum(1 for _, _, _, kp in traj[1:] if kp)
        print(f"[gl] SUMMARY: {first:.1f} -> best {last:.1f} over {len(traj)-1} fill-cycles "
              f"({n_kept} kept @ >=1%/cycle). "
              f"{'COMPOUNDING' if last > first else 'no net compounding'}", flush=True)


def main():
    phase = os.environ.get("GL_PHASE", "")
    if phase == "grow":
        phase_grow()
    elif phase == "harvest":
        phase_harvest()
    else:
        orchestrate()


if __name__ == "__main__":
    main()