From 42d49dd8bd4c273df21c2375cc0282a9060d13de Mon Sep 17 00:00:00 2001 From: Dom Date: Fri, 10 Apr 2026 09:01:52 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20L=C3=A9a=20personnalit=C3=A9=20?= =?UTF-8?q?=E2=80=94=20langage=20m=C3=A9tier=20multi-domaines?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Aspect 4/4 Léa : Léa parle le langage du métier, pas du robot. DomainContext enrichi avec 5 domaines : - tim_codage : CIM-10, CCAM, GHM, DP/DAS (enrichi) - comptabilite : factures HT/TVA/TTC, OCR, lettrage, PCG - rh_paie : bulletins, DSN, brut/net, congés, IJSS - stocks_logistique : BC/BL/BR, SKU, inventaires, picking - generic : fallback Nouvelle API DomainContext : - summarize_action(action, params) — click "DP" → "saisir le diagnostic principal" - pose_clarification_question(context) — question pertinente quand Léa bloque - describe_workflow_outcome(...) — rapport final en langage métier Exemples : TIM : "J'ai codé 14 dossiers sur 15. 1 en attente — codes CIM-10 ambigus." Compta : "Je ne trouve pas le champ montant de TVA. C'est bien la facture F2026-0145 ?" Intégration ui/messages.py : - Import lazy (pas de dépendance circulaire) - formatter_cible_non_trouvee utilise les templates de clarification métier - Rétro-compat : tous les anciens appels sans domain_id fonctionnent 47 nouveaux tests, 0 régression. Co-Authored-By: Claude Opus 4.6 (1M context) --- agent_v0/agent_v1/ui/messages.py | 151 ++++- agent_v0/server_v1/domain_context.py | 855 +++++++++++++++++++++++++- tests/unit/test_domain_personality.py | 543 ++++++++++++++++ 3 files changed, 1525 insertions(+), 24 deletions(-) create mode 100644 tests/unit/test_domain_personality.py diff --git a/agent_v0/agent_v1/ui/messages.py b/agent_v0/agent_v1/ui/messages.py index 19bbdf071..27514a956 100644 --- a/agent_v0/agent_v1/ui/messages.py +++ b/agent_v0/agent_v1/ui/messages.py @@ -19,7 +19,45 @@ from __future__ import annotations import re from dataclasses import dataclass from enum import Enum -from typing import Optional +from typing import Any, Mapping, Optional + + +# ---------------------------------------------------------------------------- +# Accès paresseux au DomainContext +# ---------------------------------------------------------------------------- +# +# On importe le module à l'appel pour éviter toute dépendance circulaire +# avec `agent_v0.server_v1.domain_context` (qui ne doit pas importer l'UI). +# Si l'import échoue (contexte client sans server_v1), on retombe sur None +# et les formatters gardent leur comportement générique historique. + + +def _get_domain_ctx(domain_id: Optional[str]): + """Récupérer un DomainContext si possible, sinon None (fallback).""" + if not domain_id: + return None + try: + from agent_v0.server_v1.domain_context import get_domain_context # lazy + return get_domain_context(domain_id) + except Exception: + return None + + +def _friendly_target(description: str, domain_id: Optional[str] = None) -> str: + """Transformer une description technique en langage métier si possible. + + Ex (tim_codage) : "DP" → "diagnostic principal" + Ex (comptabilite) : "TVA" → "montant de TVA" + Retombe sur la description nettoyée si aucun domaine ne matche. + """ + base = _nettoyer_description_cible(description) + ctx = _get_domain_ctx(domain_id) + if ctx is None or not base: + return base + try: + return ctx._apply_synonyms(base) + except Exception: + return base class NiveauMessage(Enum): @@ -123,19 +161,57 @@ def _nettoyer_description_cible(description: str) -> str: def formatter_cible_non_trouvee( description_cible: str, titre_fenetre: Optional[str] = None, + domain_id: Optional[str] = None, + params: Optional[Mapping[str, Any]] = None, ) -> MessageUtilisateur: """Message quand Léa ne trouve pas un élément à cliquer. + Si un domaine métier est fourni, la description de la cible est + transformée en langage métier via le DomainContext : + - tim_codage + "DP" → "diagnostic principal" + - comptabilite + "TVA" → "montant de TVA" + Exemple avant : target_not_found: 'bonjour' dans *bonjour, – Bloc-notes Exemple après : Léa a besoin d'aide Je ne trouve pas "bonjour" dans le Bloc-notes. Peux-tu cliquer dessus toi-même ? Je reprends ensuite. + + Args: + description_cible: Description brute de la cible. + titre_fenetre: Titre de la fenêtre active (pour extraire l'app). + domain_id: Domaine métier pour enrichir la sortie (optionnel). + params: Paramètres du workflow (nom_patient, num_facture...) + utilisés par les templates de clarification métier. """ - cible = _nettoyer_description_cible(description_cible) or "l'élément" + cible = _friendly_target(description_cible, domain_id) or "l'élément" app = _extraire_nom_application(titre_fenetre or "") + # Si un domaine et un template de clarification existent, préférer la + # question métier (plus pertinente que le message générique). + ctx = _get_domain_ctx(domain_id) + if ctx is not None and ctx.clarification_templates: + try: + corps = ctx.pose_clarification_question( + { + "blocked_on": "target_not_found", + "target": description_cible or "", + "app": app, + "params": dict(params or {}), + } + ) + except Exception: + corps = "" + if corps: + return MessageUtilisateur( + niveau=NiveauMessage.BLOCAGE, + titre="Léa a besoin d'aide", + corps=corps, + duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE], + persistent=True, + ) + if app: corps = ( f"Je ne trouve pas « {cible} » dans {app}. " @@ -312,8 +388,63 @@ def formatter_fin_workflow( nom_workflow: str = "", nb_etapes: int = 0, duree_s: float = 0.0, + domain_id: Optional[str] = None, + items_count: int = 0, + failed_count: int = 0, + params: Optional[Mapping[str, Any]] = None, ) -> MessageUtilisateur: - """Message à la fin d'un workflow.""" + """Message à la fin d'un workflow. + + Si un domaine métier est fourni (et qu'il expose des summary_templates), + on utilise `DomainContext.describe_workflow_outcome` pour formuler un + rapport en langage métier (ex: "J'ai codé 14 dossiers sur 15"). + + Args: + succes: True si l'ensemble du workflow a réussi. + nom_workflow: Nom du workflow. + nb_etapes: Nombre d'étapes techniques (pour fallback générique). + duree_s: Durée totale en secondes. + domain_id: Domaine métier (optionnel). + items_count: Nombre d'items métier traités (ex: 15 dossiers). + failed_count: Nombre d'items en échec. + params: Infos supplémentaires passées aux templates. + """ + ctx = _get_domain_ctx(domain_id) + if ctx is not None and ctx.summary_templates: + try: + corps = ctx.describe_workflow_outcome( + workflow_name=nom_workflow, + success=succes, + items_count=items_count or max(1, nb_etapes), + failed_count=failed_count, + elapsed_s=duree_s, + extra=dict(params or {}), + ) + except Exception: + corps = "" + if corps: + if succes and failed_count == 0: + return MessageUtilisateur( + niveau=NiveauMessage.INFO, + titre="Léa a terminé", + corps=corps, + duree_s=6, + ) + if succes and failed_count > 0: + return MessageUtilisateur( + niveau=NiveauMessage.ATTENTION, + titre="Léa a terminé partiellement", + corps=corps, + duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION], + ) + return MessageUtilisateur( + niveau=NiveauMessage.BLOCAGE, + titre="Léa s'arrête", + corps=corps, + duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE], + persistent=True, + ) + if succes: if nom_workflow and nb_etapes > 0: corps = ( @@ -342,11 +473,17 @@ def formatter_fin_workflow( ) -def formatter_erreur_generique(message_technique: str) -> MessageUtilisateur: +def formatter_erreur_generique( + message_technique: str, + domain_id: Optional[str] = None, + params: Optional[Mapping[str, Any]] = None, +) -> MessageUtilisateur: """Formater un message d'erreur technique non catégorisé. On essaie de détecter les motifs connus dans le message technique pour le router vers le bon formatter spécialisé, sinon on emballe le message. + Si `domain_id` est fourni, il est propagé aux formatters spécialisés + pour produire un message en langage métier. """ if not message_technique: return MessageUtilisateur( @@ -363,7 +500,7 @@ def formatter_erreur_generique(message_technique: str) -> MessageUtilisateur: # Essayer d'extraire la description après le ':' match = re.match(r"target_not_found[:\s]*(.*)", message_technique, re.IGNORECASE) desc = match.group(1).strip() if match else "" - return formatter_cible_non_trouvee(desc) + return formatter_cible_non_trouvee(desc, domain_id=domain_id, params=params) # Fenêtre incorrecte: 'X' (attendu: 'Y') if "fenêtre incorrecte" in msg_lower or "fenetre incorrecte" in msg_lower: @@ -380,7 +517,9 @@ def formatter_erreur_generique(message_technique: str) -> MessageUtilisateur: # Policy abort / supervise if "policy_abort" in msg_lower or "visual_resolve_failed" in msg_lower: - return formatter_cible_non_trouvee(message_technique) + return formatter_cible_non_trouvee( + message_technique, domain_id=domain_id, params=params + ) # Fallback : message technique tronqué msg_tronque = message_technique.strip() diff --git a/agent_v0/server_v1/domain_context.py b/agent_v0/server_v1/domain_context.py index 1ae5a4327..7d7530059 100644 --- a/agent_v0/server_v1/domain_context.py +++ b/agent_v0/server_v1/domain_context.py @@ -3,35 +3,81 @@ Contexte métier pour les appels VLM — rend Léa experte du domaine. Chaque workflow est associé à un domaine métier (médical, comptable, etc.) -qui enrichit TOUS les prompts VLM (Observer, Critic, acteur, enrichissement). +qui enrichit TOUS les prompts VLM (Observer, Critic, acteur, enrichissement) +ET la personnalité de Léa (résumés, questions de clarification, rapports). Un gemma4 qui sait qu'il regarde un DPI et que l'utilisateur fait du codage -CIM-10 prend des décisions bien meilleures qu'un VLM générique. +CIM-10 prend des décisions bien meilleures qu'un VLM générique. Et Léa qui +dit "J'ai codé 14 dossiers sur 15" plutôt que "J'ai exécuté 112 clics" est +bien plus utile pour un TIM. -Premier domaine : TIM (Technicien d'Information Médicale) -- Logiciels DPI/DMS (dossier patient informatisé) -- Codage CIM-10 / CCAM / GHM -- Lecture de comptes rendus médicaux -- Validation des séjours / RSS / RSA +Domaines pré-configurés : +- tim_codage : TIM, codage CIM-10 / CCAM / PMSI, DPI +- comptabilite : factures, TVA, OCR, plans comptables +- rh_paie : fiches de paie, employés, charges sociales +- stocks_logistique : bons, commandes, réceptions, inventaires +- generic : fallback bureautique -Usage : +Usage basique : ctx = get_domain_context("tim_codage") - prompt = f"{ctx.system_prompt}\n\n{user_prompt}" + prompt = ctx.enrich_prompt(user_prompt, role="actor") + +Usage langage métier : + ctx = get_domain_context("tim_codage") + phrase = ctx.summarize_action("click", {"target": "DP"}) + # → "saisir le diagnostic principal" + + question = ctx.pose_clarification_question( + {"blocked_on": "target_not_found", "target": "Fichier patient", + "params": {"nom_patient": "Mme Durand"}} + ) + # → "Je ne trouve pas le dossier de Mme Durand..." + + rapport = ctx.describe_workflow_outcome( + workflow_name="Codage séjours janvier", + success=True, + items_count=15, + failed_count=1, + ) + # → "J'ai codé 14 dossiers sur 15..." """ +from __future__ import annotations + import logging +import os +import re +import unicodedata from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Mapping, Optional logger = logging.getLogger(__name__) +def _strip_accents(s: str) -> str: + """Supprimer les accents pour les comparaisons insensibles aux diacritiques.""" + if not s: + return "" + nkfd = unicodedata.normalize("NFKD", s) + return "".join(c for c in nkfd if not unicodedata.combining(c)) + + +# ========================================================================= +# Data class +# ========================================================================= + + @dataclass class DomainContext: - """Contexte métier pour un domaine spécifique.""" - domain_id: str # Identifiant unique (tim_codage, comptabilite, etc.) - name: str # Nom lisible (Codage médical TIM) - description: str # Description courte du métier + """Contexte métier pour un domaine spécifique. + + Contient à la fois les hints pour les prompts VLM et les éléments de + personnalité de Léa (langage métier, questions, rapports). + """ + + domain_id: str # tim_codage, comptabilite, ... + name: str # Nom lisible + description: str # Description courte # Prompt système injecté dans TOUS les appels VLM system_prompt: str = "" @@ -39,18 +85,47 @@ class DomainContext: # Vocabulaire métier (termes que le VLM doit connaître) vocabulary: List[str] = field(default_factory=list) - # Applications connues (noms de logiciels que le VLM peut rencontrer) + # Applications connues known_apps: List[str] = field(default_factory=list) - # Écrans types (descriptions des écrans courants du métier) + # Écrans types screen_patterns: Dict[str, str] = field(default_factory=dict) + # --- Personnalité Léa ------------------------------------------------- + + # Mapping d'actions techniques (click/type/key_combo) vers description métier, + # indexé par un mot-clé lisible trouvé dans la cible/texte. + # Format : { (action_type, keyword_lower) : "description métier" } + # Exemple : ("click", "dp") → "saisir le diagnostic principal" + common_actions: Dict[str, str] = field(default_factory=dict) + + # Synonymes métier : technique → forme lisible + # Exemple : {"dp": "diagnostic principal", "das": "diagnostics associés"} + vocabulary_synonyms: Dict[str, str] = field(default_factory=dict) + + # Templates de questions de clarification (selon la raison de blocage). + # Clé = identifiant de blocage ("target_not_found", "ambiguous_field", ...) + # Valeur = template f-string (champs: {target}, {app}, {nom_patient}, ...) + clarification_templates: Dict[str, str] = field(default_factory=dict) + + # Templates de résumés de fin de workflow. + # Clés attendues : + # - "success" : tout a marché + # - "partial" : succès partiel (failed_count > 0) + # - "failure" : échec complet + # - "success_one" : cas 1 élément (pour éviter "1 dossiers") + # - "item_singular" : libellé d'un item ("dossier") + # - "item_plural" : libellé au pluriel ("dossiers") + summary_templates: Dict[str, str] = field(default_factory=dict) + + # ------------------------------------------------------------------ API + def enrich_prompt(self, prompt: str, role: str = "") -> str: """Enrichir un prompt avec le contexte métier. Args: prompt: Le prompt original - role: Le rôle du VLM (observer, critic, actor, enrichment) + role: Le rôle du VLM (observer, critic, actor, enrichment) """ parts = [] @@ -65,6 +140,310 @@ class DomainContext: parts.append(prompt) return "\n\n".join(parts) + # ------------------------------------------------------------------ + # Personnalité : résumé d'action en langage métier + # ------------------------------------------------------------------ + + def summarize_action( + self, + action: str, + params: Optional[Mapping[str, Any]] = None, + ) -> str: + """Résumer une action technique en langage métier. + + Args: + action: Type d'action ("click", "type", "key_combo", "wait", "scroll") + params: Paramètres de l'action (target, text, keys, ...) + + Returns: + Phrase en français orientée métier. Fallback générique si aucun + mapping ne correspond. + + Exemples (domaine tim_codage) : + click sur "DP" → "saisir le diagnostic principal" + type "E11.9" → "saisir le code CIM-10 E11.9" + click sur "Valider" → "valider le codage" + """ + params = dict(params or {}) + target = str(params.get("target") or params.get("description") or "").strip() + text = str(params.get("text") or "").strip() + keys = params.get("keys") or [] + + haystack = _strip_accents(f"{target} {text}".lower()) + + # 1) Essayer un match mot-clé dans common_actions. + # Clés sous la forme "click:mot" ou "type:mot". + # Comparaison insensible à la casse ET aux accents. + for key, label in self.common_actions.items(): + if ":" not in key: + continue + k_action, k_word = key.split(":", 1) + if k_action != action: + continue + k_word_norm = _strip_accents(k_word.lower()) + if k_word_norm and k_word_norm in haystack: + return label + + # 2) Essayer une substitution via vocabulary_synonyms dans la cible. + friendly_target = self._apply_synonyms(target) + + if action == "click": + if friendly_target: + return f"cliquer sur {friendly_target}" + return "cliquer" + + if action == "type": + if text and friendly_target: + return f"saisir « {text} » dans {friendly_target}" + if text: + return f"saisir « {text} »" + return "saisir du texte" + + if action == "key_combo": + if isinstance(keys, (list, tuple)) and keys: + return f"utiliser le raccourci {'+'.join(str(k) for k in keys)}" + return "utiliser un raccourci clavier" + + if action == "wait": + return "attendre le chargement de l'écran" + + if action == "scroll": + return "faire défiler l'écran" + + # Fallback ultime + return f"effectuer l'action {action}" + + # ------------------------------------------------------------------ + # Personnalité : question de clarification + # ------------------------------------------------------------------ + + def pose_clarification_question( + self, + context: Optional[Mapping[str, Any]] = None, + ) -> str: + """Générer une question pertinente quand Léa bloque. + + Cherche un template dans clarification_templates selon : + - context["blocked_on"] (ex: "target_not_found", "ambiguous_field") + - context["target"] (la cible visée) + - paramètres du workflow (params) disponibles pour substitution + + Args: + context: Dictionnaire libre contenant au minimum 'blocked_on' ou + 'target'. Peut contenir 'params' pour la substitution. + + Returns: + Question en français. Fallback générique si aucun template ne + correspond. + """ + ctx = dict(context or {}) + blocked_on = str(ctx.get("blocked_on") or "").strip() + target = str(ctx.get("target") or "").strip() + params = dict(ctx.get("params") or {}) + + # Dictionnaire de substitution : champs du context + params + target + subs: Dict[str, Any] = { + "target": target, + "target_friendly": self._apply_synonyms(target) or target or "cet élément", + "app": ctx.get("app", ""), + } + subs.update(params) + + # 1) Essai par clé exacte de blocage + template = self.clarification_templates.get(blocked_on, "") + + # 2) Essai par cible (si la cible matche un mot-clé connu) + if not template and target: + low = target.lower() + for key, tpl in self.clarification_templates.items(): + if key.startswith("target:") and key.split(":", 1)[1].lower() in low: + template = tpl + break + + # 3) Template générique du domaine + if not template: + template = self.clarification_templates.get("default", "") + + if template: + try: + return template.format_map(_SafeDict(subs)) + except Exception as e: # pragma: no cover - format inattendu + logger.warning("clarification template format error: %s", e) + + # 4) Fallback ultime cross-domaine + friendly = subs["target_friendly"] + return ( + f"Je ne trouve pas {friendly}. " + f"Peux-tu me le montrer ou me confirmer que c'est le bon écran ?" + ) + + # ------------------------------------------------------------------ + # Personnalité : rapport final + # ------------------------------------------------------------------ + + def describe_workflow_outcome( + self, + workflow_name: str = "", + success: bool = True, + items_count: int = 1, + failed_count: int = 0, + elapsed_s: float = 0.0, + extra: Optional[Mapping[str, Any]] = None, + use_llm: bool = False, + ) -> str: + """Générer un rapport de fin de workflow en langage métier. + + Args: + workflow_name: Nom du workflow ("Codage janvier"). + success: True si le workflow a globalement réussi. + items_count: Nombre d'items traités (ex: 15 dossiers). 1 par défaut. + failed_count: Nombre d'items en échec. + elapsed_s: Durée totale (secondes). + extra: Infos supplémentaires (hint pour le LLM). + use_llm: Si True, tenter un appel à gemma4 pour produire le + rapport. Fallback sur les templates en cas d'échec. + + Returns: + Rapport en français. Toujours une chaîne, jamais None. + """ + extra = dict(extra or {}) + done = max(0, items_count - failed_count) + + item_sg = self.summary_templates.get("item_singular", "élément") + item_pl = self.summary_templates.get("item_plural", "éléments") + item_word = item_sg if done <= 1 else item_pl + + # Données disponibles pour les templates + subs = { + "workflow_name": workflow_name or "le workflow", + "items_count": items_count, + "done": done, + "failed": failed_count, + "item_singular": item_sg, + "item_plural": item_pl, + "item_word": item_word, + "elapsed_s": int(elapsed_s), + } + subs.update(extra) + + # Choisir le template adéquat + if not success and failed_count >= items_count: + key = "failure" + elif failed_count > 0: + key = "partial" + elif items_count == 1: + key = "success_one" if "success_one" in self.summary_templates else "success" + else: + key = "success" + + template = self.summary_templates.get(key, "") + + # Optionnel : raffiner via gemma4 + if use_llm: + llm_text = self._llm_refine_summary(template, subs, success) + if llm_text: + return llm_text + + if template: + try: + return template.format_map(_SafeDict(subs)) + except Exception as e: # pragma: no cover + logger.warning("summary template format error: %s", e) + + # Fallback générique + if success: + if items_count <= 1: + return f"C'est fait, j'ai terminé « {workflow_name or 'le workflow'} »." + return ( + f"J'ai traité {done} {item_word} sur {items_count}" + + (f", {failed_count} en échec." if failed_count else ".") + ) + return ( + f"Je n'ai pas pu terminer « {workflow_name or 'le workflow'} ». " + f"Je te rends la main." + ) + + # ------------------------------------------------------------------ + # Helpers internes + # ------------------------------------------------------------------ + + def _apply_synonyms(self, text: str) -> str: + """Remplacer les sigles/termes techniques par leur forme métier. + + Cherche mots entiers (word boundaries) en insensible à la casse. + """ + if not text or not self.vocabulary_synonyms: + return text + result = text + for short, full in self.vocabulary_synonyms.items(): + if not short: + continue + pattern = r"\b" + re.escape(short) + r"\b" + result = re.sub(pattern, full, result, flags=re.IGNORECASE) + return result + + def _llm_refine_summary( + self, + template: str, + subs: Dict[str, Any], + success: bool, + ) -> str: + """Tenter un raffinement du rapport via gemma4. + + Appel best-effort : toute erreur retourne "" et le caller retombe sur + le template brut. Isolé dans une méthode pour pouvoir le monkey-patcher + dans les tests. + """ + try: + import requests as _requests + except Exception: + return "" + + port = os.environ.get("GEMMA4_PORT", "11435") + url = f"http://localhost:{port}/api/chat" + + base = "" + if template: + try: + base = template.format_map(_SafeDict(subs)) + except Exception: + base = "" + + prompt = ( + f"Tu es Léa, une assistante RPA dans le domaine : {self.name}.\n" + f"Tu viens de terminer un workflow. Résume en UNE à DEUX phrases " + f"en langage métier, chaleureux mais professionnel, en français.\n\n" + f"Données :\n" + f"- workflow : {subs.get('workflow_name', '')}\n" + f"- items traités : {subs.get('done', 0)} / {subs.get('items_count', 0)}\n" + f"- échecs : {subs.get('failed', 0)}\n" + f"- succès global : {'oui' if success else 'non'}\n" + f"- durée : {subs.get('elapsed_s', 0)}s\n\n" + f"Base suggérée (tu peux la reformuler) : {base or '(aucune)'}\n\n" + f"Ta phrase :" + ) + + try: + resp = _requests.post( + url, + json={ + "model": "gemma4:e4b", + "messages": [{"role": "user", "content": prompt}], + "stream": False, + "options": {"temperature": 0.3, "num_predict": 200}, + }, + timeout=30, + ) + if not resp.ok: + return "" + content = resp.json().get("message", {}).get("content", "").strip() + # Nettoyage basique : supprimer guillemets typographiques en bord + content = content.strip("\"' \n") + return content + except Exception as e: + logger.debug("gemma4 refine summary failed: %s", e) + return "" + def to_dict(self) -> Dict[str, Any]: return { "domain_id": self.domain_id, @@ -72,9 +451,24 @@ class DomainContext: "description": self.description, "known_apps": self.known_apps, "vocabulary_count": len(self.vocabulary), + "common_actions_count": len(self.common_actions), + "has_clarification_templates": bool(self.clarification_templates), + "has_summary_templates": bool(self.summary_templates), } +# ========================================================================= +# Utilitaires +# ========================================================================= + + +class _SafeDict(dict): + """dict pour str.format_map qui retourne "" pour les clés manquantes.""" + + def __missing__(self, key): # type: ignore[override] + return "" + + # Hints par rôle VLM — adaptés au contexte métier _ROLE_HINTS = { "observer": ( @@ -100,6 +494,7 @@ _ROLE_HINTS = { # Domaines pré-configurés # ========================================================================= + _TIM_CODAGE = DomainContext( domain_id="tim_codage", name="Codage médical TIM", @@ -156,8 +551,405 @@ _TIM_CODAGE = DomainContext( "recherche_code": "Recherche de code CIM-10 ou CCAM (champ de recherche + arborescence)", "validation_ghm": "Écran de validation du groupage avec GHM calculé et valorisation", }, + vocabulary_synonyms={ + "DP": "diagnostic principal", + "DAS": "diagnostics associés", + "CMA": "complication associée", + "UM": "unité médicale", + "CR": "compte rendu", + "RSS": "résumé de sortie", + "RSA": "résumé anonymisé", + "GHM": "groupe homogène de malades", + "IPP": "identifiant patient", + }, + common_actions={ + "click:dp": "saisir le diagnostic principal", + "click:diagnostic principal": "saisir le diagnostic principal", + "click:das": "ajouter un diagnostic associé", + "click:ccam": "saisir un acte CCAM", + "click:valider": "valider le codage", + "click:valider le codage": "valider le codage", + "click:grouper": "calculer le GHM", + "click:ghm": "consulter le groupage GHM", + "click:dossier patient": "ouvrir le dossier patient", + "click:fiche patient": "ouvrir la fiche patient", + "click:compte rendu": "consulter le compte rendu", + "click:cr": "consulter le compte rendu", + "click:rechercher": "rechercher un code CIM-10", + "type:cim": "saisir un code CIM-10", + }, + clarification_templates={ + "default": ( + "Je ne trouve pas {target_friendly}. " + "Tu peux me montrer où il se trouve dans le dossier ?" + ), + "target_not_found": ( + "Je ne trouve pas {target_friendly}. " + "Le dossier de {nom_patient} est peut-être déjà codé ou archivé ?" + ), + "target:fichier patient": ( + "Je ne trouve pas le dossier de {nom_patient}. " + "Il est peut-être archivé ? Tu peux me le montrer ?" + ), + "target:dossier": ( + "Je ne trouve pas le dossier de {nom_patient}. " + "Il est peut-être archivé ? Tu peux me le montrer ?" + ), + "ambiguous_code": ( + "Le compte rendu mentionne plusieurs codes possibles. " + "Est-ce le code CIM-10 {code_a} ou {code_b} que tu préfères ?" + ), + "no_cr": ( + "Je ne trouve pas de compte rendu pour {nom_patient}. " + "Tu veux que je saute ce dossier ou que je continue sans ?" + ), + }, + summary_templates={ + "item_singular": "dossier", + "item_plural": "dossiers", + "success_one": ( + "J'ai codé le dossier de {nom_patient} en {elapsed_s}s. " + "Tu peux vérifier le groupage GHM." + ), + "success": ( + "J'ai codé {done} dossiers sur {items_count}. " + "Tout est passé sans erreur, tu peux valider le groupage." + ), + "partial": ( + "J'ai codé {done} dossiers sur {items_count}. " + "{failed} sont en attente — codes CIM-10 ambigus, à valider manuellement." + ), + "failure": ( + "Je n'ai pas pu coder les dossiers de {workflow_name}. " + "Je te rends la main, les comptes rendus sont peut-être inaccessibles." + ), + }, ) + +_COMPTABILITE = DomainContext( + domain_id="comptabilite", + name="Comptabilité", + description=( + "Comptable : saisie de factures fournisseurs et clients, lettrage, " + "rapprochement bancaire, déclarations de TVA, bilans, immobilisations." + ), + system_prompt=( + "Tu es un assistant expert en comptabilité d'entreprise. " + "L'utilisateur est un comptable qui utilise un logiciel de saisie comptable " + "(Sage, Cegid, EBP, Quadra, Isacompta) pour saisir des factures, faire " + "les rapprochements bancaires, préparer la TVA et les bilans.\n\n" + "Vocabulaire du métier :\n" + "- Facture : justificatif de vente ou d'achat (numéro, date, HT, TVA, TTC)\n" + "- HT/TVA/TTC : montants hors taxes, taxe, toutes taxes\n" + "- Compte comptable : numéro du plan comptable général (PCG), ex 401 (fournisseurs), 411 (clients)\n" + "- Journal : journal de saisie (achats, ventes, banque, OD)\n" + "- Lettrage : association d'une facture avec son paiement\n" + "- Rapprochement : comparaison compte comptable / relevé bancaire\n" + "- OCR / LAD : reconnaissance automatique des factures scannées\n" + "- Écriture : ligne comptable (débit/crédit)\n" + "- Exercice : période comptable annuelle\n" + "- Bilan / compte de résultat : états financiers\n" + "- CA : chiffre d'affaires\n\n" + "Écrans courants :\n" + "- Saisie d'écritures (numéro de compte, libellé, débit, crédit)\n" + "- Import OCR de factures fournisseurs\n" + "- Lettrage / rapprochement\n" + "- Brouillard / journal\n" + "- Balance / grand livre" + ), + vocabulary=[ + "facture", "HT", "TVA", "TTC", "compte", "journal", "lettrage", + "rapprochement", "OCR", "LAD", "écriture", "débit", "crédit", + "exercice", "bilan", "compte de résultat", "CA", "immobilisation", + "fournisseur", "client", "PCG", "plan comptable", + ], + known_apps=[ + "Sage", "Cegid", "EBP", "Quadra", "Isacompta", "Ciel Compta", + "Odoo", "Pennylane", "Dext", "Agicap", + ], + screen_patterns={ + "saisie_ecriture": "Saisie d'écriture comptable (compte, libellé, débit, crédit)", + "ocr_facture": "Import OCR : zone image + champs extraits (numéro, date, HT, TVA, TTC, fournisseur)", + "lettrage": "Liste d'écritures à lettrer (débit vs crédit)", + "rapprochement": "Comparaison compte banque / relevé", + "balance": "Balance comptable (comptes agrégés avec soldes)", + }, + vocabulary_synonyms={ + "HT": "montant hors taxes", + "TVA": "montant de TVA", + "TTC": "montant toutes taxes", + "CA": "chiffre d'affaires", + "PCG": "plan comptable général", + "OD": "opération diverse", + }, + common_actions={ + "click:valider": "valider l'écriture", + "click:enregistrer": "enregistrer la saisie", + "click:lettrer": "lettrer les écritures", + "click:rapprocher": "rapprocher avec la banque", + "click:ocr": "lancer la reconnaissance OCR", + "click:facture": "ouvrir la facture", + "click:compte": "sélectionner le compte comptable", + "type:ht": "saisir le montant hors taxes", + "type:tva": "saisir le montant de TVA", + "type:ttc": "saisir le montant toutes taxes", + }, + clarification_templates={ + "default": ( + "Je ne trouve pas {target_friendly}. " + "C'est bien la facture {num_facture} que tu veux saisir ?" + ), + "target_not_found": ( + "Je ne trouve pas le champ {target_friendly}. " + "C'est bien la facture {num_facture} qui doit être saisie ?" + ), + "target:montant": ( + "Je ne trouve pas le champ « Montant HT ». " + "C'est bien la facture {num_facture} que tu veux saisir ?" + ), + "target:tva": ( + "Je ne trouve pas le champ TVA. Est-ce une facture à taux {taux_tva} % ?" + ), + "ambiguous_account": ( + "Je ne sais pas sur quel compte imputer : {compte_a} ou {compte_b} ?" + ), + }, + summary_templates={ + "item_singular": "facture", + "item_plural": "factures", + "success_one": ( + "J'ai saisi la facture {num_facture} en {elapsed_s}s." + ), + "success": ( + "J'ai saisi {done} factures sur {items_count}. " + "Tout est en brouillard, tu peux valider." + ), + "partial": ( + "J'ai saisi {done} factures sur {items_count}. " + "{failed} factures sont en attente — imputation comptable à vérifier." + ), + "failure": ( + "Je n'ai pas pu saisir les factures de {workflow_name}. " + "L'OCR n'a peut-être pas fonctionné, je te rends la main." + ), + }, +) + + +_RH_PAIE = DomainContext( + domain_id="rh_paie", + name="Ressources humaines et paie", + description=( + "Gestionnaire RH / paie : fiches employés, contrats, bulletins de salaire, " + "déclarations sociales (DSN), charges, congés, absences." + ), + system_prompt=( + "Tu es un assistant expert en gestion RH et paie française. " + "L'utilisateur est un gestionnaire RH ou de paie qui utilise un logiciel " + "(Silae, Sage Paie, Cegid, ADP, PayFit) pour éditer des bulletins de salaire, " + "gérer les contrats, les absences, et envoyer les DSN.\n\n" + "Vocabulaire du métier :\n" + "- Bulletin de paie : fiche de salaire mensuelle\n" + "- DSN : Déclaration Sociale Nominative (mensuelle, transmise à l'URSSAF)\n" + "- Brut / Net : salaire avant et après charges\n" + "- Charges sociales / patronales : cotisations employeur et salarié\n" + "- CDI / CDD : types de contrats\n" + "- Période de paie : mois concerné par le bulletin\n" + "- SMIC : salaire minimum\n" + "- IJSS : indemnités journalières sécurité sociale\n" + "- Congés payés : solde de congés\n" + "- RTT : réduction du temps de travail\n" + "- Saisie sur salaire : retenue judiciaire\n" + "- Solde de tout compte : dernier bulletin d'un salarié qui part\n\n" + "Écrans courants :\n" + "- Fiche employé (identité, contrat, poste, salaire)\n" + "- Saisie des variables (heures, absences, primes)\n" + "- Bulletin de paie (aperçu avant validation)\n" + "- Déclaration DSN\n" + "- Gestion des absences / congés" + ), + vocabulary=[ + "bulletin", "salaire", "brut", "net", "charges sociales", "DSN", + "CDI", "CDD", "congés", "RTT", "SMIC", "IJSS", "URSSAF", + "employé", "salarié", "contrat", "prime", "heures supplémentaires", + "absence", "solde de tout compte", "STC", + ], + known_apps=[ + "Silae", "Sage Paie", "Cegid Paie", "ADP", "PayFit", "Nibelis", + "Cegedim SRH", "Lucca", "HR Access", + ], + screen_patterns={ + "fiche_employe": "Fiche employé avec identité, contrat, poste", + "saisie_variables": "Saisie des variables de paie (heures, absences, primes)", + "apercu_bulletin": "Aperçu du bulletin de paie avant validation", + "dsn": "Écran DSN (déclaration sociale nominative)", + "conges": "Gestion des absences et congés", + }, + vocabulary_synonyms={ + "DSN": "déclaration sociale", + "RTT": "réduction du temps de travail", + "STC": "solde de tout compte", + "IJSS": "indemnités journalières", + "CP": "congés payés", + }, + common_actions={ + "click:valider": "valider le bulletin", + "click:editer": "éditer le bulletin", + "click:bulletin": "ouvrir le bulletin de paie", + "click:employe": "ouvrir la fiche employé", + "click:dsn": "lancer la DSN", + "click:conges": "gérer les congés", + "click:absence": "saisir une absence", + "type:heures": "saisir les heures travaillées", + "type:prime": "saisir une prime", + }, + clarification_templates={ + "default": ( + "Je ne trouve pas {target_friendly} pour {nom_employe}. " + "Tu peux me confirmer la période de paie ?" + ), + "target_not_found": ( + "Je ne trouve pas {target_friendly} dans la fiche de {nom_employe}. " + "Le contrat est peut-être clôturé ?" + ), + "target:employe": ( + "Je ne trouve pas {nom_employe} dans la liste. " + "Est-il encore actif dans l'entreprise ?" + ), + "ambiguous_period": ( + "Est-ce la période {periode_a} ou {periode_b} que tu veux traiter ?" + ), + }, + summary_templates={ + "item_singular": "bulletin", + "item_plural": "bulletins", + "success_one": ( + "J'ai édité le bulletin de {nom_employe} en {elapsed_s}s." + ), + "success": ( + "J'ai édité {done} bulletins sur {items_count}. " + "La paie est prête pour validation." + ), + "partial": ( + "J'ai édité {done} bulletins sur {items_count}. " + "{failed} sont en attente — variables de paie à compléter." + ), + "failure": ( + "Je n'ai pas pu éditer les bulletins de {workflow_name}. " + "Il y a peut-être un blocage côté logiciel de paie." + ), + }, +) + + +_STOCKS_LOGISTIQUE = DomainContext( + domain_id="stocks_logistique", + name="Stocks et logistique", + description=( + "Gestionnaire de stocks / logistique : bons de commande, bons de livraison, " + "réceptions, inventaires, mouvements de stock, expéditions." + ), + system_prompt=( + "Tu es un assistant expert en gestion de stocks et logistique. " + "L'utilisateur utilise un ERP ou WMS (SAP, Dynamics, Odoo, Sage, Divalto) " + "pour gérer les commandes, les réceptions, les expéditions et les inventaires.\n\n" + "Vocabulaire du métier :\n" + "- BC : Bon de Commande (achat ou vente)\n" + "- BL : Bon de Livraison\n" + "- BR : Bon de Réception\n" + "- Article / Référence / SKU : produit en stock\n" + "- Emplacement : localisation physique (allée, rayon, emplacement)\n" + "- Mouvement de stock : entrée, sortie, transfert\n" + "- Inventaire : comptage physique pour recaler le stock théorique\n" + "- FIFO / LIFO : ordre de sortie des stocks\n" + "- ERP : progiciel de gestion intégré\n" + "- WMS : Warehouse Management System\n" + "- Picking : préparation de commande\n" + "- Quantité en stock / disponible / réservée\n\n" + "Écrans courants :\n" + "- Saisie de bon de commande / réception\n" + "- Liste des articles (avec photo, quantité, emplacement)\n" + "- Inventaire (comptage)\n" + "- Mouvements de stock\n" + "- Picking list (liste de préparation)" + ), + vocabulary=[ + "bon de commande", "BC", "bon de livraison", "BL", "bon de réception", "BR", + "article", "référence", "SKU", "emplacement", "stock", "inventaire", + "mouvement", "entrée", "sortie", "picking", "FIFO", "LIFO", "ERP", "WMS", + "fournisseur", "client", "quantité", "disponible", "réservé", + ], + known_apps=[ + "SAP", "Dynamics", "Odoo", "Sage X3", "Divalto", "Cegid", + "Oracle NetSuite", "Reflex WMS", "Infolog", + ], + screen_patterns={ + "bon_commande": "Saisie de bon de commande (fournisseur, lignes d'articles, quantités)", + "reception": "Bon de réception (rapprochement avec la commande)", + "inventaire": "Saisie d'inventaire (article, emplacement, quantité comptée)", + "picking": "Liste de préparation avec articles et emplacements", + "mouvement": "Mouvement de stock (entrée/sortie/transfert)", + }, + vocabulary_synonyms={ + "BC": "bon de commande", + "BL": "bon de livraison", + "BR": "bon de réception", + "SKU": "référence produit", + "WMS": "gestion d'entrepôt", + "ERP": "progiciel de gestion", + }, + common_actions={ + "click:valider": "valider le bon", + "click:commande": "ouvrir le bon de commande", + "click:livraison": "ouvrir le bon de livraison", + "click:reception": "saisir la réception", + "click:inventaire": "démarrer l'inventaire", + "click:article": "sélectionner un article", + "click:picking": "démarrer la préparation", + "type:quantite": "saisir la quantité", + "type:reference": "saisir la référence article", + }, + clarification_templates={ + "default": ( + "Je ne trouve pas {target_friendly}. " + "C'est bien la commande {num_bc} qu'on traite ?" + ), + "target_not_found": ( + "Je ne trouve pas {target_friendly}. " + "La commande {num_bc} est peut-être déjà clôturée ?" + ), + "target:article": ( + "Je ne trouve pas l'article {ref_article}. " + "Il est peut-être archivé ou mal référencé ?" + ), + "quantity_mismatch": ( + "La quantité reçue ({qte_recue}) ne correspond pas à la commande " + "({qte_commandee}). Je saisis un écart ou tu vérifies ?" + ), + }, + summary_templates={ + "item_singular": "bon", + "item_plural": "bons", + "success_one": ( + "J'ai traité le bon {num_bc} en {elapsed_s}s." + ), + "success": ( + "J'ai traité {done} bons sur {items_count}. " + "Les mouvements de stock sont validés." + ), + "partial": ( + "J'ai traité {done} bons sur {items_count}. " + "{failed} bons sont en attente — écarts de quantité à vérifier." + ), + "failure": ( + "Je n'ai pas pu traiter les bons de {workflow_name}. " + "L'ERP a peut-être refusé une ligne, je te rends la main." + ), + }, +) + + _GENERIC = DomainContext( domain_id="generic", name="Bureautique générale", @@ -166,11 +958,37 @@ _GENERIC = DomainContext( "Tu es un assistant RPA qui observe des applications bureautiques. " "Décris précisément ce que tu vois à l'écran." ), + summary_templates={ + "item_singular": "action", + "item_plural": "actions", + "success_one": "C'est fait, j'ai terminé « {workflow_name} » en {elapsed_s}s.", + "success": ( + "J'ai terminé « {workflow_name} » : {done} {item_word} exécutées " + "sur {items_count}." + ), + "partial": ( + "J'ai terminé « {workflow_name} » partiellement : " + "{done} {item_word} sur {items_count} ({failed} en échec)." + ), + "failure": ( + "Je n'ai pas pu terminer « {workflow_name} ». Je te rends la main." + ), + }, + clarification_templates={ + "default": ( + "Je ne trouve pas {target_friendly} à l'écran. " + "Tu peux me le montrer ?" + ), + }, ) + # Registre des domaines disponibles _DOMAINS: Dict[str, DomainContext] = { "tim_codage": _TIM_CODAGE, + "comptabilite": _COMPTABILITE, + "rh_paie": _RH_PAIE, + "stocks_logistique": _STOCKS_LOGISTIQUE, "generic": _GENERIC, } @@ -179,7 +997,8 @@ def get_domain_context(domain_id: str = "generic") -> DomainContext: """Récupérer le contexte métier par ID. Args: - domain_id: Identifiant du domaine (tim_codage, generic, etc.) + domain_id: Identifiant du domaine (tim_codage, comptabilite, rh_paie, + stocks_logistique, generic, etc.) Returns: DomainContext correspondant, ou generic si non trouvé. diff --git a/tests/unit/test_domain_personality.py b/tests/unit/test_domain_personality.py new file mode 100644 index 000000000..85f17ceb2 --- /dev/null +++ b/tests/unit/test_domain_personality.py @@ -0,0 +1,543 @@ +"""Tests unitaires pour la personnalité métier de Léa. + +Couvre : +- summarize_action : résumé d'actions en langage métier par domaine +- pose_clarification_question : questions contextuelles quand Léa bloque +- describe_workflow_outcome : rapports de fin en langage métier +- Fallback domaine inconnu / vocabulaire synonyme +- Intégration avec agent_v0.agent_v1.ui.messages (formatters enrichis) +- Appel gemma4 mocké pour le raffinement de résumé + +Tous les tests sont 100% offline : aucun appel réseau réel. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +# Assurer que la racine du projet est dans le path (comme les autres tests unit) +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from agent_v0.server_v1.domain_context import ( + DomainContext, + get_domain_context, + list_domains, + register_domain, +) + + +# ============================================================================ +# Domaines pré-configurés +# ============================================================================ + + +class TestDomainesDisponibles: + """Tous les domaines prévus doivent être enregistrés.""" + + def test_tim_codage_present(self): + ctx = get_domain_context("tim_codage") + assert ctx.domain_id == "tim_codage" + assert "CIM-10" in ctx.vocabulary + assert ctx.common_actions # non vide + assert ctx.clarification_templates + assert ctx.summary_templates + + def test_comptabilite_present(self): + ctx = get_domain_context("comptabilite") + assert ctx.domain_id == "comptabilite" + assert "facture" in ctx.vocabulary + assert ctx.summary_templates["item_plural"] == "factures" + + def test_rh_paie_present(self): + ctx = get_domain_context("rh_paie") + assert ctx.domain_id == "rh_paie" + assert "bulletin" in ctx.vocabulary + assert ctx.summary_templates["item_plural"] == "bulletins" + + def test_stocks_logistique_present(self): + ctx = get_domain_context("stocks_logistique") + assert ctx.domain_id == "stocks_logistique" + assert "BC" in ctx.vocabulary or "bon de commande" in ctx.vocabulary + assert ctx.summary_templates["item_plural"] == "bons" + + def test_generic_fallback(self): + """Un domaine inconnu retourne le contexte générique.""" + ctx = get_domain_context("n_existe_pas_42") + assert ctx.domain_id == "generic" + + def test_list_domains_contains_all(self): + ids = {d["domain_id"] for d in list_domains()} + assert { + "tim_codage", + "comptabilite", + "rh_paie", + "stocks_logistique", + "generic", + }.issubset(ids) + + +# ============================================================================ +# summarize_action — résumé d'actions +# ============================================================================ + + +class TestSummarizeAction: + """Résumés en langage métier par domaine.""" + + def test_tim_click_dp_saisir_diagnostic_principal(self): + ctx = get_domain_context("tim_codage") + phrase = ctx.summarize_action("click", {"target": "DP"}) + assert phrase == "saisir le diagnostic principal" + + def test_tim_click_valider_codage(self): + ctx = get_domain_context("tim_codage") + phrase = ctx.summarize_action("click", {"target": "Valider le codage"}) + assert phrase == "valider le codage" + + def test_tim_click_dossier_patient(self): + ctx = get_domain_context("tim_codage") + phrase = ctx.summarize_action( + "click", {"target": "Ouvrir le dossier patient"} + ) + assert phrase == "ouvrir le dossier patient" + + def test_compta_type_ht(self): + ctx = get_domain_context("comptabilite") + phrase = ctx.summarize_action( + "type", {"target": "Montant HT", "text": "1500"} + ) + # La mention "ht" dans la cible déclenche le mapping + assert phrase == "saisir le montant hors taxes" + + def test_compta_click_lettrer(self): + ctx = get_domain_context("comptabilite") + phrase = ctx.summarize_action("click", {"target": "Lettrer"}) + assert phrase == "lettrer les écritures" + + def test_rh_click_bulletin(self): + ctx = get_domain_context("rh_paie") + phrase = ctx.summarize_action("click", {"target": "Bulletin de paie"}) + assert phrase == "ouvrir le bulletin de paie" + + def test_stocks_type_quantite(self): + ctx = get_domain_context("stocks_logistique") + phrase = ctx.summarize_action( + "type", {"target": "Quantité reçue", "text": "42"} + ) + assert phrase == "saisir la quantité" + + def test_generic_click_fallback(self): + ctx = get_domain_context("generic") + phrase = ctx.summarize_action("click", {"target": "Bouton quelconque"}) + # Pas de mapping mais une description → "cliquer sur ..." + assert "cliquer sur" in phrase + + def test_unknown_domain_click(self): + """Un domaine inconnu ne plante pas.""" + ctx = get_domain_context("inconnu") + phrase = ctx.summarize_action("click", {"target": "Quelque chose"}) + assert phrase # non vide + assert "cliquer" in phrase + + def test_tim_synonymes_dp_dans_cible_longue(self): + """Si aucun mapping exact mais la cible contient DP → substitution synonyme.""" + ctx = get_domain_context("tim_codage") + # Aucun mapping direct "saisir le" mais "DP" est dans les synonymes + phrase = ctx.summarize_action("click", {"target": "Saisir le DP"}) + assert phrase == "saisir le diagnostic principal" + + def test_key_combo_generic(self): + ctx = get_domain_context("generic") + phrase = ctx.summarize_action("key_combo", {"keys": ["ctrl", "s"]}) + assert "ctrl+s" in phrase + + def test_wait_and_scroll(self): + ctx = get_domain_context("tim_codage") + assert "attendre" in ctx.summarize_action("wait", {}) + assert "défiler" in ctx.summarize_action("scroll", {}) + + def test_type_no_target(self): + ctx = get_domain_context("generic") + phrase = ctx.summarize_action("type", {"text": "hello"}) + assert "hello" in phrase + + +# ============================================================================ +# pose_clarification_question — questions de blocage +# ============================================================================ + + +class TestClarification: + """Questions posées par Léa en cas de blocage.""" + + def test_tim_fichier_patient_avec_nom(self): + ctx = get_domain_context("tim_codage") + question = ctx.pose_clarification_question( + { + "blocked_on": "target_not_found", + "target": "Fichier patient", + "params": {"nom_patient": "Mme Durand"}, + } + ) + assert "Mme Durand" in question + # Langage métier : mention "dossier" (pas juste "fichier") + assert "dossier" in question.lower() + + def test_compta_montant_avec_num_facture(self): + ctx = get_domain_context("comptabilite") + question = ctx.pose_clarification_question( + { + "blocked_on": "target_not_found", + "target": "Montant HT", + "params": {"num_facture": "F2026-0145"}, + } + ) + assert "F2026-0145" in question + assert "Montant HT" in question or "Montant" in question + + def test_rh_employe_non_trouve(self): + ctx = get_domain_context("rh_paie") + question = ctx.pose_clarification_question( + { + "blocked_on": "target_not_found", + "target": "Fiche employé", + "params": {"nom_employe": "Jean Martin"}, + } + ) + assert "Jean Martin" in question + + def test_stocks_article_non_trouve(self): + ctx = get_domain_context("stocks_logistique") + question = ctx.pose_clarification_question( + { + "blocked_on": "target_not_found", + "target": "Article", + "params": {"ref_article": "REF-4242", "num_bc": "BC-2026-042"}, + } + ) + # Un des deux identifiants au moins apparaît + assert "REF-4242" in question or "BC-2026-042" in question + + def test_ambiguous_code_tim(self): + ctx = get_domain_context("tim_codage") + question = ctx.pose_clarification_question( + { + "blocked_on": "ambiguous_code", + "params": {"code_a": "E11.9", "code_b": "E11.8"}, + } + ) + assert "E11.9" in question + assert "E11.8" in question + + def test_clarification_unknown_domain_fallback(self): + """Domaine inconnu → message générique, jamais de crash.""" + ctx = get_domain_context("inconnu") + question = ctx.pose_clarification_question( + {"blocked_on": "target_not_found", "target": "Un champ"} + ) + assert question + assert "trouve pas" in question.lower() + + def test_clarification_empty_context(self): + """Pas de contexte du tout → fallback.""" + ctx = get_domain_context("tim_codage") + question = ctx.pose_clarification_question(None) + assert question # non vide + assert isinstance(question, str) + + def test_clarification_missing_params_no_crash(self): + """Si un template mentionne {nom_patient} mais qu'il n'est pas fourni, + on ne plante pas — les champs manquants sont vides.""" + ctx = get_domain_context("tim_codage") + question = ctx.pose_clarification_question( + { + "blocked_on": "target_not_found", + "target": "Fichier patient", + # pas de nom_patient + } + ) + assert isinstance(question, str) + assert question + + +# ============================================================================ +# describe_workflow_outcome — rapports finaux +# ============================================================================ + + +class TestWorkflowOutcome: + """Rapports de fin de workflow en langage métier.""" + + def test_tim_succes_complet(self): + ctx = get_domain_context("tim_codage") + rapport = ctx.describe_workflow_outcome( + workflow_name="Codage janvier", + success=True, + items_count=15, + failed_count=0, + ) + assert "15 dossiers" in rapport + assert "codé" in rapport + + def test_tim_succes_partiel(self): + ctx = get_domain_context("tim_codage") + rapport = ctx.describe_workflow_outcome( + workflow_name="Codage janvier", + success=True, + items_count=15, + failed_count=1, + ) + assert "14 dossiers" in rapport + assert "15" in rapport + assert "1" in rapport # nombre en attente + + def test_tim_echec_complet(self): + ctx = get_domain_context("tim_codage") + rapport = ctx.describe_workflow_outcome( + workflow_name="Codage janvier", + success=False, + items_count=15, + failed_count=15, + ) + assert "Codage janvier" in rapport + assert "pas" in rapport.lower() or "rends la main" in rapport.lower() + + def test_compta_success_factures(self): + ctx = get_domain_context("comptabilite") + rapport = ctx.describe_workflow_outcome( + workflow_name="Saisie factures mars", + success=True, + items_count=30, + failed_count=0, + ) + assert "30 factures" in rapport + + def test_rh_success_bulletins(self): + ctx = get_domain_context("rh_paie") + rapport = ctx.describe_workflow_outcome( + workflow_name="Paie avril", + success=True, + items_count=50, + failed_count=2, + ) + assert "48" in rapport + assert "50" in rapport + assert "bulletins" in rapport + + def test_stocks_success_bons(self): + ctx = get_domain_context("stocks_logistique") + rapport = ctx.describe_workflow_outcome( + workflow_name="Réceptions semaine 14", + success=True, + items_count=12, + failed_count=0, + ) + assert "12 bons" in rapport + + def test_generic_fallback(self): + """Domaine inconnu → rapport générique cohérent.""" + ctx = get_domain_context("inconnu") + rapport = ctx.describe_workflow_outcome( + workflow_name="Mon workflow", + success=True, + items_count=5, + failed_count=0, + ) + assert rapport + assert "Mon workflow" in rapport or "5" in rapport + + def test_tim_success_one_avec_nom_patient(self): + """Cas 1 item : utilise success_one avec un paramètre métier.""" + ctx = get_domain_context("tim_codage") + rapport = ctx.describe_workflow_outcome( + workflow_name="Codage urgent", + success=True, + items_count=1, + failed_count=0, + elapsed_s=42, + extra={"nom_patient": "M. Dupont"}, + ) + assert "M. Dupont" in rapport + assert "42" in rapport + + +class TestWorkflowOutcomeLLM: + """Tests du raffinement LLM (gemma4) pour le rapport final.""" + + def test_use_llm_success_mocked(self): + """Quand use_llm=True et gemma4 répond, on utilise sa réponse.""" + ctx = get_domain_context("tim_codage") + + def fake_refine(self, template, subs, success): + return "Voilà, j'ai codé tous tes dossiers, bon café !" + + with patch.object(DomainContext, "_llm_refine_summary", fake_refine): + rapport = ctx.describe_workflow_outcome( + workflow_name="Codage", success=True, + items_count=10, use_llm=True, + ) + assert "bon café" in rapport + + def test_use_llm_failure_falls_back_to_template(self): + """Si l'appel LLM retourne "" → on retombe sur le template.""" + ctx = get_domain_context("tim_codage") + + def fake_refine(self, template, subs, success): + return "" # simulate failure + + with patch.object(DomainContext, "_llm_refine_summary", fake_refine): + rapport = ctx.describe_workflow_outcome( + workflow_name="Codage", success=True, + items_count=10, failed_count=0, use_llm=True, + ) + assert "10 dossiers" in rapport + + def test_llm_refine_network_error_safe(self): + """_llm_refine_summary ne doit jamais lever, même si requests échoue.""" + ctx = get_domain_context("tim_codage") + + fake_requests = MagicMock() + fake_requests.post.side_effect = RuntimeError("boom") + + with patch.dict("sys.modules", {"requests": fake_requests}): + out = ctx._llm_refine_summary( + template="ok", subs={"workflow_name": "x"}, success=True + ) + assert out == "" + + +# ============================================================================ +# Domaine custom enregistré dynamiquement +# ============================================================================ + + +class TestRegisterDomain: + def test_register_custom_domain(self): + custom = DomainContext( + domain_id="test_custom_xyz", + name="Test", + description="test", + common_actions={"click:foo": "faire foo"}, + summary_templates={ + "item_singular": "truc", + "item_plural": "trucs", + "success": "J'ai fait {done} trucs sur {items_count}.", + "partial": "Partiel : {done}/{items_count}.", + "failure": "Echec.", + }, + ) + register_domain(custom) + fetched = get_domain_context("test_custom_xyz") + assert fetched.name == "Test" + assert fetched.summarize_action("click", {"target": "FOO"}) == "faire foo" + + +# ============================================================================ +# Intégration avec ui.messages +# ============================================================================ + + +class TestMessagesIntegration: + """Les formatters de messages utilisent le domaine quand fourni.""" + + def test_cible_non_trouvee_domain_tim(self): + from agent_v0.agent_v1.ui.messages import formatter_cible_non_trouvee + + msg = formatter_cible_non_trouvee( + description_cible="Fichier patient", + titre_fenetre="DxCare", + domain_id="tim_codage", + params={"nom_patient": "Mme Durand"}, + ) + assert "Mme Durand" in msg.corps + + def test_cible_non_trouvee_domain_comptabilite(self): + from agent_v0.agent_v1.ui.messages import formatter_cible_non_trouvee + + msg = formatter_cible_non_trouvee( + description_cible="Montant HT", + titre_fenetre="Sage", + domain_id="comptabilite", + params={"num_facture": "F2026-007"}, + ) + assert "F2026-007" in msg.corps + + def test_cible_non_trouvee_sans_domain_retrocompat(self): + """Sans domain_id, comportement historique conservé.""" + from agent_v0.agent_v1.ui.messages import formatter_cible_non_trouvee + + msg = formatter_cible_non_trouvee( + description_cible="bonjour", + titre_fenetre="Test – Bloc-notes", + ) + assert "bonjour" in msg.corps + assert "Bloc-notes" in msg.corps + + def test_fin_workflow_tim_partiel(self): + from agent_v0.agent_v1.ui.messages import ( + NiveauMessage, + formatter_fin_workflow, + ) + + msg = formatter_fin_workflow( + succes=True, + nom_workflow="Codage janvier", + nb_etapes=120, + duree_s=900, + domain_id="tim_codage", + items_count=15, + failed_count=1, + ) + # Langage métier, pas "120 étapes" + assert "14 dossiers" in msg.corps + assert msg.niveau == NiveauMessage.ATTENTION # succès partiel + + def test_fin_workflow_tim_complet(self): + from agent_v0.agent_v1.ui.messages import ( + NiveauMessage, + formatter_fin_workflow, + ) + + msg = formatter_fin_workflow( + succes=True, + nom_workflow="Codage janvier", + nb_etapes=120, + duree_s=900, + domain_id="tim_codage", + items_count=15, + failed_count=0, + ) + assert "15 dossiers" in msg.corps + assert msg.niveau == NiveauMessage.INFO + + def test_fin_workflow_sans_domain_retrocompat(self): + from agent_v0.agent_v1.ui.messages import formatter_fin_workflow + + msg = formatter_fin_workflow( + succes=True, nom_workflow="Demo", nb_etapes=5, duree_s=10 + ) + assert "Demo" in msg.corps + assert "5 étapes" in msg.corps + + def test_erreur_generique_propagate_domain(self): + from agent_v0.agent_v1.ui.messages import formatter_erreur_generique + + msg = formatter_erreur_generique( + "target_not_found: Montant HT", + domain_id="comptabilite", + params={"num_facture": "F-001"}, + ) + assert "F-001" in msg.corps + + def test_friendly_target_tim_synonyme(self): + from agent_v0.agent_v1.ui.messages import _friendly_target + + assert _friendly_target("DP", "tim_codage") == "diagnostic principal" + assert _friendly_target("DP", None) == "DP" # pas de domaine → identique + assert _friendly_target("DP", "domaine_inexistant") == "DP"