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

"""System-level coherence governor for TinyMind subsystems."""

from __future__ import annotations

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


DEFAULT_INPUTS = {
    "dataset": "reports/dataset_quality_governor_apex/dataset_quality_governor_manifest.json",
    "axiomflow": "reports/axiomflow/axiomflow_bench_report.json",
    "tensor_layer": "reports/tensor_layer_plan_1803/tensor_layer_plan.json",
    "champion": "reports/system_auto_tuner/champion_adapter.json",
    "runtime": "reports/runtime_selector/runtime_selector_report.json",
}


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


def build_system_coherence_governor(out_dir: str | Path, inputs: dict[str, str] | None = None) -> dict[str, Any]:
    inputs = inputs or dict(DEFAULT_INPUTS)
    reports = {name: _load(path) for name, path in inputs.items()}
    dataset = reports.get("dataset", {})
    axiomflow = reports.get("axiomflow", {})
    tensor_layer = reports.get("tensor_layer", {})
    champion = reports.get("champion", {})
    runtime = reports.get("runtime", {})

    dataset_gate = dataset.get("purity_intensity_gate", {})
    axiom_coherence = axiomflow.get("coherence_gate", {})
    intensity = axiomflow.get("intensity_pressure", {})
    tensor_stability = tensor_layer.get("stability_gate", {})
    tensor_feasibility = tensor_layer.get("feasibility_gate", {})
    champion_row = champion.get("champion", champion)

    blockers: list[str] = []
    if dataset.get("claim_gate", {}).get("train_allowed") is not True and dataset_gate.get("training_intensity_ready") is not True:
        blockers.append("dataset_training_intensity")
    if axiom_coherence.get("passed") is not True or axiom_coherence.get("zero_work_lanes"):
        blockers.append("axiomflow_zero_work")
    if intensity.get("anti_collapse_gate", {}).get("passed") is not True:
        blockers.append("route_collapse")
    if intensity.get("lane_diversity_gate", {}).get("passed") is not True:
        blockers.append("lane_duplication")
    if tensor_stability.get("passed") is not True:
        blockers.append("tensor_layer_instability")
    if champion_row.get("status") != "measured" or int(champion_row.get("eval_records", 0) or 0) < 128:
        blockers.append("champion_eval_not_credible")

    no_zero_work_passed = "axiomflow_zero_work" not in blockers and "lane_duplication" not in blockers
    flexibility_axes = {
        "bounded_memory": axiomflow.get("bounded_memory_gate", {}).get("passed") is True,
        "virtual_depth": int(tensor_layer.get("active_layers_tensor", 0) or 0) >= 1803,
        "apex_data_mix": dataset_gate.get("training_intensity_ready") is True,
        "runtime_optional": bool(runtime) or True,
    }
    stability_axes = {
        "forward_finite": axiomflow.get("forward_finite") is True,
        "backward_finite": axiomflow.get("backward_finite") is True,
        "tensor_stability": tensor_stability.get("passed") is True,
        "rtx3090_feasible": tensor_feasibility.get("rtx_3090_planning_safe") is True,
        "champion_eval_credible": "champion_eval_not_credible" not in blockers,
    }
    precision_axes = {
        "per_sample_loss_normalization": dataset.get("training_contract", {}).get("loss_normalization") == "per_sample_token_normalized",
        "route_anti_collapse": intensity.get("anti_collapse_gate", {}).get("passed") is True,
        "lane_diversity": intensity.get("lane_diversity_gate", {}).get("passed") is True,
        "tensor_stability_score": float(tensor_stability.get("stability_score", 0.0) or 0.0) >= 0.95,
    }

    report = {
        "schema_version": "tinymind-system-coherence-governor-v1",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "inputs": inputs,
        "coherence_gate": {
            "system_coherence_ready": not blockers,
            "no_zero_work_passed": no_zero_work_passed,
            "blockers": blockers,
            "zero_work_lanes": axiom_coherence.get("zero_work_lanes", []),
            "productive_lane_count": axiom_coherence.get("productive_lane_count", 0),
            "mutual_support_score": axiom_coherence.get("mutual_support_score", 0.0),
        },
        "flexibility_gate": {
            "passed": all(flexibility_axes.values()),
            "axes": flexibility_axes,
            "meaning": "System can move between data, memory, virtual depth, and runtime modes without relying on one brittle path.",
        },
        "stability_gate": {
            "passed": all(stability_axes.values()),
            "axes": stability_axes,
            "meaning": "Core measured components are finite, bounded, and have credible local eval state.",
        },
        "precision_gate": {
            "passed": all(precision_axes.values()),
            "axes": precision_axes,
            "meaning": "Training/data pressure and route/lane pressure are aligned against collapse and waste.",
        },
        "coordination_contract": {
            "data_to_training": "Apex governed data sets per-sample normalized loss weights before QLoRA/continued training.",
            "training_to_model": "AxiomFlow intensity pressure can be added as auxiliary loss for no-zero-work routing.",
            "model_to_memory": "Local exact window handles recent details while long context stays in exact ledger/retrieval, not full KV.",
            "depth_to_runtime": "Total Layers Tensor 1803 is virtualized through tensors_per_layer=44 and physical_layers=41.",
            "runtime_to_claims": "Claim gates remain conservative until raw/external benchmarks pass.",
        },
        "claim_gate": {
            "absolute_all_environment_claim_allowed": False,
            "world_best_claim_allowed": False,
            "reason": "The governor can enforce local coherence and reduce waste, but cannot prove every environment or world-best status without external evidence.",
        },
    }
    out = Path(out_dir)
    out.mkdir(parents=True, exist_ok=True)
    json_path = out / "system_coherence_governor_report.json"
    md_path = out / "system_coherence_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 _markdown(report: dict[str, Any]) -> str:
    lines = [
        "# TinyMind System Coherence Governor",
        "",
        f"- System coherence ready: {report['coherence_gate']['system_coherence_ready']}",
        f"- No-zero-work passed: {report['coherence_gate']['no_zero_work_passed']}",
        f"- Flexibility gate: {report['flexibility_gate']['passed']}",
        f"- Stability gate: {report['stability_gate']['passed']}",
        f"- Precision gate: {report['precision_gate']['passed']}",
        f"- World-best claim allowed: {report['claim_gate']['world_best_claim_allowed']}",
        "",
        "## Blockers",
        "",
    ]
    blockers = report["coherence_gate"]["blockers"]
    if blockers:
        lines.extend(f"- {row}" for row in blockers)
    else:
        lines.append("- None")
    lines.extend(["", "## Coordination Contract", ""])
    for key, value in report["coordination_contract"].items():
        lines.append(f"- {key}: {value}")
    return "\n".join(lines) + "\n"