Caption key corrected per the campaign record: qwen-deepfashion-fused stays THE dataset (r2 exp011-013 design of record); caption = caption_joycaption with prompt_fused fallback (dexp011:154), materialized as a derived column; exp011 audit gate + id dedup in the split. The earlier ft1 retarget was wrong — the fix was the key, not the dataset.

colab/anima_closeout.ipynb

@@ -138,30 +138,30 @@
     "print(\"Anima split_files present \u2014 NC weights; derived ckpts inherit NC\")\n",
     "\n",
     "# -- 5. dataset columns, verified BEFORE anything spends ------------------\n",
-    "# exp004's actual source: diffusion-pretrain-set-ft1, deepfashion config \u2014\n",
-    "# the dataset whose shards were the pod's ft1_shards and which carries\n",
-    "# caption_vlm_json. (qwen-deepfashion-fused does NOT have that column; the\n",
-    "# first campaign run caught this exactly here, at zero cost.) The repo\n",
-    "# holds 582 parquet files across many configs \u2014 filter to deepfashion or\n",
-    "# the split cell would download all of them.\n",
-    "DATASET = \"AbstractPhil/diffusion-pretrain-set-ft1\"\n",
-    "DATASET_SUBDIR = \"data/deepfashion/\"\n",
+    "# THE DATASET IS qwen-deepfashion-fused (the fused line's design of record:\n",
+    "# r2 exp011/012/013 all trained on it). The CAPTION KEY per the record is\n",
+    "# caption_joycaption with prompt_fused fallback (pod2/dexp011:154; results\n",
+    "# cond = \"caption_joycaption nl77\"). caption_vlm_json belongs to the older\n",
+    "# ft1 shards and does NOT exist here \u2014 the first campaign run's pre-spend\n",
+    "# column check caught exactly that; the fix was the key, not the dataset.\n",
+    "DATASET = \"AbstractPhil/qwen-deepfashion-fused\"\n",
     "from huggingface_hub import HfApi\n",
     "SHARDS = sorted(f for f in HfApi().list_repo_files(\n",
-    "    DATASET, repo_type=\"dataset\")\n",
-    "    if f.endswith(\".parquet\") and f.startswith(DATASET_SUBDIR))\n",
-    "assert SHARDS, f\"no parquet shards under {DATASET_SUBDIR} in {DATASET}\"\n",
-    "print(f\"deepfashion shards: {len(SHARDS)}\")\n",
+    "    DATASET, repo_type=\"dataset\") if f.endswith(\".parquet\"))\n",
+    "assert SHARDS, f\"no parquet shards in {DATASET}\"\n",
+    "print(f\"shards: {len(SHARDS)} (incl. the shard_r0_* series)\")\n",
     "import pyarrow.parquet as pq\n",
     "_p0 = hf_hub_download(DATASET, SHARDS[0], repo_type=\"dataset\")\n",
     "_cols = set(pq.read_schema(_p0).names)\n",
-    "_missing = {\"image\", \"caption_vlm_json\"} - _cols\n",
+    "_missing = {\"image\", \"caption_joycaption\", \"id\"} - _cols\n",
     "assert not _missing, f\"dataset lacks {_missing}; has {sorted(_cols)[:24]}\"\n",
     "ANIMA.update(DATASET=DATASET, SHARDS=SHARDS,\n",
     "             HAS_WH={\"image_width\", \"image_height\"} <= _cols,\n",
-    "             ID_COL=\"id\" if \"id\" in _cols else None)\n",
-    "print(f\"dataset OK: {len(SHARDS)} shard(s); wh cols {ANIMA['HAS_WH']}; \"\n",
-    "      f\"id col {ANIMA['ID_COL']}\")\n",
+    "             HAS_FUSED_JSON=\"fused_json\" in _cols,\n",
+    "             HAS_PROMPT_FUSED=\"prompt_fused\" in _cols,\n",
+    "             ID_COL=\"id\")\n",
+    "print(f\"dataset OK: wh cols {ANIMA['HAS_WH']}; fused_json \"\n",
+    "      f\"{ANIMA['HAS_FUSED_JSON']}; prompt_fused {ANIMA['HAS_PROMPT_FUSED']}\")\n",
     "print(\"\\n=== READY ===\")\n"
    ]
   },
@@ -171,11 +171,16 @@
    "execution_count": null,
    "outputs": [],
    "source": [
-    "# \u2550\u2550\u2550 CELL 1 \u2014 materialize the held-out split + write all TOMLs \u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\n",
-    "# Deterministic row-hash split: 512 eval rows, disjoint by construction,\n",
-    "# recorded in split_manifest.json (the reproducibility anchor). Both sides\n",
-    "# read the SAME materialized parquet. caption_type='text' = exp004's\n",
-    "# verbatim-JSON recipe (flattened-'json' is the recorded alternative).\n",
+    "# \u2550\u2550\u2550 CELL 1 \u2014 materialize the gated held-out split + write TOMLs \u2550\u2550\u2550\u2550\u2550\u2550\u2550\u2550\n",
+    "# Row pipeline, exactly the campaign record:\n",
+    "#   AUDIT GATE (pod2/dexp011:98-105): a row passes when its age_audit/audit\n",
+    "#     value contains \"approved\" or \"pass\" (case-insensitive); gated rows are\n",
+    "#     counted and reported loudly.\n",
+    "#   DEDUP by id (first occurrence wins) \u2014 guards the stray rank0.part00\n",
+    "#     shard alongside the shard_r0_* series.\n",
+    "#   CAPTION = caption_joycaption or prompt_fused fallback (dexp011:154),\n",
+    "#     materialized as a `caption` column so the fork reads one key.\n",
+    "#   SPLIT: sha256(id) ascending; first EVAL_N surviving rows are eval.\n",
     "import hashlib, json\n",
     "import pyarrow as pa\n",
     "import pyarrow.parquet as pq\n",
@@ -186,40 +191,72 @@
     "EVAL_PARQUET = DATA / \"eval.parquet\"\n",
     "MANIFEST = DATA / \"split_manifest.json\"\n",
     "\n",
+    "def _passes_gate(row_get):\n",
+    "    vals = [str(row_get(c)) for c in (\"age_audit\", \"audit\")\n",
+    "            if row_get(c) is not None]\n",
+    "    return (not vals) or any((\"approved\" in v.lower() or \"pass\" in v.lower())\n",
+    "                             for v in vals)\n",
+    "\n",
     "if not MANIFEST.exists():\n",
-    "    _paths = [hf_hub_download(ANIMA[\"DATASET\"], s, repo_type=\"dataset\")\n",
-    "              for s in ANIMA[\"SHARDS\"]]\n",
-    "\n",
-    "    def _row_key(si, ri, batch, j):\n",
-    "        if ANIMA[\"ID_COL\"]:\n",
-    "            return str(batch.column(ANIMA[\"ID_COL\"])[j].as_py())\n",
-    "        cap = str(batch.column(\"caption_vlm_json\")[j].as_py())[:256]\n",
-    "        return f\"{si}:{ri}:{hashlib.sha1(cap.encode()).hexdigest()[:12]}\"\n",
-    "\n",
-    "    _key_cols = [c for c in (ANIMA[\"ID_COL\"], \"caption_vlm_json\") if c]\n",
-    "    ranked = []\n",
-    "    for si, p in enumerate(_paths):\n",
+    "    _paths = [hf_hub_download(ANIMA[\"DATASET\"], sh, repo_type=\"dataset\")\n",
+    "              for sh in ANIMA[\"SHARDS\"]]\n",
+    "\n",
+    "    # pass 1: gate + dedup + rank survivors by sha256(id)\n",
+    "    _key_cols = [c for c in (\"id\", \"caption_joycaption\", \"prompt_fused\",\n",
+    "                             \"age_audit\", \"audit\")\n",
+    "                 if True]\n",
+    "    seen_ids, ranked = set(), []\n",
+    "    n_seen = n_gated = n_dup = 0\n",
+    "    for si, pth in enumerate(_paths):\n",
+    "        pf = pq.ParquetFile(pth)\n",
+    "        avail = [c for c in _key_cols if c in pf.schema_arrow.names]\n",
     "        ri = 0\n",
-    "        for batch in pq.ParquetFile(p).iter_batches(batch_size=256,\n",
-    "                                                    columns=_key_cols):\n",
+    "        for batch in pf.iter_batches(batch_size=256, columns=avail):\n",
+    "            cols = {c: batch.column(c) for c in avail}\n",
     "            for j in range(batch.num_rows):\n",
-    "                k = _row_key(si, ri, batch, j)\n",
-    "                ranked.append(\n",
-    "                    (hashlib.sha256(k.encode()).hexdigest(), si, ri, k))\n",
+    "                n_seen += 1\n",
+    "                get = lambda c: (cols[c][j].as_py() if c in cols else None)\n",
+    "                rid = str(get(\"id\"))\n",
+    "                if not _passes_gate(get):\n",
+    "                    n_gated += 1\n",
+    "                elif rid in seen_ids:\n",
+    "                    n_dup += 1\n",
+    "                else:\n",
+    "                    seen_ids.add(rid)\n",
+    "                    ranked.append((hashlib.sha256(rid.encode()).hexdigest(),\n",
+    "                                   si, ri, rid))\n",
     "                ri += 1\n",
     "    ranked.sort()\n",
+    "    assert len(ranked) > 4 * EVAL_N, (\n",
+    "        f\"only {len(ranked)} rows survive the audit gate \"\n",
+    "        f\"(saw {n_seen}, gated {n_gated}, dup {n_dup}) \u2014 inspect before \"\n",
+    "        \"training on so little\")\n",
     "    eval_set = {(si, ri) for _h, si, ri, _k in ranked[:EVAL_N]}\n",
-    "    total = len(ranked)\n",
+    "    keep_set = {(si, ri) for _h, si, ri, _k in ranked}\n",
     "\n",
+    "    # pass 2: route rows, appending the derived caption column\n",
     "    ew = tw = None\n",
-    "    for si, p in enumerate(_paths):\n",
+    "    for si, pth in enumerate(_paths):\n",
+    "        pf = pq.ParquetFile(pth)\n",
+    "        names = pf.schema_arrow.names\n",
     "        ri = 0\n",
-    "        for batch in pq.ParquetFile(p).iter_batches(batch_size=64):\n",
+    "        for batch in pf.iter_batches(batch_size=64):\n",
     "            tbl = pa.Table.from_batches([batch])\n",
-    "            emask = [(si, ri + j) in eval_set for j in range(batch.num_rows)]\n",
-    "            ri += batch.num_rows\n",
+    "            jc = tbl.column(\"caption_joycaption\")\n",
+    "            pfu = (tbl.column(\"prompt_fused\")\n",
+    "                   if \"prompt_fused\" in names else None)\n",
+    "            caps, kmask, emask = [], [], []\n",
+    "            for j in range(tbl.num_rows):\n",
+    "                c = jc[j].as_py() or (pfu[j].as_py() if pfu is not None\n",
+    "                                      else None) or \"\"\n",
+    "                caps.append(str(c))\n",
+    "                kmask.append((si, ri + j) in keep_set)\n",
+    "                emask.append((si, ri + j) in eval_set)\n",
+    "            ri += tbl.num_rows\n",
+    "            tbl = tbl.append_column(\"caption\", pa.array(caps))\n",
     "            esub = tbl.filter(pa.array(emask))\n",
-    "            tsub = tbl.filter(pa.array([not m for m in emask]))\n",
+    "            tsub = tbl.filter(pa.array([k and not e for k, e\n",
+    "                                        in zip(kmask, emask)]))\n",
     "            if esub.num_rows:\n",
     "                if ew is None:\n",
     "                    ew = pq.ParquetWriter(EVAL_PARQUET, esub.schema)\n",
@@ -233,16 +270,24 @@
     "        if w:\n",
     "            w.close()\n",
     "    MANIFEST.write_text(json.dumps(\n",
-    "        {\"dataset\": ANIMA[\"DATASET\"], \"total_rows\": total, \"eval_n\": EVAL_N,\n",
+    "        {\"dataset\": ANIMA[\"DATASET\"], \"rows_seen\": n_seen,\n",
+    "         \"gated_out\": n_gated, \"duplicate_ids\": n_dup,\n",
+    "         \"rows_kept\": len(ranked), \"eval_n\": EVAL_N,\n",
     "         \"eval_ids\": [k for _h, _s, _r, k in ranked[:EVAL_N]],\n",
-    "         \"rule\": \"sha256(row_key) ascending; first EVAL_N rows are eval\"},\n",
+    "         \"caption_rule\": \"caption_joycaption or prompt_fused fallback \"\n",
+    "                         \"(dexp011:154; cond = caption_joycaption nl77)\",\n",
+    "         \"gate_rule\": \"age_audit/audit must contain approved|pass \"\n",
+    "                      \"(dexp011:98-105)\",\n",
+    "         \"split_rule\": \"sha256(id) ascending; first EVAL_N are eval\"},\n",
     "        indent=1))\n",
-    "    print(f\"split: {total} rows -> {total - EVAL_N} train + {EVAL_N} eval\")\n",
+    "    print(f\"split: seen {n_seen} -> gated {n_gated}, dup {n_dup}, kept \"\n",
+    "          f\"{len(ranked)} = {len(ranked) - EVAL_N} train + {EVAL_N} eval\")\n",
     "else:\n",
-    "    print(\"split already materialized:\",\n",
-    "          json.loads(MANIFEST.read_text())[\"total_rows\"], \"rows\")\n",
+    "    _m = json.loads(MANIFEST.read_text())\n",
+    "    print(f\"split already materialized: kept {_m['rows_kept']} \"\n",
+    "          f\"(gated {_m['gated_out']}, dup {_m['duplicate_ids']})\")\n",
     "\n",
-    "# -- dataset TOMLs (the exp004 recipe) ------------------------------------\n",
+    "# -- dataset TOMLs (fused recipe: derived caption column) -----------------\n",
     "_wh = (\"image_width_column = 'image_width'\\n\"\n",
     "       \"image_height_column = 'image_height'\\n\") if ANIMA[\"HAS_WH\"] else \"\"\n",
     "\n",
@@ -260,7 +305,7 @@
     "type = 'parquet'\n",
     "parquet_files = '{files}'\n",
     "image_column = 'image'\n",
-    "{_wh}caption_column = 'caption_vlm_json'\n",
+    "{_wh}caption_column = 'caption'\n",
     "caption_type = 'text'\n",
     "num_repeats = 1\n",
     "skip_empty_caption = true\n",
@@ -269,7 +314,6 @@
     "(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",
-    "\n",
     "# -- training TOMLs per arm -----------------------------------------------\n",
     "_AP = WORK / \"models\" / \"anima\" / \"split_files\"\n",
     "\n",