oto-enterprise-os-dtp/05_deliverables_mvp/qa/acceptance/acclib/builder.py

"""Assemblage de la matrice d'acceptation à partir du spec + de la réalité CI.

La matrice mappe chaque promesse roadmap → preuve gated OU hors-périmètre sourcé.
La liste des modules-preuve n'est PAS crue sur parole : elle est confrontée à
`.gitea/workflows/ci.yml` (via q4lib.registry.parse_ci, réutilisé) de façon
BIJECTIVE, et la fenêtre de sprint de chaque module est LUE dans le registre de
l'auditeur 4Big (zéro re-déclaration · anti-dérive).

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

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 la matrice elle-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 (job minimal en cas de collision)."""
    out: dict[str, str] = {}
    for job, path in ci.get("job_to_path", {}).items():
        if path not in out or job < out[path]:
            out[path] = job
    return out


def module_sprint_map(spec: dict, audit_sprints: dict) -> dict[str, str]:
    """{module -> sprint}, dérivé du registre 4Big + complété par le spec.

    Le registre de l'auditeur est la source ; l'auditeur lui-même n'y figure pas
    (SoD) → son sprint vient de `extra_module_sprint` (avec source). Une entrée
    `extra` qui écraserait un module déjà daté par le registre est un conflit :
    on la garde ici mais l'incohérence est signalée par l'invariant du CLI."""
    out = dict(audit_sprints)
    for path, meta in spec.get("extra_module_sprint", {}).items():
        out.setdefault(path, meta["sprint"])
    return out


def all_deliverables(spec: dict) -> list[dict]:
    """Livrables de sprint + métriques MVP, dans l'ordre du spec."""
    return list(spec.get("sprint_deliverables", [])) + list(spec.get("mvp_metrics", []))


def _resolve_row(d: dict, gated_set: set, p2j: dict, deps_mod) -> dict[str, Any]:
    """Une ligne de matrice : preuves résolues + drapeaux de cohérence."""
    ev_modules = list(d.get("evidence_modules", []))
    modules = [
        {"module": m, "ci_job": p2j.get(m), "gated": m in gated_set}
        for m in ev_modules
    ]
    artifacts = [
        {"path": a, "exists": deps_mod.artifact_exists(a)}
        for a in d.get("evidence_artifacts", [])
    ]
    oos = [
        {"item": o["item"], "source": o.get("source")}
        for o in d.get("out_of_scope", [])
    ]
    has_evidence = bool(ev_modules) or bool(d.get("evidence_artifacts"))
    return {
        "id": d["id"],
        "kind": d["kind"],
        "sprint": d.get("sprint"),
        "roadmap_line": d["roadmap_line"],
        "statement": d["statement"],
        "status": d["status"],
        "modules": modules,
        "artifacts": artifacts,
        "out_of_scope": oos,
        "n_modules": len(ev_modules),
        "all_modules_gated": all(m in gated_set for m in ev_modules),
        "all_artifacts_exist": all(a["exists"] for a in artifacts),
        "has_evidence": has_evidence,
    }


def build_matrix(spec: dict, ci: dict, audit_sprints: dict, deps_mod) -> dict[str, Any]:
    """Consolide spec + CI + registre 4Big en matrice + manifeste agrégé.

    Ne lève JAMAIS : les incohérences (module manquant / en trop / partition de
    sprint fausse / artefact absent) sont REMONTÉES dans `manifest.coverage`
    pour que le CLI/les tests les signalent explicitement. Une recette ne doit
    pas oublier silencieusement une promesse."""
    self_module = spec["self_module"]
    gated = gated_modules(ci, self_module)
    gated_set = set(gated)
    p2j = _path_to_job(ci)
    ms_map = module_sprint_map(spec, audit_sprints)

    deliverables = all_deliverables(spec)
    rows = [_resolve_row(d, gated_set, p2j, deps_mod) for d in deliverables]

    # --- Couverture BIJECTIVE (cœur anti-invention · #6) ------------------
    # cited = tout module-preuve des lignes IN_REPO ; doit égaler l'ensemble
    # gated (hors self). Un module gated non cité = promesse orpheline ; un
    # module cité non gated = preuve fantôme.
    cited: set[str] = set()
    for d in deliverables:
        if d.get("status") == "in_repo":
            cited.update(d.get("evidence_modules", []))
    missing = sorted(gated_set - cited)   # gated mais non tracé
    phantom = sorted(cited - gated_set)   # tracé mais non gated
    bijective = not missing and not phantom

    # --- Partition par sprint --------------------------------------------
    # Chaque livrable de sprint SX doit citer EXACTEMENT les modules gated dont
    # la fenêtre est SX (ni plus, ni moins). Prouve que la matrice répartit les
    # livraisons par sprint sans trou ni chevauchement.
    gated_by_sprint: dict[str, set] = {}
    unknown_sprint = sorted(m for m in gated if m not in ms_map)
    for m in gated:
        if m in ms_map:
            gated_by_sprint.setdefault(ms_map[m], set()).add(m)

    sprint_partition: list[dict] = []
    for d in spec.get("sprint_deliverables", []):
        sx = d["sprint"]
        want = gated_by_sprint.get(sx, set())
        have = set(d.get("evidence_modules", []))
        sprint_partition.append({
            "sprint": sx,
            "expected": sorted(want),
            "cited": sorted(have),
            "missing": sorted(want - have),
            "extra": sorted(have - want),
            "exact": want == have,
        })
    partition_ok = all(p["exact"] for p in sprint_partition) and not unknown_sprint

    # --- Artefacts hors-CI (docs) ----------------------------------------
    missing_artifacts = sorted(
        a["path"]
        for r in rows for a in r["artifacts"] if not a["exists"]
    )

    # --- Complétude roadmap ----------------------------------------------
    sprint_ids = [d["id"] for d in spec.get("sprint_deliverables", [])]
    metric_ids = [d["id"] for d in spec.get("mvp_metrics", [])]
    oos_count = sum(len(r["out_of_scope"]) for r in rows)
    in_repo = sum(1 for r in rows if r["status"] == "in_repo")
    out_of_scope_full = sum(1 for r in rows if r["status"] == "out_of_scope")

    manifest = {
        "generated_from": f"acceptance_spec.json v{spec.get('version')}",
        "reference_cadre": spec.get("reference_cadre"),
        "roadmap_ref": spec.get("roadmap_ref"),
        "cible_portage": spec.get("cible_portage"),
        "self_module": self_module,
        "counts": {
            "sprint_deliverables": len(sprint_ids),
            "mvp_metrics": len(metric_ids),
            "deliverables_total": len(deliverables),
            "in_repo": in_repo,
            "out_of_scope_full": out_of_scope_full,
            "gated_modules": len(gated),
            "cited_modules": len(cited),
            "out_of_scope_items": oos_count,
        },
        "coverage": {
            "ci_modules_count": len(gated),
            "cited_modules_count": len(cited),
            "bijective": bijective,
            "missing_in_matrix": missing,
            "phantom_evidence": phantom,
            "partition_ok": partition_ok,
            "sprint_partition": sprint_partition,
            "unknown_sprint": unknown_sprint,
            "missing_artifacts": missing_artifacts,
            "self_module_excluded": self_module,
        },
        "roadmap": {
            "sprint_ids": sprint_ids,
            "metric_ids": metric_ids,
        },
    }

    verdict = bijective and partition_ok and not missing_artifacts

    return {"matrix": rows, "manifest": manifest, "verdict": verdict}