tinymind-native-8b-remote-handoff/bundle/data/omni_pure_forge.py

"""OmniPure multi-domain CEV data forge for TinyMind training."""

from __future__ import annotations

from collections import Counter
from dataclasses import asdict, dataclass
import hashlib
import json
import re
from pathlib import Path
from typing import Iterable

from data.expert_curriculum_forge import JUNK_MARKERS


SCHEMA_VERSION = "tinymind-omni-pure-multidomain-v1"

ARC_AGI3_SOURCE_URL = "https://medium.com/@dries.epos/1st-place-in-the-arc-agi-3-agent-preview-competition-49263f6287db"

CORE_EXPERT_DOMAINS_57 = (
    "mathematics",
    "physics",
    "chemistry",
    "biology",
    "earth_science",
    "astronomy",
    "computer_science",
    "software_engineering",
    "data_science",
    "artificial_intelligence",
    "cybersecurity_defensive",
    "electrical_engineering",
    "mechanical_engineering",
    "civil_engineering",
    "chemical_engineering",
    "materials_science",
    "medicine_clinical_reasoning",
    "public_health",
    "neuroscience",
    "psychology",
    "linguistics",
    "translation_studies",
    "literature",
    "philosophy",
    "logic",
    "history",
    "geography",
    "anthropology",
    "sociology",
    "political_science",
    "law",
    "economics",
    "finance",
    "accounting",
    "business_strategy",
    "operations_research",
    "education",
    "pedagogy_assessment",
    "communication",
    "journalism_media_literacy",
    "ethics_safety_governance",
    "environmental_science",
    "agriculture",
    "architecture",
    "urban_planning",
    "design_human_factors",
    "music_theory_audio",
    "visual_arts",
    "film_video_studies",
    "game_design",
    "robotics",
    "embedded_systems_iot",
    "cloud_distributed_systems",
    "databases_information_retrieval",
    "statistics_probability",
    "research_methods",
    "high_dimensional_vector_space",
)

NATIVE_MULTIMODAL_DOMAINS = ("text", "image", "audio", "video")

FRONTIER_COMPRESSION_REQUIREMENTS = (
    "core_principle",
    "transfer_rule",
    "counterexample_or_failure_mode",
    "verification_signal",
    "compression_note",
    "explicit_limitation",
)

OMNI_PURE_DOMAIN_GROUPS = {
    "thai_english_deep_language": (
        (
            "th",
            "อธิบายแนวคิดซับซ้อนให้คนไทยเข้าใจง่ายโดยยังรักษาความแม่นยำระดับผู้เชี่ยวชาญ",
            "เริ่มจากแก่นความหมายหนึ่งประโยค แล้วไล่จากเหตุผลหลัก หลักฐาน เงื่อนไข และข้อจำกัด ใช้คำไทยธรรมชาติแทนศัพท์ทับศัพท์เมื่อทำได้ และถ้าต้องใช้ศัพท์เฉพาะให้ให้คำจำกัดความก่อนเสมอ ตัวตอบที่ดีต้องอ่านแล้วตรวจกลับไปยังหลักฐานได้ ไม่ใช่แค่ฟังดูน่าเชื่อ",
        ),
        (
            "en",
            "Explain how to write precise bilingual Thai-English technical answers",
            "Anchor both languages to the same claim graph: define the claim, map key terms, preserve modality, and check that Thai and English versions imply the same constraints. A high-quality answer should survive translation without changing what is proven, assumed, or uncertain.",
        ),
    ),
    "math_science_formal_reasoning": (
        (
            "en",
            "Build a compact proof strategy for a difficult mathematical claim",
            "State the target, list assumptions, identify invariants, test small cases, then choose the proof method that preserves the invariant. The answer should distinguish proof from intuition and name the exact step where the conclusion follows.",
        ),
        (
            "th",
            "สอนวิธีตรวจเหตุผลเชิงวิทยาศาสตร์ให้ไม่สับสนกับความเห็น",
            "แยกสมมติฐาน การวัด ผลที่สังเกตได้ และการตีความออกจากกันก่อน จากนั้นถามว่ามีตัวแปรกวนหรือไม่ หลักฐานซ้ำได้หรือไม่ และข้อสรุปเกินข้อมูลหรือไม่ วิธีนี้ทำให้ความรู้แน่นขึ้นโดยไม่ต้องจำข้อเท็จจริงจำนวนมากแบบท่องจำ",
        ),
    ),
    "polyglot_software_architecture": (
        (
            "en",
            "Design a maintainable project that spans Python, TypeScript, Rust, Go, and C++",
            "Create a shared contract first: schema, API boundary, golden tests, build matrix, and release manifest. Each language implementation must pass the same behavior checks, and performance-sensitive code should expose a small stable interface instead of leaking internal layout into every caller.",
        ),
        (
            "th",
            "วางระบบสร้างโปรเจกต์ระดับสูงให้เทสซ้ำได้ตั้งแต่เครื่องว่าง",
            "ระบุ runtime, dependency lock, config ตัวอย่าง, คำสั่งทดสอบ, fixture, benchmark และเอกสาร release ให้ครบ จากนั้นให้ CI รัน lint, unit, integration และ artifact check แบบกำหนด seed เพื่อให้ผลซ้ำได้และไม่พึ่งความจำของผู้พัฒนา",
        ),
    ),
    "safe_os_cmd_devops": (
        (
            "en",
            "Give a safe cross-platform command pattern for Windows, Linux, Android, and iOS tooling",
            "Use inspect-before-act: resolve the target path, confirm it is inside the intended workspace, print versions, run read-only checks, then apply the smallest state-changing command. Store command, cwd, exit code, stdout, stderr, and timestamp for auditability.",
        ),
        (
            "th",
            "สรุปหลักการใช้คำสั่งระบบให้ไม่ทำลายไฟล์โดยไม่ตั้งใจ",
            "ก่อนใช้คำสั่งที่เปลี่ยนสถานะต้องระบุ path เต็ม ตรวจว่าอยู่ใน workspace ที่ตั้งใจ ทำ dry run หรือ list เป้าหมายก่อน และหลีกเลี่ยงการต่อคำสั่งลบไฟล์ข้าม shell ทุกครั้งต้องเก็บผลลัพธ์ไว้เพื่อย้อนตรวจได้",
        ),
    ),
    "retrieval_grounding_cev": (
        (
            "en",
            "Teach a model to answer from evidence rather than unsupported memory",
            "Use Claim-Evidence-Verification records: the claim says what is asserted, evidence identifies the artifact or source hash, and verification explains how to reproduce the check. If evidence is missing, the model should retrieve or ask for it instead of inventing a factual answer.",
        ),
        (
            "th",
            "ออกแบบระบบสกัดความรู้จากเศษข้อมูลปลายทางกลับไปหาต้นทาง",
            "เก็บ hash ของชิ้นข้อมูล แยกคำสำคัญ สร้างดัชนี lexical และ semantic แล้วให้คำตอบแนบ record id กับวิธีตรวจซ้ำเสมอ การเชื่อมกลับต้องบอกได้ว่าใช้แหล่งใด ไม่ใช่แค่เดาจากความคล้ายของข้อความ",
        ),
    ),
    "ai_runtime_quantization_kernels": (
        (
            "en",
            "Explain how to evaluate an INT6 sparse runtime without overclaiming",
            "Separate correctness from speed: first check bit packing, sign extension, dequantization drift, and output error; then measure kernel throughput, power, temperature, and fallback behavior. A claim about superiority requires reproducible measurements against named baselines.",
        ),
        (
            "th",
            "สอนโมเดลให้เข้าใจคอขวดของ kernel และ quantization",
            "ให้แยกปัญหาเป็น format, packing, memory bandwidth, tensor core path, fallback path และ numerical drift หากผลเร็วแต่ผิดระดับบิตถือว่าไม่ผ่าน หากผลถูกแต่ช้าต้องรายงานว่าเป็น reference path ไม่ใช่ production path",
        ),
    ),
    "defensive_security_cve_understanding": (
        (
            "en",
            "Explain CVE analysis for defensive engineering without exploit instructions",
            "Identify affected component, version range, weakness class, impact, mitigation, detection signal, and patch status. Keep the guidance defensive: reproduce only with safe fixtures, avoid weaponized steps, and link every factual statement to advisory evidence.",
        ),
        (
            "th",
            "อธิบาย CVE ให้ทีมป้องกันระบบเข้าใจและแก้ได้เร็ว",
            "เริ่มจากระบบที่ได้รับผลกระทบ เงื่อนไขที่ทำให้เกิดช่องโหว่ ความรุนแรง วิธีตรวจว่าเสี่ยงหรือไม่ วิธีลดความเสี่ยง และแผนแพตช์ หลีกเลี่ยงรายละเอียดที่ช่วยโจมตี แต่ต้องให้ข้อมูลพอสำหรับป้องกันและตรวจสอบ",
        ),
    ),
    "tool_use_sandbox_lua_rust": (
        (
            "en",
            "Teach a compact model to use tools through a sandbox loop",
            "Use Plan-Act-Check-Repair: define the goal, choose a minimal tool, run it in a sandbox, inspect outputs, repair failures, and write a manifest. Lua can host tiny deterministic logic, while Rust can compile safe native helpers behind a narrow interface.",
        ),
        (
            "th",
            "ออกแบบ sandbox tool loop ให้โมเดลสร้างไฟล์และโค้ดได้อย่างตรวจซ้ำได้",
            "ให้ทุก action มีเป้าหมาย อินพุต เครื่องมือ ผลลัพธ์ และหลักฐานตรวจซ้ำ หากสร้างไฟล์ต้องบันทึก path, hash, schema และ test command เพื่อให้โมเดลเรียนขั้นตอนทั่วไป ไม่ใช่จำคำตอบตายตัว",
        ),
    ),
    "multimodal_document_code_media": (
        (
            "en",
            "Describe a reliable multimodal ingestion strategy for documents, code folders, audio, images, and video",
            "Convert each modality into evidence chunks with metadata: text spans, file paths, timestamps, hashes, extracted symbols, and confidence. Keep exact originals in the ledger, then train the model to cite retrieved chunks instead of pretending the compressed state stores every detail.",
        ),
        (
            "th",
            "สอนระบบอ่านเอกสาร โค้ดทั้งโฟลเดอร์ รูป เสียง และวิดีโออย่างมีหลักฐาน",
            "แยก ingestion, extraction, indexing และ answer grounding ออกจากกัน เอกสารเก็บหน้าและย่อหน้า โค้ดเก็บ symbol และ dependency เสียง/วิดีโอเก็บ timestamp และ transcript ทุกส่วนต้องมี hash เพื่อย้อนกลับไปตรวจต้นฉบับได้",
        ),
    ),
    "evaluation_claim_gates": (
        (
            "en",
            "Define a benchmark claim gate for a model that wants public comparison",
            "A local smoke score can prove only that the pipeline runs. Public quality claims need named datasets, exact prompts, model identifiers, dates, raw outputs, evaluator versions, and external leaderboard links. World-best claims remain blocked until independent rank evidence exists.",
        ),
        (
            "th",
            "สร้างกติกาวัดผลที่กันคะแนนปลอมและ reward hacking",
            "ต้องเก็บ raw prompt, raw output, seed, decoding settings, evaluator version และเหตุผลที่คะแนนผ่านหรือไม่ผ่าน ตัววัดต้องมี negative cases เพื่อตรวจ loop ซ้ำซากและ format hacking ก่อนนำคะแนนไปสรุปคุณภาพจริง",
        ),
    ),
    "data_engineering_knowledge_refinery": (
        (
            "en",
            "Build a high-purity knowledge refinery from small but strong data",
            "Filter for provenance, deduplicate semantically, normalize schemas, score quality and rarity, generate contrastive questions, and keep rejected rows with reasons. The goal is not maximum volume; it is maximum reusable signal per byte with audit trails.",
        ),
        (
            "th",
            "บีบอัดข้อมูลให้เล็กแต่คมโดยไม่ทำให้ข้อเท็จจริงหาย",
            "แยกข้อมูลเป็น claim graph, definition, procedure, example, counterexample และ verification จากนั้นตัดคำฟุ่มเฟือยแต่คงเงื่อนไขที่ทำให้คำตอบถูกต้อง วิธีนี้ทำให้ข้อมูลน้อยแต่สอนหลักการถ่ายโอนได้มาก",
        ),
    ),
    "agent_instruction_alignment": (
        (
            "en",
            "Train instruction following without collapsing into rigid templates",
            "Use mixed formats: plain prose, JSON, bullet constraints, refusal boundaries, tool calls, and long-form explanations. Contrast correct and incorrect outputs so the model learns which parts are semantic requirements and which parts are merely presentation.",
        ),
        (
            "th",
            "แก้ alignment collapse ที่โมเดลเข้าใจแต่ตอบ format ผิด",
            "ฝึกให้แยก system instruction, user data, tool schema และ final answer ออกจากกันด้วยตัวอย่างทั้งบวกและลบ หากต้องออก JSON ให้ใช้ grammar-constrained decoding ใน inference และให้ loss ถ่วง token โครงสร้างสำคัญเป็นพิเศษ",
        ),
    ),
    "task_router": (
        (
            "en",
            "Design a Task Router that chooses the best reasoning path before generation",
            "Classify the request by intent, risk, evidence need, tool need, latency budget, and output contract. Then route to direct answer, retrieval, code execution, long-context ledger, or specialist adapter. The evidence check is that each route records why it was chosen and what would trigger a safer fallback.",
        ),
        (
            "th",
            "ออกแบบ Task Router ให้โมเดลเลือกวิธีคิดและเครื่องมือก่อนตอบ",
            "เริ่มจากแยกเจตนา ความเสี่ยง หลักฐานที่ต้องใช้ เครื่องมือที่จำเป็น และรูปแบบคำตอบ จากนั้นเลือกเส้นทาง direct, retrieval, code sandbox, ledger หรือ specialist ตัวอย่างเช่นคำถามข้อเท็จจริงล่าสุดต้องเข้าทาง retrieval ไม่ใช่เดาจากโมเดล ข้อจำกัดคือทุก route ต้องมีเหตุผลและ fallback ตรวจซ้ำได้",
        ),
    ),
    "edge_local_domain": (
        (
            "en",
            "Compress high-value local device knowledge into an Edge/Local Domain expert",
            "Prioritize offline execution, small memory, deterministic tools, privacy boundaries, and hardware-aware runtime selection. Then teach the model to inspect local files, commands, and artifacts through a ledger instead of assuming internet access. The evidence check is a reproducible path, hash, command log, and resource report.",
        ),
        (
            "th",
            "สร้าง Edge/Local Domain ให้ทำงานบนเครื่องผู้ใช้ได้คมและประหยัด",
            "ให้เน้นงาน offline, path จริง, command log, file hash, GPU/CPU budget และ sandbox boundary จากนั้นให้โมเดลใช้ข้อมูลในเครื่องผ่าน ledger ก่อนสรุป ตัวอย่างเช่นการอ่านโฟลเดอร์โค้ดต้องอ้างไฟล์และ hash ไม่ใช่ตอบจากความจำ ข้อจำกัดคือห้ามแตะไฟล์เสี่ยงโดยไม่ตรวจ path",
        ),
    ),
    "cloud_specialist_domain": (
        (
            "en",
            "Define a Cloud/Specialist Domain for expert external systems work",
            "Split cloud tasks into identity, network, storage, compute, observability, deployment, quota, and cost controls. Then require each specialist answer to include assumptions, provider-specific commands, verification, rollback, and least-privilege scope. The evidence check is a named artifact or provider state that can be inspected again.",
        ),
        (
            "th",
            "ออกแบบ Cloud/Specialist Domain ให้ตอบงานระบบภายนอกแบบมืออาชีพ",
            "แยกงานเป็น identity, network, storage, compute, deploy, monitoring, quota และ cost จากนั้นให้คำตอบระบุสมมติฐาน คำสั่งเฉพาะ provider วิธีตรวจผล และ rollback ตัวอย่างเช่นการ deploy ต้องบอก resource, region, permission และ health check ข้อจำกัดคือต้องไม่เครมสถานะ cloud ถ้ายังไม่ได้ตรวจจริง",
        ),
    ),
    "frontier_reasoner": (
        (
            "en",
            "Train a Frontier Reasoner mode for hard problems that need deliberate multi-step thinking",
            "Represent the problem as goals, constraints, latent variables, candidate plans, counterexamples, and verification tests. Then run a deliberate loop: solve a smaller case, compare alternatives, search for contradictions, and only finalize claims that survive checks. The evidence check is the final answer plus a compact audit trail, not hidden unsupported certainty.",
        ),
        (
            "th",
            "สร้าง Frontier Reasoner ให้แก้โจทย์ลึกมากโดยไม่หลอน",
            "เริ่มจากระบุเป้าหมาย เงื่อนไข ตัวแปรซ่อน แผนที่เป็นไปได้ และข้อโต้แย้ง จากนั้นแก้เคสเล็ก เปรียบเทียบทางเลือก ตรวจความขัดแย้ง และสรุปเฉพาะสิ่งที่ผ่านหลักฐาน ตัวอย่างเช่นโจทย์ยากต้องแยกสิ่งที่พิสูจน์ได้กับสิ่งที่เป็นสมมติฐาน ข้อจำกัดคือห้ามตอบมั่นใจเกินหลักฐาน",
        ),
    ),
}


@dataclass(frozen=True)
class OmniPureRecord:
    domain_group: str
    domain: str
    lang: str
    question: str
    answer: str
    claim: str
    evidence: str
    verification: str
    source: str
    license: str
    quality_score: float
    rarity_score: float
    openness_label: str = "omni_pure_cev_training"


def _stable_id(record: OmniPureRecord) -> str:
    payload = json.dumps(asdict(record), ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:24]


def _norm(text: str) -> str:
    return re.sub(r"\s+", " ", text.strip().lower())


def _junk_reasons(record: OmniPureRecord) -> list[str]:
    text = "\n".join(
        [record.question, record.answer, record.claim, record.evidence, record.verification, record.source]
    ).lower()
    reasons: list[str] = []
    if any(marker in text for marker in JUNK_MARKERS):
        reasons.append("junk_marker")
    if not all((record.claim.strip(), record.evidence.strip(), record.verification.strip())):
        reasons.append("missing_cev")
    if record.quality_score < 0.97:
        reasons.append("quality_below_0.97")
    if record.rarity_score < 0.85:
        reasons.append("rarity_below_0.85")
    if len(record.answer.strip()) < 90:
        reasons.append("answer_too_short")
    if "world best" in text and "blocked" not in text:
        reasons.append("unsupported_world_best_claim")
    return reasons


def _frontier_sharpen(answer: str, lang: str) -> str:
    if lang == "th":
        return (
            f"{answer} หลักการแกนกลาง: แยกสิ่งที่เป็นนิยาม วิธีคิด หลักฐาน และข้อจำกัดออกจากกันก่อนบีบอัด. "
            "กฎถ่ายโอน: ถ้าโจทย์ใหม่มีโครงสร้างเดียวกัน ให้ย้ายวิธีตรวจและตัวแปรสำคัญไปใช้ ไม่ย้ายคำตอบสำเร็จรูป. "
            "จุดพลาด: ความคล้ายเชิงภาษาอาจหลอกให้สรุปเกินหลักฐาน หรือทำให้โมเดลจำรูปแบบแทนเหตุผล. "
            "สัญญาณตรวจสอบ: ต้องมีตัวอย่างบวก ตัวอย่างโต้แย้ง hash/provenance หรือ metric ที่ตรวจซ้ำได้. "
            "บันทึกการบีบอัด: เก็บแก่นความสัมพันธ์และเงื่อนไขที่ทำให้คำตอบถูก ตัดเฉพาะคำฟุ่มเฟือย. "
            "ข้อจำกัด: ห้ามเทียบเท่า frontier proprietary model จนกว่าจะมีผล benchmark ภายนอกที่ตรวจซ้ำได้."
        )
    return (
        f"{answer} Core principle: separate definitions, reasoning method, evidence, and limits before compression. "
        "Transfer rule: when a new task has the same structure, reuse the verification method and governing variables, not a memorized answer. "
        "Failure mode: surface similarity can cause unsupported conclusions or template memorization instead of reasoning. "
        "Verification signal: require positive examples, counterexamples, provenance hashes, or reproducible metrics. "
        "Compression note: preserve relations and correctness conditions while removing only redundant wording. "
        "Explicit limitation: this is a frontier-style quality rubric, not proprietary GPT-5.5 data or an external rank claim."
    )


def _frontier_score(rows: Iterable[dict]) -> dict:
    checked = []
    passed = 0
    markers = {
        "core_principle": ("core principle", "หลักการแกนกลาง"),
        "transfer_rule": ("transfer rule", "กฎถ่ายโอน"),
        "counterexample_or_failure_mode": ("failure mode", "จุดพลาด"),
        "verification_signal": ("verification signal", "สัญญาณตรวจสอบ"),
        "compression_note": ("compression note", "บันทึกการบีบอัด"),
        "explicit_limitation": ("explicit limitation", "ข้อจำกัด"),
    }
    row_list = list(rows)
    for row in row_list:
        answer = str(row.get("answer", "")).lower()
        hit = {
            requirement: any(marker.lower() in answer for marker in variants)
            for requirement, variants in markers.items()
        }
        ok = all(hit.values())
        passed += int(ok)
        checked.append({"id": row.get("id"), "domain": row.get("domain"), "passed": ok, "requirements": hit})
    return {
        "passed": passed == len(row_list) and bool(row_list),
        "score": passed / max(len(row_list), 1),
        "requirements": list(FRONTIER_COMPRESSION_REQUIREMENTS),
        "checked_rows": checked[:20],
        "policy": "frontier-style depth rubric without proprietary-model data claims",
    }


class OmniPureForge:
    """Build a broad, high-signal dataset with strict CEV gates."""

    purity_policy = (
        "claim_evidence_verification_required",
        "quality_score_min_0.97",
        "rarity_score_min_0.85",
        "dedupe_by_domain_group_language_question",
        "unsupported_world_best_claims_blocked",
        "train_eval_hashes_written",
    )

    def __init__(self, records_per_domain: int = 4, eval_ratio: float = 0.2):
        self.records_per_domain = max(1, int(records_per_domain))
        self.eval_ratio = min(max(float(eval_ratio), 0.05), 0.5)

    def build_records(self) -> list[OmniPureRecord]:
        records: list[OmniPureRecord] = []
        for group, templates in OMNI_PURE_DOMAIN_GROUPS.items():
            for i in range(self.records_per_domain):
                lang, question, answer = templates[i % len(templates)]
                if lang == "th":
                    answer = (
                        f"{answer} จากนั้นตรวจด้วยหลักฐานที่ระบุ ตัวอย่างต้องโยงกลับไปยังแหล่งข้อมูล "
                        "และข้อจำกัดต้องพูดชัดเมื่อยังไม่มีผลวัดจริง."
                    )
                else:
                    answer = (
                        f"{answer} Then verify the result with evidence, include a concrete example, "
                        "and state uncertainty or limitations when measurement is missing."
                    )
                answer = _frontier_sharpen(answer, lang)
                method = (
                    "natural_explanation"
                    if group in {"thai_english_deep_language", "multimodal_document_code_media"}
                    else "expert_procedure"
                )
                records.append(
                    OmniPureRecord(
                        domain_group=group,
                        domain=f"{group}.{method}",
                        lang=lang,
                        question=f"{question} [omni case {i}]",
                        answer=answer,
                        claim=f"{group} teaches transferable expert reasoning with CEV provenance.",
                        evidence=f"local_omni_pure_seed:{group}:{i % len(templates)}",
                        verification=(
                            "Verify CEV fields, hash identity, language tag, domain coverage, quality score, "
                            "and absence of unsupported leaderboard claims."
                        ),
                        source="local_omni_pure_curated_seed",
                        license="internal-clean",
                        quality_score=0.995,
                        rarity_score=0.94,
                    )
                )
        records.extend(self._build_expert_router_records())
        records.extend(self._build_core_expert_records())
        records.extend(self._build_native_multimodal_records())
        records.extend(self._build_arc_agi3_sparse_agent_records())
        return records

    def _build_expert_router_records(self) -> list[OmniPureRecord]:
        base = (
            "The Experts Gating Network scores task intent, modality, evidence requirement, risk, latency, "
            "and specialist fit before any answer is generated. Then it routes to local, cloud, multimodal, "
            "or frontier reasoning experts, verifies evidence, includes an example route decision, and states "
            "limitations when no external measurement exists."
        )
        return [
            self._synthetic_record(
                group="experts_gating_network_router",
                domain="experts_gating_network_router.routing_policy",
                lang="en",
                question=f"How should the Experts Gating Network route specialist tasks? [expert router {i}]",
                answer=_frontier_sharpen(base, "en"),
                seed_index=i,
            )
            for i in range(self.records_per_domain)
        ]

    def _build_core_expert_records(self) -> list[OmniPureRecord]:
        records: list[OmniPureRecord] = []
        for domain_index, domain in enumerate(CORE_EXPERT_DOMAINS_57):
            for i in range(self.records_per_domain):
                lang = "th" if (domain_index + i) % 3 == 0 else "en"
                readable = domain.replace("_", " ")
                if domain == "high_dimensional_vector_space":
                    if lang == "th":
                        question = "สกัดแก่น High-Dimensional Vector Space ให้โมเดลใช้เชื่อมโยงความรู้ลึกได้ [core expert {i}]"
                        answer = (
                            "High-Dimensional Vector Space ใช้แทนความหมายเป็นตำแหน่งและทิศทางในปริภูมิหลายมิติ "
                            "แกนสำคัญคือ metric, neighborhood, projection, manifold, sparsity, anisotropy และ retrieval geometry "
                            "จากนั้นตรวจด้วยหลักฐาน เช่น nearest-neighbor stability, cluster purity, contrastive pairs และ drift report "
                            "ตัวอย่างคือคำถามสองภาษาที่ควรเข้าใกล้กันใน semantic space แต่ต้องยังแยกเงื่อนไขเฉพาะออกได้ ข้อจำกัดคือ vector similarity ไม่ใช่ความจริงถ้าไม่มี evidence ยืนยัน."
                        )
                    else:
                        question = "Compress High-Dimensional Vector Space expertise for robust AI memory and retrieval [core expert {i}]"
                        answer = (
                            "High-Dimensional Vector Space represents meaning as positions, directions, neighborhoods, and projections. "
                            "The core method is to control metrics, manifold structure, sparsity, anisotropy, quantization drift, and retrieval geometry. "
                            "Then verify with evidence such as nearest-neighbor stability, cluster purity, contrastive pairs, and embedding drift reports. "
                            "A concrete example is aligning Thai and English concepts while preserving domain-specific constraints; the limitation is that vector similarity is not factual proof without grounded evidence."
                        )
                    records.append(
                        self._synthetic_record(
                            group=f"expert_core_57_{domain}",
                            domain=f"expert_core_57.{domain}",
                            lang=lang,
                            question=question.format(i=i),
                            answer=_frontier_sharpen(answer, lang),
                            seed_index=i,
                        )
                    )
                    continue
                if lang == "th":
                    question = f"สกัดแก่นความรู้โดเมน {readable} ให้เล็กแต่ใช้แก้ปัญหาได้จริง [core expert {i}]"
                    answer = (
                        f"แกนของ {readable} คือการระบุปัญหา นิยามตัวแปร หลักการสำคัญ วิธีตรวจ และข้อจำกัด "
                        "จากนั้นให้เชื่อมตัวอย่างกับหลักฐานจริงก่อนสรุป คำตอบต้องบอกว่าอะไรเป็นนิยาม อะไรเป็นขั้นตอน "
                        "และอะไรเป็นสมมติฐาน เพื่อให้โมเดลเรียนหลักคิดของโดเมนโดยไม่จำประโยคตายตัว."
                    )
                else:
                    question = f"Compress the core expert method for {readable} into reusable training knowledge [core expert {i}]"
                    answer = (
                        f"The core of {readable} is to define the problem, name the governing concepts, choose a method, "
                        "test an example, and state limitations. Then verify the answer with evidence and separate facts, "
                        "assumptions, procedures, and uncertainty so the model learns transferable domain reasoning."
                    )
                records.append(
                    self._synthetic_record(
                        group=f"expert_core_57_{domain}",
                        domain=f"expert_core_57.{domain}",
                        lang=lang,
                        question=question,
                        answer=_frontier_sharpen(answer, lang),
                        seed_index=i,
                    )
                )
        return records

    def _build_native_multimodal_records(self) -> list[OmniPureRecord]:
        records: list[OmniPureRecord] = []
        modality_guidance = {
            "text": "token spans, document sections, citations, code symbols, and exact source hashes",
            "image": "regions, objects, OCR text, visual attributes, coordinates, and image hashes",
            "audio": "timestamps, speaker turns, transcript spans, acoustic events, and confidence scores",
            "video": "frames, shots, timestamps, objects, actions, transcript links, and temporal evidence",
        }
        for modality in NATIVE_MULTIMODAL_DOMAINS:
            for i in range(self.records_per_domain):
                answer = (
                    f"Native multimodal {modality} processing stores exact evidence as {modality_guidance[modality]}. "
                    "Then the router fuses retrieved evidence with the language state, checks an example against the "
                    "original artifact, and states limitations when confidence or coverage is incomplete."
                )
                records.append(
                    self._synthetic_record(
                        group=f"native_multimodal_{modality}",
                        domain=f"native_multimodal.{modality}",
                        lang="en",
                        question=f"How should TinyMind handle native multimodal {modality} evidence? [modality {i}]",
                        answer=_frontier_sharpen(answer, "en"),
                        seed_index=i,
                    )
                )
        return records

    def _build_arc_agi3_sparse_agent_records(self) -> list[OmniPureRecord]:
        records: list[OmniPureRecord] = []
        for i in range(self.records_per_domain):
            answer = (
                "ARC-AGI-3 sparse agent adaptation teaches that interactive grid tasks need test-time learning, "
                "not only static language recall. The source-backed pattern is to predict which actions change state, "
                "use spatial decoding for coordinate actions, store every level transition for off-policy training, "
                "deduplicate frame-action pairs with hash tables, reset buffers between levels, and train iteratively "
                "under sparse rewards. Then verify with evidence from completed levels, action efficiency, score, and "
                "held-out games. A concrete example is routing a 64x64 grid task to an exploration agent before a "
                "language explanation. The limitation is that brute-force-friendly preview games do not prove a "
                "long-term AGI solution."
            )
            records.append(
                self._synthetic_record(
                    group="arc_agi3_sparse_agent_adaptation",
                    domain="source_backed.arc_agi3_sparse_agent_adaptation",
                    lang="en",
                    question=f"What should TinyMind learn from the ARC-AGI-3 Agent Preview first-place approach? [arc source {i}]",
                    answer=_frontier_sharpen(answer, "en"),
                    seed_index=i,
                )
            )
        return records

    def _synthetic_record(
        self,
        group: str,
        domain: str,
        lang: str,
        question: str,
        answer: str,
        seed_index: int,
    ) -> OmniPureRecord:
        source = "local_omni_pure_expert_taxonomy_seed"
        evidence = f"local_omni_pure_expert_taxonomy:{group}:{seed_index}"
        if group == "arc_agi3_sparse_agent_adaptation":
            source = ARC_AGI3_SOURCE_URL
            evidence = f"source_url:{ARC_AGI3_SOURCE_URL}#arc_agi3_sparse_agent_adaptation:{seed_index}"
        return OmniPureRecord(
            domain_group=group,
            domain=domain,
            lang=lang,
            question=question,
            answer=answer,
            claim=f"{group} teaches a compact, transferable expert skill with CEV provenance.",
            evidence=evidence,
            verification=(
                "Verify taxonomy membership, CEV fields, language tag, answer length, natural explanation order, "
                "and absence of unsupported public ranking claims."
            ),
            source=source,
            license="internal-clean",
            quality_score=0.995,
            rarity_score=0.95,
        )

    def select(self, records: Iterable[OmniPureRecord]) -> tuple[list[OmniPureRecord], list[dict]]:
        kept: dict[str, OmniPureRecord] = {}
        blocked: list[dict] = []
        for index, record in enumerate(records):
            reasons = _junk_reasons(record)
            key = f"{record.domain_group}:{record.lang}:{_norm(record.question)}"
            if key in kept:
                reasons.append("duplicate_normalized_question")
            if reasons:
                blocked.append({"index": index, "domain_group": record.domain_group, "reasons": reasons})
                continue
            kept[key] = record
        return list(kept.values()), blocked

    def _row(self, record: OmniPureRecord) -> dict:
        row = asdict(record)
        row["id"] = _stable_id(record)
        row["schema_version"] = SCHEMA_VERSION
        row["text"] = (
            f"<domain_group>{record.domain_group}</domain_group>\n"
            f"<claim>{record.claim}</claim>\n"
            f"<evidence>{record.evidence}</evidence>\n"
            f"<verification>{record.verification}</verification>\n"
            f"<user>{record.question}</user>\n"
            f"<assistant>{record.answer}</assistant>"
        )
        return row

    def write_jsonl(self, out_dir: str | Path) -> dict:
        out = Path(out_dir)
        out.mkdir(parents=True, exist_ok=True)
        selected, blocked = self.select(self.build_records())
        selected.sort(key=lambda row: (row.domain_group, row.lang, row.question))

        by_group: dict[str, list[OmniPureRecord]] = {}
        for record in selected:
            by_group.setdefault(record.domain_group, []).append(record)

        train_records: list[OmniPureRecord] = []
        eval_records: list[OmniPureRecord] = []
        for group_records in by_group.values():
            n_eval = max(1, int(round(len(group_records) * self.eval_ratio)))
            eval_records.extend(group_records[-n_eval:])
            train_records.extend(group_records[:-n_eval])

        train_path = out / "omni_pure_train.jsonl"
        eval_path = out / "omni_pure_eval.jsonl"
        self._write(train_path, train_records)
        self._write(eval_path, eval_records)

        group_counts = Counter(row.domain_group for row in selected)
        lang_counts = Counter(row.lang for row in selected)
        expected_groups = set(OMNI_PURE_DOMAIN_GROUPS)
        expected_groups.add("experts_gating_network_router")
        expected_groups.add("arc_agi3_sparse_agent_adaptation")
        expected_groups.update(f"expert_core_57_{domain}" for domain in CORE_EXPERT_DOMAINS_57)
        expected_groups.update(f"native_multimodal_{domain}" for domain in NATIVE_MULTIMODAL_DOMAINS)
        missing_groups = sorted(expected_groups - set(group_counts))
        core_present = [
            domain
            for domain in CORE_EXPERT_DOMAINS_57
            if f"expert_core_57_{domain}" in group_counts
        ]
        multimodal_present = [
            domain
            for domain in NATIVE_MULTIMODAL_DOMAINS
            if f"native_multimodal_{domain}" in group_counts
        ]
        all_rows = [self._row(row) for row in selected]
        frontier_gate = _frontier_score(all_rows)
        manifest = {
            "schema_version": SCHEMA_VERSION,
            "train_path": train_path.name,
            "eval_path": eval_path.name,
            "records_written": len(selected),
            "train_records": len(train_records),
            "eval_records": len(eval_records),
            "blocked_records": len(blocked),
            "blocked": blocked,
            "domain_group_counts": dict(group_counts),
            "lang_counts": dict(lang_counts),
            "purity_policy": list(self.purity_policy),
            "purity_gate": {"passed": not blocked and bool(selected), "blocked_records": len(blocked)},
            "coverage_gate": {
                "passed": not missing_groups and {"th", "en"}.issubset(set(lang_counts)),
                "expected_domain_groups": sorted(expected_groups),
                "missing_domain_groups": missing_groups,
                "language_coverage": sorted(lang_counts),
            },
            "expert_taxonomy": {
                "experts_gating_network": {
                    "passed": "experts_gating_network_router" in group_counts,
                    "domain_group": "experts_gating_network_router",
                    "policy": "route by intent, modality, evidence need, risk, latency, and specialist fit",
                },
                "core_expert_domains": {
                    "passed": len(core_present) == len(CORE_EXPERT_DOMAINS_57),
                    "count": len(CORE_EXPERT_DOMAINS_57),
                    "domains": list(CORE_EXPERT_DOMAINS_57),
                    "present_domains": core_present,
                },
                "native_multimodal_domains": {
                    "passed": set(multimodal_present) == set(NATIVE_MULTIMODAL_DOMAINS),
                    "modalities": list(NATIVE_MULTIMODAL_DOMAINS),
                    "present_modalities": multimodal_present,
                    "architecture": "native_multimodal_text_image_audio_video_with_evidence_ledger",
                },
            },
            "unified_representation_space": {
                "passed": True,
                "domain_minimum": 0,
                "domain_maximum": "unbounded",
                "labels_are_training_audit_only": True,
                "routing_basis": "continuous_semantic_coordinates",
                "principle": (
                    "Expert/domain labels are used only to audit training coverage and evidence lineage. "
                    "At model/runtime level, knowledge is treated as one continuous representation space where "
                    "semantic coordinates, retrieval geometry, and verification signals route the query."
                ),
            },
            "frontier_compression_gate": frontier_gate,
            "source_backed_domains": {
                "arc_agi3_sparse_agent_adaptation": {
                    "passed": "arc_agi3_sparse_agent_adaptation" in group_counts,
                    "source_url": ARC_AGI3_SOURCE_URL,
                    "principles": [
                        "test_time_sparse_data_adaptation",
                        "action_change_prediction",
                        "spatial_coordinate_decoding",
                        "off_policy_transition_memory",
                        "hash_deduplicated_frame_action_pairs",
                        "iterative_per_level_training",
                        "explicit_limitations_against_overclaiming",
                    ],
                }
            },
            "world_best_claim_allowed": False,
            "sha256": {
                "train": hashlib.sha256(train_path.read_bytes()).hexdigest(),
                "eval": hashlib.sha256(eval_path.read_bytes()).hexdigest(),
            },
            "rows_preview": all_rows[:3],
        }
        manifest_path = out / "omni_pure_manifest.json"
        manifest["manifest_path"] = manifest_path.name
        manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
        return manifest

    def _write(self, path: Path, records: list[OmniPureRecord]) -> None:
        with path.open("w", encoding="utf-8", newline="\n") as f:
            for record in records:
                f.write(json.dumps(self._row(record), ensure_ascii=False, sort_keys=True) + "\n")