[DTP-Worker] Sprint 7 · Générateur Scénarios démo (run-sheet P07 banquier / P05 client) (CRM+Faisabilité · roadmap L68)

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>
This commit is contained in:
Claude Code DTP Worker
2026-07-30 10:39:24 +00:00
parent a517619432
commit cac65386ab
19 changed files with 1691 additions and 11 deletions
@@ -0,0 +1,9 @@
"""Bibliothèque du générateur de scénarios démo (run-sheet de pitch).
Découpage :
- deps : réutilisation des briques déjà livrées (validateur maison +
preuve de couverture CI) · zéro duplication (#5).
- evidence : chargement des hand-offs `out/` amont + résolution de pointeurs
RFC 6901 · cœur anti-invention (#6).
- builder : assemblage déterministe du run-sheet + du manifeste.
"""
@@ -0,0 +1,123 @@
"""Assemblage déterministe du run-sheet démo + du manifeste (hand-off).
Le builder est PUR : mêmes entrées (spec + artefacts `out/` sur disque) ⇒ même
sortie. Aucun horodatage implicite, aucun ordre non déterministe. L'ordre des
beats est dérivé de la position dans le spec (jamais d'un champ `ordre` en dur,
pour éviter toute dérive). Les modules cités sont recoupés avec l'ensemble des
modules gated par le CI (via `deps.ci_module_paths`) — une démo ne s'appuie que
sur du validé.
"""
from __future__ import annotations
from . import deps, evidence
SELF_MODULE = "demo/scenarios"
def _build_beat(loader: evidence.Loader, beat: dict, ordre: int) -> dict:
cites = []
for ev in beat["evidence"]:
module, file, pointer = ev["module"], ev["file"], ev["pointer"]
valeur, resolu = evidence.resolve(loader, module, file, pointer)
cites.append({
"module": module,
"file": file,
"pointer": pointer,
"label": ev["label"],
"valeur": valeur,
"resolu": resolu,
})
return {
"ordre": ordre,
"id": beat["id"],
"titre": beat["titre"],
"role_narratif": beat["role_narratif"],
"duree_min": beat["duree_min"],
"evidence": cites,
}
def _build_scenario(loader: evidence.Loader, spec: dict, sc: dict) -> dict:
beats = [_build_beat(loader, b, i + 1) for i, b in enumerate(sc["beats"])]
modules = sorted({c["module"] for b in beats for c in b["evidence"]})
duree = sum(b["duree_min"] for b in beats)
return {
"id": sc["id"],
"projet": sc["projet"],
"projet_libelle": spec["projets"][sc["projet"]],
"audience": sc["audience"],
"titre": sc["titre"],
"objectif": sc["objectif"],
"duree_min": duree,
"modules_cites": modules,
"beats": beats,
}
def build_run_sheet(spec: dict, generated_at: str | None = None) -> dict:
"""Construit le run-sheet complet (le hand-off principal)."""
loader = evidence.Loader()
scenarios = [_build_scenario(loader, spec, sc) for sc in spec["scenarios"]]
return {
"doc": "OTO · Run-sheet démo (méta-livrable)",
"spec_version": spec["spec_version"],
"roadmap_ref": spec["roadmap_ref"],
"generated_at": generated_at,
"scenarios": scenarios,
}
def _all_citations(run_sheet: dict) -> list[dict]:
return [c for sc in run_sheet["scenarios"] for b in sc["beats"] for c in b["evidence"]]
def build_manifest(spec: dict, run_sheet: dict) -> dict:
"""Manifeste diagnostic + preuve de couverture CI + drapeau `pret`."""
cites = _all_citations(run_sheet)
non_resolues = [c for c in cites if not c["resolu"]]
modules_cites = sorted({c["module"] for c in cites})
gated = deps.ci_module_paths() # modules réellement gated par le CI
non_gated = sorted(m for m in modules_cites if m not in gated)
scen_diag = []
for sc in run_sheet["scenarios"]:
sc_cites = [c for b in sc["beats"] for c in b["evidence"]]
scen_diag.append({
"id": sc["id"],
"projet": sc["projet"],
"audience": sc["audience"],
"beats": len(sc["beats"]),
"citations": len(sc_cites),
"citations_resolues": sum(1 for c in sc_cites if c["resolu"]),
"duree_min": sc["duree_min"],
"modules_cites": sc["modules_cites"],
})
couverture_ok = not non_gated
pret = (not non_resolues) and couverture_ok
return {
"deliverable": "Scénarios démo · run-sheet de pitch",
"sprint": "Sprint 7 · CRM + Faisabilité",
"roadmap_ref": spec["roadmap_ref"],
"generated_from": "scenario_spec.json",
"deliverables_root": "05_deliverables_mvp",
"self_module": SELF_MODULE,
"generated_at": run_sheet["generated_at"],
"scenarios": scen_diag,
"counts": {
"scenarios": len(run_sheet["scenarios"]),
"beats_total": sum(len(sc["beats"]) for sc in run_sheet["scenarios"]),
"citations_total": len(cites),
"citations_non_resolues": len(non_resolues),
"modules_cites_uniques": len(modules_cites),
},
"modules_cites": modules_cites,
"couverture_ci": {
"ok": couverture_ok,
"modules_cites_non_gated": non_gated,
},
"pret": pret,
}
@@ -0,0 +1,44 @@
"""Réutilisation des briques déjà livrées (workflow #5 · zéro duplication).
L'orchestrateur démo ne redéfinit RIEN qui existe ailleurs :
- `validate` : le validateur JSON-Schema maison du Publiciste (draft-07,
sous-ensemble) — le run-sheet est validé SANS pip, car le
gate CI Gitea Actions tourne sans réseau (CLAUDE.md #2).
- `ci_module_paths` : la PREUVE de couverture de l'auditeur 4Big
(`qa/audit_4big/q4lib/registry.py`), qui lit
`.gitea/workflows/ci.yml` pour connaître les modules
réellement gated. On la réutilise pour garantir qu'une
démo ne s'appuie QUE sur des livrables validés par le CI —
jamais sur du code non gated.
Import par manipulation de `sys.path` (idiome des modules voisins :
`banclib/deps.py`, `q4lib/deps.py`) plutôt que duplication — une seule source de
vérité pour tout le mandat.
"""
from __future__ import annotations
import os
import sys
_HERE = os.path.dirname(os.path.abspath(__file__))
# demo/scenarios/scenlib → 05_deliverables_mvp
_DELIVERABLES = os.path.normpath(os.path.join(_HERE, "..", "..", ".."))
_PUB = os.path.join(_DELIVERABLES, "publiciste")
_AUDIT = os.path.join(_DELIVERABLES, "qa", "audit_4big")
for _p in (_PUB, _AUDIT):
if _p not in sys.path:
sys.path.insert(0, _p)
from lib import validator # type: ignore # noqa: E402
from q4lib import registry # type: ignore # noqa: E402
validate = validator.validate
ci_module_paths = registry.ci_module_paths
# Racine des livrables, exposée aux autres modules scenlib.
DELIVERABLES_ROOT = _DELIVERABLES
__all__ = ["validate", "ci_module_paths", "DELIVERABLES_ROOT"]
@@ -0,0 +1,112 @@
"""Cœur anti-invention (#6) : résolution des preuves depuis les hand-offs amont.
Un scénario démo ne contient AUCUN chiffre en dur. Chaque « evidence » du spec est
un pointeur `(module, file, pointer)` vers un artefact `out/` déjà produit par un
module livré. Ce fichier :
1. charge l'artefact JSON (avec cache) et signale toute absence / JSON invalide ;
2. résout un pointeur RFC 6901 (`/0/counts/fields`) vers sa valeur verbatim ;
3. quand la valeur n'existe pas, renvoie un PLACEHOLDER `{{module:file#pointer}}`
— jamais un chiffre fabriqué, jamais un `0` posé.
La résolution est PURE et rejouable : l'invariant `INV_ANTI_INVENTION` du CLI
re-résout indépendamment chaque pointeur et compare à ce qui figure dans le
run-sheet — impossible de « poser » une valeur à la main.
"""
from __future__ import annotations
import json
import os
from typing import Any
from .deps import DELIVERABLES_ROOT
# Sentinelle « non résolu » — un placeholder lisible, jamais confondable avec une
# vraie valeur (ni un nombre, ni une chaîne métier).
_MISSING = object()
def placeholder(module: str, file: str, pointer: str) -> str:
return f"{{{{{module}:{file}#{pointer}}}}}"
def is_placeholder(value: Any) -> bool:
return isinstance(value, str) and value.startswith("{{") and value.endswith("}}")
def artifact_path(module: str, file: str) -> str:
"""Chemin absolu de l'artefact `out/` (jamais hors des livrables)."""
return os.path.join(DELIVERABLES_ROOT, module, file)
class Loader:
"""Charge et met en cache les artefacts amont pour une passe de build."""
def __init__(self) -> None:
self._cache: dict[str, Any] = {}
self._errors: dict[str, str] = {}
def load(self, module: str, file: str) -> Any:
key = f"{module}/{file}"
if key in self._cache or key in self._errors:
return self._cache.get(key, _MISSING)
path = artifact_path(module, file)
if not os.path.isfile(path):
self._errors[key] = "artefact absent"
return _MISSING
try:
with open(path, encoding="utf-8") as fh:
obj = json.load(fh)
except (OSError, ValueError) as exc:
self._errors[key] = f"JSON invalide : {exc}"
return _MISSING
self._cache[key] = obj
return obj
def errors(self) -> dict[str, str]:
return dict(self._errors)
def _split_pointer(pointer: str) -> list[str]:
"""Découpe un pointeur RFC 6901 (`/a/0/b`) en tokens déséchappés."""
if pointer == "":
return []
if not pointer.startswith("/"):
raise ValueError(f"pointeur RFC 6901 invalide (doit commencer par '/') : {pointer!r}")
tokens = pointer.split("/")[1:]
return [t.replace("~1", "/").replace("~0", "~") for t in tokens]
def _descend(obj: Any, token: str) -> Any:
"""Un pas de descente : clé d'objet ou index de liste (jamais d'erreur)."""
if isinstance(obj, dict):
return obj.get(token, _MISSING)
if isinstance(obj, list):
if token.lstrip("-").isdigit():
idx = int(token)
if -len(obj) <= idx < len(obj):
return obj[idx]
return _MISSING
return _MISSING
def resolve(loader: Loader, module: str, file: str, pointer: str) -> tuple[Any, bool]:
"""Résout un pointeur en `(valeur, resolu)`.
`resolu=False` ⇒ `valeur` est un placeholder `{{…}}` (artefact absent / JSON
invalide / chemin inexistant). `resolu=True` ⇒ valeur verbatim non `None`.
Une valeur littérale `null` dans l'artefact est traitée comme NON résolue :
un chiffre « à confirmer » ne doit jamais alimenter un pitch (#6).
"""
root = loader.load(module, file)
if root is _MISSING:
return placeholder(module, file, pointer), False
node: Any = root
for token in _split_pointer(pointer):
node = _descend(node, token)
if node is _MISSING:
return placeholder(module, file, pointer), False
if node is None:
return placeholder(module, file, pointer), False
return node, True