219 lines
9.3 KiB
Python
219 lines
9.3 KiB
Python
"""Format de skill commun (Fusion ACC-0.3) — lecture, validation,
|
|
conversion (ACC-4.5).
|
|
|
|
Le contrat figé vit dans `dracodev-commons/contrats/skill-format.md` :
|
|
un dossier par skill, un `SKILL.md` à frontmatter YAML (`name`,
|
|
`description` = déclencheur, `version`, `apps`, `domaines`, `outils`,
|
|
section `ecritures` obligatoire dès qu'il y a modification) et un
|
|
corps en trois sections ordonnées — **Quand l'utiliser**,
|
|
**Procédure**, **Garde-fous**.
|
|
|
|
Ce module est la part Flowdeck de la preuve du format (ACC-4.5) :
|
|
|
|
- ``analyser_skill_md`` : SKILL.md → structure (métadonnées + sections) ;
|
|
- ``valider`` : contrôle du contrat, liste d'erreurs lisibles ;
|
|
- ``vers_champs_flowdeck`` : conversion vers les champs d'un skill
|
|
Flowdeck (`agent_skills` / galerie) — les outils `flowdeck.*`
|
|
deviennent les `allowed_tools`, la Procédure devient le
|
|
`prompt_template` (précédée du déclencheur et suivie des garde-fous) ;
|
|
- ``depuis_preset_flowdeck`` : conversion inverse d'un preset de la
|
|
galerie vers un SKILL.md au format commun (point de départ d'une
|
|
conversion relue par un humain).
|
|
|
|
Les skills convertis de la galerie vivent dans `skills-communs/` à la
|
|
racine du dépôt ; ils sont installables tels quels via
|
|
``skill_gallery.upsert_skill(vers_champs_flowdeck(...))``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
import yaml
|
|
|
|
#: Sections du corps, dans l'ordre imposé par le contrat.
|
|
SECTIONS = ("Quand l'utiliser", "Procédure", "Garde-fous")
|
|
|
|
#: Valeurs admises pour l'approbation d'une écriture.
|
|
APPROBATIONS = ("toujours", "selon-portee", "jamais")
|
|
|
|
#: Champs de frontmatter obligatoires (contrat § « En-tête »).
|
|
CHAMPS_REQUIS = ("name", "description", "version", "apps", "domaines")
|
|
|
|
_KEBAB = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$")
|
|
_TITRE_SECTION = re.compile(r"^#{1,3}\s+(.+?)\s*$", re.MULTILINE)
|
|
|
|
|
|
def analyser_skill_md(texte: str) -> dict:
|
|
"""SKILL.md → {meta, sections, corps}. Lève ValueError si le
|
|
frontmatter est absent ou illisible."""
|
|
brut = (texte or "").replace("\r\n", "\n")
|
|
if not brut.startswith("---\n"):
|
|
raise ValueError("SKILL.md doit commencer par un frontmatter YAML (---)")
|
|
fin = brut.find("\n---", 4)
|
|
if fin < 0:
|
|
raise ValueError("Frontmatter YAML non fermé (---)")
|
|
try:
|
|
meta = yaml.safe_load(brut[4:fin]) or {}
|
|
except yaml.YAMLError as exc:
|
|
raise ValueError(f"Frontmatter YAML illisible : {exc}") from exc
|
|
if not isinstance(meta, dict):
|
|
raise ValueError("Frontmatter YAML : un objet est attendu")
|
|
corps = brut[fin + 4:].lstrip("\n")
|
|
sections: dict[str, str] = {}
|
|
marques = [(m.start(), m.group(1).strip()) for m in _TITRE_SECTION.finditer(corps)]
|
|
for i, (pos, titre) in enumerate(marques):
|
|
if titre in SECTIONS:
|
|
debut = corps.find("\n", pos) + 1
|
|
fin_s = marques[i + 1][0] if i + 1 < len(marques) else len(corps)
|
|
sections[titre] = corps[debut:fin_s].strip()
|
|
return {"meta": meta, "sections": sections, "corps": corps}
|
|
|
|
|
|
def _outils_flowdeck(skill: dict) -> list[str]:
|
|
"""Noms d'outils Flowdeck déclarés (préfixe `flowdeck.` retiré),
|
|
lectures (`outils`) puis écritures (`ecritures`), sans doublons."""
|
|
meta = skill["meta"]
|
|
noms: list[str] = []
|
|
for entree in meta.get("outils") or []:
|
|
nom = str(entree).strip()
|
|
if nom.startswith("flowdeck."):
|
|
noms.append(nom[len("flowdeck."):])
|
|
for entree in meta.get("ecritures") or []:
|
|
if isinstance(entree, dict):
|
|
nom = str(entree.get("action") or "").strip()
|
|
if nom.startswith("flowdeck."):
|
|
noms.append(nom[len("flowdeck."):])
|
|
return list(dict.fromkeys(noms))
|
|
|
|
|
|
def valider(skill: dict) -> list[str]:
|
|
"""Contrôle le contrat ACC-0.3 → liste d'erreurs (vide = conforme)."""
|
|
erreurs: list[str] = []
|
|
meta = skill.get("meta") or {}
|
|
for champ in CHAMPS_REQUIS:
|
|
valeur = meta.get(champ)
|
|
if valeur is None or (isinstance(valeur, str) and not valeur.strip()) \
|
|
or (isinstance(valeur, list) and not valeur):
|
|
erreurs.append(f"Champ obligatoire manquant ou vide : {champ}")
|
|
nom = str(meta.get("name") or "")
|
|
if nom and not _KEBAB.match(nom):
|
|
erreurs.append(f"name doit être en kebab-case : {nom!r}")
|
|
for champ in ("apps", "domaines"):
|
|
valeur = meta.get(champ)
|
|
if valeur is not None and (not isinstance(valeur, list)
|
|
or not all(isinstance(v, str) for v in valeur)):
|
|
erreurs.append(f"{champ} doit être une liste de chaînes")
|
|
outils = meta.get("outils") or []
|
|
if not isinstance(outils, list) or not all(isinstance(o, str) for o in outils):
|
|
erreurs.append("outils doit être une liste de noms (app.outil)")
|
|
ecritures = meta.get("ecritures") or []
|
|
if not isinstance(ecritures, list):
|
|
erreurs.append("ecritures doit être une liste d'objets {action, approbation}")
|
|
ecritures = []
|
|
for entree in ecritures:
|
|
if not isinstance(entree, dict) or not str(entree.get("action") or "").strip():
|
|
erreurs.append("Chaque écriture doit déclarer une action")
|
|
continue
|
|
approbation = str(entree.get("approbation") or "selon-portee")
|
|
if approbation not in APPROBATIONS:
|
|
erreurs.append(
|
|
f"approbation inconnue pour {entree['action']} : {approbation!r} "
|
|
f"(attendu : {', '.join(APPROBATIONS)})")
|
|
# Règle du contrat : « ecritures » obligatoire dès que le skill
|
|
# modifie quelque chose — pour la part Flowdeck, un outil déclaré
|
|
# qui est un outil d'écriture interne impose la section.
|
|
if not ecritures and "flowdeck" in (meta.get("apps") or []):
|
|
from app.services.permission_manager import DESTRUCTIVE_TOOLS, WRITE_TOOLS
|
|
|
|
modifiants = [o for o in _outils_flowdeck(skill)
|
|
if o in WRITE_TOOLS or o in DESTRUCTIVE_TOOLS]
|
|
if modifiants:
|
|
erreurs.append(
|
|
"Section ecritures obligatoire : le skill utilise des outils "
|
|
f"d'écriture Flowdeck ({', '.join(modifiants)})")
|
|
sections = skill.get("sections") or {}
|
|
for titre in SECTIONS:
|
|
if not sections.get(titre):
|
|
erreurs.append(f"Section de corps manquante ou vide : « {titre} »")
|
|
return erreurs
|
|
|
|
|
|
def vers_champs_flowdeck(skill: dict) -> dict:
|
|
"""Skill au format commun → champs Flowdeck (`agent_skills`).
|
|
|
|
Le `prompt_template` assemble le déclencheur, la procédure et les
|
|
garde-fous ; les écritures déclarées restent visibles dans le
|
|
prompt (le skill ne s'exécute jamais sans la politique d'agent).
|
|
"""
|
|
erreurs = valider(skill)
|
|
if erreurs:
|
|
raise ValueError("Skill non conforme au format commun : " + " ; ".join(erreurs))
|
|
meta = skill["meta"]
|
|
sections = skill["sections"]
|
|
morceaux = [
|
|
"Quand l'utiliser\n" + sections["Quand l'utiliser"],
|
|
"Procédure\n" + sections["Procédure"],
|
|
"Garde-fous\n" + sections["Garde-fous"],
|
|
]
|
|
ecritures = meta.get("ecritures") or []
|
|
if ecritures:
|
|
lignes = "\n".join(
|
|
f"- {e['action']} : approbation "
|
|
f"{e.get('approbation') or 'selon-portee'}"
|
|
for e in ecritures)
|
|
morceaux.append(f"Écritures déclarées (format commun)\n{lignes}")
|
|
return {
|
|
"name": str(meta["name"]),
|
|
"description": str(meta["description"]).strip(),
|
|
"prompt_template": "\n\n".join(morceaux),
|
|
"allowed_tools": _outils_flowdeck(skill),
|
|
}
|
|
|
|
|
|
def _slug(nom: str) -> str:
|
|
import unicodedata
|
|
|
|
plat = unicodedata.normalize("NFD", nom.lower())
|
|
plat = "".join(c for c in plat if unicodedata.category(c) != "Mn")
|
|
slug = re.sub(r"[^a-z0-9]+", "-", plat).strip("-")
|
|
return slug or "skill"
|
|
|
|
|
|
def depuis_preset_flowdeck(preset: dict, *, domaines: list[str] | None = None) -> str:
|
|
"""Preset de la galerie Flowdeck → SKILL.md au format commun.
|
|
|
|
Point de départ d'une conversion (ACC-4.5) : les outils
|
|
d'écriture du preset deviennent la section `ecritures`
|
|
(approbation `selon-portee`, le régime par défaut du contrat) ;
|
|
le texte est à relire par un humain avant diffusion.
|
|
"""
|
|
from app.services.permission_manager import WRITE_TOOLS
|
|
|
|
outils = [str(o) for o in preset.get("allowed_tools") or []]
|
|
lectures = [o for o in outils if o not in WRITE_TOOLS]
|
|
ecritures = [o for o in outils if o in WRITE_TOOLS]
|
|
meta: dict = {
|
|
"name": _slug(str(preset.get("name") or "skill")),
|
|
"description": str(preset.get("description") or "").strip(),
|
|
"version": "1.0.0",
|
|
"apps": ["flowdeck"],
|
|
"domaines": domaines or ["pro"],
|
|
"outils": [f"flowdeck.{o}" for o in lectures],
|
|
}
|
|
if ecritures:
|
|
meta["ecritures"] = [
|
|
{"action": f"flowdeck.{o}", "approbation": "selon-portee"}
|
|
for o in ecritures
|
|
]
|
|
front = yaml.safe_dump(meta, allow_unicode=True, sort_keys=False).strip()
|
|
corps = (
|
|
f"## Quand l'utiliser\n\n{preset.get('description') or ''}\n\n"
|
|
f"## Procédure\n\n{preset.get('prompt_template') or ''}\n\n"
|
|
"## Garde-fous\n\n"
|
|
"- N'agir que dans les domaines déclarés ; un besoin croisé "
|
|
"entre domaines est signalé, jamais contourné.\n"
|
|
"- Ne jamais inventer un contenu absent des sources lues.\n"
|
|
)
|
|
return f"---\n{front}\n---\n\n{corps}"
|