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

from __future__ import annotations

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


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


def _exists(path: str | Path | None) -> bool:
    return bool(path) and Path(path).exists()


def _score(report: dict[str, Any]) -> dict[str, Any]:
    axes = report["evo_axes"]
    gates = report["evidence_gates"]
    purity = 1.0 if gates["purity_manifest_present"] else 0.0
    adapter = 1.0 if gates["adapter_manifest_present"] else 0.0
    web = 1.0 if gates["web_knowledge_present"] else 0.0
    gguf = 1.0 if gates["gguf_runtime_present"] else 0.0
    compaction = 1.0 if gates["compaction_present"] else 0.0
    curriculum = min(1.0, len(axes["mutation_curriculum"]) / 6.0)
    total = 0.22 * purity + 0.24 * adapter + 0.18 * web + 0.14 * gguf + 0.12 * compaction + 0.10 * curriculum
    return {
        "purity_axis": round(purity * 100, 3),
        "adapter_axis": round(adapter * 100, 3),
        "web_knowledge_axis": round(web * 100, 3),
        "gguf_runtime_axis": round(gguf * 100, 3),
        "compaction_axis": round(compaction * 100, 3),
        "curriculum_axis": round(curriculum * 100, 3),
        "evo_cross_species_score": round(total * 100, 3),
    }


def build_evo_cross_species_report(
    out_dir: str | Path,
    *,
    adapter_manifest: str | Path | None = None,
    data_manifest: str | Path | None = None,
    compaction_manifest: str | Path | None = None,
    compaction_probe_report: str | Path | None = None,
    gguf_manifest: str | Path | None = None,
    web_knowledge_report: str | Path | None = None,
    output_training_jsonl: str | Path | None = None,
) -> dict[str, Any]:
    adapter = _read_json(adapter_manifest)
    data = _read_json(data_manifest)
    compaction = _read_json(compaction_manifest)
    compaction_probe = _read_json(compaction_probe_report)
    gguf = _read_json(gguf_manifest)
    web = _read_json(web_knowledge_report)

    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    sft_path = Path(output_training_jsonl) if output_training_jsonl else out / "evo_cross_species_bridge_sft.jsonl"

    mutation_curriculum = [
        {
            "species": "weight_adapter",
            "skill": "preserve raw 20/20 probe behavior while learning new evidence access format",
            "source": str(adapter_manifest) if adapter_manifest else None,
        },
        {
            "species": "data_body",
            "skill": "reject noise, secrets, raw traces, and duplicate pressure before training",
            "source": str(data_manifest) if data_manifest else None,
        },
        {
            "species": "web_evidence",
            "skill": "search/fetch/hash/cache external knowledge before answering current facts",
            "source": str(web_knowledge_report) if web_knowledge_report else None,
        },
        {
            "species": "runtime_gguf",
            "skill": "use runtime law and decoding controls for grounded local inference",
            "source": str(gguf_manifest) if gguf_manifest else None,
        },
        {
            "species": "compression",
            "skill": "make smaller candidates only when quality-per-MB improves",
            "source": str(compaction_manifest) if compaction_manifest else None,
        },
        {
            "species": "claim_gate",
            "skill": "separate local proof, model-plus-runtime proof, and official external proof",
            "source": "internal_gate",
        },
    ]

    evidence_gates = {
        "adapter_manifest_present": _exists(adapter_manifest),
        "purity_manifest_present": _exists(data_manifest),
        "compaction_present": _exists(compaction_manifest),
        "gguf_runtime_present": _exists(gguf_manifest),
        "web_knowledge_present": _exists(web_knowledge_report),
        "training_jsonl_written": False,
    }

    report: dict[str, Any] = {
        "schema_version": "tinymind-evo-cross-species-v1",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "definition": "Evo cross-species means coordinated mutation across adapter weights, data body, retrieval/web evidence, GGUF runtime, compression, and claim gates without treating any single layer as magic.",
        "inputs": {
            "adapter_manifest": str(adapter_manifest) if adapter_manifest else None,
            "data_manifest": str(data_manifest) if data_manifest else None,
            "compaction_manifest": str(compaction_manifest) if compaction_manifest else None,
            "compaction_probe_report": str(compaction_probe_report) if compaction_probe_report else None,
            "gguf_manifest": str(gguf_manifest) if gguf_manifest else None,
            "web_knowledge_report": str(web_knowledge_report) if web_knowledge_report else None,
        },
        "evidence_gates": evidence_gates,
        "evo_axes": {
            "adapter_body": {
                "base_model": adapter.get("base_model"),
                "adapter": adapter.get("adapter"),
                "eval_loss": adapter.get("eval_loss"),
                "perplexity": adapter.get("perplexity"),
                "eval_credible": adapter.get("eval_credible"),
            },
            "data_body": {
                "kept_records": data.get("kept_records") or data.get("stats", {}).get("kept"),
                "rejected_records": data.get("rejected_records"),
                "domain_counts": data.get("domain_counts") or data.get("stats", {}).get("domain_counts"),
            },
            "web_body": {
                "status": web.get("status"),
                "evidence_count": len(web.get("evidence", [])) if web else 0,
                "answer_grounded": web.get("claim_gate", {}).get("answer_grounded"),
            },
            "gguf_body": {
                "model_name": gguf.get("model_name"),
                "context_window_requested": gguf.get("runtime_upgrade", {}).get("context_window_requested"),
                "can_claim_runtime_quality_upgrade": gguf.get("promotion_gate", {}).get("can_claim_runtime_quality_upgrade"),
            },
            "compression_body": {
                "source_mb": compaction.get("size", {}).get("source_mb"),
                "output_mb": compaction.get("size", {}).get("output_mb"),
                "reduction_ratio": compaction.get("size", {}).get("reduction_ratio"),
                "probe_score": max(
                    [int(row.get("total_score", 0)) for row in (compaction_probe.get("adapters") or {}).values()] or [0]
                ),
                "probe_max_score": max(
                    [int(row.get("max_score", 0)) for row in (compaction_probe.get("adapters") or {}).values()] or [0]
                ),
            },
            "mutation_curriculum": mutation_curriculum,
        },
        "claim_gate": {
            "evo_cross_species_pipeline_ready": False,
            "can_continue_training": False,
            "can_claim_species_jump": False,
            "world_best_claim_allowed": False,
            "reason": "All available species layers must have evidence before calling this a complete Evo jump; external official eval is still required for broad claims.",
        },
    }

    scores = _score(report)
    complete = (
        evidence_gates["adapter_manifest_present"]
        and evidence_gates["purity_manifest_present"]
        and evidence_gates["web_knowledge_present"]
    )
    report["scores"] = scores
    report["claim_gate"]["evo_cross_species_pipeline_ready"] = bool(complete)
    report["claim_gate"]["can_continue_training"] = bool(complete)
    report["claim_gate"]["can_claim_species_jump"] = scores["evo_cross_species_score"] >= 90 and bool(complete)

    with sft_path.open("w", encoding="utf-8", newline="\n") as f:
        for item in mutation_curriculum:
            f.write(
                json.dumps(
                    {
                        "source": "evo_cross_species_bridge",
                        "metadata": {"species": item["species"], "source": item["source"]},
                        "messages": [
                            {
                                "role": "system",
                                "content": "You are TinyMind Evo. Coordinate adapter, data, retrieval, runtime, compression, and claim gates. Never fake evidence.",
                            },
                            {"role": "user", "content": f"Activate Evo species layer: {item['species']}"},
                            {"role": "assistant", "content": f"{item['species']}: {item['skill']} Evidence source: {item['source'] or 'missing; keep gate closed'}."},
                        ],
                    },
                    ensure_ascii=False,
                    sort_keys=True,
                )
                + "\n"
            )
    report["evidence_gates"]["training_jsonl_written"] = True
    report["training_jsonl"] = str(sft_path)

    path = out / "evo_cross_species_report.json"
    md = out / "evo_cross_species_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:
    lines = [
        "# TinyMind Evo Cross-Species",
        "",
        f"- Score: {report['scores']['evo_cross_species_score']:.3f}",
        f"- Pipeline ready: {report['claim_gate']['evo_cross_species_pipeline_ready']}",
        f"- Continue training: {report['claim_gate']['can_continue_training']}",
        f"- Species-jump claim: {report['claim_gate']['can_claim_species_jump']}",
        f"- World-best claim: {report['claim_gate']['world_best_claim_allowed']}",
        "",
        "## Mutation Curriculum",
    ]
    for item in report["evo_axes"]["mutation_curriculum"]:
        lines.append(f"- {item['species']}: {item['skill']}")
    return "\n".join(lines) + "\n"