Claude Code DTP Worker
58ed555db0
Agrégateur de méta-niveau au-dessus du run-book RBAC : ordonne le déploiement VPS de TOUS les livrables gated en 7 phases (Prérequis → DocTypes → RBAC → Workflow/métier → Frontend → Contenu → Vérification QA), avec graphe de dépendances inter-phases acyclique et confirmations préalables sourcées. Anti-invention (#6) : périmètre dérivé du CI (parse_ci réutilisé), couverture bijective module→phase (un module gated non planifié OU un module planifié non gated → refus), SoD (auto-exclusion), zéro chiffre métier (confirmations sourcées via audit_5d D1.1/D1.2/D1.3/D2.3 + endpoint OTOIA). Vérifs : 29/29 tests module (dont 14 injections négatives) · audit_4big PASS 20/20 à 100 · régression run 20/20 suites · 503 tests · 0 échec · gate CI local vert. CI job devops-deploy-runbook-tests + gate ; enregistrement audit_4big (19→20) ; plan régression régénéré (19→20 suites). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
297 lines
12 KiB
Python
297 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Run-book de déploiement VPS unifié (Sprint 8 · DevOps).
|
|
|
|
Volet « Deployment production complet » réalisable en-repo. Une couche AU-DESSUS
|
|
du run-book RBAC (`rbac/apply_plan`) : celui-ci n'ordonne que les 3 volets RBAC ;
|
|
ce module-ci ordonne le déploiement VPS de TOUS les livrables gated du mandat en
|
|
un seul plan de phases.
|
|
|
|
Périmètre PROUVÉ, pas déclaré : l'ensemble des modules à déployer est DÉRIVÉ de
|
|
`.gitea/workflows/ci.yml` (réutilise `q4lib.registry.parse_ci` — zéro duplication
|
|
· #5), puis mis en correspondance BIJECTIVE avec le `module_phase` du spec. Tout
|
|
livrable ajouté au CI sans entrée de phase casse la couverture → génération
|
|
refusée (aucune omission silencieuse).
|
|
|
|
Cible de portage VPS : agent DevOps / ERPNext Backend. Ce worker n'écrit JAMAIS
|
|
sur le VPS (CLAUDE.md #8) — il produit en-repo un plan ordonné que l'agent suit.
|
|
|
|
Produit :
|
|
- deploy_runbook.json → plan ordonné de phases (modules gated assignés,
|
|
dépendances inter-phases, confirmations préalables
|
|
sourcées).
|
|
- MANIFEST.json → manifeste agrégé (comptes, preuve de couverture
|
|
bijective vs CI, graphe des phases, confirmations
|
|
ouvertes).
|
|
|
|
Sous-commandes :
|
|
build [-o OUT] → écrit deploy_runbook.json / MANIFEST.json (après validation)
|
|
validate [-o OUT] → (re)génère en mémoire, valide vs deploy.schema.json +
|
|
invariants — sort en erreur si un invariant casse.
|
|
|
|
Sortie déterministe (ordre de phases fixe, listes triées, aucun horodatage).
|
|
Anti-invention (#6) : aucun chiffre métier ; les paramètres non confirmés
|
|
restent des confirmations sourcées (owner + source), jamais fabriquées.
|
|
"""
|
|
|
|
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 deploylib import builder, deps # noqa: E402
|
|
|
|
_SPEC_PATH = os.path.join(_HERE, "deploy_spec.json")
|
|
_SCHEMA_PATH = os.path.join(_HERE, "deploy.schema.json")
|
|
_DEFAULT_OUT = os.path.join(_HERE, "out")
|
|
|
|
|
|
def _eprint(*args) -> None:
|
|
print(*args, file=sys.stderr)
|
|
|
|
|
|
def _load(path: str) -> dict:
|
|
with open(path, encoding="utf-8") as fh:
|
|
return json.load(fh)
|
|
|
|
|
|
def _write_json(path: str, data) -> None:
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
json.dump(data, fh, ensure_ascii=False, indent=2)
|
|
fh.write("\n")
|
|
|
|
|
|
def _build_bundle() -> dict:
|
|
"""Charge le spec, lit la réalité du CI, assemble le run-book."""
|
|
spec = _load(_SPEC_PATH)
|
|
ci = deps.parse_ci()
|
|
return builder.build_runbook(spec, ci)
|
|
|
|
|
|
def validate_bundle(bundle: dict, spec: dict, ci: dict) -> list[str]:
|
|
"""Valide le run-book contre deploy.schema.json + invariants de déploiement.
|
|
|
|
Fonction pure sur (bundle, spec, ci) → les tests injectent des dictionnaires
|
|
synthétiques pour prouver le basculement en erreur (tests négatifs).
|
|
"""
|
|
schema = _load(_SCHEMA_PATH)
|
|
errors = list(deps.validate(bundle, schema))
|
|
|
|
runbook = bundle["runbook"]
|
|
manifest = bundle["manifest"]
|
|
self_module = spec["self_module"]
|
|
|
|
# INV1 · phases séquentielles 1..N, ids uniques, alignées sur le spec.
|
|
orders = [p["order"] for p in runbook]
|
|
if orders != list(range(1, len(runbook) + 1)):
|
|
errors.append(f"phases: champ `order` non séquentiel {orders}.")
|
|
ids = [p["id"] for p in runbook]
|
|
if len(set(ids)) != len(ids):
|
|
errors.append(f"phases: ids en double {ids}.")
|
|
if ids != [p["id"] for p in spec["phases"]]:
|
|
errors.append("phases: ordre/ids divergents du spec.")
|
|
|
|
# INV2 · graphe inter-phases : toute dépendance existe ET pointe vers une
|
|
# phase STRICTEMENT antérieure (déploiement applicable dans l'ordre, sans
|
|
# cycle ni renvoi en avant).
|
|
order_of = {p["id"]: p["order"] for p in runbook}
|
|
for p in runbook:
|
|
for dep in p["depends_on"]:
|
|
if dep not in order_of:
|
|
errors.append(f"phase {p['id']}: dépendance inconnue {dep!r}.")
|
|
elif order_of[dep] >= p["order"]:
|
|
errors.append(f"phase {p['id']}: dépend d'une phase postérieure {dep!r}.")
|
|
|
|
# INV3 · COUVERTURE BIJECTIVE (cœur anti-invention · #6) : les modules
|
|
# planifiés == les modules gated par le CI (hors self). Aucun oubli, aucune
|
|
# invention. On RECALCULE indépendamment depuis le CI, on ne fait pas
|
|
# confiance au manifeste.
|
|
gated = set(builder.gated_modules(ci, self_module))
|
|
mapped = set(spec["module_phase"])
|
|
missing = sorted(gated - mapped)
|
|
extra = sorted(mapped - gated)
|
|
if missing:
|
|
errors.append(f"couverture: modules gated non planifiés {missing}.")
|
|
if extra:
|
|
errors.append(f"couverture: modules planifiés non gated {extra}.")
|
|
cov = manifest["coverage"]
|
|
if cov["missing_in_map"] != missing or cov["extra_in_map"] != extra:
|
|
errors.append("coverage: diagnostic du manifeste ≠ recalcul indépendant.")
|
|
if cov["bijective"] != (not missing and not extra):
|
|
errors.append("coverage.bijective incohérent avec les écarts calculés.")
|
|
|
|
# INV4 · séparation des pouvoirs (ISA 315) : le run-book ne se déploie pas
|
|
# lui-même et ne se compte pas.
|
|
if self_module in mapped:
|
|
errors.append(f"SoD: le module {self_module!r} ne doit pas figurer dans module_phase.")
|
|
for p in runbook:
|
|
if any(m["module"] == self_module for m in p["modules"]):
|
|
errors.append(f"SoD: {self_module!r} présent dans la phase {p['id']}.")
|
|
|
|
# INV5 · chaque module gated est assigné à EXACTEMENT une phase, et chaque
|
|
# phase du run-book est non vide (schéma garantit minItems=1, on revérifie
|
|
# la partition).
|
|
assigned: dict[str, int] = {}
|
|
for p in runbook:
|
|
for m in p["modules"]:
|
|
assigned[m["module"]] = assigned.get(m["module"], 0) + 1
|
|
dups = sorted(m for m, n in assigned.items() if n > 1)
|
|
if dups:
|
|
errors.append(f"partition: modules assignés à plusieurs phases {dups}.")
|
|
if set(assigned) != gated:
|
|
errors.append("partition: l'union des phases ≠ modules gated.")
|
|
|
|
# INV6 · toute phase déclare une phase connue pour chaque module (pas de
|
|
# `module_phase` pointant vers une phase absente du spec).
|
|
if manifest["coverage"]["unknown_phase"]:
|
|
errors.append(
|
|
f"module_phase: phases inconnues {manifest['coverage']['unknown_phase']}."
|
|
)
|
|
|
|
# INV7 · CONFIRMATIONS sourcées + intègres (anti-invention · #6) :
|
|
# - chaque confirmation référencée par une phase existe au catalogue ;
|
|
# - aucune entrée de catalogue orpheline (non référencée) ;
|
|
# - chaque confirmation porte owner + source NON vides ;
|
|
# - AUCUN champ de valeur chiffrée (`valeur`/`value`/`montant`/`taux`) :
|
|
# un paramètre non confirmé reste une confirmation, jamais un chiffre.
|
|
catalogue = spec["confirmations"]
|
|
if manifest["graph"]["unknown_confirmations"]:
|
|
errors.append(
|
|
f"confirmations inconnues référencées {manifest['graph']['unknown_confirmations']}."
|
|
)
|
|
if manifest["graph"]["orphan_confirmations"]:
|
|
errors.append(
|
|
f"confirmations orphelines (non référencées) {manifest['graph']['orphan_confirmations']}."
|
|
)
|
|
_VALUE_KEYS = {"valeur", "value", "montant", "taux", "amount", "rate"}
|
|
for cid, entry in catalogue.items():
|
|
if not entry.get("owner"):
|
|
errors.append(f"confirmation {cid!r}: owner manquant.")
|
|
if not entry.get("source"):
|
|
errors.append(f"confirmation {cid!r}: source manquante (anti-invention #6).")
|
|
bad = _VALUE_KEYS & set(entry)
|
|
if bad:
|
|
errors.append(f"confirmation {cid!r}: valeur chiffrée interdite {sorted(bad)} (#6).")
|
|
for p in runbook:
|
|
for c in p["confirmations"]:
|
|
if c["id"] not in catalogue:
|
|
errors.append(f"phase {p['id']}: confirmation inconnue {c['id']!r}.")
|
|
elif not c["source"] or not c["owner"]:
|
|
errors.append(f"phase {p['id']}: confirmation {c['id']!r} non sourcée.")
|
|
|
|
# INV8 · unicité job CI par module (le parseur ne doit pas mapper 2 jobs sur
|
|
# le même module — sinon `ci_job` serait ambigu).
|
|
j2p = ci.get("job_to_path", {})
|
|
path_jobs: dict[str, list[str]] = {}
|
|
for job, path in j2p.items():
|
|
path_jobs.setdefault(path, []).append(job)
|
|
for path in gated:
|
|
jobs = path_jobs.get(path, [])
|
|
if len(jobs) != 1:
|
|
errors.append(f"module {path!r}: {len(jobs)} job(s) CI (attendu 1) {sorted(jobs)}.")
|
|
|
|
# INV9 · chaque module du run-book porte un `ci_job` résolu (non null).
|
|
for p in runbook:
|
|
for m in p["modules"]:
|
|
if not m["ci_job"]:
|
|
errors.append(f"phase {p['id']}: module {m['module']!r} sans ci_job.")
|
|
|
|
# INV10 · déterminisme : modules triés par phase, confirmations_ouvertes triée.
|
|
for p in runbook:
|
|
names = [m["module"] for m in p["modules"]]
|
|
if names != sorted(names):
|
|
errors.append(f"phase {p['id']}: modules non triés (déterminisme).")
|
|
co = manifest["confirmations_ouvertes"]
|
|
if co != sorted(co):
|
|
errors.append("confirmations_ouvertes non triée (déterminisme).")
|
|
|
|
# INV11 · cohérence des comptes du manifeste avec le contenu.
|
|
c = manifest["counts"]
|
|
if c["phases"] != len(runbook):
|
|
errors.append("counts.phases ≠ nombre de phases.")
|
|
if c["modules"] != len(gated):
|
|
errors.append("counts.modules ≠ modules gated.")
|
|
if c["confirmations"] != len(catalogue):
|
|
errors.append("counts.confirmations ≠ taille du catalogue.")
|
|
|
|
return errors
|
|
|
|
|
|
def _validate_current() -> tuple[dict, list[str]]:
|
|
spec = _load(_SPEC_PATH)
|
|
ci = deps.parse_ci()
|
|
bundle = builder.build_runbook(spec, ci)
|
|
return bundle, validate_bundle(bundle, spec, ci)
|
|
|
|
|
|
def cmd_build(args: argparse.Namespace) -> int:
|
|
bundle, errors = _validate_current()
|
|
if errors:
|
|
_eprint("❌ Run-book invalide — génération refusée (anti-régression) :")
|
|
for e in errors:
|
|
_eprint(f" - {e}")
|
|
return 1
|
|
|
|
out = os.path.abspath(args.out)
|
|
os.makedirs(out, exist_ok=True)
|
|
_write_json(os.path.join(out, "deploy_runbook.json"), bundle["runbook"])
|
|
_write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"])
|
|
|
|
m = bundle["manifest"]
|
|
c = m["counts"]
|
|
print(f"✅ Run-book de déploiement généré dans {out}")
|
|
print(
|
|
f" deploy_runbook.json : {c['phases']} phases · {c['modules']} modules gated "
|
|
f"(couverture bijective vs CI) · MANIFEST.json : {c['confirmations']} confirmations."
|
|
)
|
|
for p in bundle["runbook"]:
|
|
mods = ", ".join(x["module"] for x in p["modules"])
|
|
conf = f" · ⚠ {len(p['confirmations'])} conf." if p["confirmations"] else ""
|
|
print(f" {p['order']}. [{p['responsable']}] {p['titre']}{conf}")
|
|
print(f" modules: {mods}")
|
|
if m["confirmations_ouvertes"]:
|
|
print(f" Confirmations préalables VPS : {', '.join(m['confirmations_ouvertes'])}")
|
|
return 0
|
|
|
|
|
|
def cmd_validate(args: argparse.Namespace) -> int:
|
|
bundle, errors = _validate_current()
|
|
if errors:
|
|
_eprint("❌ Validation KO :")
|
|
for e in errors:
|
|
_eprint(f" - {e}")
|
|
return 1
|
|
m = bundle["manifest"]
|
|
print(
|
|
f"✅ Validation OK — {m['counts']['phases']} phases, couverture bijective "
|
|
f"{m['coverage']['mapped_modules_count']}/{m['coverage']['ci_modules_count']} "
|
|
f"modules gated, schéma + 11 familles d'invariants verts."
|
|
)
|
|
return 0
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
p = argparse.ArgumentParser(
|
|
description="Run-book de déploiement VPS unifié (tous livrables gated → 1 plan de phases)."
|
|
)
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
|
|
pb = sub.add_parser("build", help="génère deploy_runbook.json / MANIFEST.json")
|
|
pb.add_argument("-o", "--out", default=_DEFAULT_OUT, help="dossier de sortie (défaut: ./out)")
|
|
pb.set_defaults(func=cmd_build)
|
|
|
|
pv = sub.add_parser("validate", help="valide le plan (schéma + invariants) sans écrire")
|
|
pv.add_argument("-o", "--out", default=_DEFAULT_OUT, help="ignoré (compat)")
|
|
pv.set_defaults(func=cmd_validate)
|
|
|
|
args = p.parse_args(argv)
|
|
return args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|