code/self_healing_finetuning/reporting/run_improvement_summary.py

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any


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


def fmt_pct(value: Any) -> str:
    if value is None:
        return "n/a"
    try:
        return f"{float(value):.4f}%"
    except (TypeError, ValueError):
        return str(value)


def fmt_num(value: Any) -> str:
    if value is None:
        return "n/a"
    try:
        return f"{float(value):.4f}"
    except (TypeError, ValueError):
        return str(value)


def print_vital_kpis(prefix: str, *, baseline: dict[str, Any], candidate: dict[str, Any], improvement: dict[str, Any], model_quality: dict[str, Any]) -> None:
    print(
        f"{prefix}   "
        f"MODEL_QUALITY={str(model_quality.get('quality_signal', 'unknown')).upper()} "
        f"CERTIFIED={model_quality.get('eligible_for_promotion', model_quality.get('ok'))} "
        f"CANDIDATE_AGGREGATE={fmt_num(candidate.get('aggregate'))} "
        f"AGGREGATE_DELTA={fmt_num(improvement.get('aggregate_abs'))} "
        f"CRITICAL_PASS={fmt_num(candidate.get('critical_pass_rate'))} "
        f"CRITICAL_DELTA={fmt_num(improvement.get('critical_pass_rate_abs'))} "
        f"WIN_RATE={fmt_num(improvement.get('pairwise_win_rate'))} "
        f"LOSS_RATE={fmt_num(improvement.get('pairwise_loss_rate'))} "
        f"BASELINE_AGGREGATE={fmt_num(baseline.get('aggregate'))}"
    )


def print_cycle_summary(run_dir: Path) -> None:
    cycle_summary = load_json(run_dir / "heal_decisions" / "cycle_summary.json")
    print("[SHFT improvement summary] self-healing cycle evidence (fixture/orchestration only)")
    if not cycle_summary:
        print("[SHFT improvement summary]   no cycle_summary.json found")
        return
    cycles = cycle_summary.get("cycles", [])
    if not cycles:
        print("[SHFT improvement summary]   no cycles recorded")
        return
    for cycle in cycles:
        evidence_path = Path(str(cycle.get("evidence_path", "")))
        evidence = load_json(evidence_path) if evidence_path.exists() else None
        if not evidence:
            print(
                "[SHFT improvement summary]   "
                f"cycle={cycle.get('cycle')} iteration={cycle.get('iteration_id')} "
                f"gate={cycle.get('gate_result')} aggregate_delta_pct={fmt_pct(cycle.get('aggregate_improvement_pct'))} "
                f"private_replay_delta_pct={fmt_pct(cycle.get('private_replay_improvement_pct'))}"
            )
            continue
        current = evidence.get("scores", {}).get("current_iteration", {})
        previous_delta = evidence.get("improvement_vs_previous", {})
        prod_delta = evidence.get("improvement_vs_prod", {})
        print(
            "[SHFT improvement summary]   "
            f"cycle={evidence.get('cycle')} iteration={evidence.get('iteration_id')} fixture_gate={evidence.get('gate_result')} "
            f"quality_signal={evidence.get('scoring', {}).get('quality_signal', 'unknown')} "
            f"repair={evidence.get('repair_reason')} fixture_aggregate={fmt_num(current.get('aggregate'))} "
            f"delta_prev={fmt_pct(previous_delta.get('aggregate', {}).get('pct'))} "
            f"delta_prod={fmt_pct(prod_delta.get('aggregate', {}).get('pct'))} "
            f"private_replay_delta_prod={fmt_pct(prod_delta.get('private_prompt_replay', {}).get('pct'))}"
        )
    print("[SHFT improvement summary]   live_model_quality_artifact=eval/paired_eval_report.json status=pending_until_proof")


def print_training_summary(run_dir: Path) -> None:
    plan = load_json(run_dir / "remote_artifacts" / "training_plan.json")
    result = load_json(run_dir / "remote_artifacts" / "training_result.json")
    print("[SHFT improvement summary] live training evidence")
    if not plan and not result:
        print("[SHFT improvement summary]   no live training artifacts found")
        return
    if plan:
        readiness = plan.get("readiness", {})
        print(
            "[SHFT VITAL TRAINING]   "
            f"TRAIN_RECORDS={plan.get('train_records')} "
            f"VALID_RECORDS={plan.get('valid_records')} "
            f"MAX_STEPS={plan.get('hyperparameters', {}).get('max_steps')} "
            f"PRODUCTION_CANDIDATE={readiness.get('production_candidate', 'unknown')}"
        )
        print(
            "[SHFT improvement summary]   "
            f"base_model={plan.get('base_model_id')} model_candidate={plan.get('model_candidate')} "
            f"train_records={plan.get('train_records')} valid_records={plan.get('valid_records')} "
            f"max_steps={plan.get('hyperparameters', {}).get('max_steps')} "
            f"production_candidate={readiness.get('production_candidate', 'unknown')}"
        )
        for warning in readiness.get("warnings", []):
            print(f"[SHFT improvement summary]   WARNING {warning}")
    if result:
        print(
            "[SHFT VITAL TRAINING]   "
            f"TRAINING_STATUS={str(result.get('status')).upper()} "
            f"TRAIN_LOSS={fmt_num(result.get('train_loss'))}"
        )
        print(
            "[SHFT improvement summary]   "
            f"training_status={result.get('status')} train_loss={fmt_num(result.get('train_loss'))} "
            f"adapter_dir={result.get('adapter_dir')}"
        )


def print_paired_eval_summary(run_dir: Path) -> None:
    report = load_json(run_dir / "eval" / "paired_eval_report.json")
    print("[SHFT improvement summary] paired model-vs-model proof")
    if not report:
        print("[SHFT improvement summary]   no paired_eval_report.json found")
        return
    baseline = report.get("baseline", {})
    candidate = report.get("candidate", {})
    improvement = report.get("improvement", {})
    promotion = report.get("promotion_gate", {})
    final_quality_gate = load_json(run_dir / "eval" / "model_quality_gate.json")
    model_quality = final_quality_gate or report.get("model_quality_gate") or promotion
    print_vital_kpis("[SHFT VITAL MODEL QUALITY]", baseline=baseline, candidate=candidate, improvement=improvement, model_quality=model_quality)
    print(
        "[SHFT improvement summary]   "
        f"baseline_aggregate={fmt_num(baseline.get('aggregate'))} "
        f"candidate_aggregate={fmt_num(candidate.get('aggregate'))} "
        f"aggregate_abs={fmt_num(improvement.get('aggregate_abs'))} "
        f"aggregate_pct={fmt_pct(improvement.get('aggregate_pct'))}"
    )
    print(
        "[SHFT improvement summary]   "
        f"baseline_critical_pass={fmt_num(baseline.get('critical_pass_rate'))} "
        f"candidate_critical_pass={fmt_num(candidate.get('critical_pass_rate'))} "
        f"critical_pass_delta={fmt_num(improvement.get('critical_pass_rate_abs'))}"
    )
    print(
        "[SHFT improvement summary]   "
        f"wins={improvement.get('wins')} ties={improvement.get('ties')} losses={improvement.get('losses')} "
        f"win_rate={fmt_num(improvement.get('pairwise_win_rate'))} "
        f"loss_rate={fmt_num(improvement.get('pairwise_loss_rate'))} "
        f"promotion_eligible={promotion.get('eligible_for_promotion')}"
    )
    print(
        "[SHFT improvement summary]   "
        f"model_quality_signal={model_quality.get('quality_signal', 'unknown')} "
        f"model_quality_eligible={model_quality.get('eligible_for_promotion', model_quality.get('ok'))}"
    )
    for error in model_quality.get("errors", [])[:5]:
        print(f"[SHFT improvement summary]   model_quality_blocker={error}")
    for warning in promotion.get("warnings", []):
        print(f"[SHFT improvement summary]   WARNING {warning}")
    for rule in promotion.get("rules", []):
        print(f"[SHFT improvement summary]   gate_rule={rule}")


def main() -> int:
    parser = argparse.ArgumentParser(description="Print a compact SHFT model-improvement summary from run artifacts.")
    parser.add_argument("--run-dir", required=True)
    args = parser.parse_args()

    run_dir = Path(args.run_dir)
    print("")
    print("============================================================")
    print("[SHFT] Model improvement extract")
    print("============================================================")
    print(f"[SHFT improvement summary] run_dir={run_dir}")
    if not run_dir.exists():
        print("[SHFT improvement summary] run directory does not exist")
        return 2
    print_cycle_summary(run_dir)
    print_training_summary(run_dir)
    print_paired_eval_summary(run_dir)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())