Add provenance/build_mix_distill_payload.py

provenance/build_mix_distill_payload.py

@@ -0,0 +1,789 @@
+#!/usr/bin/env python3
+"""Build Mix Distillation 1ep/2ep payload for Q4B-THINK-MIX (Li et al. 2025 Mix-Large).
+
+30B-A3B-Thinking : 4B-Thinking = 0.2 : 0.8 by row/problem count. One teacher
+trace per problem. Dose target 32,436,894 assistant tokens/ep (±1%), physical
+2ep concat same order. V4 recipe otherwise unchanged.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import random
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+ROOT = Path("/workspace/code-sft-runs/t30b2507-q4b-tn-v2-lr1e4-e1-20260907")
+ARM = ROOT / "arms" / "Q4B-THINK-MIX"
+DATA = ARM / "data"
+P30_2EP = ROOT / "data" / "THINK_PAIRED_V4_2EP.parquet"
+P30_2EP_SHA = "c160713fea76eae124dbbc3b5c619d3164862da83f61d91d3970ede990a12540"
+P30_1EP_ROWS = 4715
+P4 = ROOT / "arms" / "Q4B-THINK-T4BDATA" / "data" / "joint-full-t4b_tail151643.parquet"
+P4_SHA = "d09c4d6272de7e50c9b2df0e401878864f09bda33db05a5833c43575969fa524"
+P4_ROWS = 3866
+OUT_1EP = DATA / "mix_l0.2_s0.8_tail151643.parquet"
+OUT_2EP = DATA / "mix_l0.2_s0.8_tail151643_2EP.parquet"
+READY = DATA / "mix_l0.2_s0.8_tail151643_2EP_READY.json"
+CENSUS = DATA / "TOKEN_CENSUS_MIX.json"
+MQ0_OUT = DATA / "MQ0_NEAR_DUP.json"
+V4_POLICY = ROOT / "arms" / "Q4B-THINK" / "POLICY.json"
+V4_CFG = ROOT / "arms" / "Q4B-THINK-T4BDATA" / "TRAINING_CONFIG.json"
+V4_LAUNCH = ROOT / "arms" / "Q4B-THINK-T4BDATA" / "control" / "LAUNCH.json"
+WORKER = Path("/workspace/tools/rstar_dual_worker_v4.py")
+RUN_ID = "t30b2507-q4b-think-mix-l0.2-s0.8-v4-tail151643"
+ARM_ID = "Q4B-THINK-MIX"
+SEED = 42
+TARGET_1EP = 32_436_894
+TOL = 0.01
+RATIO_30, RATIO_4 = 1, 4
+
+sys.path.insert(0, str(ROOT / "scripts"))
+sys.path.insert(0, "/workspace/code-sft-infra/src")
+sys.path.insert(0, "/workspace/tools")
+sys.path.insert(0, "/workspace/code_think_30b_main_20260906/scripts")
+
+from code_sft_infra.contracts import document_digest, sha256_file  # noqa: E402
+from code_sft_infra.evaluation import build_lcb_prompt_messages  # noqa: E402
+from code_sft_infra.rendering import render_completion_only  # noqa: E402
+from code_sft_infra.source_census import load_hash_denylist  # noqa: E402
+import materialize_t30b_g2_fenced as think_m  # noqa: E402
+import near_dup_scan  # noqa: E402
+import select_core  # noqa: E402
+import v4_render_lib  # noqa: E402
+
+
+def _sha256_bytes_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(8 << 20), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _pct(values: list[int], p: float) -> float:
+    if not values:
+        return float("nan")
+    xs = sorted(values)
+    if len(xs) == 1:
+        return float(xs[0])
+    k = (len(xs) - 1) * p / 100.0
+    lo = int(math.floor(k))
+    hi = int(math.ceil(k))
+    if lo == hi:
+        return float(xs[lo])
+    return xs[lo] + (xs[hi] - xs[lo]) * (k - lo)
+
+
+def mq0_all_hits(rows: list[dict[str, Any]]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+    summary = think_m.run_mq0(rows)
+    if summary.get("coverage") != "6_OF_6":
+        raise SystemExit(f"MQ0 coverage not 6_OF_6: {summary}")
+    if summary.get("status") in {"blocked_before_train"}:
+        raise SystemExit(f"MQ0 blocked: {summary}")
+    denylist, _ = load_hash_denylist(think_m.DENYLIST_PATH)
+    extra, _ = think_m.load_eval_extra_signatures()
+    union = set(denylist) | extra
+    hits: list[dict[str, Any]] = []
+    for row in rows:
+        overlap = sorted(think_m.train_signatures(row) & union)
+        if overlap:
+            hits.append(
+                {
+                    "problem_hash": row["problem_hash"],
+                    "problem_id": row.get("problem_id"),
+                    "overlap_n": len(overlap),
+                    "overlap_head": overlap[:3],
+                    "reason": "mq0_intersection",
+                }
+            )
+    if len(hits) != int(summary.get("intersection") or 0):
+        raise SystemExit(
+            f"MQ0 hit count mismatch full={len(hits)} summary={summary.get('intersection')}"
+        )
+    return summary, hits
+
+
+def _index(rows: list[dict[str, Any]], label: str) -> dict[str, dict[str, Any]]:
+    out: dict[str, dict[str, Any]] = {}
+    pids: dict[str, str] = {}
+    for row in rows:
+        h = str(row["problem_hash"])
+        pid = str(row["problem_id"])
+        if h in out:
+            raise SystemExit(f"{label} duplicate problem_hash {h}")
+        if pid in pids:
+            raise SystemExit(f"{label} duplicate problem_id {pid}")
+        out[h] = row
+        pids[pid] = h
+    return out
+
+
+def _fill(
+    union: list[str],
+    preferred: dict[str, str],
+    only30: set[str],
+    only4: set[str],
+    n30: int,
+    n4: int,
+) -> tuple[list[str], list[str]]:
+    picked_30: list[str] = []
+    picked_4: list[str] = []
+    for h in union:
+        if len(picked_30) >= n30 and len(picked_4) >= n4:
+            break
+        if h in only30:
+            if len(picked_30) < n30:
+                picked_30.append(h)
+            continue
+        if h in only4:
+            if len(picked_4) < n4:
+                picked_4.append(h)
+            continue
+        pref = preferred[h]
+        if pref == "30b" and len(picked_30) < n30:
+            picked_30.append(h)
+        elif pref == "4b" and len(picked_4) < n4:
+            picked_4.append(h)
+        elif pref == "30b" and len(picked_4) < n4:
+            picked_4.append(h)
+        elif pref == "4b" and len(picked_30) < n30:
+            picked_30.append(h)
+    return picked_30, picked_4
+
+
+def _sample_record(i: int, row: dict[str, Any], tok: Any) -> dict[str, Any]:
+    reason = str(row.get("reasoning") or "")
+    answer = str(row.get("final_answer") or "")
+    rendered = v4_render_lib.render_qwen(
+        tok,
+        mode="think",
+        prompt=str(row["prompt"] or ""),
+        reasoning=reason,
+        answer=answer,
+    )
+    rec = {
+        "i": i,
+        "teacher_source": row["teacher_source"],
+        "problem_id": row["problem_id"],
+        "problem_hash": row["problem_hash"],
+        "prompt_head": str(row["prompt"])[:160].replace("\n", " "),
+        "think_in_reason": "<think>" in reason,
+        "think_close_in_reason": "</think>" in reason,
+        "code_fence_in_answer": "```" in answer,
+        "final_answer_head": answer[:120].replace("\n", "\\n"),
+        "assistant_target_tokens": int(row["assistant_target_tokens"]),
+        "full_tokens": int(row["full_tokens"]),
+        "rendered_think_open": rendered.target_text.lower().count("<think>"),
+        "rendered_think_close": rendered.target_text.lower().count("</think>"),
+        "rendered_code_fence": "```" in rendered.target_text,
+        "last_token": rendered.last_token_id,
+        "active_match": rendered.active_tokens == int(row["assistant_target_tokens"]),
+        "full_match": rendered.full_tokens == int(row["full_tokens"]),
+    }
+    if rec["rendered_think_open"] != 1 or rec["rendered_think_close"] != 1:
+        raise SystemExit(f"think pair not 1/1 i={i} {rec}")
+    if not rec["rendered_code_fence"]:
+        raise SystemExit(f"missing code fence i={i} {rec}")
+    if rec["last_token"] != 151643:
+        raise SystemExit(f"tail not 151643 i={i} last={rec['last_token']}")
+    return rec
+
+
+def main() -> int:
+    DATA.mkdir(parents=True, exist_ok=True)
+    for sub in ("out", "status", "logs", "control"):
+        (ARM / sub).mkdir(parents=True, exist_ok=True)
+
+    sha30 = _sha256_bytes_file(P30_2EP)
+    if sha30 != P30_2EP_SHA:
+        raise SystemExit(f"30B 2EP sha mismatch {sha30} != {P30_2EP_SHA}")
+    sha4 = _sha256_bytes_file(P4)
+    if sha4 != P4_SHA:
+        raise SystemExit(f"4B 1ep sha mismatch {sha4} != {P4_SHA}")
+
+    t30_all = pq.read_table(P30_2EP)
+    if t30_all.num_rows != P30_1EP_ROWS * 2:
+        raise SystemExit(f"30B 2EP rows {t30_all.num_rows} != {P30_1EP_ROWS * 2}")
+    t30 = t30_all.slice(0, P30_1EP_ROWS)
+    t4 = pq.read_table(P4)
+    if t4.num_rows != P4_ROWS:
+        raise SystemExit(f"4B rows {t4.num_rows} != {P4_ROWS}")
+    if list(t30.schema.names) != list(t4.schema.names):
+        raise SystemExit(f"schema names differ 30B={t30.schema.names} 4B={t4.schema.names}")
+    for name in t30.schema.names:
+        if str(t30.schema.field(name).type) != str(t4.schema.field(name).type):
+            raise SystemExit(f"schema type differ {name}")
+
+    rows30 = t30.to_pylist()
+    rows4 = t4.to_pylist()
+    tok30 = sum(int(r["assistant_target_tokens"]) for r in rows30)
+    tok4_pool = sum(int(r["assistant_target_tokens"]) for r in rows4)
+    if tok30 != TARGET_1EP:
+        raise SystemExit(f"30B 1ep tokens {tok30} != {TARGET_1EP}")
+    if tok4_pool != 30_327_765:
+        raise SystemExit(f"4B 1ep tokens {tok4_pool} != 30327765")
+
+    map30 = _index(rows30, "30B")
+    map4 = _index(rows4, "4B")
+    h30, h4 = set(map30), set(map4)
+    inter = h30 & h4
+    only30 = h30 - h4
+    only4 = h4 - h30
+    union_set = h30 | h4
+    pid30 = {str(map30[h]["problem_id"]) for h in h30}
+    pid4 = {str(map4[h]["problem_id"]) for h in h4}
+    if len(pid30 & pid4) != len(inter):
+        raise SystemExit("problem_id intersection size != problem_hash intersection")
+    for h in inter:
+        if str(map30[h]["problem_id"]) != str(map4[h]["problem_id"]):
+            raise SystemExit(f"hash/id mismatch {h}")
+
+    print(
+        "UNIVERSE",
+        json.dumps(
+            {
+                "union": len(union_set),
+                "intersect": len(inter),
+                "only30": len(only30),
+                "only4": len(only4),
+                "pid_intersect": len(pid30 & pid4),
+                "pid_only30": len(pid30 - pid4),
+                "pid_only4": len(pid4 - pid30),
+            }
+        ),
+    )
+
+    mean30 = tok30 / len(rows30)
+    mean4 = tok4_pool / len(rows4)
+    n30_est = max(1, int(round(TARGET_1EP / (mean30 * RATIO_30 + mean4 * RATIO_4))))
+    n4_est = n30_est * RATIO_4
+    print("EST", n30_est, n4_est, "mean30", mean30, "mean4", mean4)
+    need4_tok = 0.8 * TARGET_1EP
+    shortfall_notes: list[str] = []
+    if n4_est > len(h4):
+        shortfall_notes.append(
+            f"4B pool rows {len(h4)} < needed {n4_est} for 1:4; not changing ratio"
+        )
+    if tok4_pool < need4_tok:
+        shortfall_notes.append(
+            f"4B pool tokens {tok4_pool} < 0.8*target {need4_tok}; not changing ratio"
+        )
+    if shortfall_notes:
+        print("SHORTFALL", shortfall_notes)
+
+    rng = random.Random(SEED)
+    union = sorted(union_set)
+    rng.shuffle(union)
+    preferred: dict[str, str] = {}
+    for h in union:
+        if h in inter:
+            preferred[h] = "30b" if rng.choice((0, 1)) == 0 else "4b"
+
+    def tokens_of(p30: list[str], p4: list[str]) -> int:
+        return sum(int(map30[h]["assistant_target_tokens"]) for h in p30) + sum(
+            int(map4[h]["assistant_target_tokens"]) for h in p4
+        )
+
+    candidates: list[tuple[int, int, list[str], list[str], int]] = []
+    n30_max = min(len(h30), len(h4) // RATIO_4)
+    for n30 in range(1, n30_max + 1):
+        n4 = n30 * RATIO_4
+        p30, p4 = _fill(union, preferred, only30, only4, n30, n4)
+        if len(p30) != n30 or len(p4) != n4:
+            continue
+        tok = tokens_of(p30, p4)
+        candidates.append((abs(tok - TARGET_1EP), n30, p30, p4, tok))
+
+    if not candidates:
+        raise SystemExit("no exact 1:4 fill succeeded")
+    in_band = [c for c in candidates if abs(c[4] - TARGET_1EP) <= TARGET_1EP * TOL]
+    pool = in_band if in_band else candidates
+    pool.sort(key=lambda c: (c[0], abs(c[1] - n30_est)))
+    _, n30, picked_30, picked_4, active_1ep = pool[0]
+    n4 = n30 * RATIO_4
+    rel = abs(active_1ep - TARGET_1EP) / TARGET_1EP
+    print(
+        "PICKED",
+        json.dumps(
+            {
+                "n30": n30,
+                "n4": n4,
+                "active_1ep": active_1ep,
+                "rel_err": rel,
+                "in_band": rel <= TOL,
+                "n30_est": n30_est,
+                "candidates_in_band": len(in_band),
+            }
+        ),
+    )
+    if rel > TOL:
+        raise SystemExit(f"token dose {active_1ep} outside ±1% of {TARGET_1EP}")
+
+    mixed: list[dict[str, Any]] = []
+    for h in picked_30:
+        row = dict(map30[h])
+        row["teacher_source"] = "30b"
+        mixed.append(row)
+    for h in picked_4:
+        row = dict(map4[h])
+        row["teacher_source"] = "4b"
+        mixed.append(row)
+    if len({r["problem_hash"] for r in mixed}) != len(mixed):
+        raise SystemExit("duplicate problem in mix")
+    order_rng = random.Random(SEED)
+    order_rng.shuffle(mixed)
+
+    mq0_summary, mq0_hits = mq0_all_hits(mixed)
+    print(
+        "mq0",
+        mq0_summary.get("status"),
+        "intersection",
+        mq0_summary.get("intersection"),
+        "coverage",
+        mq0_summary.get("coverage"),
+    )
+    if mq0_hits or int(mq0_summary.get("intersection") or 0) != 0:
+        raise SystemExit(f"MQ0 intersection nonzero: {mq0_hits[:5]}")
+    if mq0_summary.get("coverage") != "6_OF_6":
+        raise SystemExit(f"MQ0 coverage {mq0_summary.get('coverage')}")
+
+    near = near_dup_scan.scan_training_rows(mixed)
+    print("near_dup hits", len(near["hit_hashes"]), "grey", near["grey_zone_count"])
+    if near["hit_hashes"]:
+        raise SystemExit(f"near-dup hits: {near['hit_hashes']}")
+
+    tok = v4_render_lib.load_qwen_tokenizer()
+    idx30 = [i for i, r in enumerate(mixed) if r["teacher_source"] == "30b"][:3]
+    idx4 = [i for i, r in enumerate(mixed) if r["teacher_source"] == "4b"][:3]
+    samples = [_sample_record(i, mixed[i], tok) for i in idx30 + idx4]
+    if len(samples) != 6:
+        raise SystemExit(f"expected 6 samples, got {len(samples)}")
+
+    v4_policy = json.loads(V4_POLICY.read_text())
+    renderer = v4_policy["arms"][0]["renderer"]
+    authority = v4_policy["arms"][0]["renderer_authority"]
+    contract = {"renderer": renderer, "sequence": {"max_sequence_length": 32768}}
+    for rec in samples:
+        row = mixed[rec["i"]]
+        worker = render_completion_only(
+            contract,
+            renderer_id="qwen3-8b-think-lcb-system-v2",
+            prompt_messages=build_lcb_prompt_messages(contract, row["prompt"]),
+            reasoning=row["reasoning"],
+            answer=row["final_answer"],
+        )
+        if worker.labels[-1] != 151643:
+            raise SystemExit(f"worker last label {worker.labels[-1]} i={rec['i']}")
+        rec["worker_last"] = worker.labels[-1]
+        rec["worker_active"] = sum(x != -100 for x in worker.labels)
+        rec["worker_full"] = len(worker.input_ids)
+        print("SAMPLE", json.dumps(rec, ensure_ascii=False))
+
+    schema = t30.schema
+    arrays = []
+    for name in schema.names:
+        arrays.append(pa.array([r[name] for r in mixed], type=schema.field(name).type))
+    arrays.append(pa.array([r["teacher_source"] for r in mixed], type=pa.string()))
+    names = list(schema.names) + ["teacher_source"]
+    one = pa.table(arrays, names=names)
+    two = pa.concat_tables([one, one])
+    if OUT_1EP.exists() or OUT_2EP.exists():
+        raise SystemExit("refusing to overwrite existing mix parquet")
+    pq.write_table(one, OUT_1EP)
+    pq.write_table(two, OUT_2EP)
+    sha1 = _sha256_bytes_file(OUT_1EP)
+    sha2 = _sha256_bytes_file(OUT_2EP)
+    actives = [int(r["assistant_target_tokens"]) for r in mixed]
+    fulls = [int(r["full_tokens"]) for r in mixed]
+    actives30 = [int(r["assistant_target_tokens"]) for r in mixed if r["teacher_source"] == "30b"]
+    actives4 = [int(r["assistant_target_tokens"]) for r in mixed if r["teacher_source"] == "4b"]
+    active_2ep = active_1ep * 2
+    print("wrote", OUT_1EP, one.num_rows, sha1, "active1ep", active_1ep)
+    print("wrote", OUT_2EP, two.num_rows, sha2, "active2ep", active_2ep)
+
+    stream_rows = [
+        {"problem_hash": r["problem_hash"], "active_tokens": int(r["assistant_target_tokens"])}
+        for r in mixed
+    ]
+    derived = select_core.derive_after_materialize(stream_rows)
+    if derived["planned_active_tokens_2ep"] != active_2ep:
+        raise SystemExit("derived 2ep tokens != 2*1ep")
+    print(
+        "derived",
+        json.dumps(
+            {
+                k: derived[k]
+                for k in (
+                    "unique_rows",
+                    "physical_rows_2ep",
+                    "unique_1ep_think_active_tokens",
+                    "planned_active_tokens_2ep",
+                    "optimizer_updates",
+                    "warmup_target_tokens",
+                    "milestone_updates_right_3",
+                    "checkpoint_target_tokens_right_3",
+                )
+            }
+        ),
+    )
+
+    mq0_body = {
+        "mq0": {
+            "status": mq0_summary.get("status"),
+            "coverage": mq0_summary.get("coverage"),
+            "intersection": mq0_summary.get("intersection"),
+            "denylist_path": mq0_summary.get("denylist_path"),
+            "denylist_sha256": mq0_summary.get("denylist_sha256"),
+        },
+        "near_dup_hit_hashes": near["hit_hashes"],
+        "near_dup_grey_zone_count": near["grey_zone_count"],
+        "dropped_n": 0,
+        "kept_n": len(mixed),
+    }
+    MQ0_OUT.write_text(json.dumps(mq0_body, indent=2, ensure_ascii=False) + "\n")
+
+    census = {
+        "schema": "TOKEN_CENSUS_MIX_L0.2_S0.8_TAIL151643",
+        "seed": SEED,
+        "n_rows": len(mixed),
+        "n_problems": len(mixed),
+        "parquet_1ep": str(OUT_1EP),
+        "parquet_1ep_sha256": sha1,
+        "parquet_2ep": str(OUT_2EP),
+        "parquet_2ep_sha256": sha2,
+        "supervised_tail": [151643],
+        "active_1ep": active_1ep,
+        "active_2ep": active_2ep,
+        "full_1ep": int(sum(fulls)),
+        "target_1ep": TARGET_1EP,
+        "rel_err": rel,
+        "ratio_rows_30_to_4": f"{n30}:{n4}",
+        "universe": {
+            "union": len(union_set),
+            "intersect": len(inter),
+            "only30": len(only30),
+            "only4": len(only4),
+        },
+        "teacher": {
+            "n30": n30,
+            "n4": n4,
+            "tok30": int(sum(actives30)),
+            "tok4": int(sum(actives4)),
+            "row_frac_30": n30 / len(mixed),
+            "row_frac_4": n4 / len(mixed),
+            "tok_frac_30": sum(actives30) / active_1ep,
+            "tok_frac_4": sum(actives4) / active_1ep,
+            "n30_from_intersect": sum(1 for h in picked_30 if h in inter),
+            "n30_from_only30": sum(1 for h in picked_30 if h in only30),
+            "n4_from_intersect": sum(1 for h in picked_4 if h in inter),
+            "n4_from_only4": sum(1 for h in picked_4 if h in only4),
+        },
+        "shortfall_notes": shortfall_notes,
+        "active": {
+            "n": len(actives),
+            "sum": active_1ep,
+            "min": min(actives),
+            "p10": _pct(actives, 10),
+            "p50": _pct(actives, 50),
+            "p90": _pct(actives, 90),
+            "p95": _pct(actives, 95),
+            "max": max(actives),
+        },
+        "active_30b": {
+            "n": len(actives30),
+            "sum": int(sum(actives30)),
+            "min": min(actives30),
+            "p50": _pct(actives30, 50),
+            "max": max(actives30),
+        },
+        "active_4b": {
+            "n": len(actives4),
+            "sum": int(sum(actives4)),
+            "min": min(actives4),
+            "p50": _pct(actives4, 50),
+            "max": max(actives4),
+        },
+        "derived": {
+            k: derived[k]
+            for k in (
+                "unique_rows",
+                "physical_rows_2ep",
+                "unique_1ep_think_active_tokens",
+                "planned_active_tokens_2ep",
+                "optimizer_updates",
+                "warmup_target_tokens",
+                "milestone_updates_right_3",
+                "checkpoint_target_tokens_right_3",
+            )
+        },
+        "samples": samples,
+        "sources": {
+            "30b_2ep": str(P30_2EP),
+            "30b_2ep_sha256": sha30,
+            "30b_1ep_rows_used": P30_1EP_ROWS,
+            "4b_1ep": str(P4),
+            "4b_1ep_sha256": sha4,
+        },
+    }
+    CENSUS.write_text(json.dumps(census, indent=2, ensure_ascii=False) + "\n")
+
+    ready_body = {
+        "physical_2ep": "concat(mix_table, mix_table) same order; tail [151643]; teacher_source kept",
+        "schema": "CODE_SFT_OR1RECIPE_READY_V1",
+        "status": "READY",
+        "training_payload": {
+            "active_tokens": active_2ep,
+            "path": OUT_2EP.name,
+            "rows": two.num_rows,
+            "sha256": sha2,
+        },
+    }
+    READY.write_text(json.dumps(ready_body, indent=2, sort_keys=True) + "\n")
+    ready_sha = sha256_file(READY)
+
+    cfg = json.loads(V4_CFG.read_text())
+    cfg["status"] = "READY_NOT_STARTED"
+    cfg["config_id"] = ARM_ID
+    cfg["arm_id"] = ARM_ID
+    cfg["corrects"] = "Q4B-THINK Mix Distillation 0.2:0.8 (30B:4B by rows)"
+    cfg["checkpoint"]["milestone_updates"] = derived["milestone_updates_right_3"]
+    cfg["checkpoint"]["checkpoint_target_tokens"] = derived["checkpoint_target_tokens_right_3"]
+    cfg["data"]["path"] = str(OUT_2EP)
+    cfg["data"]["sha256"] = sha2
+    cfg["data"]["unique_rows"] = len(mixed)
+    cfg["data"]["payload_rows"] = two.num_rows
+    cfg["data"]["physical_rows"] = two.num_rows
+    cfg["data"]["physical_2ep_rule"] = (
+        "concat(mix_l0.2_s0.8, mix_l0.2_s0.8) same order; 1ep is unique mixed rows"
+    )
+    cfg["data"]["planned_active_tokens"] = active_2ep
+    cfg["data"]["active_tokens_1ep"] = active_1ep
+    cfg["data"]["full_tokens_1ep"] = int(sum(fulls))
+    cfg["data"]["optimizer_updates"] = derived["optimizer_updates"]
+    cfg["data"]["census"] = str(CENSUS)
+    cfg["data"]["token_columns_in_parquet"] = "supervised tail [151643]; matches v4 worker render"
+    cfg["data"]["source_1ep_sha256"] = sha1
+    cfg["data"].pop("source_old_payload", None)
+    cfg["data"].pop("source_old_sha256", None)
+    cfg["optimization"]["warmup_target_tokens"] = derived["warmup_target_tokens"]
+    cfg["optimization"]["scheduler_horizon_optimizer_updates"] = derived["optimizer_updates"]
+    cfg["optimization"]["target_assistant_tokens"] = active_2ep
+    worker_sha = _sha256_bytes_file(WORKER)
+    cfg["runner"]["sha256"] = worker_sha
+    (ARM / "TRAINING_CONFIG.json").write_text(json.dumps(cfg, indent=2, ensure_ascii=False) + "\n")
+
+    (ARM / "TOKEN_ROWS_META.json").write_text(
+        json.dumps(
+            {
+                "schema": "TOKEN_ROWS_META_V1",
+                "ids": [151643, 151667, 151668],
+                "dtype": "float32",
+                "tied_before": True,
+                "untied": True,
+                "worker": "rstar_dual_worker_v4",
+                "hidden": 2560,
+                "base_rev": "906bfd4b4dc7f14ee4320094d8b41684abff8539",
+            },
+            indent=2,
+        )
+        + "\n"
+    )
+
+    identity = {
+        "schema": "RUN_IDENTITY_V1",
+        "run_id": RUN_ID,
+        "corrects": "Q4B-THINK Mix Distillation 0.2:0.8 (30B-A3B-Thinking : 4B-Thinking by rows)",
+        "notes": (
+            "Mix Distillation Mix-Large (Li et al. 2025 arXiv 2502.12143): "
+            "30B-A3B-Thinking : 4B-Thinking = 0.2 : 0.8 by row/problem; one teacher "
+            f"trace per problem; dose target {TARGET_1EP}/ep ±1%, actual 1ep "
+            f"{active_1ep}, 2ep {active_2ep}. 30B source={P30_2EP} sha={sha30} "
+            f"(2EP first half {P30_1EP_ROWS} rows). 4B source={P4} sha={sha4}. "
+            "V4 recipe unchanged (LoRA r64/α128, lr 1e-4, both-sides B-row "
+            "[151643,151667,151668], physical 2ep concat, seed 42). "
+            "PI approved 2026-09-09."
+        ),
+        "model": {
+            "hf_id": "Qwen/Qwen3-4B-Base",
+            "revision": "906bfd4b4dc7f14ee4320094d8b41684abff8539",
+            "local_path": "/workspace/code-sft-infra/models/qwen3-4b-base",
+            "tied_embeddings": True,
+        },
+        "data": {
+            "path": str(OUT_2EP),
+            "rows": two.num_rows,
+            "sha256": sha2,
+            "census": str(CENSUS),
+            "physical_2ep": "concat(mix, mix) same order; teacher_source 30b/4b; tail 151643",
+            "source_1ep_path": str(OUT_1EP),
+            "source_1ep_sha256": sha1,
+            "source_1ep_rows": one.num_rows,
+            "teacher_30b_path": str(P30_2EP),
+            "teacher_30b_sha256": sha30,
+            "teacher_4b_path": str(P4),
+            "teacher_4b_sha256": sha4,
+            "mix_ratio_rows": "0.2:0.8",
+            "n30": n30,
+            "n4": n4,
+            "active_tokens_1ep": active_1ep,
+        },
+        "seeds": {"train": 42, "eval": 3407, "mix": 42},
+        "parameters": {
+            "epochs": 2,
+            "target_assistant_tokens": active_2ep,
+            "active_tokens_1ep": active_1ep,
+            "checkpoint_target_tokens": derived["checkpoint_target_tokens_right_3"],
+            "milestone_updates": derived["milestone_updates_right_3"],
+            "warmup_target_tokens": derived["warmup_target_tokens"],
+            "learning_rate": 0.0001,
+            "row_lr": 0.0001,
+            "optimizer": "adamw",
+            "optimizer_betas": [0.9, 0.95],
+            "optimizer_eps": 1e-08,
+            "lora_rank": 64,
+            "lora_alpha": 128,
+            "lora_dropout": 0.0,
+            "lora_target_modules": [
+                "q_proj",
+                "k_proj",
+                "v_proj",
+                "o_proj",
+                "gate_proj",
+                "up_proj",
+                "down_proj",
+            ],
+            "lora_weight_decay": 0.1,
+            "row_weight_decay": 0.0,
+            "max_grad_norm": 1.0,
+            "tokens_per_optimizer_update": 65536,
+            "packing": False,
+            "truncation": False,
+            "repeat_examples": False,
+            "precision": "bfloat16",
+            "trainable_dtype": "float32",
+            "gradient_checkpointing": True,
+            "eos_weight": 1.0,
+            "scheduler": "cosine_by_assistant_target_token_dose",
+            "worker_lr_scale": "rstar_dual_worker_v4._lr_scale",
+            "supervised_tail": [151643],
+            "trainable_token_indices": {
+                "embed_tokens": [151643, 151667, 151668],
+                "lm_head": [151643, 151667, 151668],
+            },
+            "untie_tied_embeddings": True,
+            "context": 32768,
+            "eval": {
+                "suite": "DEV256",
+                "seed": 3407,
+                "temperature": 0.6,
+                "top_p": 0.95,
+                "top_k": 20,
+                "ctx": 32768,
+                "stop_ids": [151643, 151645],
+                "base_reeval_same_mode": True,
+            },
+        },
+        "code": {
+            "path": str(WORKER),
+            "commit_or_sha256": worker_sha,
+            "prep": str(Path(__file__).resolve()),
+            "prep_sha256": _sha256_bytes_file(Path(__file__).resolve()),
+        },
+        "started_at": None,
+        "endpoint": f"2EP_ONLY @{active_2ep} assistant tokens; no checkpoint picking",
+        "status": "READY_NOT_STARTED",
+    }
+    (ARM / "RUN_IDENTITY.json").write_text(json.dumps(identity, indent=2, ensure_ascii=False) + "\n")
+
+    v4_launch = json.loads(V4_LAUNCH.read_text())
+    policy = {
+        "schema": "CODE_SFT_RSTAR_DUAL_TRAINING_POLICY_V1",
+        "status": "SEALED",
+        "owner_authorization": "PI 2026-09-09 Q4B-THINK-MIX Mix Distillation 0.2:0.8",
+        "generation_behavior_gate": None,
+        "arms": [
+            {
+                "arm_id": ARM_ID,
+                "gpu": 2,
+                "model_id": "qwen3-4b-base",
+                "source_id": "rstar-coder",
+                "renderer_id": "qwen3-8b-think-lcb-system-v2",
+                "renderer": renderer,
+                "renderer_authority": authority,
+                "model": json.loads(json.dumps(v4_policy["arms"][0]["model"])),
+                "data": {
+                    "assistant_target_tokens": active_2ep,
+                    "manifest_path": str(READY),
+                    "manifest_file_sha256": ready_sha,
+                    "manifest_sha256": ready_sha,
+                    "payload_rows": two.num_rows,
+                    "payload_sha256": sha2,
+                    "source_revision": "3a7a0a0636ec96e3c1ec42ebe79ade467caa040d",
+                    "max_assistant_target_tokens_per_example": 16384,
+                },
+            }
+        ],
+        "recipe": json.loads(json.dumps(v4_policy["recipe"])),
+        "runtime": {
+            "state_root": str(ARM),
+            "checkpoint_root": str(ARM / "out"),
+        },
+    }
+    policy["recipe"]["target_assistant_tokens"] = active_2ep
+    policy["recipe"]["warmup_target_tokens"] = derived["warmup_target_tokens"]
+    policy["recipe"]["checkpoint_target_tokens"] = derived["checkpoint_target_tokens_right_3"]
+    policy_path = ARM / "POLICY.json"
+    policy_path.write_text(json.dumps(policy, indent=2, sort_keys=True) + "\n")
+    policy_sha = sha256_file(policy_path)
+
+    job = {
+        "arm_id": ARM_ID,
+        "assistant_target_tokens": active_2ep,
+        "gpu": 2,
+        "materialized_manifest_path": str(READY),
+        "materialized_manifest_sha256": ready_sha,
+        "model_hf_id": "Qwen/Qwen3-4B-Base",
+        "model_revision": "906bfd4b4dc7f14ee4320094d8b41684abff8539",
+        "payload_sha256": sha2,
+        "renderer_authority_sha256": authority["sha256"],
+        "renderer_id": "qwen3-8b-think-lcb-system-v2",
+        "source_id": "rstar-coder",
+    }
+    job["job_sha256"] = document_digest(job, "job_sha256")
+    launch = {
+        "schema": "CODE_SFT_RSTAR_DUAL_LAUNCH_V1",
+        "status": "ATOMICALLY_RELEASED",
+        "run_id": RUN_ID,
+        "automation_policy_sha256": v4_launch["automation_policy_sha256"],
+        "delegation_sha256": v4_launch["delegation_sha256"],
+        "launch_critical_go_sha256": v4_launch["launch_critical_go_sha256"],
+        "release_sha256": v4_launch["release_sha256"],
+        "release_nonce": v4_launch["release_nonce"],
+        "release_claim_path": v4_launch["release_claim_path"],
+        "release_claim_file_sha256": v4_launch["release_claim_file_sha256"],
+        "evidence_path": v4_launch["evidence_path"],
+        "evidence_file_sha256": v4_launch["evidence_file_sha256"],
+        "training_policy_path": str(policy_path),
+        "training_policy_sha256": policy_sha,
+        "jobs": [job],
+    }
+    launch["launch_sha256"] = document_digest(launch, "launch_sha256")
+    launch_path = ARM / "control" / "LAUNCH.json"
+    launch_path.write_text(json.dumps(launch, indent=2, sort_keys=True) + "\n")
+    if document_digest(json.loads(launch_path.read_text()), "launch_sha256") != launch["launch_sha256"]:
+        raise SystemExit("launch digest drifted after write")
+    print("wrote", policy_path, policy_sha)
+    print("wrote", launch_path, launch["launch_sha256"])
+    print("PREP_OK", datetime.now(timezone.utc).isoformat())
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())