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

from __future__ import annotations

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


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


def _score(data_manifest: dict[str, Any], compaction_manifest: dict[str, Any]) -> dict[str, float]:
    kept = float(data_manifest.get("kept_records", 0))
    rejected = float(data_manifest.get("rejected_records", 0))
    total = kept + rejected
    purity_pressure = rejected / total if total else 0.0
    domain_count = len(data_manifest.get("domain_counts", {}) or {})
    reduction = float((compaction_manifest.get("size") or {}).get("reduction_ratio", 0.0))
    smaller = 1.0 if (compaction_manifest.get("claim_gate") or {}).get("smaller_adapter_created") else 0.0
    return {
        "purity_pressure": round(purity_pressure, 6),
        "domain_coverage_proxy": round(min(domain_count / 8.0, 1.0), 6),
        "size_reduction_ratio": round(max(0.0, reduction), 6),
        "smaller_artifact_gate": smaller,
        "whole_body_evo_score": round(0.35 * purity_pressure + 0.25 * min(domain_count / 8.0, 1.0) + 0.30 * max(0.0, reduction) + 0.10 * smaller, 6),
    }


def build_evo_whole_body_report(
    out_dir: str | Path,
    *,
    data_manifest: str | Path,
    compaction_manifest: str | Path | None = None,
    active_training_pid: int | None = None,
) -> dict[str, Any]:
    data = _load_json(data_manifest)
    compaction = _load_json(compaction_manifest) if compaction_manifest else {}
    scores = _score(data, compaction)
    report = {
        "schema_version": "tinymind-evo-whole-body-report-v1",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "goal": "make the system purer, smaller, and more capable per resource without overwriting evidence",
        "active_training_pid": active_training_pid,
        "inputs": {
            "data_manifest": str(data_manifest),
            "compaction_manifest": str(compaction_manifest) if compaction_manifest else None,
        },
        "scores": scores,
        "evo_axes": {
            "data_body": {
                "kept_records": data.get("kept_records", 0),
                "rejected_records": data.get("rejected_records", 0),
                "domain_counts": data.get("domain_counts", {}),
                "reject_counts": data.get("reject_counts", {}),
            },
            "model_body": {
                "adapter_compaction": bool(compaction),
                "source_mb": (compaction.get("size") or {}).get("source_mb", 0.0),
                "output_mb": (compaction.get("size") or {}).get("output_mb", 0.0),
                "target_rank": compaction.get("target_rank"),
            },
            "runtime_body": {
                "strategy": "QLoRA chain -> compact adapter candidate -> eval/drift gate -> promote only if quality per MB improves",
                "rtx3090_safe_queue": active_training_pid is not None,
            },
        },
        "next_required_evidence": [
            "eval_loss_perplexity_before_after",
            "generation_repetition_rate_before_after",
            "Thai-English and code holdout comparison",
            "adapter drift comparison",
            "VRAM and latency measurement",
        ],
        "claim_gate": {
            "whole_body_evo_packet_ready": True,
            "promote_compacted_adapter_allowed": False,
            "world_best_or_frontier_claim_allowed": False,
            "reason": "The report proves an Evo pipeline exists; promotion requires measured quality preservation or improvement.",
        },
    }
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    path = out / "evo_whole_body_report.json"
    md = out / "evo_whole_body_report.md"
    report["json_path"] = str(path)
    report["markdown_path"] = str(md)
    path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
    md.write_text(_markdown(report), encoding="utf-8")
    return report


def _markdown(report: dict[str, Any]) -> str:
    scores = report["scores"]
    lines = [
        "# TinyMind Evo Whole Body",
        "",
        f"- Whole-body Evo score: {scores['whole_body_evo_score']:.6f}",
        f"- Purity pressure: {scores['purity_pressure']:.6f}",
        f"- Size reduction ratio: {scores['size_reduction_ratio']:.6f}",
        f"- Promote compacted adapter: {report['claim_gate']['promote_compacted_adapter_allowed']}",
        f"- World-best/frontier claim: {report['claim_gate']['world_best_or_frontier_claim_allowed']}",
        "",
        "## Next Evidence",
    ]
    for item in report["next_required_evidence"]:
        lines.append(f"- {item}")
    return "\n".join(lines) + "\n"