code/self_healing_finetuning/approvals/promotion.py

from __future__ import annotations

from pathlib import Path
from typing import Any

from approvals.policy import validate_approvals
from approvals.proof_chain import validate_promotion_proof_chain
from n21.config import write_json
from observability.audit_log import utc_now
from rollback.anchor import ensure_last_good_anchor, validate_rollback_anchor


def plan_promotion(
    *,
    implementation_root: Path,
    run_dir: Path,
    run_id: str,
    env: str,
    approvals: dict[str, str],
    risk_tier: str,
    model_id: str,
    live: bool = False,
) -> dict[str, Any]:
    approval_errors = validate_approvals(approvals, env=env, risk_tier=risk_tier)
    proof_chain = validate_promotion_proof_chain(run_dir)
    rollback_anchor = ensure_last_good_anchor(implementation_root, env=env, model_id=model_id)
    rollback_anchor_errors = validate_rollback_anchor(rollback_anchor)
    promotion_errors = [*approval_errors, *proof_chain["errors"], *rollback_anchor_errors]
    manifest = {
        "run_id": run_id,
        "env": env,
        "status": "promotion_blocked" if promotion_errors else ("promotion_planned" if not live else "promotion_requested"),
        "approvals": approvals,
        "approval_errors": approval_errors,
        "proof_chain": proof_chain,
        "rollback_anchor": rollback_anchor,
        "rollback_anchor_errors": rollback_anchor_errors,
        "promotion_errors": promotion_errors,
        "risk_tier": risk_tier,
        "rollback_target": {
            "type": "last_good",
            "anchor_id": rollback_anchor.get("anchor_id"),
            "model_id": rollback_anchor.get("model_id"),
            "deployment_id": rollback_anchor.get("deployment_id"),
            "checksum_sha256": rollback_anchor.get("checksum_sha256"),
            "path": str(implementation_root / "rollback" / "last_good"),
        },
        "dry_run": not live,
        "created_at": utc_now(),
    }
    write_json(run_dir / "manifests" / "promotion_manifest.json", manifest)
    write_json(implementation_root / "registry" / "promotions" / f"{run_id}_{env}.json", manifest)
    return manifest