oto-enterprise-os-dtp/05_deliverables_mvp/devops/deploy_runbook/deploylib/builder.py

"""Assemblage du run-book de déploiement VPS à partir du spec + de la réalité CI.

Le run-book ordonne le déploiement de TOUS les livrables gated du mandat en un
plan de phases. La liste des modules à déployer n'est PAS écrite à la main : elle
est dérivée de `.gitea/workflows/ci.yml` (via q4lib.registry.parse_ci, réutilisé)
puis mise en correspondance BIJECTIVE avec le `module_phase` du spec.

Fonctions PURES (aucune I/O, aucun horodatage) → sortie déterministe, diffable,
re-générable bit-à-bit. Le CLI leur injecte le spec chargé et le résultat de
`parse_ci()` ; les tests injectent des dictionnaires synthétiques (facile à
mettre en défaut → tests négatifs).

Anti-invention (#6) : aucun chiffre métier ; les paramètres non confirmés restent
des CONFIRMATIONS sourcées (owner + source), jamais une valeur fabriquée.
"""

from __future__ import annotations

from typing import Any


def gated_modules(ci: dict, self_module: str) -> list[str]:
    """Chemins de modules gated par le CI, hors le run-book lui-même (SoD)."""
    paths = set(ci.get("job_to_path", {}).values())
    paths.discard(self_module)
    return sorted(paths)


def _path_to_job(ci: dict) -> dict[str, str]:
    """Inverse job→path en path→job. Suppose l'unicité (vérifiée en invariant)."""
    out: dict[str, str] = {}
    for job, path in ci.get("job_to_path", {}).items():
        # En cas de collision on garde le job le plus petit (déterministe) ; la
        # collision elle-même est signalée par l'invariant INV9 du CLI.
        if path not in out or job < out[path]:
            out[path] = job
    return out


def build_runbook(spec: dict, ci: dict) -> dict[str, Any]:
    """Consolide spec + CI en run-book de phases + manifeste agrégé.

    Ne lève pas sur une incohérence de couverture (module manquant / en trop) :
    elle est REMONTÉE dans `manifest.coverage` pour que le CLI/les tests la
    signalent explicitement (un déploiement ne doit pas oublier silencieusement
    un livrable).
    """
    self_module = spec["self_module"]
    module_phase = spec["module_phase"]
    phases_meta = spec["phases"]
    catalogue = spec["confirmations"]

    gated = gated_modules(ci, self_module)
    gated_set = set(gated)
    mapped_set = set(module_phase)
    p2j = _path_to_job(ci)

    # Couverture bijective : les modules mappés == les modules gated (hors self).
    missing_in_map = sorted(gated_set - mapped_set)   # gated mais non planifié
    extra_in_map = sorted(mapped_set - gated_set)     # planifié mais non gated
    bijective = not missing_in_map and not extra_in_map

    phase_ids = [p["id"] for p in phases_meta]
    phase_id_set = set(phase_ids)

    # Modules dont la phase déclarée n'existe pas dans `phases` (spec incohérent).
    unknown_phase = sorted(
        m for m, ph in module_phase.items() if ph not in phase_id_set
    )

    # Assemblage du run-book : une phase = métadonnées + modules gated assignés +
    # confirmations résolues depuis le catalogue.
    runbook: list[dict[str, Any]] = []
    for p in phases_meta:
        mods = sorted(
            m for m, ph in module_phase.items()
            if ph == p["id"] and m in gated_set
        )
        modules = [{"module": m, "ci_job": p2j.get(m)} for m in mods]
        confs = [
            {
                "id": cid,
                "libelle": catalogue.get(cid, {}).get("libelle"),
                "owner": catalogue.get(cid, {}).get("owner"),
                "source": catalogue.get(cid, {}).get("source"),
            }
            for cid in p.get("confirmations", [])
        ]
        runbook.append({
            "order": p["order"],
            "id": p["id"],
            "titre": p["titre"],
            "responsable": p["responsable"],
            "rationale": p["rationale"],
            "depends_on": list(p.get("depends_on", [])),
            "modules": modules,
            "confirmations": confs,
        })

    # Confirmations effectivement référencées par ≥1 phase (détection d'orphelin).
    referenced: set[str] = set()
    for p in phases_meta:
        referenced.update(p.get("confirmations", []))
    orphan_confirmations = sorted(set(catalogue) - referenced)
    unknown_confirmations = sorted(referenced - set(catalogue))

    manifest = {
        "generated_from": f"deploy_spec.json v{spec.get('version')}",
        "reference_cadre": spec.get("reference_cadre"),
        "cible_portage": spec.get("cible_portage"),
        "self_module": self_module,
        "counts": {
            "phases": len(phases_meta),
            "modules": len(gated),
            "modules_mapped": len(mapped_set),
            "confirmations": len(catalogue),
            "confirmations_referenced": len(referenced),
        },
        "coverage": {
            "ci_modules_count": len(gated),
            "mapped_modules_count": len(mapped_set),
            "bijective": bijective,
            "missing_in_map": missing_in_map,
            "extra_in_map": extra_in_map,
            "unknown_phase": unknown_phase,
            "self_module_excluded": self_module,
        },
        "graph": {
            "phase_order": phase_ids,
            "orphan_confirmations": orphan_confirmations,
            "unknown_confirmations": unknown_confirmations,
        },
        "confirmations_ouvertes": sorted(catalogue),
    }

    return {"runbook": runbook, "manifest": manifest}