#!/usr/bin/env python3
"""
Strict 8-Gate Validator for Multi-Harness Agentic Python-Kernel + In-Voice Persona Dataset (v5)
Enforces:
Gate 1: Verbatim `code == cells.source` and `tool.content == transcript tool_output` + 0 fake user prompts.
Gate 2: Zero honest-isms & 'own my mistake' / sycophancy (`HONEST_SLOP_RE`).
Gate 3: Zero structural markdown slop (bullets, numbered lists, # headers, **bold**: labels, non-fence : intros).
Gate 4: Zero lexical cliches (`LEXICAL_SLOP_RE`), zero Gen-Z slang (`GENZ_SLANG_RE`), zero rubric `<think>`.
Gate 5: Strict lowercase (`\bI\b` banned outside backticks), zero Unicode emoji, zero kaomoji inside `<think>`,
and `loss_mask=0` context turns strictly `reasoning_content == ""`.
Gate 6: Global 3-gram `<think>` opener diversity (<= 6 across entire dataset) + 100% unique `<think>` strings (`max == 1`).
Gate 7: Backtick grounding gate on `loss_mask=1` replies + AST undefined-name check + zero anti-patterns on `loss_mask=1` python cells.
Gate 8: Exact Strata server encoding path (`literal_tags` + `mark_think_literals` + `ChatTemplate` + `unmark_think_literals` + `tok.encode(plain=...)`),
verifying exact token-prefix & `<|im_end|>` span boundaries (`len(upto_toks) - 1`) + 100% unique session-split FTPO/DPO pairs.
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from collections import Counter
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "scripts"))
from extract_agent_persona_candidates import ( # noqa: E402
EMOJI_RE,
GENZ_SLANG_RE,
HONEST_SLOP_RE,
KAOMOJI_IN_THINK_RE,
LEXICAL_SLOP_RE,
RUBRIC_THINK_RE,
SECRET_RE,
check_backtick_grounding,
check_lowercase_clean,
get_defined_and_undef,
has_structural_markdown_slop,
is_clean_idiomatic_code,
)
DATA_DIR = ROOT / "data" / "agent_persona_v1"
SHARD_DIR = DATA_DIR / "shards"
def check_think_item(th: str, window_msgs: list[dict], msg_idx: int) -> list[str]:
errs = []
th = (th or "").strip()
if len(th) < 18 or len(th) > 420:
errs.append(f"<think> length {len(th)} out of range [18, 420]")
for name, rx in (
("honest-ism/sycophancy", HONEST_SLOP_RE),
("lexical cliche", LEXICAL_SLOP_RE),
("gen-z slang", GENZ_SLANG_RE),
("rubric/meta narration", RUBRIC_THINK_RE),
("unicode emoji", EMOJI_RE),
("kaomoji in <think>", KAOMOJI_IN_THINK_RE),
("secret pattern", SECRET_RE),
):
th_check = re.sub(r"`[^`\n]+`", "", th) if name == "kaomoji in <think>" else th
m = rx.search(th_check)
if m:
errs.append(f"<think> contains banned {name} ({m.group(0)!r})")
if re.search(r"\blet['’]?s\b", th, re.IGNORECASE):
errs.append("<think> contains banned 'let's'")
if has_structural_markdown_slop(th):
errs.append("<think> has structural markdown slop")
lc_reason = check_lowercase_clean(th)
if lc_reason is not None:
errs.append(f"<think> {lc_reason}")
turn_m = window_msgs[msg_idx]
ctx_parts = [json.dumps(turn_m.get("tool_calls") or ""), turn_m.get("content") or ""]
for pm in window_msgs[max(1, msg_idx - 2) : msg_idx]:
ctx_parts.append((pm.get("content") or "")[:1200])
if pm.get("tool_calls"):
ctx_parts.append(json.dumps(pm["tool_calls"]))
ctx_blob = " ".join(ctx_parts).lower()
ctx_tokens = {
tok
for tok in re.findall(r"[a-zA-Z0-9_./:-]{3,}", ctx_blob)
if tok not in {"the", "and", "for", "with", "that", "this", "from", "into", "function", "type", "python", "code", "arguments", "call", "true", "false", "none", "null"}
}
th_tokens = {tok.lower() for tok in re.findall(r"[a-zA-Z0-9_./:-]{3,}", th)}
overlap = th_tokens & ctx_tokens
if len(overlap) < 2 and not re.search(r"`[^`\n]+`", th):
errs.append(f"<think> not grounded in turn tokens (overlap={overlap})")
return errs
def check_reply_item(rep: str, window_msgs: list[dict], msg_idx: int) -> list[str]:
errs = []
rep = (rep or "").strip()
if len(rep) < 15:
errs.append(f"reply too short ({len(rep)} chars)")
for name, rx in (
("honest-ism/sycophancy", HONEST_SLOP_RE),
("lexical cliche", LEXICAL_SLOP_RE),
("gen-z slang", GENZ_SLANG_RE),
("unicode emoji", EMOJI_RE),
("secret pattern", SECRET_RE),
):
m = rx.search(rep)
if m:
errs.append(f"reply contains banned {name} ({m.group(0)!r})")
if has_structural_markdown_slop(rep):
errs.append("reply has structural markdown slop")
lc_reason = check_lowercase_clean(rep)
if lc_reason is not None:
errs.append(f"reply {lc_reason}")
ctx_blob = "\n".join(
(m.get("content") or "") + "\n" + json.dumps(m.get("tool_calls") or "")
for m in window_msgs[1:msg_idx]
)
if not check_backtick_grounding(rep, ctx_blob):
errs.append("reply contains ungrounded backticked identifier not in turn context")
return errs
def merge_and_validate_final() -> int:
systems = json.loads((DATA_DIR / "systems.json").read_text(encoding="utf-8"))
sel_windows = [
json.loads(l)
for l in (DATA_DIR / "selected_candidates.jsonl").read_text(encoding="utf-8").splitlines()
if l.strip()
]
sys.path.insert(0, str(ROOT / "vendor" / "strata"))
sys.path.insert(0, str(ROOT / "vendor" / "strata" / "tools"))
from serve.frontend import ( # noqa: E402
ChatTemplate,
literal_tags,
mark_think_literals,
openai_to_messages,
unmark_think_literals,
)
import strata_tokenizer as ST # noqa: E402
tok_dir = ROOT / "models" / "K512_iq2_xs" / "tokenizer"
vocab = json.loads((tok_dir / "vocab.json").read_text(encoding="utf-8"))
toks = [None] * len(vocab)
for t_s, i_s in vocab.items():
toks[i_s] = t_s
token_types = json.loads((tok_dir / "token_type.json").read_text(encoding="utf-8"))
tok = ST.Tokenizer(
toks,
(tok_dir / "merges.txt").read_text(encoding="utf-8").splitlines(),
token_types,
)
controls = [toks[i] for i, t in enumerate(token_types) if t == 3]
lt = literal_tags(controls)
tpl = ChatTemplate(str(tok_dir / "chat_template.jinja"))
def encode_server_exact(msgs: list[dict], tools: list[dict], add_generation_prompt: bool) -> list[int]:
cm, ct, kw = openai_to_messages({"messages": msgs, "tools": tools, "reasoning_effort": "low"})
mk, mt, ch = mark_think_literals(cm, ct, lt)
rendered = tpl.render(mk, tools=mt, add_generation_prompt=add_generation_prompt, **kw)
if ch:
rendered, plain = unmark_think_literals(rendered, lt)
return list(tok.encode(rendered, parse_special=True, plain=plain))
return list(tok.encode(rendered, parse_special=True))
all_errors: list[str] = []
openers_3gram: Counter[str] = Counter()
exact_thinks: Counter[str] = Counter()
merged_by_split: dict[str, list[dict]] = {"train": [], "val_id": [], "val_ood_repo": []}
total_sup_turns = 0
total_py_cells = 0
for w in sel_windows:
wid = w["id"]
msgs = w["messages"]
if any(m.get("role") == "user" and m.get("content") == "inspect the workspace and run the next step." for m in msgs):
all_errors.append(f"{wid}: contains banned placeholder user prompt")
prior_defs: set[str] = set()
for idx, m in enumerate(msgs):
if m.get("role") != "assistant":
continue
if m.get("loss_mask") == 0:
if (m.get("reasoning_content") or "").strip() != "":
all_errors.append(f"{wid}::{idx}: loss_mask=0 turn has non-empty <think>: {m.get('reasoning_content')!r}")
if m.get("tool_calls"):
for tc in m["tool_calls"]:
if tc["function"]["name"] == "python":
du = get_defined_and_undef(tc["function"]["arguments"]["code"], prior_defs)
if du is not None:
prior_defs.update(du[0])
elif m.get("loss_mask") == 1:
total_sup_turns += 1
th = (m.get("reasoning_content") or "").strip()
all_errors.extend(f"{wid}::{idx}: {e}" for e in check_think_item(th, msgs, idx))
words = re.findall(r"[a-z0-9_']+", th.lower())
if len(words) >= 3:
openers_3gram[" ".join(words[:3])] += 1
exact_thinks[th] += 1
if m.get("content") is not None:
all_errors.extend(f"{wid}::{idx}: {e}" for e in check_reply_item(m["content"], msgs, idx))
if m.get("tool_calls"):
for tc in m["tool_calls"]:
fn = tc["function"]["name"]
if fn == "python":
total_py_cells += 1
code = tc["function"]["arguments"]["code"]
if not is_clean_idiomatic_code(code):
all_errors.append(f"{wid}::{idx}: loss_mask=1 python cell has anti-pattern")
du = get_defined_and_undef(code, prior_defs)
if du is None:
all_errors.append(f"{wid}::{idx}: loss_mask=1 python cell has syntax error")
else:
defs, undef = du
if undef:
all_errors.append(f"{wid}::{idx}: loss_mask=1 python cell has undefined AST names {undef}")
prior_defs.update(defs)
# Gate 8: Exact server path render + token-prefix & <|im_end|> span check
sys_txt = systems[msgs[0]["system_id"]]
clean_msgs = [
{"role": "system", "content": sys_txt}
if idx == 0
else {k: v for k, v in x.items() if k not in ("loss_mask", "think_provenance", "reply_provenance")}
for idx, x in enumerate(msgs)
]
try:
full_toks = encode_server_exact(clean_msgs, w.get("tools") or [], add_generation_prompt=False)
for m_idx, m in enumerate(clean_msgs):
if m.get("role") == "assistant":
upto_toks = encode_server_exact(clean_msgs[: m_idx + 1], w.get("tools") or [], add_generation_prompt=False)
if full_toks[: len(upto_toks)] != upto_toks:
all_errors.append(f"{wid}::{m_idx}: upto token prefix mismatch in ChatTemplate")
if msgs[m_idx].get("loss_mask") == 1:
pre_toks = encode_server_exact(clean_msgs[:m_idx], w.get("tools") or [], add_generation_prompt=True)
if full_toks[: len(pre_toks)] != pre_toks:
all_errors.append(f"{wid}::{m_idx}: loss_mask=1 pre token prefix mismatch in ChatTemplate")
if toks[upto_toks[-2]] != "<|im_end|>" or toks[upto_toks[-1]] != "Ċ":
all_errors.append(f"{wid}::{m_idx}: unexpected turn end tokens {toks[upto_toks[-2]]:!r}, {toks[upto_toks[-1]]:!r}")
except Exception as e:
all_errors.append(f"{wid}: ChatTemplate render failed: {e}")
merged_by_split[w["split"]].append(w)
for op, cnt in openers_3gram.items():
if cnt > 6:
all_errors.append(f"Global 3-gram opener {op!r} appears {cnt} times (max allowed across dataset is 6)")
for th, cnt in exact_thinks.items():
if cnt > 1:
all_errors.append(f"Duplicate <think> across dataset ({cnt}x): {th[:70]!r}")
# Check zero session overlap across SFT splits
s_tr = {w["session_id"] for w in merged_by_split["train"]}
s_vi = {w["session_id"] for w in merged_by_split["val_id"]}
s_vo = {w["session_id"] for w in merged_by_split["val_ood_repo"]}
if (s_tr & s_vi) or (s_tr & s_vo) or (s_vi & s_vo):
all_errors.append("Session overlap across SFT splits")
# Validate FTPO/DPO pairs (including 100% uniqueness + zero train/val leakage + server-exact encoding)
ftpo_path = DATA_DIR / "ftpo_dpo_pairs.jsonl"
if ftpo_path.exists():
ftpo_rows = [json.loads(l) for l in ftpo_path.read_text(encoding="utf-8").splitlines() if l.strip()]
seen_ftpo_sigs: set[str] = set()
seen_chosen_6c: set[str] = set()
f_tr_sess: set[str] = set()
f_va_sess: set[str] = set()
f_tr_users: set[str] = set()
f_va_users: set[str] = set()
for p in ftpo_rows:
pid = p["id"]
c_m = p["chosen_assistant"]
r_m = p["rejected_assistant"]
p_msgs = p["messages"] + [c_m]
all_errors.extend(f"{pid}::chosen: {e}" for e in check_think_item(c_m.get("reasoning_content", ""), p_msgs, len(p_msgs) - 1))
if c_m.get("content") is not None:
all_errors.extend(f"{pid}::chosen: {e}" for e in check_reply_item(c_m["content"], p_msgs, len(p_msgs) - 1))
if c_m == r_m:
all_errors.append(f"{pid}: chosen == rejected")
psig = json.dumps([p["messages"][-1], c_m, r_m], sort_keys=True)
if psig in seen_ftpo_sigs:
all_errors.append(f"{pid}: duplicate FTPO/DPO pair")
seen_ftpo_sigs.add(psig)
if p["category"] == "6C_dpo_anti_slop_voice":
csig = json.dumps(c_m, sort_keys=True)
if csig in seen_chosen_6c:
all_errors.append(f"{pid}: duplicate 6C chosen assistant turn")
seen_chosen_6c.add(csig)
u_txt = next((m.get("content", "") for m in reversed(p["messages"]) if m.get("role") == "user"), "")
if p["split"] == "train":
f_tr_sess.add(p["session_id"])
f_tr_users.add(u_txt)
else:
f_va_sess.add(p["session_id"])
f_va_users.add(u_txt)
if f_tr_sess & f_va_sess:
all_errors.append(f"FTPO session overlap between train and val_id: {len(f_tr_sess & f_va_sess)}")
if f_tr_users & f_va_users:
all_errors.append(f"FTPO user overlap between train and val_id: {len(f_tr_users & f_va_users)}")
if all_errors:
print(f"FAILED --final validation: {len(all_errors)} errors found (showing first 30):")
for e in all_errors[:30]:
print(" -", e)
return 1
for sp_name, rows in merged_by_split.items():
out_p = DATA_DIR / f"sft_{sp_name}.jsonl"
with out_p.open("w", encoding="utf-8") as fh:
for r in rows:
fh.write(json.dumps(r, ensure_ascii=False) + "\n")
print(
f"PASSED --final validation 100%!\n"
f" - Verified {len(sel_windows)} SFT windows ({total_sup_turns} loss_mask=1 turns, {total_py_cells} loss_mask=1 python cells)\n"
f" - Verified 100% unique <think> strings (max count = {max(exact_thinks.values()) if exact_thinks else 0})\n"
f" - Verified {len(ftpo_rows) if ftpo_path.exists() else 0} 100% unique, zero-leakage contrastive FTPO/DPO pairs\n"
f" - Top global 3-gram openers: {openers_3gram.most_common(3)}\n"
f" - Splits written: sft_train.jsonl ({len(merged_by_split['train'])}), "
f"sft_val_id.jsonl ({len(merged_by_split['val_id'])}), "
f"sft_val_ood_repo.jsonl ({len(merged_by_split['val_ood_repo'])})"
)
return 0
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--final", action="store_true")
args = ap.parse_args()
if args.final:
return merge_and_validate_final()
print("Specify --final")
return 1
if __name__ == "__main__":
raise SystemExit(main())