From c1a144c6739dc39918ef47145349216e99fa9719 Mon Sep 17 00:00:00 2001 From: Dom Date: Fri, 29 May 2026 11:28:25 +0200 Subject: [PATCH] feat(vwb): expose competence yaml catalog --- core/competences/__init__.py | 14 ++ core/competences/catalog.py | 201 ++++++++++++++++++ tests/unit/test_competence_catalog_loader.py | 34 +++ .../backend/catalog_routes_v2_vlm.py | 56 ++++- 4 files changed, 302 insertions(+), 3 deletions(-) create mode 100644 core/competences/__init__.py create mode 100644 core/competences/catalog.py create mode 100644 tests/unit/test_competence_catalog_loader.py diff --git a/core/competences/__init__.py b/core/competences/__init__.py new file mode 100644 index 000000000..d926390de --- /dev/null +++ b/core/competences/__init__.py @@ -0,0 +1,14 @@ +"""Competence catalogue helpers.""" + +from .catalog import ( + CompetenceSummary, + load_competence_catalog_actions, + load_competences, +) + +__all__ = [ + "CompetenceSummary", + "load_competence_catalog_actions", + "load_competences", +] + diff --git a/core/competences/catalog.py b/core/competences/catalog.py new file mode 100644 index 000000000..9a7d68ba8 --- /dev/null +++ b/core/competences/catalog.py @@ -0,0 +1,201 @@ +"""Load Lea competence YAML files as runtime catalogue entries.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Iterable + +import yaml + + +REPO_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_COMPETENCE_ROOT = REPO_ROOT / "data" / "competences" +KNOWN_STATES = ("candidate", "supervised", "stable", "observed") + + +@dataclass(frozen=True) +class CompetenceSummary: + """Small, UI-safe projection of a persisted competence YAML.""" + + id: str + name: str + learning_state: str + intent_fr: str + source_path: str + methods: tuple[dict[str, Any], ...] + success_marker: dict[str, Any] + failure_message_template: dict[str, Any] + t2_known_gaps: tuple[dict[str, Any], ...] + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "name": self.name, + "learning_state": self.learning_state, + "intent_fr": self.intent_fr, + "source_path": self.source_path, + "methods": list(self.methods), + "success_marker": self.success_marker, + "failure_message_template": self.failure_message_template, + "t2_known_gaps": list(self.t2_known_gaps), + } + + +def load_competences( + *, + root: Path | str = DEFAULT_COMPETENCE_ROOT, + states: Iterable[str] | None = None, +) -> list[CompetenceSummary]: + """Load all competence YAML files under ``data/competences``. + + ``states`` filters by directory/``learning_state`` value. Returned entries + are sorted by state maturity first, then by id, to make catalogue output + deterministic. + """ + + competence_root = Path(root) + state_filter = set(states or KNOWN_STATES) + summaries: list[CompetenceSummary] = [] + + for state in KNOWN_STATES: + if state not in state_filter: + continue + state_dir = competence_root / state + if not state_dir.exists(): + continue + for path in sorted(state_dir.glob("*.yaml")): + summary = load_competence_file(path, repo_root=REPO_ROOT) + if summary.learning_state in state_filter: + summaries.append(summary) + + return sorted(summaries, key=lambda item: (KNOWN_STATES.index(item.learning_state), item.id)) + + +def load_competence_file(path: Path | str, *, repo_root: Path = REPO_ROOT) -> CompetenceSummary: + competence_path = Path(path) + with competence_path.open("r", encoding="utf-8") as handle: + data = yaml.safe_load(handle) or {} + + if not isinstance(data, dict): + raise ValueError(f"{competence_path} must contain a YAML mapping") + + competence_id = _required_text(data, "id", competence_path) + learning_state = _required_text(data, "learning_state", competence_path) + name = str(data.get("name") or competence_id) + intent = data.get("intent") if isinstance(data.get("intent"), dict) else {} + intent_fr = str(intent.get("fr") or name) + methods = _method_summaries(data.get("methods")) + success_marker = data.get("success_marker") if isinstance(data.get("success_marker"), dict) else {} + failure_template = ( + data.get("failure_message_template") + if isinstance(data.get("failure_message_template"), dict) + else {} + ) + promotion = data.get("promotion") if isinstance(data.get("promotion"), dict) else {} + gaps = promotion.get("t2_known_gaps") if isinstance(promotion.get("t2_known_gaps"), list) else [] + + try: + source_path = str(competence_path.resolve().relative_to(repo_root.resolve())) + except ValueError: + source_path = str(competence_path) + + return CompetenceSummary( + id=competence_id, + name=name, + learning_state=learning_state, + intent_fr=intent_fr, + source_path=source_path, + methods=tuple(methods), + success_marker=success_marker, + failure_message_template=failure_template, + t2_known_gaps=tuple(gap for gap in gaps if isinstance(gap, dict)), + ) + + +def load_competence_catalog_actions( + *, + root: Path | str = DEFAULT_COMPETENCE_ROOT, + states: Iterable[str] | None = ("candidate", "supervised", "stable"), +) -> list[dict[str, Any]]: + """Expose competences in the VWB action-catalogue shape.""" + + return [competence_to_catalog_action(item) for item in load_competences(root=root, states=states)] + + +def competence_to_catalog_action(summary: CompetenceSummary) -> dict[str, Any]: + method_labels = ", ".join( + str(method.get("kind") or method.get("primitive_ref") or method.get("id")) + for method in summary.methods + ) + description = f"Compétence Léa {summary.learning_state}: {summary.intent_fr}" + if method_labels: + description = f"{description} ({method_labels})" + + return { + "id": f"lea_competence_{summary.id}", + "name": summary.intent_fr, + "description": description, + "category": "lea_competence", + "icon": "🧠", + "source": "competence_yaml", + "competence_id": summary.id, + "learning_state": summary.learning_state, + "source_path": summary.source_path, + "parameters": { + "competence_id": { + "type": "string", + "required": True, + "default": summary.id, + "description": "Identifiant de la compétence Léa à tester ou rejouer", + }, + "supervised": { + "type": "boolean", + "required": False, + "default": True, + "description": "Exécuter en mode supervisé humain", + }, + }, + "methods": list(summary.methods), + "success_marker": summary.success_marker, + "failure_message_template": summary.failure_message_template, + "t2_known_gaps": list(summary.t2_known_gaps), + "examples": [ + { + "name": "Tester en supervision", + "description": f"Rejouer la compétence {summary.id} avec validation humaine", + "parameters": { + "competence_id": summary.id, + "supervised": True, + }, + } + ], + } + + +def _required_text(data: dict[str, Any], key: str, path: Path) -> str: + value = data.get(key) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"{path} missing required text field {key!r}") + return value.strip() + + +def _method_summaries(methods: Any) -> list[dict[str, Any]]: + if not isinstance(methods, list): + return [] + + summaries: list[dict[str, Any]] = [] + for method in methods: + if not isinstance(method, dict): + continue + summaries.append( + { + "id": method.get("id"), + "kind": method.get("kind"), + "primitive_ref": method.get("primitive_ref"), + "description": method.get("description"), + "parameters": method.get("parameters") if isinstance(method.get("parameters"), dict) else {}, + } + ) + return summaries + diff --git a/tests/unit/test_competence_catalog_loader.py b/tests/unit/test_competence_catalog_loader.py new file mode 100644 index 000000000..d76816e21 --- /dev/null +++ b/tests/unit/test_competence_catalog_loader.py @@ -0,0 +1,34 @@ +from core.competences.catalog import load_competence_catalog_actions, load_competences + + +def test_load_candidate_competences_from_yaml_catalog(): + competences = load_competences(states=("candidate",)) + + ids = {competence.id for competence in competences} + assert "open_windows_search" in ids + assert "key_win_r_wait_explorer_exe" in ids + assert "key_ctrl_s_wait_notepad_exe" in ids + assert "key_alt_f4_wait_windowsterminal_exe" in ids + assert all(competence.learning_state == "candidate" for competence in competences) + + +def test_competence_catalog_actions_include_runtime_gap_metadata(): + actions = load_competence_catalog_actions(states=("candidate",)) + by_competence_id = {action["competence_id"]: action for action in actions} + + alt_f4 = by_competence_id["key_alt_f4_wait_windowsterminal_exe"] + assert alt_f4["id"] == "lea_competence_key_alt_f4_wait_windowsterminal_exe" + assert alt_f4["category"] == "lea_competence" + assert alt_f4["learning_state"] == "candidate" + assert alt_f4["source"] == "competence_yaml" + assert "fermer la fenêtre Bloc-notes" in alt_f4["name"] + assert alt_f4["parameters"]["supervised"]["default"] is True + assert alt_f4["t2_known_gaps"][0]["id"] == "alt_f4_confirmation_dialog_not_covered" + + +def test_competence_catalog_actions_are_deterministic(): + first = load_competence_catalog_actions(states=("candidate",)) + second = load_competence_catalog_actions(states=("candidate",)) + + assert [action["id"] for action in first] == [action["id"] for action in second] + diff --git a/visual_workflow_builder/backend/catalog_routes_v2_vlm.py b/visual_workflow_builder/backend/catalog_routes_v2_vlm.py index a3ceb0c0e..1b1552e94 100644 --- a/visual_workflow_builder/backend/catalog_routes_v2_vlm.py +++ b/visual_workflow_builder/backend/catalog_routes_v2_vlm.py @@ -84,6 +84,14 @@ except ImportError: cv2 = None np = None +try: + from core.competences.catalog import load_competence_catalog_actions + COMPETENCE_CATALOG_AVAILABLE = True +except ImportError as e: + print(f"⚠️ Catalogue compétences Léa non disponible: {e}") + COMPETENCE_CATALOG_AVAILABLE = False + load_competence_catalog_actions = None + # ============================================================================ # OmniParser (Microsoft) - Détection d'éléments UI avancée @@ -1068,6 +1076,18 @@ catalog_bp = Blueprint('catalog', __name__, url_prefix='/api/vwb/catalog') _screen_capturer_instance = None +def _load_lea_competence_actions() -> List[Dict[str, Any]]: + """Load YAML-backed Lea competences for the VWB catalogue.""" + + if not COMPETENCE_CATALOG_AVAILABLE or load_competence_catalog_actions is None: + return [] + try: + return load_competence_catalog_actions() + except Exception as e: + print(f"⚠️ Erreur chargement catalogue compétences Léa: {e}") + return [] + + def get_screen_capturer(): """ Obtient l'instance du ScreenCapturer (initialisation paresseuse). @@ -1553,6 +1573,9 @@ def list_actions(): } ] + # Ajouter les compétences YAML Léa comme actions testables. + available_actions.extend(_load_lea_competence_actions()) + # Filtrer par catégorie if category_filter: available_actions = [ @@ -1701,7 +1724,18 @@ Cette action permet d'attendre qu'un élément apparaisse ou disparaisse de l'é """ } - action['documentation'] = documentation.get(action_id, "Documentation non disponible") + if action.get("source") == "competence_yaml": + action['documentation'] = """ +# Compétence Léa + +Cette entrée vient de `data/competences/*.yaml`. Elle représente une compétence apprise +ou candidate, avec méthode, marqueur de succès, message d'échec et gaps runtime. + +Le premier usage attendu est le replay supervisé : Léa tente la compétence, l'humain +valide ou corrige, puis les résultats alimentent l'historique d'apprentissage. + """ + else: + action['documentation'] = documentation.get(action_id, "Documentation non disponible") return jsonify({ "success": True, @@ -2727,6 +2761,8 @@ def list_categories(): } """ try: + competence_action_count = len(_load_lea_competence_actions()) + # Définir les catégories disponibles avec métadonnées available_categories = [ { @@ -2757,6 +2793,17 @@ def list_categories(): "isEnabled": True } ] + + if competence_action_count: + available_categories.append({ + "id": "lea_competence", + "name": "Compétences Léa", + "description": "Compétences YAML candidates ou supervisées à tester", + "icon": "🧠", + "actionCount": competence_action_count, + "color": "#673ab7", + "isEnabled": True + }) return jsonify({ "success": True, @@ -2793,9 +2840,12 @@ def catalog_health(): # Vérifier les services screen_capturer = get_screen_capturer() + competence_action_count = len(_load_lea_competence_actions()) + services_status = { "screen_capturer": screen_capturer is not None, - "actions": 7, # Nombre d'actions disponibles + "actions": 7 + competence_action_count, + "lea_competences": competence_action_count, "screen_capturer_method": getattr(screen_capturer, 'method', 'unavailable') if screen_capturer else 'unavailable' } @@ -2834,4 +2884,4 @@ def register_catalog_routes(app): print(" - GET /api/vwb/catalog/actions/") print(" - POST /api/vwb/catalog/execute") print(" - POST /api/vwb/catalog/validate") - print(" - GET /api/vwb/catalog/health") \ No newline at end of file + print(" - GET /api/vwb/catalog/health")