mini-beatrix-3/code/pod/write_mission_json.py

"""Writes the mission's config (the notebook's decision block as JSON for train/mission.py) from the screens' ledgers on this pod.
FUSION: the fusion screen's verdict call (a dict, or null = the byte trunk unfused; the atlas table as this pod's local copy).
ANNEAL_LR_SCALE: the anneal screen's verdict multiplier once the 4090's gauge pass has filled it, else 'from_screen' (the mission
re-reads this file at every boundary, so the number lands here later without a relaunch; the same for the arm certification flag).
The rest from the record (DEPTH 24, DATA_SCALE 4.0, EPOCH_CAP 2.0, REBALANCE_TO generators, LAWFUL_16x8 uncertified until the
4090's lawful chain settles, HEAD_BIRTH revival_birth, the coverage audit report on the hub, the L6 guard ledger on the hub).
Env: V3_MISSION_JSON (out), V3_RUNS (the ledgers' dir), V3_TABLE (the atlas table), V3_ARM_CERT=1 (certified), V3_ANNEAL (a number
overrides the ledger). Refuses to write without a fusion verdict on the ledger."""
import json
import os
import sys

RUNS = os.environ.get("V3_RUNS", "/root/v3/runs")
OUT = os.environ.get("V3_MISSION_JSON", "/root/v3/mission.json")
TABLE = os.environ.get("V3_TABLE", "/root/v3/tools/v3_atlas_entropy_table.npz")
COVERAGE = "hf://AbstractPhil/alephllm-mini-beatrix-training/mini-beatrix-3/v3_preflight/coverage_audit.json"

fs_path = os.path.join(RUNS, "v3_fs_fusion_screen.json")
fs = json.load(open(fs_path, encoding="utf-8"))
v = fs.get("verdict") or {}
if "FUSION_call" not in v:
    print(f"no fusion verdict on {fs_path}: {json.dumps(v)[:300]} — mission.json NOT written")
    sys.exit(1)
fusion = v["FUSION_call"]
if isinstance(fusion, dict) and str(fusion.get("table", "")).startswith("hf://"):
    assert os.path.exists(TABLE), TABLE
    fusion = dict(fusion, table=TABLE)

anneal = "from_screen"
if os.environ.get("V3_ANNEAL"):
    anneal = float(os.environ["V3_ANNEAL"])
else:
    ap = os.path.join(RUNS, "v3_anneal_screen.json")
    if os.path.exists(ap):
        av = json.load(open(ap, encoding="utf-8")).get("verdict") or {}
        if av.get("status") == "COMPLETE" and isinstance(av.get("multiplier"), (int, float)):
            anneal = float(av["multiplier"])

cfg = {
    "craft": "mini-beatrix-3",
    "depth": 32, "data_scale": 4.0, "epoch_cap": 2.0, "rebalance_to": "generators",
    "anneal_lr_scale": anneal,
    "arm_spec": "LAWFUL_16x8", "arm_spec_certified": os.environ.get("V3_ARM_CERT") == "1",
    "fusion": fusion,
    "coverage_audit": COVERAGE,
    "head_birth": "revival_birth",
    "quiet_trunk_grad": False,
    "resume_after_halt": False,
    "micro_batch": 2, "grad_accum": 16,
    "max_hours": 1000.0,
    "out_dir": "/root/v3/runs/mission",
    "l6_ledger": "v3_preflight/l6_guards/v3_l6_guards_ledger.json",
    "local": False,
    "fresh": False,
    "_provenance": {"fusion": f"{fs_path} verdict.FUSION_call ({v.get('call_note', '')})",
                    "anneal_lr_scale": "the anneal screen's verdict.multiplier (the 4090's gauge pass) or from_screen",
                    "written": __import__("time").strftime("%Y-%m-%d %H:%M UTC")},
}
json.dump(cfg, open(OUT, "w", encoding="utf-8"), indent=1)
print(f"wrote {OUT}: fusion {'dict' if isinstance(fusion, dict) else fusion} · anneal {anneal} · arms certified {cfg['arm_spec_certified']}")