"""AI assistant skills & slash-commands. A *skill* is a reusable prompt/workflow the user can trigger from the assistant composer with ``/``. ObsiGate ships a set of built-in skills; users can create their own with ``/create-new-skill``. Built-in skills are code constants, while user skills are persisted per-user in ``data/skills.json``. The same module also exposes the metadata for the *admin* commands (``/help``, ``/providers``, ``/provider``, ``/model``, ``/keys``). Those are executed client-side (they only touch the picker / display info), but listing them here keeps the ``/`` menu single-sourced. """ from __future__ import annotations import json import logging import re from pathlib import Path from typing import Any logger = logging.getLogger("obsigate.skills") SKILLS_FILE = Path("data/skills.json") MAX_USER_SKILLS = 100 MAX_PROMPT_CHARS = 8000 _SKILL_ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,47}$") # ── Built-in skills ───────────────────────────────────────────────────────── # ``prompt`` is appended to the assistant system prompt when the skill is # selected. Keep prompts concise and language-agnostic: the model answers in # the user's language. BUILTIN_SKILLS: list[dict[str, Any]] = [ { "id": "research", "label": "Recherche structurée", "icon": "🔎", "type": "skill", "description": "Recherche structurée + recommandation", "prompt": ( "Applique un mode RECHERCHE STRUCTURÉE. Structure ta réponse en : " "1) Contexte et question reformulée, 2) Constats appuyés sur le contenu fourni, " "3) Options/approches avec avantages et limites, 4) Recommandation argumentée. " "Cite les sources (fichiers) utilisées." ), }, { "id": "create-new-skill", "label": "Créer un skill", "icon": "🛠️", "type": "skill", "special": "create_skill", "description": "Crée un workflow réutilisable (skill)", "prompt": "", }, { "id": "resume", "label": "Résumé", "icon": "📄", "type": "skill", "description": "Résumé / synthèse structurée", "prompt": ( "Produis un RÉSUMÉ structuré du contenu : idées clés, points importants, " "conclusions. Utilise des titres et des puces concises." ), }, { "id": "actions", "label": "Actions & to-dos", "icon": "✅", "type": "skill", "description": "Extraire les actions & to-dos", "prompt": ( "Extrais les ACTIONS et TO-DOS du contenu. Rends une liste de tâches markdown " "`- [ ] ...`, avec responsable et échéance si mentionnés, sinon `(à préciser)`." ), }, { "id": "reformuler", "label": "Reformuler", "icon": "✍️", "type": "skill", "description": "Réécriture clarté / ton", "prompt": ( "RÉÉCRIS le contenu pour améliorer la clarté et le ton, en préservant le sens. " "Retourne uniquement le texte reformulé." ), }, { "id": "correction", "label": "Correction", "icon": "🔤", "type": "skill", "description": "Correction grammaire / orthographe / style", "prompt": ( "CORRIGE la grammaire, l'orthographe et le style. Retourne le texte corrigé, " "puis une courte liste des corrections notables." ), }, { "id": "brainstorm", "label": "Brainstorm", "icon": "💡", "type": "skill", "description": "Générer des idées, angles, variantes", "prompt": ( "Mode BRAINSTORM : génère un maximum d'idées, angles et variantes pertinents. " "Regroupe-les par thème, sans juger, puis signale les plus prometteuses." ), }, { "id": "plan", "label": "Planifier", "icon": "🧭", "type": "skill", "description": "Planifier / structurer un document", "prompt": ( "PLANIFIE et structure un document : propose un plan détaillé (sections, " "sous-sections, objectif de chaque partie) et une progression logique." ), }, { "id": "ask", "label": "Q&R", "icon": "💬", "type": "skill", "description": "Q&A sur un contenu référencé", "prompt": ( "Mode QUESTION/RÉPONSE : réponds précisément à la question en te basant " "strictement sur le contenu référencé. Cite les passages/fichiers utilisés et " "dis clairement si l'information est absente." ), }, { "id": "meeting-note", "label": "Note de réunion", "icon": "📝", "type": "skill", "description": "Compte-rendu / note de réunion", "prompt": ( "Rédige une NOTE DE RÉUNION : participants, ordre du jour, décisions, " "points d'action (`- [ ] ...`), questions ouvertes et prochaines étapes." ), }, { "id": "livrable", "label": "Livrable", "icon": "📨", "type": "skill", "description": "Email / compte-rendu / message Slack", "prompt": ( "Rédige un LIVRABLE de communication (email, compte-rendu ou message Slack) " "clair et prêt à envoyer, adapté au canal et au destinataire indiqués." ), }, ] # ── Admin commands (handled client-side) ──────────────────────────────────── ADMIN_COMMANDS: list[dict[str, Any]] = [ { "id": "help", "label": "Aide", "icon": "❓", "type": "admin", "description": "Liste des commandes", }, { "id": "providers", "label": "Fournisseurs", "icon": "🔌", "type": "admin", "description": "Liste les fournisseurs actifs", }, { "id": "provider", "label": "Changer de fournisseur", "icon": "🔀", "type": "admin", "usage": "/provider ", "description": "Changer de fournisseur LLM", }, { "id": "model", "label": "Changer de modèle", "icon": "🧠", "type": "admin", "usage": "/model ", "description": "Changer de modèle LLM", }, { "id": "keys", "label": "Clés API", "icon": "🔑", "type": "admin", "description": "Fournisseurs avec clé API enregistrée", }, ] def list_builtin_skills() -> list[dict[str, Any]]: """Return a copy of the built-in skills.""" return [dict(skill) for skill in BUILTIN_SKILLS] def list_admin_commands() -> list[dict[str, Any]]: """Return a copy of the admin command metadata.""" return [dict(cmd) for cmd in ADMIN_COMMANDS] def _read_store() -> dict[str, list[dict[str, Any]]]: if not SKILLS_FILE.exists(): return {} try: data = json.loads(SKILLS_FILE.read_text(encoding="utf-8")) return data if isinstance(data, dict) else {} except Exception as exc: # pragma: no cover - corrupted file logger.warning("Cannot read skills store: %s", exc) return {} def _write_store(store: dict[str, list[dict[str, Any]]]) -> None: SKILLS_FILE.parent.mkdir(parents=True, exist_ok=True) tmp = SKILLS_FILE.with_suffix(".tmp") tmp.write_text(json.dumps(store, indent=2, ensure_ascii=False), encoding="utf-8") tmp.replace(SKILLS_FILE) def _username(user: dict | None) -> str: if not user: return "anonymous" return str(user.get("username") or "anonymous") def list_user_skills(user: dict | None) -> list[dict[str, Any]]: """Return the persisted custom skills for a user.""" store = _read_store() skills = store.get(_username(user), []) return [dict(s) for s in skills if isinstance(s, dict)] def list_skills(user: dict | None) -> dict[str, list[dict[str, Any]]]: """Return built-in skills, admin commands and the user's custom skills.""" return { "skills": list_builtin_skills() + list_user_skills(user), "commands": list_admin_commands(), } def get_skill_prompt(skill_id: str | None, user: dict | None) -> str | None: """Resolve a skill id to its prompt, searching built-ins then user skills.""" if not skill_id: return None for skill in BUILTIN_SKILLS: if skill["id"] == skill_id: return skill.get("prompt") or None for skill in list_user_skills(user): if skill.get("id") == skill_id: return skill.get("prompt") or None return None def create_user_skill(user: dict | None, payload: dict[str, Any]) -> dict[str, Any]: """Create and persist a custom skill for a user. Raises: ValueError: when the payload is invalid (bad id, duplicate, too many). """ skill_id = str(payload.get("id") or "").strip().lower() label = str(payload.get("label") or "").strip() prompt = str(payload.get("prompt") or "").strip() description = str(payload.get("description") or "").strip() if not _SKILL_ID_RE.match(skill_id): raise ValueError("Identifiant invalide (a-z, 0-9, '-', '_', max 48)") if any(s["id"] == skill_id for s in BUILTIN_SKILLS): raise ValueError(f"L'identifiant '{skill_id}' est réservé") if not label: raise ValueError("Le nom du skill est requis") if not prompt: raise ValueError("Le prompt du skill est requis") if len(prompt) > MAX_PROMPT_CHARS: raise ValueError("Le prompt est trop long") username = _username(user) store = _read_store() user_skills = store.get(username, []) if any(s.get("id") == skill_id for s in user_skills): raise ValueError(f"Le skill '{skill_id}' existe déjà") if len(user_skills) >= MAX_USER_SKILLS: raise ValueError("Trop de skills personnalisés") skill = { "id": skill_id, "label": label, "icon": str(payload.get("icon") or "🧩").strip() or "🧩", "type": "skill", "custom": True, "description": description or label, "prompt": prompt, } user_skills.append(skill) store[username] = user_skills _write_store(store) return skill def delete_user_skill(user: dict | None, skill_id: str) -> bool: """Delete a custom skill. Returns True when a skill was removed.""" username = _username(user) store = _read_store() user_skills = store.get(username, []) remaining = [s for s in user_skills if s.get("id") != skill_id] if len(remaining) == len(user_skills): return False store[username] = remaining _write_store(store) return True