recipe+sizing artifacts

_recipe/year_branch_train.py

@@ -0,0 +1,345 @@
+"""FY2001 additive year-plugin SCAFFOLD (plan §2, r=512). Skeleton + DRY mode.
+AdditiveYearBranch = independent bottleneck FFN E_t (d->r->d, r=512) + scalar gate
+g_t=σ(w_t·h + v_t·c_t), hooked at insertion layers. Frozen backbone; ONLY E_t/g_t train.
+Hard year-key isolation (this-year tokens only) + cutoff g:=0 -> byte-exact backbone.
+DRY: wire onto frozen base, train a few steps on tiny FY2001 set, prove (a) branch reduces
+NLL, (b) backbone params byte-unchanged (base_equal). Real training gated on Manager review.
+"""
+import os, sys, json, math, urllib.request, random
+
+BLUE="/blue/fire-finai/lq62.yale/annulus"; CODE_V11=f"{BLUE}/code_v11"
+BASE_CKPT=f"{BLUE}/checkpoints/train/v11-33e-2001/weights"
+TOK_PATH=f"{BLUE}/checkpoints/train/v11-33e-2001/context/qwen2-annulus-v5f"
+EDGAR=f"{BLUE}/data/edgar_multiyear_v2"; OUT_DIR=f"{BLUE}/residprep_uf"
+DISCORD_CHAN="1516845719955439636"; MANAGER_UID="1518372899575169215"
+MAXTOK=512; R=int(os.environ.get("BRANCH_R") or "128"); SEED=1234
+INSERT_LAYERS=[int(x) for x in (os.environ.get("INSERT_LAYERS") or "12,23").split(",")]  # plan §1: early+mid
+DRY=os.environ.get("DRY","1")=="1"; DRY_DOCS=8; DRY_STEPS=int(os.environ.get("DRY_STEPS") or "400")
+LR=float(os.environ.get('LR') or '3e-4'); FISCAL_YEAR=2001
+DATA_YEAR=int(os.environ.get("DATA_YEAR") or "2001")
+DATA_N=int(os.environ.get("DATA_N") or "3000")
+REAL_STEPS=int(os.environ.get("REAL_STEPS") or "3000")
+FILTERED_DIR=os.environ.get("FILTERED_DIR") or "/blue/fire-finai/lq62.yale/annulus/data/01_filtered/parts"
+
+os.environ["ANNULUS_ROUTED_EXPERTS"]="33"; os.environ["ANNULUS_SHARED_EXPERTS"]="1"
+os.environ["ANNULUS_SHARED_FFN"]="2048"; os.environ["ANNULUS_LAYERS"]="24"
+os.environ["ANNULUS_TOPK"]="8"; os.environ["ANNULUS_GROUPED_GEMM"]="0"; os.environ["ANNULUS_GROUP_AUX"]="1"
+os.environ["RANK"]="0"; os.environ["WORLD_SIZE"]="1"; os.environ["LOCAL_RANK"]="0"
+os.environ.setdefault("MASTER_ADDR","localhost")
+os.environ["MASTER_PORT"]=str(20000+(int(os.environ.get("SLURM_JOB_ID","0") or 0) or os.getpid())%40000)
+for p in [f"{CODE_V11}/nemo/src", f"{BLUE}/.pylibs", CODE_V11, f"{CODE_V11}/eval"]:
+    try: sys.path.remove(p)
+    except ValueError: pass
+    sys.path.insert(0,p)
+
+def discord_post(msg):
+    t=os.environ.get("DISCORD_BOT_TOKEN","")
+    if not t: print(f"[discord_post] no token — {msg}"); return
+    try:
+        urllib.request.urlopen(urllib.request.Request(
+            f"https://discord.com/api/v10/channels/{DISCORD_CHAN}/messages",
+            data=json.dumps({"content":msg}).encode(),
+            headers={"Authorization":f"Bot {t}","Content-Type":"application/json"},method="POST"),timeout=15)
+    except Exception as e: print(f"[discord_post] failed: {e}")
+
+import torch
+torch.manual_seed(SEED); random.seed(SEED)
+
+class AdditiveYearBranch(torch.nn.Module):
+    """E_t + scalar content-gate. Init to no-op (fc2=0) => backbone byte-exact until trained."""
+    def __init__(self, d, r, c_t):
+        super().__init__()
+        self.fc1=torch.nn.Linear(d, r); self.fc2=torch.nn.Linear(r, d, bias=False)
+        torch.nn.init.zeros_(self.fc2.weight)
+        self.w=torch.nn.Parameter(torch.zeros(d)); self.v=torch.nn.Parameter(torch.zeros(d))
+        self.register_buffer("c_t", c_t.detach().float())  # frozen year-context vector
+        self.enabled=True   # cutoff: set False -> g:=0 -> h_out=h_frozen byte-exact
+    def forward(self, h):
+        if not self.enabled: return h
+        wd=self.fc1.weight.dtype
+        g=torch.sigmoid((h.to(wd)*self.w).sum(-1,keepdim=True) + (self.c_t*self.v).sum())  # σ(w·h+v·c_t)
+        e=self.fc2(torch.nn.functional.silu(self.fc1(h.to(wd))))
+        return h + (g*e).to(h.dtype)
+
+print(f"[branch] loading base v11-32e (mask e32), R={R}, insert L{INSERT_LAYERS}, DRY={DRY}", flush=True)
+import icl_eval_v5 as V
+core, tok = V.build_v5_model_and_tokenizer(BASE_CKPT, TOK_PATH)
+nm=0
+for name,param in core.named_parameters():
+    if ("mlp.experts.linear_fc" in name) and name.endswith("weight32"): param.data.zero_(); nm+=1
+assert nm>0
+for p in core.parameters(): p.requires_grad=False   # FREEZE backbone
+core.eval()
+# snapshot backbone for base_equal (byte-exact freeze proof)
+snap={n:p.detach().clone() for n,p in core.named_parameters()}
+
+# year-context c_t = [Y{yr}] token embedding (real year anchor)
+d_model=1024
+YEAR_TOK_ID=tok(f"[Y{FISCAL_YEAR}]",add_special_tokens=False)["input_ids"]
+assert len(YEAR_TOK_ID)==1, f"[Y{FISCAL_YEAR}] not single token: {YEAR_TOK_ID}"
+YEAR_TOK_ID=YEAR_TOK_ID[0]
+_emb=None
+for _n,_p in core.named_parameters():
+    if _p.dim()==2 and _p.shape[1]==d_model and _p.shape[0]>=150000 and ("embed" in _n.lower() or "word" in _n.lower()):
+        _emb=_p; break
+if _emb is None:
+    for _n,_p in core.named_parameters():
+        if _p.dim()==2 and _p.shape[1]==d_model and _p.shape[0]>=150000: _emb=_p; break
+assert _emb is not None, "word embedding [vocab,d_model] not found"
+c_t=_emb[YEAR_TOK_ID].detach().float().cuda()
+_ctnorm=float(c_t.norm())
+print(f"[branch] special_token_added=True YEAR_TOK[Y{FISCAL_YEAR}]={YEAR_TOK_ID} c_t_norm={_ctnorm:.4f} emb={tuple(_emb.shape)}",flush=True)
+discord_post(f"\U0001f9e9 <@{MANAGER_UID}> FY2001 anchor: [Y{FISCAL_YEAR}]token=YES(id={YEAR_TOK_ID}, prepended to每篇), c_t=emb[{YEAR_TOK_ID}] norm={_ctnorm:.4f} "+("YEAR-ANCHOR-OK" if _ctnorm>1e-6 else "c_t=0!"))
+
+
+tf=[m for n,m in core.named_modules() if m.__class__.__name__=="TransformerLayer"]
+branches=[]; hooks=[]
+for li in INSERT_LAYERS:
+    br=AdditiveYearBranch(d_model, R, c_t).cuda(); branches.append(br)
+    def mk(b):
+        def hook(mod,inp,out):
+            hs=out[0] if isinstance(out,tuple) else out; m=b(hs)
+            return (m,)+out[1:] if isinstance(out,tuple) else m
+        return hook
+    hooks.append(tf[li].register_forward_hook(mk(br)))
+n_branch_params=sum(p.numel() for b in branches for p in b.parameters())
+print(f"[branch] {len(branches)} branches, {n_branch_params} trainable params (~{n_branch_params/1e6:.2f}M)", flush=True)
+# ③ cutoff invariant HARD GATE (module-level, noise-free): enabled=False => branch(x)==x bit-for-bit
+_b=branches[0]; _b.enabled=False
+for _dt in (torch.bfloat16, torch.float32):
+    _x=torch.randn(5,1,d_model,device="cuda",dtype=_dt)
+    assert torch.equal(_b(_x),_x), f"CUTOFF ③ FAIL dt={_dt}: branch(x)!=x when disabled"
+_b.enabled=True
+print("[branch] ③ CUTOFF_INVARIANT_PASS (enabled=False -> bit-exact)", flush=True)
+
+docs=[]; n2001=0
+import glob as _g3
+import pyarrow.parquet as _pq3
+_cand=sorted(_g3.glob(f"{FILTERED_DIR}/year_{FISCAL_YEAR}_*.parquet"))
+_yearsplit=bool(_cand)
+if not _cand: _cand=sorted(_g3.glob(f"{FILTERED_DIR}/*.parquet"))
+assert _cand, f"no parquet in {FILTERED_DIR}"
+_sch=_pq3.ParquetFile(_cand[0]).schema_arrow
+_tx=next((x for x in ("text","content","body","document","output","input","raw") if x in _sch.names),None)
+_dc=next((x for x in ("date","date_filed","fiscal_year","filing_year","year","time") if x in _sch.names),None)
+assert _tx, f"cols={list(_sch.names)} no text col"
+for _p in _cand:
+    if len(docs)>=DATA_N: break
+    _cols=[_tx]+([_dc] if _dc and not _yearsplit else [])
+    _t=_pq3.read_table(_p,columns=_cols)
+    _tl=_t.column(_tx).to_pylist()
+    _dl=_t.column(_dc).to_pylist() if (_dc and not _yearsplit) else [None]*len(_tl)
+    for _tv,_dv in zip(_tl,_dl):
+        if len(docs)>=DATA_N: break
+        if _yearsplit or (not _dc) or str(_dv)[:4]==str(FISCAL_YEAR):
+            n2001+=1
+            if _tv and len(_tv)>1500:
+                ids=[YEAR_TOK_ID]+tok(_tv,add_special_tokens=False)["input_ids"][:MAXTOK-1]
+                if len(ids)>=16: docs.append(ids)
+print(f"[branch] parts={len(_cand)} yearsplit={_yearsplit} text={_tx} date={_dc} FY{FISCAL_YEAR}_rows={n2001} usable={len(docs)}",flush=True)
+discord_post(f"\U0001f9e9 <@{MANAGER_UID}> 01_filtered 落地读通:parts={len(_cand)} yearsplit={_yearsplit} FY{FISCAL_YEAR}行={n2001} 可用docs={len(docs)}")
+assert len(docs)>=8, f"only {len(docs)} usable (rows={n2001}) — STOP ask Lingfei"
+
+def nll(ds):
+    tot,nt=0.0,0
+    with torch.no_grad():
+        for ids in ds:
+            inp=torch.tensor([ids],device="cuda"); pos=torch.arange(len(ids),device="cuda").unsqueeze(0)
+            out=core(input_ids=inp,position_ids=pos,attention_mask=None)
+            lg=(out[0] if isinstance(out,tuple) else out).float()
+            lp=torch.log_softmax(lg[0,:-1],dim=-1); tgt=torch.tensor(ids[1:],device="cuda")
+            tot+=float((-lp.gather(1,tgt[:,None]).squeeze(1)).mean()); nt+=1
+    return tot/nt
+
+# ---- DIAG mode: load trained BEST plugin, compare ON/OFF NLL in train() vs eval(), dump c_t/gate ----
+if os.environ.get("DIAG"):
+    import glob as _gd
+    _cks=sorted(_gd.glob(f"{OUT_DIR}/plugin_FY{DATA_YEAR}_r{R}_*BEST.pt"))  # FIX: 带R精确匹配,别把r32/r106混选
+    assert _cks, f"no BEST ckpt in {OUT_DIR}"
+    _sd=torch.load(_cks[-1],map_location="cuda")
+    for _j,_b in enumerate(branches): _b.load_state_dict(_sd[f"branch{_j}"])
+    print(f"[diag] loaded {_cks[-1]}",flush=True)
+    _msg=[f"loaded {_cks[-1].split('/')[-1]}"]
+    _msg.append(f"c_t.norm(emb[{YEAR_TOK_ID}])={float(branches[0].c_t.norm()):.4f}")
+    for _j,_b in enumerate(branches[:2]):
+        print(f"[diag] branch{_j} w={float(_b.w.norm()):.3f} v={float(_b.v.norm()):.3f} fc2w={float(_b.fc2.weight.norm()):.3f}",flush=True)
+    if os.environ.get("NONINT"):
+        import json as _nj
+        _HF="/blue/fire-finai/lq62.yale/annulus/heldout/eval_heldout.jsonl"
+        _pre=[]; _y2000=[]
+        for _ln in open(_HF):
+            _o=_nj.loads(_ln); _t=_o.get("text") or ""
+            try: _yv=int(str(_o.get("year"))[:4])
+            except: _yv=None
+            if _yv is None or len(_t)<=800: continue
+            _bucket=_pre if _yv<2000 else (_y2000 if _yv==2000 else None)
+            if _bucket is None: continue
+            if len(_bucket)>=80: continue
+            _ids=[YEAR_TOK_ID]+tok(_t,add_special_tokens=False)["input_ids"][:MAXTOK-1]
+            if len(_ids)>=16: _bucket.append(_ids)
+        core.eval()
+        _res=[]
+        for _nm,_ds in [("pre-2000",_pre),("year2000",_y2000)]:
+            if not _ds: continue
+            for _b in branches: _b.enabled=False
+            _off=nll(_ds)
+            for _b in branches: _b.enabled=True
+            _on=nll(_ds)
+            _res.append(f"{_nm}(n={len(_ds)}): OFF={_off:.4f} ON={_on:.4f} dNLL={_on-_off:+.4f}")
+            print(f"[nonint] {_res[-1]}",flush=True)
+        try: discord_post("\U0001f9ea <@1329102249854500885> FY2001 BEST插件 非干扰测(修好版,core.eval,ON=强制开插件):\n"+"\n".join(_res)+"\n≈0=非干扰成立(插件对旧文本几乎不动);明显正=还有扰动。注:按cutoff设计旧年插件本应关(bit-exact),这是'强制开'的上界。")
+        except Exception: pass
+        raise SystemExit(0)
+    _dd=docs[:6]
+    for _mode in ("train","eval"):
+        getattr(core,_mode)()
+        for _b in branches: _b.enabled=False
+        _off=nll(_dd)
+        for _b in branches: _b.enabled=True
+        _on=nll(_dd)
+        _line=f"mode={_mode}: OFF={_off:.4f} ON={_on:.4f} dNLL={_on-_off:+.4f}"
+        print(f"[diag] {_line}",flush=True); _msg.append(_line)
+    try:
+        discord_post("\U0001f50e <@1518372899575169215> FY2001 插件 DIAG(train vs eval forward):\n"+"\n".join(_msg)+"\n判读:train-mode 负/eval-mode 正 = train/eval forward 不一致(MoE模式)就是根因;两 mode 都正=另查;c_t.norm≈0 = [Y2001] emb 是垃圾。")
+    except Exception: pass
+    raise SystemExit(0)
+
+train=docs[:DRY_DOCS] if DRY else docs
+opt=torch.optim.AdamW([p for b in branches for p in b.parameters()], lr=LR, weight_decay=float(os.environ.get('WD') or '0.01'))
+import math as _math
+WARMUP=int(os.environ.get("WARMUP") or "500")
+def _lr_at(_st,_tot):
+    if _st<WARMUP: return LR*(_st+1)/max(1,WARMUP)
+    _p=(_st-WARMUP)/max(1,_tot-WARMUP)
+    return LR*0.5*(1+_math.cos(_math.pi*min(1.0,_p)))
+core.eval()   # global align (base_nll/held-off/loop all eval-mode)
+_trs=train[:300]   # fixed train sample for decisive train-set ΔNLL
+base_nll=nll(_trs)
+# ---- held-out early-stop setup (Lingfei same-source heldout_2001) ----
+import json as _json
+_HELD=os.environ.get("HELDOUT_PATH") or "/blue/fire-finai/lq62.yale/annulus/data/rung1_heldout_2001/heldout_2001_samesource.jsonl"
+_hd=None; _held_off=None; _best_delta=1e9; _best_state=None; _best_step=0
+try:
+    _tmp=[]
+    with open(_HELD) as _hf:
+        for _ln in _hf:
+            if len(_tmp)>=int(os.environ.get("HELD_N") or "150"): break
+            _o=_json.loads(_ln); _tv=_o.get("text") or ""
+            if _tv and len(_tv)>1500:
+                _hi=[YEAR_TOK_ID]+tok(_tv,add_special_tokens=False)["input_ids"][:MAXTOK-1]
+                if len(_hi)>=16: _tmp.append(_hi)
+    assert len(_tmp)>=16, f"held-out only {len(_tmp)}"
+    _hd=_tmp
+    for _b in branches: _b.enabled=False
+    _held_off=nll(_hd)
+    for _b in branches: _b.enabled=True
+    print(f"[held] loaded={len(_hd)} OFF_nll={_held_off:.4f}", flush=True)
+    discord_post(f"📐 <@{MANAGER_UID}> FY{DATA_YEAR} 重训 held-out={len(_hd)}篇 OFF-NLL={_held_off:.4f} (早停基线,目标ΔNLL变负)")
+except Exception as _e:
+    _hd=None
+    try: discord_post(f"⚠ held-out 加载失败: {_e} — 早停关闭")
+    except Exception: pass
+core.eval()   # FIX(train/eval forward align): backbone in eval-mode so plugin calibrates to deployment forward (dropout=0 but year_fused_loss/router self.training branches differ)
+STEPS=DRY_STEPS if DRY else REAL_STEPS
+ACCUM=1 if DRY else int(os.environ.get("ACCUM") or "16")   # grad accumulation -> effective batch
+GCLIP=float(os.environ.get("GCLIP") or "1.0")              # grad clip norm
+print(f"[branch] ACCUM={ACCUM} GCLIP={GCLIP}", flush=True)
+for st in range(STEPS):
+    for _g in opt.param_groups: _g["lr"]=_lr_at(st,STEPS)
+    ids=train[st%len(train)]
+    inp=torch.tensor([ids],device="cuda"); pos=torch.arange(len(ids),device="cuda").unsqueeze(0)
+    if st%ACCUM==0: opt.zero_grad()
+    out=core(input_ids=inp,position_ids=pos,attention_mask=None)
+    lg=(out[0] if isinstance(out,tuple) else out).float()
+    lp=torch.log_softmax(lg[0,:-1],dim=-1); tgt=torch.tensor(ids[1:],device="cuda")
+    loss=(-lp.gather(1,tgt[:,None]).squeeze(1)).mean()
+    (loss/ACCUM).backward()                                # accumulate scaled grad
+    if (st+1)%ACCUM==0:
+        torch.nn.utils.clip_grad_norm_([p for b in branches for p in b.parameters()], GCLIP)
+        opt.step()
+    if not DRY and (st+1)%100==0:
+        print(f"[train] step {st+1}/{STEPS} loss={loss.item():.4f} lr="+format(LR,".2e"), flush=True)
+        _gn=sum((p.grad.norm().item() if p.grad is not None else 0.0) for b in branches for p in b.parameters())
+        _ng=sum(1 for b in branches for p in b.parameters() if (p.grad is not None and float(p.grad.abs().max())>0))
+        _tot=sum(1 for b in branches for p in b.parameters())
+        print(f"[grad] step {st+1} branch_grad_norm={_gn:.4e} nonzero_grad_params={_ng}/{_tot}", flush=True)
+        _drift=max((core.state_dict()[n]-snap[n]).abs().max().item() for n in snap if n in core.state_dict())
+        # post-train rollback ③ (periodic): trained branch OFF must be bit-exact identity
+        _bt=branches[0]; _bt.enabled=False
+        _rb=0.0
+        for _dt in (torch.bfloat16, torch.float32):
+            _xx=torch.randn(4,1,d_model,device="cuda",dtype=_dt); _rb=max(_rb,(_bt(_xx)-_xx).abs().max().item())
+        _bt.enabled=True
+        try: discord_post(f"📉 FY{DATA_YEAR} r{R} step {st+1}/{STEPS} loss={loss.item():.4f} | base_equal 漂移={_drift:.1e}{' ✅' if _drift<1e-6 else ' ❌骨干被动!'} | 训后rollback③={_rb:.1e}{' ✅' if _rb==0 else ' ❌'}")
+        except Exception: pass
+        if _drift>=1e-6:
+            discord_post(f"🛑 <@{MANAGER_UID}> FY{DATA_YEAR} STOP: base_equal 骨干漂移={_drift:.2e}≠0 (step {st+1}) — backbone 被动了,fail-loud 停训!")
+            raise SystemExit(f"BASE_EQUAL_VIOLATION drift={_drift}")
+        if _rb!=0:
+            discord_post(f"🛑 <@{MANAGER_UID}> FY{DATA_YEAR} STOP: 训后rollback③={_rb:.2e}≠0 (step {st+1}) — cutoff 关不净,fail-loud 停训!")
+            raise SystemExit(f"ROLLBACK_VIOLATION={_rb}")
+    if not DRY and _hd is not None and (st+1)%int(os.environ.get("HELD_EVERY") or "500")==0:
+        core.eval()
+        for _b in branches: _b.enabled=True
+        _held_on=nll(_hd); _dl=_held_on-_held_off
+        core.eval()   # FIX(train/eval forward align): backbone in eval-mode so plugin calibrates to deployment forward (dropout=0 but year_fused_loss/router self.training branches differ)
+        _new=_dl<_best_delta
+        print(f"[held] step {st+1} ON={_held_on:.4f} OFF={_held_off:.4f} dNLL={_dl:+.4f} best={_best_delta:+.4f}", flush=True)
+        if _new:
+            _best_delta=_dl; _best_step=st+1
+            _best_state={f"branch{_j}":{k:v.detach().cpu().clone() for k,v in b.state_dict().items()} for _j,b in enumerate(branches)}
+        _tag=(" ↓新最优" if _new else " (best={:+.4f}@{})".format(_best_delta,_best_step))
+        try: discord_post("📈 FY{} step {}/{} held-out ΔNLL(ON-OFF)={:+.4f}{}".format(DATA_YEAR,st+1,STEPS,_dl,_tag))
+        except Exception: pass
+    if not DRY and (st+1)%1000==0:
+        _lstr2="-".join(map(str,INSERT_LAYERS)); _ckp=f"{OUT_DIR}/plugin_FY{DATA_YEAR}_r{R}_L{_lstr2}_step{st+1}.pt"
+        core.eval(); torch.save({f"branch{_j}":b.state_dict() for _j,b in enumerate(branches)}, _ckp); core.eval()   # FIX(train/eval forward align): backbone in eval-mode so plugin calibrates to deployment forward (dropout=0 but year_fused_loss/router self.training branches differ)
+        print(f"[ckpt] saved {_ckp}", flush=True)
+core.eval()
+fin_nll=nll(_trs)
+if not DRY and _best_state is not None:
+    _lstrB="-".join(map(str,INSERT_LAYERS))
+    _ckB=f"{OUT_DIR}/plugin_FY{DATA_YEAR}_r{R}_L{_lstrB}_BEST.pt"
+    torch.save(_best_state|{"meta":{"year":DATA_YEAR,"r":R,"layers":INSERT_LAYERS,"params":n_branch_params,"best_step":_best_step,"best_delta_nll":_best_delta,"held_off_nll":_held_off}}, _ckB)
+    print(f"[branch] BEST plugin saved -> {_ckB} (step{_best_step} dNLL={_best_delta:+.4f})", flush=True)
+    try: discord_post("🏆 <@{}> FY{} 早停最优: step{} held-out ΔNLL={:+.4f} -> {}".format(MANAGER_UID,DATA_YEAR,_best_step,_best_delta,_ckB))
+    except Exception: pass
+
+if not DRY:
+    _lstr="-".join(map(str,INSERT_LAYERS))
+    _ck=f"{OUT_DIR}/plugin_FY{DATA_YEAR}_r{R}_L{_lstr}.pt"
+    torch.save({f"branch{_i}":b.state_dict() for _i,b in enumerate(branches)}|{"meta":{"year":DATA_YEAR,"r":R,"layers":INSERT_LAYERS,"params":n_branch_params}}, _ck)
+    print(f"[branch] plugin saved -> {_ck}", flush=True)
+    try: discord_post(f"💾 FY{DATA_YEAR} 插件已存 {_ck} (r={R}, {n_branch_params/1e6:.2f}M, L{INSERT_LAYERS})")
+    except Exception: pass
+# base_equal: backbone params byte-unchanged
+max_drift=max((core.state_dict()[n]-snap[n]).abs().max().item() for n in snap if n in core.state_dict())
+
+# ── ③ exact-rollback = MODULE-LEVEL unit test (noise-free, no MoE) ──
+# cutoff invariant: enabled=False => branch(x) MUST equal x bit-for-bit.
+_b=branches[0]; _b.enabled=False
+_max=0.0; _ok=True
+for _dt in (torch.bfloat16, torch.float32):
+    _x=torch.randn(7,1,1024, device="cuda", dtype=_dt)
+    _y=_b(_x); _d=(_y-_x).abs().max().item(); _max=max(_max,_d)
+    if not torch.equal(_y,_x): _ok=False
+# and: enabled=True but UNTRAINED-equivalent check — trained branch ON adds nonzero (sanity)
+_b.enabled=True; _xon=torch.randn(7,1,1024,device="cuda",dtype=torch.bfloat16)
+_on_delta=(_b(_xon)-_xon).abs().max().item()
+_b.enabled=False
+rollback_max=_max; rollback_exact=_ok
+# whole-model determinism floor (context only; MoE routing argmax flip -> big, expected)
+def _logits(ids):
+    inp=torch.tensor([ids],device="cuda"); pos=torch.arange(len(ids),device="cuda").unsqueeze(0)
+    with torch.no_grad(): out=core(input_ids=inp,position_ids=pos,attention_mask=None)
+    return (out[0] if isinstance(out,tuple) else out).float()
+_pid=train[0]; _det=(_logits(_pid)-_logits(_pid)).abs().max().item()
+
+sig_roll = "✅精确复原(逐位==x)" if rollback_exact else f"❌分支没关净(max={rollback_max:.2e})"
+summary=(f"🧩 <@{MANAGER_UID}> FY2001 加法插件真训 (r={R}, L{INSERT_LAYERS})\n"
+    f"branch params={n_branch_params/1e6:.2f}M (只训这些, 骨干冻)\n"
+    f"① train-NLL {base_nll:.4f} → {fin_nll:.4f} ({'✅插件在学' if fin_nll<base_nll*0.98 else '⚠️没降'})\n"
+    f"② base_equal 骨干漂移={max_drift:.2e} ({'✅byte-frozen' if max_drift<1e-6 else '❌骨干被动'})\n"
+    f"③ cutoff 模块级单元测试: enabled=False→branch(x)逐位==x, max_diff={rollback_max:.2e} → {sig_roll}\n"
+    f"  (旁证:branch ON 对随机x的改动={_on_delta:.3f}≠0=分支有作用;整模型MoE非确定性底={_det:.2f}=路由argmax翻转,与③无关)")
+discord_post(summary); print(summary,flush=True); print("BRANCH_DRY_DONE",flush=True)