tinymind-native-8b-remote-handoff/bundle/evaluation/holistic_purity_flexibility.py

"""Holistic purity, coverage, and flexibility gate for TinyMind.

This report intentionally separates three claims that are often blurred:

1. clean, decontaminated training inputs;
2. flexible audited runtime/tool behavior;
3. raw model capability inside the checkpoint.

Only the first two can pass from local evidence alone.  Raw model and
world-best claims stay blocked until uncontrolled probes and official/hidden
evaluations support them.
"""

from __future__ import annotations

from datetime import datetime, timezone
import json
from pathlib import Path
from typing import Any


REQUIRED_AXES = {
    "thai_explain",
    "low_level_code",
    "ffi_systems",
    "math_contract",
    "eval_theory",
    "tool_schema",
    "grounding",
    "self_check",
}

AXIS_ALIASES = {
    "thai_explain": {"thai_explain", "thai_natural", "natural_dialogue"},
    "low_level_code": {"low_level_code", "code_systems"},
    "ffi_systems": {"ffi_systems", "code_systems", "security_boundary"},
    "math_contract": {"math_contract", "math_reasoning"},
    "eval_theory": {"eval_theory", "self_assessment", "data_purity"},
    "tool_schema": {"tool_schema", "tool_grounding", "agent_planning"},
    "grounding": {"grounding", "world_context", "data_purity", "retrieval_memory"},
    "self_check": {"self_check", "self_assessment"},
}


def _load(path: str | Path | None) -> dict[str, Any]:
    if not path:
        return {}
    p = Path(path)
    return json.loads(p.read_text(encoding="utf-8")) if p.exists() else {}


def _num(payload: dict[str, Any], *keys: str, default: float = 0.0) -> float:
    cur: Any = payload
    for key in keys:
        if not isinstance(cur, dict) or key not in cur:
            return default
        cur = cur[key]
    try:
        return float(cur)
    except (TypeError, ValueError):
        return default


def _bool(payload: dict[str, Any], *keys: str) -> bool:
    cur: Any = payload
    for key in keys:
        if not isinstance(cur, dict) or key not in cur:
            return False
        cur = cur[key]
    return cur is True


def _score_from_contamination(contamination: dict[str, Any]) -> float:
    if not contamination:
        return 0.0
    risk = _num(contamination, "metrics", "contamination_risk", default=_num(contamination, "contamination_risk", default=100.0))
    exact = _num(contamination, "metrics", "exact_prompt_hits", default=1.0)
    high = _num(contamination, "metrics", "high_overlap_hits", default=1.0)
    cleared = _bool(contamination, "claim_gate", "contamination_cleared")
    if cleared and risk <= 0 and exact == 0 and high == 0:
        return 100.0
    return max(0.0, min(100.0, 100.0 - risk - exact * 10.0 - high * 2.0))


def _coverage_score(manifest: dict[str, Any]) -> tuple[float, list[str], list[str]]:
    axes = {str(axis) for axis in manifest.get("axes", [])}
    covered = sorted(axis for axis, aliases in AXIS_ALIASES.items() if axes & aliases)
    missing = sorted(REQUIRED_AXES - set(covered))
    score = 100.0 * len(covered) / max(1, len(REQUIRED_AXES))
    return score, covered, missing


def _raw_baseline_score(raw_probe: dict[str, Any]) -> float:
    native = _num(raw_probe, "summary", "native_score", default=_num(raw_probe, "native", "score", default=0.0))
    baseline = _num(raw_probe, "summary", "baseline_score", default=_num(raw_probe, "baseline", "score", default=0.0))
    if baseline <= 0:
        return 0.0
    return min(100.0, 100.0 * native / baseline)


def _broad_score(broad_probe: dict[str, Any]) -> float:
    return _num(broad_probe, "percent", default=0.0)


def _runtime_flexibility_score(tool_qa: dict[str, Any]) -> float:
    checks = tool_qa.get("checks", {})
    if not isinstance(checks, dict):
        checks = {}
    check_values = [value is True for value in checks.values()]
    check_score = 100.0 * sum(check_values) / max(1, len(check_values))
    gate_bonus = 100.0 if _bool(tool_qa, "claim_gate", "tool_augmented_qa_ready") else 0.0
    return (check_score * 0.6) + (gate_bonus * 0.4)


def _collapse_penalty(train_report: dict[str, Any]) -> float:
    collapse = train_report.get("collapse_check", {})
    if not isinstance(collapse, dict):
        return 0.0
    return 25.0 if collapse.get("fixed_answer_collapse_detected") is True else 0.0


def build_holistic_purity_flexibility_report(
    out_dir: str | Path,
    *,
    dataset_manifest: str | Path | None = None,
    contamination_report: str | Path | None = None,
    train_report: str | Path | None = None,
    raw_probe_report: str | Path | None = None,
    broad_probe_report: str | Path | None = None,
    tool_qa_report: str | Path | None = None,
    leaderboard_safe_report: str | Path | None = None,
) -> dict[str, Any]:
    manifest = _load(dataset_manifest)
    contamination = _load(contamination_report)
    train = _load(train_report)
    raw = _load(raw_probe_report)
    broad = _load(broad_probe_report)
    tool_qa = _load(tool_qa_report)
    leaderboard = _load(leaderboard_safe_report)

    purity_score = _score_from_contamination(contamination)
    coverage_score, covered_axes, missing_axes = _coverage_score(manifest)
    runtime_score = _runtime_flexibility_score(tool_qa)
    raw_score = _raw_baseline_score(raw)
    broad_score = _broad_score(broad)
    collapse_penalty = _collapse_penalty(train)
    overall = (
        purity_score * 0.32
        + coverage_score * 0.24
        + runtime_score * 0.18
        + raw_score * 0.14
        + broad_score * 0.12
        - collapse_penalty
    )
    overall = max(0.0, min(100.0, overall))

    clean_data_ready = (
        purity_score >= 99.0
        and coverage_score >= 99.0
        and _bool(contamination, "claim_gate", "leaderboard_safe_training_allowed")
        and (
            _bool(manifest, "claim_gate", "designed_for_leaderboard_safe_transfer")
            or _bool(manifest, "claim_gate", "designed_for_targeted_omni_round_training")
        )
        and _bool(manifest, "claim_gate", "contains_exact_probe_prompts") is False
    )
    runtime_ready = runtime_score >= 95.0 and _bool(tool_qa, "claim_gate", "tool_augmented_qa_ready")
    raw_complete = raw_score >= 100.0 and broad_score >= 90.0 and collapse_penalty == 0.0
    leaderboard_ready = _bool(leaderboard, "claim_gate", "leaderboard_safe_improvement_ready")

    report = {
        "schema_version": "tinymind-holistic-purity-flexibility-v1",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "inputs": {
            "dataset_manifest": str(dataset_manifest) if dataset_manifest else None,
            "contamination_report": str(contamination_report) if contamination_report else None,
            "train_report": str(train_report) if train_report else None,
            "raw_probe_report": str(raw_probe_report) if raw_probe_report else None,
            "broad_probe_report": str(broad_probe_report) if broad_probe_report else None,
            "tool_qa_report": str(tool_qa_report) if tool_qa_report else None,
            "leaderboard_safe_report": str(leaderboard_safe_report) if leaderboard_safe_report else None,
        },
        "scores": {
            "purity_no_junk_input_score": round(purity_score, 6),
            "domain_coverage_score": round(coverage_score, 6),
            "runtime_flexibility_score": round(runtime_score, 6),
            "raw_vs_baseline_score": round(raw_score, 6),
            "broad_raw_capability_score": round(broad_score, 6),
            "collapse_penalty": round(collapse_penalty, 6),
            "overall_holistic_score": round(overall, 6),
        },
        "coverage": {
            "required_axes": sorted(REQUIRED_AXES),
            "covered_axes": covered_axes,
            "missing_axes": missing_axes,
        },
        "diagnostics": {
            "fixed_answer_collapse_detected": train.get("collapse_check", {}).get("fixed_answer_collapse_detected"),
            "raw_model_still_needs_training": not raw_complete,
            "runtime_tool_layer_is_not_raw_model_capability": True,
            "absolute_weight_junk_free_claim_supported": False,
        },
        "claim_gate": {
            "clean_training_input_ready": clean_data_ready,
            "flexible_tool_augmented_runtime_ready": runtime_ready,
            "holistic_data_runtime_stack_ready": clean_data_ready and runtime_ready,
            "raw_model_complete_all_domains": raw_complete,
            "leaderboard_safe_promotion_allowed": leaderboard_ready,
            "world_best_claim_allowed": False,
            "reason": (
                "Clean input and runtime flexibility can be locally audited. "
                "Raw model completeness requires uncontrolled probe wins without fixed-answer collapse; "
                "world-best claims require external official evidence."
            ),
        },
    }
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    json_path = out / "holistic_purity_flexibility_report.json"
    md_path = out / "holistic_purity_flexibility_report.md"
    report["json_path"] = str(json_path)
    report["markdown_path"] = str(md_path)
    json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    md_path.write_text(_markdown(report), encoding="utf-8")
    return report


def _markdown(report: dict[str, Any]) -> str:
    lines = [
        "# TinyMind Holistic Purity/Flexibility Gate",
        "",
        "## Scores",
        "",
    ]
    for key, value in report["scores"].items():
        lines.append(f"- {key}: {value}")
    lines.extend(
        [
            "",
            "## Gates",
            "",
        ]
    )
    for key, value in report["claim_gate"].items():
        if key != "reason":
            lines.append(f"- {key}: {value}")
    lines.extend(
        [
            "",
            "## Coverage",
            "",
            f"- covered_axes: {', '.join(report['coverage']['covered_axes'])}",
            f"- missing_axes: {', '.join(report['coverage']['missing_axes']) or 'none'}",
            "",
            f"Reason: {report['claim_gate']['reason']}",
        ]
    )
    return "\n".join(lines) + "\n"