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

from __future__ import annotations

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


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


def _sha256(path: str | Path) -> str:
    h = hashlib.sha256()
    with Path(path).open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def _artifact(path: str | Path | None) -> dict[str, Any] | None:
    if not path:
        return None
    p = Path(path)
    if not p.exists():
        return {"path": str(p), "exists": False}
    return {
        "path": str(p),
        "exists": True,
        "bytes": p.stat().st_size,
        "sha256": _sha256(p),
    }


def _provider_matrix(provider_report: dict[str, Any]) -> list[dict[str, Any]]:
    models = provider_report.get("external_eval", {}).get("models", [])
    matrix = []
    for model in models:
        rows = model.get("rows", [])
        errors = [row.get("error") for row in rows if row.get("error")]
        matrix.append(
            {
                "model_id": model.get("model_id"),
                "access_verified": len(rows) > 0 and not errors,
                "probe_score": model.get("score"),
                "probe_count": len(rows),
                "error_count": len(errors),
                "axes": [
                    {
                        "axis": row.get("axis"),
                        "score": row.get("score"),
                        "has_error": "error" in row,
                    }
                    for row in rows
                ],
            }
        )
    return matrix


def _stress_summary(soak_report: dict[str, Any]) -> dict[str, Any]:
    stress = soak_report.get("stress", {})
    soak = stress.get("soak", {})
    latency = soak.get("latency_ms", {})
    rss = soak.get("rss_mb", {})
    gpu = soak.get("gpu_memory_used_mb", {})
    return {
        "local_stress_passed": soak_report.get("claim_gate", {}).get("local_stress_passed"),
        "axiomkv_seq_lengths": stress.get("axiomkv_seq_lengths", []),
        "axiomkv_cached_token_capacity": stress.get("axiomkv_cached_token_capacity"),
        "regenesis_loops": stress.get("regenesis_dll_to_kv", {}).get("loops"),
        "kv_tokens_stored": stress.get("regenesis_dll_to_kv", {}).get("kv_tokens_stored"),
        "soak_enabled": soak.get("enabled"),
        "duration_actual_s": soak.get("duration_actual_s"),
        "iterations": soak.get("iterations"),
        "all_iterations_passed": soak.get("all_iterations_passed"),
        "latency_ms": latency,
        "rss_mb": rss,
        "gpu_memory_used_mb": gpu,
    }


def build_stress_provider_evidence(
    out_dir: str | Path,
    *,
    provider_report_path: str | Path,
    soak_report_path: str | Path,
    provider_command: str = "",
    soak_command: str = "",
) -> dict[str, Any]:
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    provider_path = Path(provider_report_path)
    soak_path = Path(soak_report_path)
    provider_report = _load_json(provider_path)
    soak_report = _load_json(soak_path)
    matrix = _provider_matrix(provider_report)
    stress = _stress_summary(soak_report)
    provider_ready = (
        provider_report.get("external_eval", {}).get("status") == "completed"
        and len(matrix) >= 3
        and sum(1 for row in matrix if row["access_verified"]) >= 3
    )
    duration = float(stress.get("duration_actual_s") or 0.0)
    latency_drift = stress.get("latency_ms", {}).get("drift")
    rss_drift = stress.get("rss_mb", {}).get("drift")
    gpu_drift = stress.get("gpu_memory_used_mb", {}).get("drift")
    soak_ready = (
        bool(stress.get("local_stress_passed"))
        and bool(stress.get("all_iterations_passed"))
        and duration >= 30 * 60
        and (rss_drift is None or float(rss_drift) <= 64.0)
        and (gpu_drift is None or float(gpu_drift) <= 256.0)
        and (latency_drift is None or float(latency_drift) <= 100.0)
    )
    samples_path = soak_report.get("stress", {}).get("soak", {}).get("samples_path")
    report = {
        "schema": "tinymind.stress_provider_access_evidence.v1",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "source_reports": {
            "provider": _artifact(provider_path),
            "soak": _artifact(soak_path),
            "soak_samples": _artifact(samples_path),
        },
        "commands": {
            "provider_probe": provider_command,
            "stress_soak": soak_command,
        },
        "provider_access": {
            "provider_kind": provider_report.get("external_eval", {}).get("provider_kind"),
            "provider": provider_report.get("external_eval", {}).get("provider"),
            "status": provider_report.get("external_eval", {}).get("status"),
            "models_tested": len(matrix),
            "models_access_verified": sum(1 for row in matrix if row["access_verified"]),
            "matrix": matrix,
        },
        "stress_soak": stress,
        "evidence_gate": {
            "provider_access_evidence_ready": provider_ready,
            "stress_soak_evidence_ready": soak_ready,
            "stress_provider_access_evidence_ready": bool(provider_ready and soak_ready),
            "official_external_claim_allowed": False,
            "world_best_claim_allowed": False,
            "reason": "This dossier proves provider access and local stress/soak behavior only. It is not an official leaderboard submission or rank.",
        },
    }
    json_path = out / "stress_provider_access_evidence.json"
    md_path = out / "stress_provider_access_evidence.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:
    gate = report["evidence_gate"]
    provider = report["provider_access"]
    stress = report["stress_soak"]
    lines = [
        "# TinyMind Stress / Provider Access Evidence",
        "",
        f"- Created: `{report['created_at']}`",
        f"- Provider access evidence ready: `{gate['provider_access_evidence_ready']}`",
        f"- Stress soak evidence ready: `{gate['stress_soak_evidence_ready']}`",
        f"- Combined evidence ready: `{gate['stress_provider_access_evidence_ready']}`",
        f"- Official/world-best claims: `blocked`",
        "",
        "## Provider Access",
        "",
        f"- Provider kind: `{provider.get('provider_kind')}`",
        f"- Status: `{provider.get('status')}`",
        f"- Models tested: `{provider.get('models_tested')}`",
        f"- Models access verified: `{provider.get('models_access_verified')}`",
        "",
        "| Model | Access | Probe score | Errors |",
        "| --- | ---: | ---: | ---: |",
    ]
    for row in provider["matrix"]:
        lines.append(
            f"| `{row['model_id']}` | `{row['access_verified']}` | `{row['probe_score']}` | `{row['error_count']}` |"
        )
    lines.extend(
        [
            "",
            "## Stress Soak",
            "",
            f"- Duration: `{stress.get('duration_actual_s')}` seconds",
            f"- Iterations: `{stress.get('iterations')}`",
            f"- All iterations passed: `{stress.get('all_iterations_passed')}`",
            f"- AxiomKV seq lengths: `{stress.get('axiomkv_seq_lengths')}`",
            f"- KV tokens stored: `{stress.get('kv_tokens_stored')}`",
            f"- Latency drift ms: `{stress.get('latency_ms', {}).get('drift')}`",
            f"- RSS drift MB: `{stress.get('rss_mb', {}).get('drift')}`",
            f"- GPU memory drift MB: `{stress.get('gpu_memory_used_mb', {}).get('drift')}`",
            "",
            "## Hash Evidence",
            "",
        ]
    )
    for name, artifact in report["source_reports"].items():
        if artifact:
            lines.append(f"- `{name}`: `{artifact.get('path')}` sha256 `{artifact.get('sha256')}`")
    lines.extend(["", "## Claim Boundary", "", gate["reason"], ""])
    return "\n".join(lines)