Claude Code DTP Worker
cac65386ab
Orchestrateur méta-niveau : compose les hand-offs out/ des 10 modules déjà livrés en un run-sheet de démo jouable, via pointeurs RFC 6901 résolus à la construction (zéro chiffre en dur · #6). Réutilise le validateur Publiciste et la preuve de couverture CI de l'auditeur 4Big (#5) — une démo ne s'appuie que sur des modules gated. Enregistre demo/scenarios dans l'audit 4Big (17→18, verdict PASS 18/18 à 100). 32 tests (9 injections négatives) · régression 474 tests verts · guards CI verts · builds déterministes. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
251 lines
11 KiB
Python
251 lines
11 KiB
Python
#!/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/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,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 # 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")
|
|
|
|
c = manifest["counts"]
|
|
_eprint(f"✅ run-sheet → {ns.out}")
|
|
_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,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())
|