"""Job 1905 (runs/CSEL/TOPIC_CLEAR_PREREG.md): DEFECT FIX - the topic boost is never cleared in the SYLLABLE commit unit. Defect (audit B, 2026-10-01): twolayer_canvas.Decoder adds the topic boost (+1.5 on every token of a topic word, per line) and documents that a line's boost for a word is cleared once that word is used in the line: put_word() does `self.topic_rows[i][X.tok_word == w] = 0.0`. The word / wak / bot units commit through put_word; the SYLLABLE unit commits cell by cell inside step_cell_or_word() and never calls put_word, so the boost stays on for the whole decode. Production (186 syllable arm, 187, 188, 189, 190's decode, 191's control and arms) decodes in the syllable unit. Result: 31/81 confirmation-189 picks use a topic word >= 4 times (humans 3/81), holding 22 of 55 judge-flagged lines (ตามใจ x13, ขวัญ x10). The fix (one change, flag --topic-clear, default OFF): in the syllable unit, when a step COMPLETES a word occurrence in line i (every cell of the word committed - the moment the base decoder writes wscore[(i, start)]), clear line i's topic boost for that word: exactly put_word's rule, applied at the same point of the word's life (its commit as a whole word). Not cleared while the word is only partly committed (first syllable placed, the rest forced); other lines keep their boost; non-topic words are a no-op (their rows are 0). The repetition penalty (flat, twolayer_canvas.lp, isin over used words) is NOT changed here - a separate, later fix. twolayer_canvas.py / twolayer_full.py / twolayer_attend.py are imported, never edited (jobs 190/191 sum them). twolayer_canvas.Decoder is replaced by TopicClearDecoder (a subclass of twolayer_full's StutterDecoder) only inside this process and only with --stutter-fix; with --topic-clear off it is a pass-through and the output is byte-identical to twolayer_full.py decode (tested; the job re-checks all 49 topics against runs/TLLONG/F/c0.jsonl on its device). Pipeline (gates / pools / score) = twolayer_attend.py's (job 191), imported unchanged; its module-level arm table is pointed at this job's one arm (T -> pool label "tpc") and pool prefix TPC inside this process only (twolayer_attend reads LABELS / PREFIX at call time): decode ... (= twolayer_full.py decode) [--topic-clear] gates OUT ATT BIGRAMS A-B CTL=DIR T=DIR 191's gates (form >= 99%, unattested stutters, final bits, echo, distinct-2) + topic-word readout -> OUT/GATES.json, OUT/TOPIC.json; prints PASSING ... pools OUT A-B HUMAN C2HUMAN CTL=DIR T=DIR 191's human-balanced pools (ctl, tpc + hum, humb; 1:1) score OUT A-B CTL=DIR T=DIR after judging runs/TPC_{CLEAN,ALIGN}{s}{q}/run1.json: decide2 on clean test CPU tests (toy invariants on a real-lexicon fake model; optional real model) """ import os, sys, json, collections, argparse sys.path[:0] = [".", "scripts"] PREFIX = "TPC" LABELS = {"CTL": "ctl", "T": "tpc"} def rows(p): return [json.loads(l) for l in open(p, encoding="utf-8") if l.strip()] def wrows(p, R): with open(p + ".tmp", "w", encoding="utf-8") as f: for r in R: f.write(json.dumps(r, ensure_ascii=False) + "\n") os.replace(p + ".tmp", p) # ---------------------------------------------------------------- decoder def make_decoder_class(): import twolayer_full as TF class TopicClearDecoder(TF.make_decoder_class()): """StutterDecoder + put_word's topic-boost clearing in the syllable unit (TOPIC_CLEAR, set on the class). n_topic_cleared = word completions whose line still had a non-zero boost for that word (= topic word uses).""" TOPIC_CLEAR = False n_topic_cleared = 0 def step_cell_or_word(self, t): if not (self.TOPIC_CLEAR and self.unit == "syllable"): return super().step_cell_or_word(t) before = dict(self.wscore) r = super().step_cell_or_word(t) for (i, s), (w, _) in self.wscore.items(): if (i, s) in before: continue # syllable unit: cells never re-open, so a key is set once m = self.X.tok_word == w if bool((self.topic_rows[i][m] != 0).any()): self.n_topic_cleared += 1 self.topic_rows[i][m] = 0.0 # = put_word: twolayer_canvas.py line 628 return r return TopicClearDecoder def decode_cli(argv): """= twolayer_full.py decode (same records, same RNG streams) + --topic-clear; extra keys only with the flag""" import torch, random as _r import twolayer_canvas as TC, twolayer_full as TF from klon8 import prosody as P ap = argparse.ArgumentParser() ap.add_argument("--model", required=True); ap.add_argument("--prompts", required=True); ap.add_argument("--out", required=True) ap.add_argument("--seed", type=int, default=0); ap.add_argument("--temp", type=float, default=0.8); ap.add_argument("--limit", type=int, default=0) ap.add_argument("--unit", choices=["syllable", "word", "wak", "bot"], default="word") ap.add_argument("--refine", type=int, default=3); ap.add_argument("--topic-boost", type=float, default=1.5) ap.add_argument("--stutter-fix", action="store_true", help="ban same-sound neighbours unless attested (default off)") ap.add_argument("--allow", default="", help="allowlist from twolayer_full.py allow (required with --stutter-fix)") ap.add_argument("--topic-clear", action="store_true", help="syllable unit: clear a line's topic boost for a word once " "that word is completed in the line (= put_word; default off)") a = ap.parse_args(argv) model, X = TC.load(a.model) if a.stutter_fix: assert a.allow, "--stutter-fix needs --allow" SD = make_decoder_class(); SD.ALLOW = TF.load_allow(a.allow, X); SD.TOPIC_CLEAR = a.topic_clear TC.Decoder = SD # decode_one builds TC.Decoder (this process only) else: assert not a.topic_clear, "--topic-clear extends --stutter-fix (the production decode)" if a.topic_clear: assert a.unit == "syllable", "--topic-clear: the defect is in the syllable unit only" rows_ = [json.loads(l) for l in open(a.prompts, encoding="utf-8")] if a.limit: rows_ = rows_[:a.limit] rng_len = _r.Random(a.seed) with open(a.out + ".tmp", "w", encoding="utf-8") as f: for n, r in enumerate(rows_): text = r.get("text") or r.get("prompt") or r.get("title") or r.get("topic") assert text, f"prompt row without text: {r}" want = TC.pick_lens(rng_len) lines, D, tries = TC.decode_one(model, X, text, unit=a.unit, temp=a.temp, lens=want, refine=a.refine, topic_boost=a.topic_boost, gen=torch.Generator().manual_seed(a.seed * 100003 + n)) poem = "\n".join(lines); sc = P.score(poem, n_bot_expected=1) got = [len(P.g2p_line(x)) for x in lines] rec = {"id": r.get("id"), "prompt": text, "sample": 0, "poem": poem, "unit": a.unit, "syl_target": want, "syl": got, "retries": tries, "forward_passes": D.n_fwd, "all_pass": sc["all_pass"], "structure": sc["structure"], "ext_rhyme": sc["ext_rhyme"], "tone": sc["tone"], "internal": sc["internal"]} if a.stutter_fix: rec["stutter_fix"] = True; rec["n_banned"] = D.n_banned; rec["stutters_left"] = len(TF.stutter_pairs(D)) if a.topic_clear: rec["topic_clear"] = True; rec["n_topic_cleared"] = D.n_topic_cleared f.write(json.dumps(rec, ensure_ascii=False) + "\n"); f.flush() print(f"{n}/{len(rows_)} all_pass={sc['all_pass']} syl={got} want={want} retries={tries}" + (f" topic cleared {D.n_topic_cleared}" if a.topic_clear else ""), file=sys.stderr, flush=True) os.replace(a.out + ".tmp", a.out) P.save_cache() # ---------------------------------------------------------------- pipeline (191's, pointed at this job's arm) def TA(): import twolayer_attend as A A.LABELS = dict(LABELS); A.PREFIX = PREFIX # this process only; the file is not edited return A def topic_readout(prompt, poem, toks): """topic content tokens (newmm, >= 2 chars) in the poem: (total, max count of any one of them, lines in which one topic token occurs >= 2 times = WITHIN-LINE repeats, what the fix removes; once per line stays allowed)""" tt = {w for w in toks(prompt) if len(w) >= 2} per = [collections.Counter(w for w in toks(ln) if w in tt) for ln in poem.split("\n")] c = sum(per, collections.Counter()) return sum(c.values()), max(c.values(), default=0), sum(max(p.values(), default=0) >= 2 for p in per) def topic_stats(R, toks): x = [topic_readout(r["prompt"], r["poem"], toks) for r in R] return {"n": len(R), "topic_tokens_per_poem": sum(a for a, _, _ in x) / len(R), "within_line_repeat_lines_per_poem": sum(c for _, _, c in x) / len(R), "max_one_topic_word_per_poem": sum(b for _, b, _ in x) / len(R), "share_one_topic_word_ge4": 100 * sum(b >= 4 for _, b, _ in x) / len(R), "share_one_topic_word_ge3": 100 * sum(b >= 3 for _, b, _ in x) / len(R)} def gates(out, att_json, bigrams, spec, *arms): """191's gates unchanged (twolayer_attend.gates; gate_fail: form >= 99%, unattested stutters <= ctl + 0.10, final bits >= ctl - 0.3, word 3x in a line <= ctl + 0.05, distinct-2 >= ctl - 0.03) + the topic-word readout and the identical T/F pair count (reported)""" A = TA(); A.gates(out, att_json, bigrams, spec, *arms) res = {} for n, d in A.arm_specs(arms): R = [r for s in A.seeds(spec) for r in rows(f"{d}/c{s}.jsonl")] res[n] = dict(topic_stats(R, A.toks), n_topic_cleared=sum(r.get("n_topic_cleared", 0) for r in R) / len(R)) print(f" {n:4s} TOPIC: topic-word tokens/poem {res[n]['topic_tokens_per_poem']:.2f}; within-line topic repeats " f"{res[n]['within_line_repeat_lines_per_poem']:.3f} lines/poem; max count of one topic word {res[n]['max_one_topic_word_per_poem']:.2f}/poem; " f"poems with one topic word >= 4x {res[n]['share_one_topic_word_ge4']:.1f}% (>= 3x {res[n]['share_one_topic_word_ge3']:.1f}%); " f"boost clears/poem {res[n]['n_topic_cleared']:.2f}") res["identical_pairs"] = len(identical(spec, *arms)); res["pairs"] = sum(len(rows(f"{dict(A.arm_specs(arms))['CTL']}/c{s}.jsonl")) for s in A.seeds(spec)) print(f" identical T/F poems: {res['identical_pairs']} of {res['pairs']} (topic, seed) pairs (label shared at scoring)") json.dump(res, open(f"{out}/TOPIC.json", "w"), indent=1) def pools(out, spec, human, c2human, *arms): return TA().pools(out, spec, human, c2human, *arms) def identical(spec, *arms): """{(topic, seed)} where T's poem is byte-identical to the control's""" A = TA(); d = dict(A.arm_specs(arms)); out = set() for s in A.seeds(spec): c = {r["id"]: r["poem"] for r in rows(f"{d['CTL']}/c{s}.jsonl")} out |= {(r["id"], s) for r in rows(f"{d['T']}/c{s}.jsonl") if c.get(r["id"]) == r["poem"]} return out def decide_fix(gates_ok, ci): """pre-registered DEFECT-FIX rule (non-inferiority), on clean T - F: REJECT iff the upper 95% bound < 0 (T significantly worse; this wins if the CI lies inside (-5, 0)); else ADOPT iff the gates pass and the lower bound > -5pp; else REPORT""" if not gates_ok: return "REJECT (gates)" if ci[1] < 0: return "REJECT" if ci[0] > -5: return "ADOPT" return "REPORT TO USER" def score(out, spec, *arms, root_pools="runs"): """191's scorer with two pre-registered changes: (1) a (topic, seed) whose T poem is byte-identical to F's carries ONE label - the control's judgement - for both arms (the tpc copy is judged only to keep the pools 1:1; its label is discarded and reported as a judge-repeatability check), so only changed pairs can be discordant; (2) the decision is decide_fix (non-inferiority), not decide2 (decide2 is reported).""" A = TA(); assert [n for n, _ in A.arm_specs(arms)] == ["CTL", "T"], "arms: CTL=DIR T=DIR" same = identical(spec, *arms); orig = A.labels; rep = {} def shared(spec_, root_pools_="runs"): lab, anchors = orig(spec_, root_pools_) for kind in ("clean", "aligned"): rep[kind] = sum(lab[kind][("tpc", i, s)] == lab[kind][("ctl", i, s)] for i, s in same) for i, s in same: lab[kind][("tpc", i, s)] = lab[kind][("ctl", i, s)] return lab, anchors A.labels = shared; void = None try: A.score(out, spec, *arms, root_pools=root_pools) except SystemExit as e: void = e.code finally: A.labels = orig R = json.load(open(f"{out}/RESULT.json")) g = json.load(open(f"{out}/GATES.json")) if os.path.exists(f"{out}/GATES.json") else None c = R["contrasts"]["tpc vs ctl"] gates_ok = g is not None and not g["dead"].get("T") and c["form"]["b"] >= 99 R["decide2_reported"] = R["decision"] dec = decide_fix(gates_ok, c["clean"]["ci"]) R["decision"] = {"tpc": dec if not void else f"provisional ({dec}) - void pools, re-judge"} R["decision_rule"] = decide_fix.__doc__.strip() n = c["clean"]["n"] R["identical"] = {"pairs": n, "identical": len(same), "changed": n - len(same), "judge_repeat_agree_on_identical": rep, "discordant_clean": c["clean"]["up"] + c["clean"]["down"], "discordant_all_three": c["all-three"]["up"] + c["all-three"]["down"]} json.dump(R, open(f"{out}/RESULT.json", "w"), indent=1, default=str) print(f" identical T/F poems {len(same)} of {n} pairs (one label each: the control's); judge repeatability on them: " f"clean {rep.get('clean')}/{len(same)}, aligned {rep.get('aligned')}/{len(same)}; discordant pairs clean " f"{R['identical']['discordant_clean']} ({c['clean']['up']} up / {c['clean']['down']} down), all-three {R['identical']['discordant_all_three']}") print(f" DECISION (pre-registered defect-fix rule, clean tpc - ctl CI [{c['clean']['ci'][0]:+.1f}, {c['clean']['ci'][1]:+.1f}], " f"gates {'pass' if gates_ok else 'FAIL'}): {R['decision']['tpc']} (decide2, reported: {R['decide2_reported']})") if void: sys.exit(void) # ---------------------------------------------------------------- tests def _test(): import tempfile, torch, contextlib, io import twolayer_canvas as TC, twolayer_full as TF import twolayer_attend as TAm t = tempfile.mkdtemp(prefix="topicfix_", dir=os.environ.get("TC_TMP") or None) lexf = "runs/TWOLAYER/data/lexicon.json" assert os.path.exists(lexf), "real lexicon missing" XR = TC.Lex(lexf); M = TAm._FakeModel(XR.NT, TC.T) SD0 = TF.make_decoder_class(); SD0.ALLOW = ({}, {}) TD0 = make_decoder_class(); TD0.ALLOW = ({}, {}) TD1 = make_decoder_class(); TD1.ALLOW = ({}, {}); TD1.TOPIC_CLEAR = True topics = ["คิดถึงบ้าน", "ตามใจ", "ฝนตก", "รักแรก", "บ้าน"] # บ้าน: a SINGLE-syllable topic word lens = [8, 9, 8, 8] ORIG = TC.Decoder # restored after every section def boosted(D): """word ids carrying a non-zero boost at decode start (the Decoder's own topic row)""" return {int(w) for w in XR.tok_word[D.topic0 != 0].unique().tolist()} def completed(D, i): """word ids with a COMPLETE occurrence in line i (every cell of the word committed)""" done = set() for j in range(D.lens[i]): k = D.idx(i, j) if k not in D.committed: continue tk_ = int(D.tok[k]); kk, nn = int(XR.tok_k[tk_]), int(XR.tok_n[tk_]); s = j - kk if all(D.idx(i, s + q) in D.committed for q in range(nn)): done.add(int(XR.tok_word[tk_])) return done def check_invariant(D, where): """topic_rows[i] == initial boost with zeros exactly on the words COMPLETED in line i (all cells committed)""" for i in range(TC.L): done = completed(D, i) want = D.topic0.clone() if done: want[torch.isin(XR.tok_word, torch.tensor(sorted(done), device=XR.tok_word.device))] = 0.0 assert torch.equal(D.topic_rows[i], want), f"{where}: line {i} topic row != put_word rule (done {done & boosted(D)})" def recording(cls, check=True): class R(cls): def __init__(s, *a, **k): super().__init__(*a, **k); s.topic0 = s.topic_rows[0].clone(); s.events = set() def step_cell_or_word(s, t): r = super().step_cell_or_word(t) if check: check_invariant(s, f"{s.unit} step {t}") # partial topic word: first syllable placed, others still forced -> boost must still be on in that line # (unless the same word is already complete elsewhere in the line) for k, ft in s.forced.items(): w = int(XR.tok_word[ft]); i = (k - TC.TL) // TC.C if w in boosted(s) and w not in completed(s, i): s.events.add(("partial", i, k, w, bool((s.topic_rows[i][XR.tok_word == w] != 0).any()))) for (i, st), (w, _) in s.wscore.items(): # a SET: each (state, line, cell, word) counted once if w in boosted(s): s.events.add(("done", i, st, w, bool((s.topic_rows[i][XR.tok_word == w] != 0).any()))) return r R.__name__ = cls.__name__; return R # 1. flag off: TopicClearDecoder == StutterDecoder byte for byte (lines, forward passes, retries, bans) base = {} for n, top in enumerate(topics): outs = [] for cls in (SD0, TD0): TC.Decoder = cls lines, Dd, tries = TC.decode_one(M, XR, top, unit="syllable", lens=lens, gen=torch.Generator().manual_seed(n)) outs.append((lines, Dd.n_fwd, tries, Dd.n_banned)) assert outs[0] == outs[1], f"flag off: TopicClearDecoder differs from StutterDecoder ({top})" base[top] = outs[0] TC.Decoder = ORIG print(f" flag off == StutterDecoder on {len(topics)} topics (fake model, real lexicon)") # 2. flag on, a strong boost so topic words are drawn: after EVERY step, each line's topic row is exactly put_word's # rule (zero iff that word is complete in that line); a partly committed multi-syllable topic word keeps its boost; # other lines are unaffected. The same invariant holds for the WORD unit's own put_word (the rule we copy). n_partial = n_done = n_multi_done = n_single_done = 0; changed = 0 for n, top in enumerate(topics): TC.Decoder = recording(TD1) lines, D1, _ = TC.decode_one(M, XR, top, unit="syllable", lens=lens, topic_boost=30.0, gen=torch.Generator().manual_seed(n)) assert boosted(D1), f"{top}: no boosted word" assert all(not on for kind, i, _, w, on in D1.events if kind == "done"), "boost still on after a word completed" assert all(on for kind, i, _, w, on in D1.events if kind == "partial"), "boost cleared before the word completed" n_partial += sum(k == "partial" for k, *_ in D1.events); n_done += sum(k == "done" for k, *_ in D1.events) n_multi_done += sum(k == "done" and len(XR.sounds[XR.words[w]]) > 1 for k, i, _, w, _ in D1.events) n_single_done += sum(k == "done" and len(XR.sounds[XR.words[w]]) == 1 for k, i, _, w, _ in D1.events) assert D1.n_topic_cleared >= 1, top # other lines unaffected: a line without that completed word keeps the full boost for it for i in range(TC.L): for w in boosted(D1): m = XR.tok_word == w if bool((D1.topic_rows[i][m] == 0).all()) and bool((D1.topic0[m] != 0).any()): assert any(int(XR.tok_word[int(D1.tok[D1.idx(i, j)])]) == w for j in range(D1.lens[i])), f"line {i} cleared without {XR.words[w]}" TC.Decoder = recording(TD0, check=False) # flag off, same boost: the boost is never cleared lines0, D0, _ = TC.decode_one(M, XR, top, unit="syllable", lens=lens, topic_boost=30.0, gen=torch.Generator().manual_seed(n)) assert all(torch.equal(D0.topic_rows[i], D0.topic0) for i in range(TC.L)), "flag off must keep the defect (no clearing)" changed += lines0 != lines Dw = recording(SD0) # word unit (put_word): the rule being copied TC.Decoder = Dw TC.decode_one(M, XR, top, unit="word", lens=lens, topic_boost=30.0, gen=torch.Generator().manual_seed(n)) TC.Decoder = ORIG assert n_partial > 0 and n_multi_done > 0 and n_single_done > 0, (n_partial, n_multi_done, n_single_done) assert changed >= 1, "the fix never changed a strong-boost decode" print(f" flag on: topic rows == put_word rule after every step ({n_done} completed topic-word occurrences: {n_multi_done} " f"multi-syllable, {n_single_done} single-syllable; {n_partial} partial-word states kept their boost); word unit obeys " f"the same invariant; {changed}/{len(topics)} decodes changed") # 2b. the word unit is untouched by the flag (put_word already clears there) for n, top in enumerate(topics[:2]): o = [] for cls in (SD0, TD1): TC.Decoder = cls lines, Dd, tries = TC.decode_one(M, XR, top, unit="word", lens=lens, gen=torch.Generator().manual_seed(n)) o.append((lines, Dd.n_fwd, tries)) assert o[0] == o[1], "flag must not touch the word unit" TC.Decoder = ORIG # 3. topic readout toks = TAm.toks tot, mx, wl = topic_readout("ตามใจ", "ตามใจตามใจ\nตามใจเธอ\nฝนตก\nตามใจฉัน", toks) assert (tot, mx, wl) == (4, 4, 1), (tot, mx, wl) st = topic_stats([{"prompt": "ตามใจ", "poem": "ตามใจตามใจ\nตามใจเธอ\nฝนตก\nตามใจฉัน"}, {"prompt": "ฝนตก", "poem": "ฝน\nตก\nก\nข"}], toks) assert st["share_one_topic_word_ge4"] == 50.0 and st["n"] == 2 and st["within_line_repeat_lines_per_poem"] == 0.5 \ and st["max_one_topic_word_per_poem"] == 2.0, st # 4. 191's pipeline pointed at T: arm table and pool prefix patched in this process only, file untouched A = TA(); assert A.arm_specs(["CTL=a", "T=b"]) == [("CTL", "a"), ("T", "b")] and A.PREFIX == "TPC" try: A.arm_specs(["CTL=a", "E=b"]); raise RuntimeError("E accepted") except AssertionError: pass NTp = 9 IDENT = (5, 6) # T's poem byte-identical to the control's on these topics for dd, badf in (("ctl", 3), ("T", 1)): os.makedirs(f"{t}/{dd}") for s in (0, 1): wrows(f"{t}/{dd}/c{s}.jsonl", [{"id": f"T{i:02d}", "prompt": f"p{i}", "unit": "syllable", "all_pass": i != badf, "poem": (f"กX{i}{s}" if i in IDENT else f"ก{dd}{i}{s}") + "\nข\nค\nง"} for i in range(NTp)]) wrows(f"{t}/hum.jsonl", [{"id": f"T{i:02d}", "prompt": f"p{i}", "poem": f"ม{i}\nข\nค\nง", "tier": 3 if i in (2, 5) else 1} for i in range(NTp)]) wrows(f"{t}/c2.jsonl", [{"id": f"G{i:03d}", "prompt": f"g{i}", "poem": f"ช{i}\nข\nค\nง"} for i in range(40)]) A6 = [f"CTL={t}/ctl", f"T={t}/T"] with contextlib.redirect_stdout(io.StringIO()): labs = pools(f"{t}/p", "0-1", f"{t}/hum.jsonl", f"{t}/c2.jsonl", *A6) assert labs == ["ctl", "tpc", "hum", "humb"], labs import build_pool, random as _r rng = _r.Random(0) for s in (0, 1): for q in range(A.NPART): for inst, al in (("CLEAN", False), ("ALIGN", True)): dd = f"{t}/runs/{PREFIX}_{inst}{s}{q}" with contextlib.redirect_stdout(io.StringIO()): km = build_pool.build(dd, [(l, f"{t}/p/{l}_s{s}q{q}.jsonl") for l in labs], nlines=4, align=al, no_drop=True) # ctl T00 judged bad; T's copy of the identical T05 judged bad too (judge noise: must be overridden by # the control's label, so it creates no discordant pair) bad = lambda v: inst == "CLEAN" and ((v["src"] == "ctl" and v["id"].startswith("T00")) or (v["src"] == "tpc" and v["id"].startswith("T05"))) res = [{"item": k, "errors": [{"severity": "major"}] if bad(v) else [], "align": rng.choice([4, 5])} for k, v in km.items()] json.dump(res, open(f"{dd}/run1.json", "w")) assert identical("0-1", *A6) == {(f"T{i:02d}", s) for i in IDENT for s in (0, 1)} os.makedirs(f"{t}/o"); json.dump({"dead": {"T": []}}, open(f"{t}/o/GATES.json", "w")) with contextlib.redirect_stdout(io.StringIO()): score(f"{t}/o", "0-1", *A6, root_pools=f"{t}/runs") assert TAm.labels.__name__ == "labels", "twolayer_attend.labels not restored" Rr = json.load(open(f"{t}/o/RESULT.json")); c = Rr["contrasts"]["tpc vs ctl"]["clean"] # ctl: T00 judged bad, T03 form-fail -> 7/9 per seed; T: T01 form-fail -> 8/9; T05's bad tpc label is replaced by ctl's assert c["n"] == 18 and abs(c["a"] - 700 / 9) < 1e-9 and abs(c["b"] - 800 / 9) < 1e-9 and c["up"] == 4 and c["down"] == 2, c I_ = Rr["identical"] assert I_["identical"] == 4 and I_["changed"] == 14 and I_["discordant_clean"] == 6 and I_["judge_repeat_agree_on_identical"]["clean"] == 2, I_ assert set(Rr["decision"]) == {"tpc"} and Rr["decision"]["tpc"] == "REJECT (gates)", Rr["decision"] # T form 8/9 < 99% assert decide_fix(True, (-4.9, 3)) == "ADOPT" and decide_fix(True, (-5, 3)) == "REPORT TO USER" assert decide_fix(True, (-12, -0.1)) == "REJECT" and decide_fix(True, (-4, -0.5)) == "REJECT" # significantly worse wins assert decide_fix(False, (1, 9)) == "REJECT (gates)" and decide_fix(True, (-8, 0)) == "REPORT TO USER" assert TAm.decide2({"diff": 6, "ci": (1, 11)}) == "CONTINUE" and TAm.decide2({"diff": 0, "ci": (-5, 4.9)}) == "STOP" print(" 191's gates/pools/score run on arm T (labels ctl/tpc + hum/humb, prefix TPC)") # 5. the REAL model, flag off through TopicClearDecoder == the control's c0 (env TC_TEST_MODEL / _REF / _ALLOW) if os.environ.get("TC_TEST_MODEL"): ref = os.environ["TC_TEST_REF"]; dev = os.environ.get("TC_TEST_PROMPTS", "data/prompts/dev49_pilot.jsonl") nlim = os.environ.get("TC_TEST_N", "3") with contextlib.redirect_stderr(io.StringIO()): decode_cli(["--model", os.environ["TC_TEST_MODEL"], "--prompts", dev, "--unit", "syllable", "--seed", "0", "--limit", nlim, "--stutter-fix", "--allow", os.environ["TC_TEST_ALLOW"], "--out", f"{t}/reg.jsonl"]) assert TC.Decoder.__name__ == "TopicClearDecoder", "the regression must run through TopicClearDecoder" assert open(f"{t}/reg.jsonl", "rb").read() == b"".join(open(ref, "rb").readlines()[:int(nlim)]), "flag-off TopicClearDecoder != control c0" TC.Decoder = ORIG print(f" real model: flag-off TopicClearDecoder decode == the control's c0 on {nlim} topics") assert TC.Decoder is ORIG print(f"twolayer_topicfix: tests pass (temp {t})") if __name__ == "__main__": cmd, args = sys.argv[1], sys.argv[2:] if cmd == "decode": decode_cli(args) elif cmd == "gates": gates(*args) elif cmd == "pools": print("LABELS " + " ".join(pools(*args))) elif cmd == "score": score(*args) elif cmd == "test": _test() else: sys.exit(f"unknown command {cmd}")