oto-enterprise-os-dtp/05_deliverables_mvp/demo/scenarios/demo_scenario_gen.py

#!/usr/bin/env python3
"""Générateur de scénarios démo (run-sheet de pitch) · Sprint 7 · CRM + Faisabilité.

Livrable **Sprint 7** (roadmap `04_roadmap/ROADMAP_8_WEEKS_OR_LESS.md` l.68 :
« CRM + Faisabilité : Scénarios démo (P07 pitch banquier · P05 client ready) »).

C'est un **orchestrateur de méta-niveau** : il ne crée aucune donnée métier, il
COMPOSE les hand-offs `out/` déjà produits par les modules livrés en un run-sheet
de démo prêt à jouer (P07 banquier · P05 client). Toute valeur chiffrée du
run-sheet est RÉSOLUE depuis un artefact amont via un pointeur RFC 6901 — jamais
saisie à la main (anti-invention CLAUDE.md #6). Une valeur introuvable devient un
placeholder `{{…}}`, jamais un chiffre fabriqué.

Chaîne de valeur :

    out/ des modules livrés (workflow, e-CF, CONFOTUR, RBAC, 4Big, portails, …)
                                   │  pointeurs RFC 6901 (scenario_spec.json)
                                   ▼
                    demo_scenario_gen ──►  out/run_sheet.json + out/run_sheet.md
                                           + out/MANIFEST.json
                                   │
                                   ▼  (VPS · hors périmètre worker · #8)
                    rendu deck / prompteur ─► démo publique otov7.com

Garanties (le CLI REFUSE d'écrire si l'une casse) :
  · chaque module cité est GATED par le CI (`.gitea/workflows/ci.yml`) — la démo
    ne s'appuie que sur du validé (réutilise la preuve de l'auditeur 4Big) ;
  · chaque valeur est re-résolue indépendamment et comparée (anti-invention) ;
  · l'artefact livré est `pret=true` (0 citation non résolue · 0 module non gated).

Sous-commandes :
  build     [-o OUT_DIR]   → OUT_DIR/run_sheet.{json,md} + OUT_DIR/MANIFEST.json
  validate                 → manifeste + invariants (sans écrire)
"""

from __future__ import annotations

import argparse
import json
import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)

from scenlib import builder, deps, evidence, render  # noqa: E402


def _eprint(*args) -> None:
    print(*args, file=sys.stderr)


def _load_spec() -> dict:
    with open(os.path.join(_HERE, "scenario_spec.json"), encoding="utf-8") as fh:
        return json.load(fh)


def _load_schema() -> dict:
    with open(os.path.join(_HERE, "scenario.schema.json"), encoding="utf-8") as fh:
        return json.load(fh)


# --------------------------------------------------------------------------- #
# Invariants — honnêteté #6 : le CLI n'écrit RIEN si l'un casse.
# --------------------------------------------------------------------------- #
def check_invariants(spec: dict, run_sheet: dict, manifest: dict) -> list[str]:
    motifs: list[str] = []

    # (a) manifeste conforme au schéma de sortie (validateur maison Publiciste).
    motifs += [f"manifest: {e}" for e in deps.validate(manifest, _load_schema())]

    # (b) identité des scénarios : ids uniques, projet/audience déclarés au spec.
    ids = [sc["id"] for sc in run_sheet["scenarios"]]
    if len(ids) != len(set(ids)):
        motifs.append(f"ids de scénario non uniques : {ids}")
    for sc in run_sheet["scenarios"]:
        if sc["projet"] not in spec["projets"]:
            motifs.append(f"{sc['id']} : projet {sc['projet']} absent du spec.projets")
        if sc["audience"] not in spec["audiences"]:
            motifs.append(f"{sc['id']} : audience {sc['audience']} hors spec.audiences")
        if sc["projet_libelle"] != spec["projets"].get(sc["projet"]):
            motifs.append(f"{sc['id']} : libellé projet incohérent avec le spec")

    # (c) structure de chaque scénario : beats ordonnés 1..n, ≥3 beats,
    #     ≥2 modules distincts (un orchestrateur DOIT tisser plusieurs livrables),
    #     et jamais s'auto-citer (pas de preuve circulaire · SoD).
    for sc in run_sheet["scenarios"]:
        ordres = [b["ordre"] for b in sc["beats"]]
        if ordres != list(range(1, len(sc["beats"]) + 1)):
            motifs.append(f"{sc['id']} : ordres de beats non contigus {ordres}")
        if len(sc["beats"]) < 3:
            motifs.append(f"{sc['id']} : {len(sc['beats'])} beat(s) (min 3)")
        mods = {c["module"] for b in sc["beats"] for c in b["evidence"]}
        if len(mods) < 2:
            motifs.append(f"{sc['id']} : {len(mods)} module cité (min 2 · orchestration)")
        if builder.SELF_MODULE in mods:
            motifs.append(f"{sc['id']} : auto-citation de {builder.SELF_MODULE} (SoD)")

    # (d) ANTI-INVENTION : re-résolution indépendante de CHAQUE citation.
    motifs += _check_anti_invention(spec, run_sheet)

    # (e) placeholder ⟺ non résolu (aucune fuite de None ni de faux chiffre).
    for sc in run_sheet["scenarios"]:
        for b in sc["beats"]:
            for c in b["evidence"]:
                ph = evidence.is_placeholder(c["valeur"])
                if c["resolu"] and (ph or c["valeur"] is None):
                    motifs.append(f"{c['module']}#{c['pointer']} : resolu mais valeur placeholder/None")
                if not c["resolu"] and not ph:
                    motifs.append(f"{c['module']}#{c['pointer']} : non résolu sans placeholder")

    # (f) comptes recalculés == manifeste (aucun compteur « posé »).
    motifs += _check_counts(run_sheet, manifest)

    # (g) couverture CI : tout module cité doit être gated + drapeau pret.
    if not manifest["couverture_ci"]["ok"] or manifest["couverture_ci"]["modules_cites_non_gated"]:
        motifs.append("module cité NON gated par le CI : "
                      f"{manifest['couverture_ci']['modules_cites_non_gated']}")
    if not manifest["pret"]:
        motifs.append("run-sheet non prêt (citations non résolues ou modules non gated) — "
                      "un livrable démo doit être prêt à jouer (#5)")

    return motifs


def _check_anti_invention(spec: dict, run_sheet: dict) -> list[str]:
    """Re-résout chaque pointeur avec un loader neuf et compare au run-sheet.

    Prouve qu'aucune valeur n'a été « posée » : elle DOIT redécouler de l'artefact
    amont sur disque. Recoupe aussi spec ↔ run-sheet (mêmes pointeurs, même ordre).
    """
    motifs: list[str] = []
    loader = evidence.Loader()
    spec_by_id = {sc["id"]: sc for sc in spec["scenarios"]}

    for sc in run_sheet["scenarios"]:
        spec_sc = spec_by_id.get(sc["id"])
        if spec_sc is None:
            motifs.append(f"{sc['id']} : scénario absent du spec")
            continue
        if len(spec_sc["beats"]) != len(sc["beats"]):
            motifs.append(f"{sc['id']} : nb de beats spec≠run-sheet")
            continue
        for spec_b, b in zip(spec_sc["beats"], sc["beats"]):
            spec_ev = spec_b["evidence"]
            if len(spec_ev) != len(b["evidence"]):
                motifs.append(f"{sc['id']}/{b['id']} : nb de citations spec≠run-sheet")
                continue
            for se, c in zip(spec_ev, b["evidence"]):
                if (se["module"], se["file"], se["pointer"]) != \
                        (c["module"], c["file"], c["pointer"]):
                    motifs.append(f"{sc['id']}/{b['id']} : citation dérivée du spec")
                    continue
                exp_val, exp_res = evidence.resolve(
                    loader, se["module"], se["file"], se["pointer"])
                if exp_res != c["resolu"] or exp_val != c["valeur"]:
                    motifs.append(
                        f"{se['module']}#{se['pointer']} : run-sheet {c['valeur']!r} "
                        f"!= re-résolution {exp_val!r} (invention/dérive)")
    return motifs


def _check_counts(run_sheet: dict, manifest: dict) -> list[str]:
    motifs: list[str] = []
    cites = [c for sc in run_sheet["scenarios"] for b in sc["beats"] for c in b["evidence"]]
    exp = {
        "scenarios": len(run_sheet["scenarios"]),
        "beats_total": sum(len(sc["beats"]) for sc in run_sheet["scenarios"]),
        "citations_total": len(cites),
        "citations_non_resolues": sum(1 for c in cites if not c["resolu"]),
        "modules_cites_uniques": len({c["module"] for c in cites}),
    }
    for cle, val in exp.items():
        if manifest["counts"].get(cle) != val:
            motifs.append(f"counts.{cle} : manifeste {manifest['counts'].get(cle)} != recalcul {val}")
    return motifs


# --------------------------------------------------------------------------- #
def _assemble(generated_at: str | None) -> tuple[dict, dict, dict]:
    spec = _load_spec()
    run_sheet = builder.build_run_sheet(spec, generated_at=generated_at)
    manifest = builder.build_manifest(spec, run_sheet)
    return spec, run_sheet, manifest


def cmd_build(ns: argparse.Namespace) -> int:
    spec, run_sheet, manifest = _assemble(ns.generated_at)
    motifs = check_invariants(spec, run_sheet, manifest)
    if motifs:
        _eprint("❌ invariants cassés — RIEN écrit :")
        for m in motifs:
            _eprint("     ·", m)
        return 1

    os.makedirs(ns.out, exist_ok=True)
    for name, obj in (("run_sheet.json", run_sheet), ("MANIFEST.json", manifest)):
        with open(os.path.join(ns.out, name), "w", encoding="utf-8") as fh:
            json.dump(obj, fh, ensure_ascii=False, indent=2)
            fh.write("\n")
    # Prompteur Markdown lisible (dérivé PUREMENT du run-sheet · #6). Reproductible
    # à l'identique (gate ci/check_artifacts.sh) car render est déterministe.
    with open(os.path.join(ns.out, "run_sheet.md"), "w", encoding="utf-8") as fh:
        fh.write(render.render_markdown(run_sheet))

    c = manifest["counts"]
    _eprint(f"✅ run-sheet → {ns.out} (+ prompteur run_sheet.md)")
    _eprint(f"   scénarios={c['scenarios']} · beats={c['beats_total']} · "
            f"citations={c['citations_total']} (non résolues={c['citations_non_resolues']}) · "
            f"modules cités={c['modules_cites_uniques']}")
    for sc in manifest["scenarios"]:
        _eprint(f"     · {sc['id']} [{sc['audience']}] {sc['projet']} — "
                f"{sc['beats']} beats · {sc['duree_min']} min · {sc['citations_resolues']}/{sc['citations']} résolues")
    _eprint(f"   pret={manifest['pret']} · couverture CI ok={manifest['couverture_ci']['ok']}")
    return 0


def cmd_validate(ns: argparse.Namespace) -> int:
    spec, run_sheet, manifest = _assemble(ns.generated_at)
    motifs = check_invariants(spec, run_sheet, manifest)
    print(json.dumps(manifest, ensure_ascii=False, indent=2))
    if motifs:
        _eprint(f"❌ {len(motifs)} invariant(s) cassé(s) :")
        for m in motifs:
            _eprint("     ·", m)
        return 1
    _eprint(f"✅ manifeste conforme + invariants OK (pret={manifest['pret']}).")
    return 0


# --------------------------------------------------------------------------- #
def build_argparser() -> argparse.ArgumentParser:
    ap = argparse.ArgumentParser(prog="demo_scenario_gen",
                                 description=__doc__.splitlines()[0])
    sub = ap.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("build", help="scenario_spec.json → out/run_sheet.{json,md} + out/MANIFEST.json")
    p.add_argument("-o", "--out", default=os.path.join(_HERE, "out"),
                   help="dossier de sortie (défaut: ./out)")
    p.add_argument("--generated-at", dest="generated_at", default=None)
    p.set_defaults(func=cmd_build)

    p = sub.add_parser("validate", help="manifeste + invariants (sans écrire)")
    p.add_argument("--generated-at", dest="generated_at", default=None)
    p.set_defaults(func=cmd_validate)
    return ap


def main(argv: list[str] | None = None) -> int:
    ns = build_argparser().parse_args(argv)
    return ns.func(ns)


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