code/self_healing_finetuning/approvals/proof_chain.py

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from eval.model_quality_gate import evaluate_model_quality_gate


def _read_json(path: Path, errors: list[str]) -> dict[str, Any] | None:
    if not path.exists():
        errors.append(f"missing proof artifact: {path}")
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8-sig"))
    except json.JSONDecodeError as exc:
        errors.append(f"invalid proof artifact JSON: {path}: {exc}")
        return None


def _roadmap_errors(label: str, evidence: dict[str, Any] | None) -> list[str]:
    if not evidence:
        return [f"{label} missing model_roadmap_evidence"]
    errors: list[str] = []
    if evidence.get("known_to_roadmap") is not True:
        errors.append(f"{label} model is not known to roadmap")
    if evidence.get("weighted_score") is None:
        errors.append(f"{label} missing weighted_score")
    if evidence.get("score_rank") is None:
        errors.append(f"{label} missing score_rank")
    if evidence.get("context_tokens") is None:
        errors.append(f"{label} missing context_tokens")
    return errors


def validate_promotion_proof_chain(run_dir: Path) -> dict[str, Any]:
    errors: list[str] = []
    warnings: list[str] = []
    cycle_path = run_dir / "heal_decisions" / "cycle_summary.json"
    cert_path = run_dir / "eval" / "certification_report.json"
    paired_eval_path = run_dir / "eval" / "paired_eval_report.json"
    training_plan_path = run_dir / "remote_artifacts" / "training_plan.json"
    training_result_path = run_dir / "remote_artifacts" / "training_result.json"
    trainer_metrics_summary_path = run_dir / "remote_artifacts" / "trainer_metrics_summary.json"
    selected_checkpoint_path = run_dir / "remote_artifacts" / "selected_checkpoint.json"
    dataset_manifest_path = run_dir / "dataset_snapshot" / "dataset_manifest.json"
    model_judge_path = run_dir / "eval" / "model_judge_report.json"
    human_review_path = run_dir / "eval" / "human_spot_check_report.json"

    cycle_summary = _read_json(cycle_path, errors)
    certification = _read_json(cert_path, errors)
    if cycle_summary is None or certification is None:
        return {"ok": False, "errors": errors, "warnings": warnings, "evidence_summary": {}}

    cycle_roadmap = cycle_summary.get("model_roadmap_evidence")
    cert_roadmap = certification.get("model_roadmap_evidence")
    errors.extend(_roadmap_errors("cycle_summary", cycle_roadmap))
    errors.extend(_roadmap_errors("certification_report", cert_roadmap))

    cycle_model = cycle_summary.get("model_candidate")
    cert_model = cert_roadmap.get("model_id") if isinstance(cert_roadmap, dict) else None
    if cycle_model and cert_model and cycle_model != cert_model:
        errors.append(f"model mismatch between cycle summary and certification: {cycle_model} != {cert_model}")

    if certification.get("gate_result") != "pass":
        errors.append(f"certification gate is not pass: {certification.get('gate_result')}")

    paired_eval = _read_json(paired_eval_path, errors)
    training_plan = _read_json(training_plan_path, errors)
    training_result = _read_json(training_result_path, errors)
    trainer_metrics_summary = _read_json(trainer_metrics_summary_path, errors)
    selected_checkpoint = _read_json(selected_checkpoint_path, errors)
    dataset_manifest = _read_json(dataset_manifest_path, errors)
    model_judge_report = _read_json(model_judge_path, errors)
    human_review_report = _read_json(human_review_path, errors)
    model_quality_gate: dict[str, Any] | None = None
    if paired_eval is not None:
        if paired_eval.get("scoring_mode") != "deterministic_heuristic_v0":
            warnings.append(f"unexpected deterministic paired eval scoring mode: {paired_eval.get('scoring_mode')}")
        model_quality_gate = evaluate_model_quality_gate(
            paired_eval=paired_eval,
            training_plan=training_plan,
            training_result=training_result,
            trainer_metrics_summary=trainer_metrics_summary,
            selected_checkpoint=selected_checkpoint,
            dataset_manifest=dataset_manifest,
            model_judge_report=model_judge_report,
            human_review_report=human_review_report,
        )
        if model_quality_gate.get("eligible_for_promotion") is not True:
            errors.append("mandatory model-quality gate is not eligible")
            errors.extend(model_quality_gate.get("errors", []))
        warnings.extend(model_quality_gate.get("warnings", []))

    cycles = cycle_summary.get("cycles", [])
    if not cycles:
        errors.append("cycle summary has no cycles")
        latest_cycle: dict[str, Any] = {}
    else:
        latest_cycle = cycles[-1]
        if not any(cycle.get("gate_result") in {"pass", "fixture_pass"} for cycle in cycles):
            errors.append("no self-healing cycle reached paired-eval readiness")

    evidence_path = latest_cycle.get("evidence_path")
    iteration_evidence: dict[str, Any] | None = None
    if evidence_path:
        iteration_evidence = _read_json(Path(evidence_path), errors)
    else:
        errors.append("latest cycle missing evidence_path")

    if iteration_evidence:
        errors.extend(_roadmap_errors("iteration_evidence", iteration_evidence.get("model_roadmap_evidence")))
        scoring = iteration_evidence.get("scoring", {})
        if scoring.get("quality_signal") == "orchestration_only":
            warnings.append("iteration evidence is fixture/orchestration-only and cannot promote by itself")
        elif iteration_evidence.get("promotion_recommendation") != "promote_to_stage":
            errors.append(f"iteration promotion recommendation is not promote_to_stage: {iteration_evidence.get('promotion_recommendation')}")

    improvement_report = certification.get("improvement_report", {})
    improvements = improvement_report.get("improvements", {})
    aggregate = improvements.get("aggregate", {})
    thresholds = certification.get("thresholds", {})
    improvement_gates = thresholds.get("improvement_gates", {})
    min_abs = improvement_gates.get("aggregate_score_delta_min_abs")
    min_pct = improvement_gates.get("aggregate_score_delta_min_pct")
    if min_abs is not None and aggregate.get("abs", 0) < min_abs:
        errors.append(f"aggregate improvement abs below promotion threshold: {aggregate.get('abs')} < {min_abs}")
    if min_pct is not None and aggregate.get("pct", 0) < min_pct:
        errors.append(f"aggregate improvement pct below promotion threshold: {aggregate.get('pct')} < {min_pct}")

    if not (run_dir / "manifests" / "deployment_manifest.json").exists():
        warnings.append("deployment manifest missing; promotion remains a plan, not a live deployment action")

    evidence_summary = {
        "model_candidate": cycle_model,
        "weighted_score": cycle_roadmap.get("weighted_score") if isinstance(cycle_roadmap, dict) else None,
        "score_rank": cycle_roadmap.get("score_rank") if isinstance(cycle_roadmap, dict) else None,
        "context_tokens": cycle_roadmap.get("context_tokens") if isinstance(cycle_roadmap, dict) else None,
        "cycles_completed": cycle_summary.get("cycles_completed"),
        "latest_cycle_gate_result": latest_cycle.get("gate_result"),
        "certification_gate_result": certification.get("gate_result"),
        "aggregate_improvement_abs": aggregate.get("abs"),
        "aggregate_improvement_pct": aggregate.get("pct"),
        "private_replay_improvement_pct": improvements.get("private_prompt_replay", {}).get("pct"),
        "paired_eval_promotion_eligible": paired_eval.get("promotion_gate", {}).get("eligible_for_promotion")
        if isinstance(paired_eval, dict)
        else None,
        "model_quality_promotion_eligible": model_quality_gate.get("eligible_for_promotion")
        if isinstance(model_quality_gate, dict)
        else None,
        "model_quality_signal": model_quality_gate.get("quality_signal") if isinstance(model_quality_gate, dict) else None,
        "paired_eval_candidate_aggregate": paired_eval.get("candidate", {}).get("aggregate") if isinstance(paired_eval, dict) else None,
        "paired_eval_candidate_critical_pass_rate": paired_eval.get("candidate", {}).get("critical_pass_rate")
        if isinstance(paired_eval, dict)
        else None,
        "certification_path": str(cert_path),
        "paired_eval_path": str(paired_eval_path),
        "training_plan_path": str(training_plan_path),
        "training_result_path": str(training_result_path),
        "trainer_metrics_summary_path": str(trainer_metrics_summary_path),
        "selected_checkpoint_path": str(selected_checkpoint_path),
        "dataset_manifest_path": str(dataset_manifest_path),
        "model_judge_path": str(model_judge_path),
        "human_review_path": str(human_review_path),
        "cycle_summary_path": str(cycle_path),
        "iteration_evidence_path": evidence_path,
    }
    return {
        "ok": not errors,
        "errors": errors,
        "warnings": warnings,
        "evidence_summary": evidence_summary,
    }