[DTP-Worker] Sprint 6 · Générateur Chat OTOIA embarqué par portail (Custom Block natif) (ERPNext Backend · roadmap L63)

Carrier natif ERPNext v15 (Custom Block) + config runtime par portail.
Ancrage RBAC : roles_allowed/knowledge_scope = surface exacte du portail,
synchronisés avec les Has Role des Workspaces. Endpoint OTOIA null (a_confirmer).
Persona Amélie + capabilities + langues FR/EN/ES sourcés (anti-invention #6).
14 invariants · 31 tests · régression 408 tests verts.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

05_deliverables_mvp/frontend/chat_otoia/chat_otoia_gen.py

@@ -0,0 +1,281 @@
+#!/usr/bin/env python3
+"""Générateur du Chat OTOIA embarqué par portail (Sprint 6 · ERPNext Backend).
+
+Livrable ERPNext Backend de la roadmap Sprint 6 (l.63 « Chat OTOIA embedded dans
+chaque portail »). Contrainte #1 « ERPNext natif = priorité absolue » : le bloc de
+contenu réutilisable inséré dans un `Workspace` v15 EST le DocType `Custom Block`.
+On produit donc un `Custom Block` par portail + sa config runtime — aucun framework
+de chat externe côté worker.
+
+Transforme (rbac_50_roles.json + chat_spec.json + portails_spec.json +
+portails/out/workspace.json + seo_spec.json) en :
+  - custom_block.json -> 1 fixture `Custom Block` (carrier natif) par portail
+  - chat_mount.json   -> 1 config runtime par portail (persona, langues, rôles
+                         autorisés, portée de connaissance, capabilities, endpoint=null)
+  - MANIFEST.json     -> traçabilité (comptes, résumé par portail, marque, hand-off)
+
+Sous-commandes :
+  build    [-o OUT]   -> écrit custom_block.json / chat_mount.json / MANIFEST.json
+  validate [-o OUT]   -> (re)génère en mémoire, valide vs chat.schema.json + invariants
+
+Sortie déterministe (tri stable, aucun horodatage) -> diffable + re-générable.
+Anti-invention (#6) : portails = portails métier ; rôles + portée de connaissance =
+surface RBAC EXACTE du portail (mêmes valeurs que les Has Role des Workspaces) ;
+endpoint OTOIA = null (a_confirmer · VPS) ; persona/capabilities/langues sourcés.
+Ce worker n'écrit JAMAIS sur le VPS (#8) : hand-off à l'agent ERPNext Backend.
+"""
+
+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 chatlib import builder, deps  # noqa: E402
+
+_DEFAULT_OUT = os.path.join(_HERE, "out")
+
+_EXPECTED_CAPS = ["aec.py", "knowledge.py", "prompt_engine.py", "chat.py"]
+
+
+def _eprint(*args) -> None:
+    print(*args, file=sys.stderr)
+
+
+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:
+    contract = deps.load_contract()
+    spec = deps.load_spec()
+    portails_spec = deps.load_portails_spec()
+    langs, default_lang = deps.load_langs()
+    return builder.build_bundle(contract, spec, portails_spec, langs, default_lang)
+
+
+def _workspace_roles(workspaces: list) -> dict[str, list[str]]:
+    """Has Role réels par Workspace (label -> rôles triés), depuis le hand-off portails."""
+    out: dict[str, list[str]] = {}
+    for ws in workspaces:
+        out[ws["label"]] = sorted(row["role"] for row in ws.get("roles", []))
+    return out
+
+
+def _validate_bundle(bundle: dict) -> list[str]:
+    schema = deps.load_json(deps.SCHEMA_PATH)
+    errors = deps.validate(bundle, schema)
+
+    contract = deps.load_contract()
+    spec = deps.load_spec()
+    portails_spec = deps.load_portails_spec()
+    langs, default_lang = deps.load_langs()
+    workspaces = deps.load_workspaces()
+
+    custom_blocks = bundle["custom_block"]
+    mounts = bundle["chat_mount"]
+    manifest = bundle["manifest"]
+
+    portails_metier = set(contract.get("portails_business", []))
+    spec_keys = [p["key"] for p in portails_spec["portails"]]
+    roles_map = deps.roles_by_portail(contract)
+    dt_map = deps.doctypes_by_portail(contract)
+    ws_roles = _workspace_roles(workspaces)
+    labels = {p["key"]: p["label"] for p in portails_spec["portails"]}
+
+    # Invariant 1 · EXACTEMENT les portails métier (roadmap « chaque portail »),
+    # la console technique `plateforme` étant exclue (comme les Workspaces).
+    mount_keys = [m["portail"] for m in mounts]
+    if set(mount_keys) != portails_metier:
+        errors.append(
+            f"portails du chat {sorted(set(mount_keys))} != portails_business {sorted(portails_metier)}."
+        )
+    if mount_keys != spec_keys:
+        errors.append(f"ordre/contenu portails {mount_keys} != portails_spec {spec_keys}.")
+
+    # Invariant 2 · 1 Custom Block ⇔ 1 mount ⇔ 1 portail ; block_name unique ; ordre stable.
+    if not (len(custom_blocks) == len(mounts) == len(portails_metier)):
+        errors.append(
+            f"comptes divergents : {len(custom_blocks)} blocks / {len(mounts)} mounts "
+            f"/ {len(portails_metier)} portails métier."
+        )
+    bn = [cb["block_name"] for cb in custom_blocks]
+    if len(bn) != len(set(bn)):
+        errors.append("block_name en doublon.")
+
+    # Invariant 8 (global) · capabilities = les 4 capacités OTOIA de la spec, sourcées.
+    spec_caps = [c["module"] for c in spec["capabilities"]]
+    if spec_caps != _EXPECTED_CAPS:
+        errors.append(f"capabilities spec {spec_caps} != {_EXPECTED_CAPS} (CLAUDE.md).")
+    for c in spec["capabilities"]:
+        if not c.get("source"):
+            errors.append(f"capability {c.get('module')!r} sans source (anti-invention).")
+
+    # Invariant 9 (global) · langues = seo_spec (source unique), aucune langue inventée.
+    if list(langs) != ["fr", "en", "es"] or default_lang != "fr":
+        errors.append(f"langues seo inattendues: {langs}/{default_lang}.")
+
+    mount_by_key = {m["portail"]: m for m in mounts}
+    cb_by_bn = {cb["block_name"]: cb for cb in custom_blocks}
+
+    for key in spec_keys:
+        m = mount_by_key[key]
+        label = labels[key]
+
+        # Invariant 3 · ANTI-INVENTION endpoint : jamais fabriqué (null · a_confirmer).
+        if m["endpoint"] is not None:
+            errors.append(f"[{key}] endpoint doit rester null (a_confirmer · VPS #8).")
+
+        # Invariant 4 · portée de connaissance = surface RBAC EXACTE du portail.
+        expected_scope = sorted(dt_map.get(key, set()))
+        if m["knowledge_scope"] != expected_scope:
+            errors.append(f"[{key}] knowledge_scope != DocTypes autorisés RBAC.")
+        if not expected_scope:
+            errors.append(f"[{key}] portée de connaissance vide (portail sans DocType ?).")
+
+        # Invariant 5 · rôles autorisés = rôles du portail (contrat), triés.
+        expected_roles = sorted(roles_map.get(key, []))
+        if m["roles_allowed"] != expected_roles:
+            errors.append(f"[{key}] roles_allowed != rôles du portail (contrat).")
+
+        # Invariant 6 · COHÉRENCE INTER-LIVRABLES : rôles du chat == Has Role du
+        # Workspace correspondant (frontend/portails/out/workspace.json).
+        if label not in ws_roles:
+            errors.append(f"[{key}] aucun Workspace '{label}' dans le hand-off portails.")
+        elif m["roles_allowed"] != ws_roles[label]:
+            errors.append(f"[{key}] roles_allowed != Has Role du Workspace '{label}'.")
+
+        # Invariant 7 · persona sourcée, identique partout.
+        if m["persona"] != {"nom": "Amélie", "voix": "multilingual_v2"}:
+            errors.append(f"[{key}] persona != Amélie/multilingual_v2 (CLAUDE.md).")
+
+        # Invariant 8 · capabilities du mount = les 4 capacités, sans ajout.
+        if m["capabilities"] != _EXPECTED_CAPS:
+            errors.append(f"[{key}] capabilities mount != {_EXPECTED_CAPS}.")
+
+        # Invariant 9 · langues + défaut du mount = source seo (aucune invention).
+        if m["langues"] != list(langs) or m["lang_defaut"] != default_lang:
+            errors.append(f"[{key}] langues/défaut != seo_spec.")
+
+        # Invariant 13 · html_id déterministe.
+        if m["html_id"] != f"otoia-chat-{key}":
+            errors.append(f"[{key}] html_id non déterministe: {m['html_id']!r}.")
+
+        # Invariant 11 · Custom Block : html de montage bien formé, sans chiffre,
+        # sans URL inventée ; block_name = préfixe + label.
+        cb_name = m["custom_block"]
+        if cb_name != f"OTOIA Chat · {label}":
+            errors.append(f"[{key}] block_name non déterministe: {cb_name!r}.")
+        if cb_name not in cb_by_bn:
+            errors.append(f"[{key}] Custom Block '{cb_name}' absent du bundle.")
+        else:
+            html = cb_by_bn[cb_name]["html"]
+            for token in (m["html_id"], f'data-portail="{key}"', 'data-assistant="Amélie"'):
+                if token not in html:
+                    errors.append(f"[{key}] html de montage sans {token!r}.")
+            if "http://" in html or "https://" in html:
+                errors.append(f"[{key}] html contient une URL (endpoint jamais fabriqué #6).")
+            if any(ch.isdigit() for ch in html):
+                errors.append(f"[{key}] html contient un chiffre (aucun chiffre émis #6).")
+
+    # Invariant 12 · comptes du manifeste cohérents.
+    c = manifest["counts"]
+    knowledge_union = sorted({dt for m in mounts for dt in m["knowledge_scope"]})
+    checks = {
+        "portails": len(mounts),
+        "custom_blocks": len(custom_blocks),
+        "mounts": len(mounts),
+        "roles_couverts": sum(len(m["roles_allowed"]) for m in mounts),
+        "knowledge_doctypes_uniques": len(knowledge_union),
+    }
+    for k, v in checks.items():
+        if c.get(k) != v:
+            errors.append(f"counts.{k} incohérent ({c.get(k)} != {v}).")
+    if manifest["endpoint_statut"] != "a_confirmer":
+        errors.append("manifest.endpoint_statut != a_confirmer.")
+
+    # Invariant 10 · tokens de marque verbatim (CLAUDE.md #4).
+    brand = manifest["brand"]
+    if brand["accent"]["valeur"] != "#f0b429" or brand["fond"]["valeur"] != "#0a0a12":
+        errors.append("brand fond/accent != CLAUDE.md #4.")
+
+    # Invariant 14 · déterminisme : deux builds successifs identiques.
+    again = _build_bundle()
+    if json.dumps(again, sort_keys=True, ensure_ascii=False) != json.dumps(bundle, sort_keys=True, ensure_ascii=False):
+        errors.append("build non déterministe (deux passes divergent).")
+
+    return errors
+
+
+def cmd_build(args: argparse.Namespace) -> int:
+    bundle = _build_bundle()
+    errors = _validate_bundle(bundle)
+    if errors:
+        _eprint("❌ Bundle 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, "custom_block.json"), bundle["custom_block"])
+    _write_json(os.path.join(out, "chat_mount.json"), bundle["chat_mount"])
+    _write_json(os.path.join(out, "MANIFEST.json"), bundle["manifest"])
+
+    c = bundle["manifest"]["counts"]
+    print(f"✅ Chat OTOIA généré dans {out}")
+    print(
+        f"   {c['portails']} portails · {c['custom_blocks']} Custom Blocks · "
+        f"{c['roles_couverts']} rôles couverts · {c['knowledge_doctypes_uniques']} "
+        f"DocTypes de connaissance (endpoint a_confirmer)"
+    )
+    for m in bundle["manifest"]["portails"]:
+        print(
+            f"     - {m['workspace']} : {m['nb_roles']} rôles / "
+            f"{m['nb_knowledge_doctypes']} DocTypes · block « {m['custom_block']} »"
+        )
+    return 0
+
+
+def cmd_validate(args: argparse.Namespace) -> int:
+    bundle = _build_bundle()
+    errors = _validate_bundle(bundle)
+    if errors:
+        _eprint("❌ Validation KO :")
+        for e in errors:
+            _eprint(f"   - {e}")
+        return 1
+    c = bundle["manifest"]["counts"]
+    print(
+        f"✅ Validation OK — {c['portails']} portails, {c['roles_couverts']} rôles, "
+        f"schéma + 14 invariants verts."
+    )
+    return 0
+
+
+def main(argv: list[str] | None = None) -> int:
+    p = argparse.ArgumentParser(
+        description="Générateur du Chat OTOIA embarqué par portail (Custom Block + config runtime)."
+    )
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    pb = sub.add_parser("build", help="génère custom_block.json / chat_mount.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 bundle (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())