anima_closeout v7: (1) rc no longer masked — the deepspeed|tee pipeline returned TEE's exit code, so the smoke's ~24s death printed rc=0; Python is now the tee and any failure (rc!=0 OR no run dir) dumps the log tail into the cell. (2) CRITICAL: train+eval dataset TOMLs shared the fork's default parquet cache root ('local_parquet' constant) — eval would clobber train's metadata/latents and the loop would silently train on the 512 eval rows; now explicit distinct path= roots under dp_cache/. (3) width_column/height_column (the real fork keys; image_*_column variants were silently ignored). (4) val bed reads the fork's REAL cache layout: per-bucket paired latents/text_embeddings_0 caches. Source-verified against the fork + exp004's as-run TOMLs by 4-agent audit.

colab/anima_closeout.ipynb

@@ -344,7 +344,14 @@
     "          f\"{_m['eval_n']} eval from {len(_m['shards_used'])} shards\")\n",
     "\n",
     "# -- dataset TOMLs (fused recipe: derived caption column) -----------------\n",
-    "def _dataset_toml(files):\n",
+    "# `path` = the latent/TE cache root, and it MUST differ between train and\n",
+    "# eval: for type='parquet' the fork's default root is the CONSTANT\n",
+    "# 'local_parquet' (utils/dataset.py:1014-1015), so without explicit paths\n",
+    "# the eval dataset overwrites the train dataset's grouped metadata and\n",
+    "# clears its latents mid-run \u2014 the train loop then silently loads the 512\n",
+    "# EVAL rows. Key names per the fork + exp004's as-run dataset TOML:\n",
+    "# width_column/height_column (image_*_column variants are silently ignored).\n",
+    "def _dataset_toml(files, cache_path):\n",
     "    return f\"\"\"# generated by anima_closeout.ipynb\n",
     "resolutions = [512]\n",
     "enable_ar_bucket = true\n",
@@ -356,10 +363,11 @@
     "\n",
     "[[directory]]\n",
     "type = 'parquet'\n",
+    "path = '{cache_path}'\n",
     "parquet_files = '{files}'\n",
     "image_column = 'image'\n",
-    "image_width_column = 'image_width'\n",
-    "image_height_column = 'image_height'\n",
+    "width_column = 'image_width'\n",
+    "height_column = 'image_height'\n",
     "caption_column = 'caption'\n",
     "caption_type = 'text'\n",
     "num_repeats = 1\n",
@@ -367,8 +375,9 @@
     "\"\"\"\n",
     "\n",
     "(TOMLS / \"dataset_train.toml\").write_text(\n",
-    "    _dataset_toml(f\"{TRAIN_DIR}/*.parquet\"))\n",
-    "(TOMLS / \"dataset_eval.toml\").write_text(_dataset_toml(str(EVAL_PARQUET)))\n",
+    "    _dataset_toml(f\"{TRAIN_DIR}/*.parquet\", WORK / \"dp_cache\" / \"train\"))\n",
+    "(TOMLS / \"dataset_eval.toml\").write_text(\n",
+    "    _dataset_toml(str(EVAL_PARQUET), WORK / \"dp_cache\" / \"heldout\"))\n",
     "# -- training TOMLs per arm -----------------------------------------------\n",
     "_AP = WORK / \"models\" / \"anima\" / \"split_files\"\n",
     "\n",
@@ -455,18 +464,42 @@
     "\n",
     "HUB_REPO = \"AbstractPhil/geolip-aleph-diffusion\"\n",
     "\n",
+    "def _die(tag, log, why):\n",
+    "    tail = \"\".join(\n",
+    "        log.read_text(errors=\"replace\").splitlines(keepends=True)[-60:]) \\\n",
+    "        if log.exists() else \"(no log written)\"\n",
+    "    raise RuntimeError(f\"{tag}: {why}\\n--- last 60 lines of {log} ---\\n\"\n",
+    "                       f\"{tail}\")\n",
+    "\n",
+    "def sh_tee(cmd, log):\n",
+    "    # NO shell pipe to tee: a pipeline returns TEE's rc, so a dead train\n",
+    "    # reported rc=0 (the smoke's exact failure). Python is the tee \u2014 the\n",
+    "    # child's rc is real, output streams live AND lands in the log.\n",
+    "    print(f\"$ {cmd}\")\n",
+    "    with open(log, \"a\") as lf:\n",
+    "        proc = subprocess.Popen(cmd, shell=True, executable=\"/bin/bash\",\n",
+    "                                stdout=subprocess.PIPE,\n",
+    "                                stderr=subprocess.STDOUT,\n",
+    "                                text=True, bufsize=1)\n",
+    "        for line in proc.stdout:\n",
+    "            print(line, end=\"\", flush=True)\n",
+    "            lf.write(line)\n",
+    "        return proc.wait()\n",
+    "\n",
     "def run_arm(tag, resume=False):\n",
     "    log = RUNS / f\"{tag}.log\"\n",
-    "    flags = \"--resume_from_checkpoint\" if resume else \"\"\n",
+    "    flags = \" --resume_from_checkpoint\" if resume else \"\"\n",
     "    t0 = time.time()\n",
-    "    rc = sh(f\"cd {FORK} && NCCL_P2P_DISABLE=1 NCCL_IB_DISABLE=1 \"\n",
-    "            f\"deepspeed --num_gpus=1 train.py --deepspeed \"\n",
-    "            f\"--config {ARMS[tag]} {flags} 2>&1 | tee -a {log}\",\n",
-    "            check=False)\n",
+    "    rc = sh_tee(f\"cd {FORK} && NCCL_P2P_DISABLE=1 NCCL_IB_DISABLE=1 \"\n",
+    "                f\"deepspeed --num_gpus=1 train.py --deepspeed \"\n",
+    "                f\"--config {ARMS[tag]}{flags}\", log)\n",
     "    print(f\"[{tag}] rc={rc} wall={(time.time() - t0) / 60:.1f} min\")\n",
     "    if rc != 0:\n",
-    "        raise RuntimeError(f\"{tag} failed \u2014 read {log}\")\n",
-    "    return newest_save(tag)\n",
+    "        _die(tag, log, f\"train exited rc={rc}\")\n",
+    "    try:\n",
+    "        return newest_save(tag)      # rc=0 but no run/save dir is ALSO death\n",
+    "    except AssertionError as e:\n",
+    "        _die(tag, log, str(e))\n",
     "\n",
     "def latest_run_dir(tag):\n",
     "    run_dirs = sorted(glob.glob(str(RUNS / tag / \"*\")))\n",
@@ -702,18 +735,24 @@
     "QUANTILES = [0.1, 0.3, 0.5, 0.7, 0.9]\n",
     "DEV = \"cuda\"\n",
     "\n",
-    "# -- read the fork's eval cache (schema discovered loudly) ----------------\n",
-    "cache_dir = CFG[\"eval_cache_dir\"]\n",
-    "shard_files = sorted(f for f in os.listdir(cache_dir)\n",
-    "                     if f.endswith(\".parquet\"))\n",
-    "assert shard_files, f\"no cache shards under {cache_dir}\"\n",
-    "schema = pq.read_schema(os.path.join(cache_dir, shard_files[0]))\n",
-    "print(\"cache columns:\", schema.names, flush=True)\n",
-    "NEEDED = [\"latents\", \"prompt_embeds\", \"attn_mask\", \"t5_input_ids\",\n",
-    "          \"t5_attn_mask\"]\n",
-    "for c in NEEDED:\n",
-    "    assert c in schema.names or c + \"__shape\" in schema.names, \\\n",
-    "        f\"cache lacks column {c} \u2014 inspect the printed schema\"\n",
+    "# -- read the fork's eval cache -------------------------------------------\n",
+    "# Layout (utils/dataset.py + parquet_cache.py): {root}/cache_{bucket}/ holds\n",
+    "# TWO separate parquet caches \u2014 'latents' (models/base.py:247) and\n",
+    "# 'text_embeddings_0' (cosmos_predict2.py:398: prompt_embeds/attn_mask/\n",
+    "# t5_input_ids/t5_attn_mask). Both map over the SAME shuffled metadata\n",
+    "# dataset and this campaign has exactly one caption per image, so rows pair\n",
+    "# by order; the per-bucket count assert guards that assumption.\n",
+    "cache_root = CFG[\"eval_cache_dir\"]\n",
+    "buckets = sorted(d for d in os.listdir(cache_root)\n",
+    "                 if d.startswith(\"cache_\")\n",
+    "                 and os.path.isdir(os.path.join(cache_root, d)))\n",
+    "assert buckets, f\"no cache_* bucket dirs under {cache_root}\"\n",
+    "\n",
+    "def read_cache(d):\n",
+    "    shards = sorted(p for p in os.listdir(d)\n",
+    "                    if p.startswith(\"shard_\") and p.endswith(\".parquet\"))\n",
+    "    assert shards, f\"no shards under {d}\"\n",
+    "    return [pq.read_table(os.path.join(d, p)) for p in shards]\n",
     "\n",
     "def decol(tbl, name, j):\n",
     "    raw = tbl.column(name)[j].as_py()\n",
@@ -722,14 +761,27 @@
     "    return torch.frombuffer(bytearray(raw),\n",
     "                            dtype=getattr(torch, dt)).reshape(shape)\n",
     "\n",
+    "LAT = [\"latents\"]\n",
+    "TE = [\"prompt_embeds\", \"attn_mask\", \"t5_input_ids\", \"t5_attn_mask\"]\n",
     "rows = []\n",
-    "for sf in shard_files:\n",
-    "    tbl = pq.read_table(os.path.join(cache_dir, sf))\n",
-    "    for j in range(tbl.num_rows):\n",
-    "        rows.append({c: decol(tbl, c, j) for c in NEEDED\n",
-    "                     if c + \"__shape\" in schema.names})\n",
+    "for b in buckets:\n",
+    "    lt = read_cache(os.path.join(cache_root, b, \"latents\"))\n",
+    "    tt = read_cache(os.path.join(cache_root, b, \"text_embeddings_0\"))\n",
+    "    for c in LAT:\n",
+    "        assert c + \"__shape\" in lt[0].schema.names, (b, lt[0].schema.names)\n",
+    "    for c in TE:\n",
+    "        assert c + \"__shape\" in tt[0].schema.names, (b, tt[0].schema.names)\n",
+    "    lrows = [(t, j) for t in lt for j in range(t.num_rows)]\n",
+    "    trows = [(t, j) for t in tt for j in range(t.num_rows)]\n",
+    "    assert len(lrows) == len(trows), \\\n",
+    "        f\"{b}: latents {len(lrows)} != text {len(trows)} \u2014 pairing unsafe\"\n",
+    "    for (ltb, lj), (ttb, tj) in zip(lrows, trows):\n",
+    "        r = {c: decol(ltb, c, lj) for c in LAT}\n",
+    "        r.update({c: decol(ttb, c, tj) for c in TE})\n",
+    "        rows.append(r)\n",
+    "print(f\"eval cache rows: {len(rows)} from {len(buckets)} bucket dir(s)\",\n",
+    "      flush=True)\n",
     "rows = rows[:int(os.environ.get(\"VB_MAX_ROWS\", \"512\"))]\n",
-    "print(f\"eval cache rows used: {len(rows)}\", flush=True)\n",
     "assert rows\n",
     "\n",
     "# -- pipeline (frozen trunk); adapters swapped per arm --------------------\n",
@@ -868,13 +920,15 @@
     "except ImportError:\n",
     "    import tomli as tomllib\n",
     "\n",
-    "_pq_files = glob.glob(str(DATA / \"**\" / \"*.parquet\"), recursive=True)\n",
-    "_cache_dirs = sorted({str(Path(f).parent) for f in _pq_files\n",
-    "                      if \"cache\" in f.lower() and \"eval\" in f.lower()})\n",
-    "print(\"eval cache candidates:\", _cache_dirs)\n",
-    "assert _cache_dirs, (\"no eval cache found under DATA \u2014 inspect the tree, \"\n",
-    "                     \"set EVAL_CACHE manually, re-run\")\n",
-    "EVAL_CACHE = _cache_dirs[0]\n",
+    "# deterministic: the eval dataset TOML pins path = dp_cache/heldout, and\n",
+    "# the fork caches under {path}/cache/{model_name} with model_name 'anima'\n",
+    "# (cosmos_predict2.py:254 renames the pipeline when llm_path is a file)\n",
+    "EVAL_CACHE = str(WORK / \"dp_cache\" / \"heldout\" / \"cache\" / \"anima\")\n",
+    "_buckets = sorted(glob.glob(str(Path(EVAL_CACHE) / \"cache_*\")))\n",
+    "assert _buckets, (f\"no cache_* bucket dirs under {EVAL_CACHE} \u2014 the arms \"\n",
+    "                  \"must have run (they build the eval cache) before the bed\")\n",
+    "print(\"eval cache:\", EVAL_CACHE,\n",
+    "      \"| buckets:\", [Path(b).name for b in _buckets])\n",
     "\n",
     "_full = tomllib.loads(ARMS[\"e004b_relay_s0\"].read_text())\n",
     "_model = {k: v for k, v in _full[\"model\"].items()\n",
@@ -887,8 +941,12 @@
     "    config.update(out=str(outp), eval_cache_dir=EVAL_CACHE,\n",
     "                  model_cfg=_model_cfg)\n",
     "    cfgp.write_text(json.dumps(config, indent=1))\n",
-    "    sh(f\"cd {FORK} && VB_CONFIG={cfgp} python {WORK}/val_bed.py \"\n",
-    "       f\"2>&1 | tee {RUNS}/{out_name}.log\")\n",
+    "    _blog = RUNS / f\"{out_name}.log\"\n",
+    "    rc = sh_tee(f\"cd {FORK} && VB_CONFIG={cfgp} python {WORK}/val_bed.py\",\n",
+    "                _blog)\n",
+    "    if rc != 0 or not outp.exists():\n",
+    "        _die(out_name, _blog, f\"val bed rc={rc}, results \"\n",
+    "             f\"{'missing' if not outp.exists() else 'present'}\")\n",
     "    return json.loads(outp.read_text())\n",
     "\n",
     "def _pt(saves, tag):\n",