tinymind-native-colab-handoff/bundle/evaluation/world_quality_governor.py

"""World quality governor for TinyMind.

This governor aggregates the strongest local evidence and refuses to collapse
protocol-wrapper success into unsupported world-best claims. It is the control
plane for pushing quality upward while keeping claim boundaries clean.
"""

from __future__ import annotations

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


DEFAULT_INPUTS = {
    "ultra_pure": "reports/ultra_pure_knowledge/ultra_pure_manifest.json",
    "ultra_lineage": "reports/ultra_pure_lineage/hyper_pure_lineage_graph.json",
    "world_class": "reports/world_class_eval/world_class_eval_report.json",
    "parity": "reports/gpt55_pro_parity/gpt55_pro_parity_report.json",
    "raw_external": "reports/raw_external_gate/raw_external_gate_report.json",
    "perfection": "reports/system_perfection_gate/system_perfection_gate.json",
    "adaptive": "reports/adaptive_alignment/adaptive_alignment_report.json",
    "core_gap": "reports/core_gap_closer/core_gap_closer_report.json",
    "memory": "reports/extreme_memory_10m/extreme_memory_report.json",
}


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


def _bool_score(value: bool) -> float:
    return 100.0 if value else 0.0


def _mean(values: list[float]) -> float:
    return sum(values) / max(len(values), 1)


def _min_score(scores: dict) -> float:
    values = [float(v) for v in scores.values()]
    return min(values) if values else 0.0


def _raw_quality(raw_external: dict) -> float:
    rows = raw_external.get("raw_rows", [])
    if not rows:
        return 0.0
    return _mean([float(row.get("raw_score", 0.0)) for row in rows])


def _external_quality(raw_external: dict) -> float:
    external = raw_external.get("external", {})
    coverage = external.get("source_coverage", {})
    axis_pass = external.get("axis_pass", {})
    coverage_score = 100.0 * _mean([1.0 if ok else 0.0 for ok in coverage.values()]) if coverage else 0.0
    axis_score = 100.0 * _mean([1.0 if ok else 0.0 for ok in axis_pass.values()]) if axis_pass else 0.0
    return _mean([coverage_score, axis_score])


def _purity_score(ultra: dict) -> float:
    input_rows = int(ultra.get("input_rows", 0) or 0)
    kept_rows = int(ultra.get("kept_rows", 0) or 0)
    blocked_rows = int(ultra.get("blocked_rows", 0) or 0)
    if ultra.get("gate", {}).get("passed") is True and blocked_rows == 0 and input_rows > 0:
        return 100.0
    return 100.0 * kept_rows / max(input_rows, 1)


def _lineage_score(lineage: dict) -> float:
    gate = lineage.get("gate", {})
    has_fragments = int(lineage.get("fragment_count", 0) or 0) > 0
    return _bool_score(gate.get("passed") is True and has_fragments)


def build_world_quality_governor(out_dir: str | Path, inputs: dict[str, str] | None = None) -> dict:
    inputs = inputs or dict(DEFAULT_INPUTS)
    reports = {name: _load(path) for name, path in inputs.items()}
    ultra = reports["ultra_pure"]
    lineage = reports["ultra_lineage"]
    world_class = reports["world_class"]
    parity = reports["parity"]
    raw_external = reports["raw_external"]
    perfection = reports["perfection"]
    adaptive = reports["adaptive"]
    core_gap = reports["core_gap"]
    memory = reports["memory"]

    parity_summary = parity.get("summary", {})
    purity_score = _purity_score(ultra)
    lineage_score = _lineage_score(lineage)
    protocol_alignment_score = _min_score(adaptive.get("scores", {}))
    core_gap_score = _min_score(core_gap.get("scores", {}))
    parity_score = float(parity_summary.get("parity_percent", 0.0))
    frontier_plus_score = float(parity_summary.get("frontier_plus_percent", 0.0))
    raw_quality_score = _raw_quality(raw_external)
    external_quality_score = _external_quality(raw_external)
    world_class_score = float(world_class.get("summary", {}).get("balanced_score", 0.0))
    operational_score = _bool_score(perfection.get("operational_integrity", {}).get("passed") is True)
    memory_score = _bool_score(
        memory.get("passkey_recall", {}).get("passed") is True
        and int(memory.get("measured_tokens", 0) or 0) >= 10_000_000
        and int(memory.get("kv_tokens_stored", 1)) == 0
    )

    dimensions = [
        {
            "axis": "ultra_pure_dataset",
            "score": purity_score,
            "scope": "local_data_integrity",
            "passed": ultra.get("gate", {}).get("passed") is True and purity_score >= 99.0,
        },
        {
            "axis": "lineage_traceability",
            "score": lineage_score,
            "scope": "local_provenance_graph",
            "passed": lineage_score >= 100.0,
        },
        {
            "axis": "protocol_alignment",
            "score": protocol_alignment_score,
            "scope": adaptive.get("scope", "system_protocol"),
            "passed": adaptive.get("claim_gate", {}).get("system_protocol_alignment_claim_allowed") is True,
        },
        {
            "axis": "core_gap_protocol",
            "score": core_gap_score,
            "scope": "system_protocol",
            "passed": core_gap.get("claim_gate", {}).get("system_protocol_gap_claim_allowed") is True,
        },
        {
            "axis": "frontier_parity_protocol",
            "score": parity_score,
            "scope": "mixed_local_and_protocol_gap_analysis",
            "passed": parity.get("claim_gate", {}).get("can_claim_gpt55_pro_equivalent") is True,
        },
        {
            "axis": "frontier_plus_protocol",
            "score": frontier_plus_score,
            "scope": "mixed_local_and_protocol_gap_analysis",
            "passed": parity.get("claim_gate", {}).get("can_claim_beyond_frontier") is True,
        },
        {
            "axis": "world_class_local_eval",
            "score": world_class_score,
            "scope": "local_eval_packet",
            "passed": world_class.get("claim_gate", {}).get("ready_for_world_top_comparison") is True,
        },
        {
            "axis": "raw_external_truth",
            "score": raw_quality_score,
            "scope": "raw_model_and_official_external",
            "passed": raw_external.get("claim_gate", {}).get("raw_external_gate_complete") is True,
        },
        {
            "axis": "official_external_coverage",
            "score": external_quality_score,
            "scope": "hf_lmarena_artificial_analysis",
            "passed": raw_external.get("claim_gate", {}).get("external_gate_passed") is True,
        },
        {
            "axis": "ten_million_exact_memory",
            "score": memory_score,
            "scope": "exact_archive_recall",
            "passed": memory_score >= 100.0,
        },
        {
            "axis": "operational_integrity",
            "score": operational_score,
            "scope": "local_system_gate",
            "passed": perfection.get("operational_integrity", {}).get("passed") is True,
        },
    ]
    local_system_quality_score = _mean([purity_score, lineage_score, protocol_alignment_score, core_gap_score, memory_score, operational_score])
    protocol_quality_score = _mean([protocol_alignment_score, core_gap_score, parity_score, frontier_plus_score])
    frontier_readiness_score = _mean([parity_score, frontier_plus_score, raw_quality_score, external_quality_score])
    weighted_score = _mean([row["score"] for row in dimensions])
    hard_blockers = [
        row["axis"]
        for row in dimensions
        if not row["passed"]
        and row["axis"] in {"raw_external_truth", "official_external_coverage", "frontier_parity_protocol", "frontier_plus_protocol"}
    ]
    local_exclusions = {"raw_external_truth", "official_external_coverage", "frontier_parity_protocol", "frontier_plus_protocol"}
    world_claim_allowed = (
        raw_external.get("claim_gate", {}).get("raw_external_gate_complete") is True
        and parity.get("claim_gate", {}).get("can_claim_beyond_frontier") is True
    )
    report = {
        "schema_version": "tinymind-world-quality-governor-v1",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "inputs": inputs,
        "dimensions": dimensions,
        "scores": {
            "local_system_quality_score": local_system_quality_score,
            "protocol_quality_score": protocol_quality_score,
            "raw_quality_score": raw_quality_score,
            "external_quality_score": external_quality_score,
            "purity_score": purity_score,
            "lineage_score": lineage_score,
            "frontier_readiness_score": frontier_readiness_score,
            "world_quality_score": weighted_score,
        },
        "world_quality_score": weighted_score,
        "hard_blockers": hard_blockers,
        "claim_gate": {
            "local_quality_stack_ready": all(row["passed"] for row in dimensions if row["axis"] not in local_exclusions),
            "world_current_quality_claim_allowed": world_claim_allowed,
            "world_best_current_claim_allowed": world_claim_allowed,
            "absolute_flawless_claim_allowed": perfection.get("perfection_claim", {}).get("absolute_100_percent_error_free_allowed") is True,
            "reason": "Current-world highest-quality claims require raw/external completion and official independent results.",
        },
        "next_actions": _next_actions(dimensions),
    }
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    json_path = out / "world_quality_governor_report.json"
    md_path = out / "world_quality_governor_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), encoding="utf-8")
    md_path.write_text(_markdown(report), encoding="utf-8")
    return report


def _next_actions(dimensions: list[dict]) -> list[str]:
    failed = {row["axis"] for row in dimensions if not row["passed"]}
    actions = []
    if "raw_external_truth" in failed:
        actions.append("Run raw model official-style benchmarks and import HF/LMArena/Artificial Analysis results.")
    if "official_external_coverage" in failed:
        actions.append("Attach independent public results from Hugging Face, LMArena, and Artificial Analysis.")
    if "frontier_parity_protocol" in failed:
        actions.append("Close remaining parity axes, especially natural_answer_style, with raw and protocol evidence split.")
    if "frontier_plus_protocol" in failed:
        actions.append("Raise every passed parity axis above frontier+ target and attach external evidence.")
    if not actions:
        actions.append("Maintain continuous regression monitoring and refresh external evidence by date.")
    return actions


def _markdown(report: dict) -> str:
    lines = [
        "# TinyMind World Quality Governor",
        "",
        f"- World quality score: {report['world_quality_score']:.2f}",
        f"- Local system quality score: {report['scores']['local_system_quality_score']:.2f}",
        f"- Frontier readiness score: {report['scores']['frontier_readiness_score']:.2f}",
        f"- Raw quality score: {report['scores']['raw_quality_score']:.2f}",
        f"- External quality score: {report['scores']['external_quality_score']:.2f}",
        f"- Local quality stack ready: {report['claim_gate']['local_quality_stack_ready']}",
        f"- Current world quality claim allowed: {report['claim_gate']['world_current_quality_claim_allowed']}",
        "",
        "## Dimensions",
        "",
        "| Axis | Score | Passed | Scope |",
        "|---|---:|---|---|",
    ]
    for row in report["dimensions"]:
        lines.append(f"| {row['axis']} | {row['score']:.2f} | {row['passed']} | {row['scope']} |")
    lines.extend(["", "## Next Actions", ""])
    for action in report["next_actions"]:
        lines.append(f"- {action}")
    return "\n".join(lines) + "\n"