code/self_healing_finetuning/orchestrator/lifecycle_proof.py

from __future__ import annotations

from pathlib import Path
from typing import Any

from approvals.promotion import plan_promotion
from eval.certification import certify
from inference.deploy import deploy
from model_policy.selector import select_model
from monitoring.canary import run_canary_monitor
from n21.config import write_json
from n21.settings import DEFAULT_MODEL_ID, SHFT_WORKSPACE_ROOT
from observability.audit_log import utc_now
from orchestrator.cycle_controller import run_self_healing_cycles
from training.launch import run_training


LIFECYCLE_MAP = [
    {"proof_step": 1, "name": "model roadmap", "original_lifecycle": ["credential readiness", "base model bootstrap"]},
    {"proof_step": 2, "name": "selection and training evidence", "original_lifecycle": ["model selection", "training plan"]},
    {"proof_step": 3, "name": "self-healing and certification evidence", "original_lifecycle": ["self-healing diagnosis", "repair", "retrain", "re-evaluate", "certification"]},
    {"proof_step": 4, "name": "promotion proof-chain gate", "original_lifecycle": ["promotion", "approval gates"]},
    {"proof_step": 5, "name": "rollback anchor", "original_lifecycle": ["rollback target", "incident readiness"]},
    {"proof_step": 6, "name": "post-promotion canary monitor", "original_lifecycle": ["monitoring", "rollback trigger"]},
    {"proof_step": 7, "name": "orchestrated proof run", "original_lifecycle": ["audit trail", "operator runbook"]},
]


def run_lifecycle_proof(
    run_dir: Path,
    *,
    run_id: str,
    env: str = "stage",
    model_candidate: str = DEFAULT_MODEL_ID,
    train_provider: str = "hf_managed",
    infer_provider: str = "hf_managed",
    max_cycles: int = 1,
    canary_mode: str = "pass",
) -> dict[str, Any]:
    run_dir.mkdir(parents=True, exist_ok=True)
    selection = select_model("finance_qa", "dev")
    write_json(SHFT_WORKSPACE_ROOT / "current" / "dev" / "model_selection.json", selection)
    train_result = run_training(
        run_dir,
        run_id=run_id,
        model_candidate=model_candidate,
        train_provider=train_provider,
        infer_provider=infer_provider,
    )
    cycle_summary = run_self_healing_cycles(
        run_dir,
        run_id=run_id,
        model_candidate=model_candidate,
        train_provider=train_provider,
        infer_provider=infer_provider,
        max_cycles=max_cycles,
    )
    certification = certify(env, "finance_qa", model_candidate=model_candidate)
    write_json(run_dir / "eval" / "certification_report.json", certification)
    deployment = deploy(run_dir, run_id=run_id, infer_provider=infer_provider, env=env, model_candidate=model_candidate)
    promotion = plan_promotion(
        implementation_root=SHFT_WORKSPACE_ROOT,
        run_dir=run_dir,
        run_id=run_id,
        env=env,
        approvals={"engineering": "approved", "compliance": "pending", "product_owner": "pending"},
        risk_tier="internal_analyst",
        model_id=model_candidate,
    )
    canary = run_canary_monitor(run_dir, run_id=run_id, env=env, mode=canary_mode)
    report = {
        "run_id": run_id,
        "env": env,
        "status": "rollback_recommended" if canary["rollback_recommended"] else "completed",
        "model_candidate": model_candidate,
        "lifecycle_map": LIFECYCLE_MAP,
        "evidence": {
            "selected_model": selection["selected_model"],
            "roadmap_weighted_score": selection["roadmap_evidence"]["weighted_score"],
            "training_gate_result": train_result["iteration_evidence"]["gate_result"],
            "cycles_completed": cycle_summary["cycles_completed"],
            "certification_gate_result": certification["gate_result"],
            "deployment_status": deployment["status"],
            "promotion_status": promotion["status"],
            "promotion_error_count": len(promotion["promotion_errors"]),
            "rollback_anchor_checksum": promotion["rollback_target"]["checksum_sha256"],
            "canary_status": canary["status"],
            "rollback_recommended": canary["rollback_recommended"],
            "rollback_reason_count": len(canary["rollback_reasons"]),
        },
        "artifact_paths": {
            "model_selection": str(SHFT_WORKSPACE_ROOT / "current" / "dev" / "model_selection.json"),
            "run_manifest": str(run_dir / "manifests" / "run_manifest.json"),
            "cycle_summary": str(run_dir / "heal_decisions" / "cycle_summary.json"),
            "certification_report": str(run_dir / "eval" / "certification_report.json"),
            "deployment_manifest": str(run_dir / "manifests" / "deployment_manifest.json"),
            "promotion_manifest": str(run_dir / "manifests" / "promotion_manifest.json"),
            "canary_report": str(run_dir / "monitoring" / "canary_report.json"),
        },
        "created_at": utc_now(),
    }
    write_json(run_dir / "lifecycle" / "lifecycle_report.json", report)
    return report