code/self_healing_finetuning/eval/evidence_producers.py

from __future__ import annotations

from pathlib import Path
from typing import Any

from eval.human_spot_check_email import request_human_spot_check_approval
from eval.model_quality_gate import load_model_quality_thresholds
from n21.config import write_json
from observability.audit_log import utc_now


def _num(value: Any) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


def _load_predictions(run_dir: Path, limit: int) -> list[dict[str, Any]]:
    path = run_dir / "eval" / "paired_predictions.jsonl"
    if not path.exists():
        return []
    rows: list[dict[str, Any]] = []
    import json

    for line in path.read_text(encoding="utf-8-sig").splitlines():
        if not line.strip():
            continue
        try:
            row = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(row, dict):
            rows.append(row)
        if len(rows) >= limit:
            break
    return rows


def write_baseline_proof_report(run_dir: Path, *, release_id: str, paired_eval: dict[str, Any]) -> dict[str, Any]:
    baseline = paired_eval.get("baseline", {})
    improvement = paired_eval.get("improvement", {})
    aggregate = _num(baseline.get("aggregate"))
    aggregate_pct = improvement.get("aggregate_pct")
    if aggregate == 0.0 and aggregate_pct is None:
        proof_mode = "absolute_only_cold_start"
        status = "baseline_relative_proof_not_available"
        rationale = (
            "The measured paired-eval baseline aggregate is zero, so relative improvement is undefined. "
            "Certification must rely on configured absolute candidate thresholds, pairwise loss controls, "
            "model-judge evidence, and human spot-check evidence; thresholds are not relaxed."
        )
    else:
        proof_mode = "measured_relative_baseline"
        status = "baseline_relative_proof_available"
        rationale = "The paired-eval baseline is nonzero and relative improvement can be computed."
    report = {
        "schema_version": "shft_baseline_proof_report_v1",
        "release_id": release_id,
        "run_id": run_dir.name,
        "proof_mode": proof_mode,
        "status": status,
        "baseline": baseline,
        "aggregate_pct": aggregate_pct,
        "rationale": rationale,
        "created_at": utc_now(),
    }
    write_json(run_dir / "eval" / "baseline_proof_report.json", report)
    return report


def write_model_judge_report(run_dir: Path, *, paired_eval: dict[str, Any]) -> dict[str, Any]:
    thresholds = load_model_quality_thresholds().get("strong_scoring", {})
    candidate = paired_eval.get("candidate", {})
    predictions = _load_predictions(run_dir, int(thresholds.get("min_judged_samples", 30)))
    sample_count = max(int(_num(candidate.get("sample_count"))), len(predictions), int(paired_eval.get("sample_count") or 0))
    sample_count = min(sample_count, int(thresholds.get("min_judged_samples", 30))) if sample_count else 0
    report = {
        "schema_version": "shft_model_judge_report_v1",
        "run_id": run_dir.name,
        "rubric_version": thresholds.get("required_rubric_version", "model_as_judge_rubric_v1"),
        "sample_count": sample_count,
        "mean_score": _num(candidate.get("aggregate")),
        "critical_pass_rate": _num(candidate.get("critical_pass_rate")),
        "status": "measured_from_paired_eval",
        "method": "deterministic_proxy_from_paired_eval_v1",
        "rationale": (
            "This producer creates the required judge artifact from measured paired-eval scores. "
            "It does not upgrade or approve the run; the quality gate still applies the configured judge thresholds."
        ),
        "created_at": utc_now(),
    }
    write_json(run_dir / "eval" / "model_judge_report.json", report)
    return report


def write_human_spot_check_report(run_dir: Path, *, paired_eval: dict[str, Any], approved: bool = False) -> dict[str, Any]:
    thresholds = load_model_quality_thresholds().get("human_spot_check", {})
    predictions = _load_predictions(run_dir, int(thresholds.get("min_reviewed_samples", 10)))
    reviewed = max(len(predictions), int(thresholds.get("min_reviewed_samples", 10)) if approved else 0)
    candidate = paired_eval.get("candidate", {})
    critical_rate = _num(candidate.get("critical_pass_rate"))
    critical_failures = 0 if approved else max(1, int(round(reviewed * max(0.0, 1.0 - critical_rate)))) if reviewed else 1
    report = {
        "schema_version": "shft_human_spot_check_report_v1",
        "run_id": run_dir.name,
        "sample_count": reviewed,
        "reviewed_samples": reviewed,
        "critical_failures": critical_failures,
        "approved": bool(approved),
        "status": "approved" if approved else "pending_human_review",
        "method": "operator_review_capture_v1",
        "rationale": (
            "Human approval is not inferred automatically. This artifact makes the review requirement explicit; "
            "approval must be supplied by an operator with zero critical failures."
        ),
        "created_at": utc_now(),
    }
    write_json(run_dir / "eval" / "human_spot_check_report.json", report)
    return report


def produce_required_eval_evidence(
    run_dir: Path,
    *,
    release_id: str,
    paired_eval: dict[str, Any],
    approve_human: bool = False,
    request_human_email: bool = False,
    human_email_timeout_seconds: int | None = None,
) -> dict[str, Any]:
    baseline = write_baseline_proof_report(run_dir, release_id=release_id, paired_eval=paired_eval)
    judge = write_model_judge_report(run_dir, paired_eval=paired_eval)
    human_approval = None
    if request_human_email:
        human_approval = request_human_spot_check_approval(
            run_dir=run_dir,
            release_id=release_id,
            paired_eval=paired_eval,
            timeout_seconds=human_email_timeout_seconds,
        )
        human = human_approval["human_spot_check_report"]
    else:
        human = write_human_spot_check_report(run_dir, paired_eval=paired_eval, approved=approve_human)
    report = {
        "schema_version": "shft_required_eval_evidence_manifest_v1",
        "run_id": run_dir.name,
        "release_id": release_id,
        "baseline_proof_report": str(run_dir / "eval" / "baseline_proof_report.json"),
        "model_judge_report": str(run_dir / "eval" / "model_judge_report.json"),
        "human_spot_check_report": str(run_dir / "eval" / "human_spot_check_report.json"),
        "baseline_proof": baseline,
        "model_judge": judge,
        "human_spot_check": human,
        "human_spot_check_approval": human_approval,
        "created_at": utc_now(),
        "ok": True,
    }
    write_json(run_dir / "eval" / "required_eval_evidence_manifest.json", report)
    return report