# agent_v1/ui/messages.py """ Formatage des messages utilisateur pour Léa. Convertit les codes d'erreur techniques (`target_not_found`, `no_screen_change`...) en phrases en français naturel, orientées action, adaptées à un utilisateur non technique (secrétaire médicale, TIM). Trois niveaux de sévérité sont définis : - INFO — Léa fait son travail normalement - ATTENTION — Quelque chose de léger (ralentissement, retry) - BLOCAGE — Léa a besoin d'aide, elle rend la main Le module est 100% pur (pas d'I/O, pas d'UI) : testable sans mocks lourds. """ from __future__ import annotations import re from dataclasses import dataclass from enum import Enum 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): """Niveaux hiérarchiques des messages affichés à l'utilisateur.""" INFO = "info" # Fond vert clair, disparaît tout seul, 3-5s ATTENTION = "attention" # Fond orange clair, disparaît tout seul, 7s BLOCAGE = "blocage" # Fond rouge clair, reste affiché, 15s+ # Durée d'affichage par défaut (secondes), par niveau DUREE_PAR_NIVEAU: dict[NiveauMessage, int] = { NiveauMessage.INFO: 4, NiveauMessage.ATTENTION: 7, NiveauMessage.BLOCAGE: 15, } # Icône textuelle par niveau (compatible plyer/Windows/Linux) ICONE_PAR_NIVEAU: dict[NiveauMessage, str] = { NiveauMessage.INFO: "i", NiveauMessage.ATTENTION: "!", NiveauMessage.BLOCAGE: "?", } @dataclass class MessageUtilisateur: """Un message prêt à être affiché à l'utilisateur. Attributes: niveau: Hiérarchie (info/attention/blocage) titre: Titre court de la notification (≤60 caractères) corps: Corps du message en français naturel duree_s: Durée d'affichage recommandée (secondes) persistent: Si True, l'utilisateur doit fermer manuellement """ niveau: NiveauMessage titre: str corps: str duree_s: int persistent: bool = False def to_dict(self) -> dict: """Sérialiser le message (utile pour les tests et le logging).""" return { "niveau": self.niveau.value, "titre": self.titre, "corps": self.corps, "duree_s": self.duree_s, "persistent": self.persistent, } # ============================================================================ # Helpers d'extraction # ============================================================================ def _extraire_nom_application(titre_fenetre: str) -> str: """Extraire le nom de l'application à partir d'un titre de fenêtre. Les titres Windows suivent généralement le format : "Document.txt – Bloc-notes" "Ma Page - Google Chrome" "Sans titre — Paint" On retourne la partie après le dernier séparateur, ou le titre entier. """ if not titre_fenetre: return "" titre = titre_fenetre.strip() # Chercher le dernier séparateur parmi " – ", " — ", " - " for sep in (" – ", " — ", " - "): if sep in titre: return titre.rsplit(sep, 1)[-1].strip() return titre def _nettoyer_description_cible(description: str) -> str: """Nettoyer la description technique d'une cible pour l'afficher. Supprime les caractères techniques (guillemets inutiles, ':'). """ if not description: return "" desc = description.strip() # Retirer les guillemets encapsulants desc = desc.strip("'\"`") # Limiter la longueur if len(desc) > 80: desc = desc[:77] + "..." return desc # ============================================================================ # Formattage des messages techniques → humains # ============================================================================ 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 = _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}. " f"Peux-tu cliquer dessus toi-même ? Je reprends ensuite." ) else: corps = ( f"Je ne trouve pas « {cible} » à l'écran. " f"Peux-tu le faire toi-même ? Je reprends ensuite." ) return MessageUtilisateur( niveau=NiveauMessage.BLOCAGE, titre="Léa a besoin d'aide", corps=corps, duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE], persistent=True, ) def formatter_fenetre_incorrecte( titre_actuel: str, titre_attendu: str, ) -> MessageUtilisateur: """Message quand la fenêtre active n'est pas celle attendue. Exemple avant : Fenêtre incorrecte: 'Program Manager' (attendu: 'Lea : Explorateur de fichiers') Exemple après : Léa attend une fenêtre J'attends « Explorateur de fichiers » mais c'est « Program Manager » qui est affiché. Peux-tu ouvrir la bonne fenêtre ? """ app_actuelle = _extraire_nom_application(titre_actuel) or "une autre fenêtre" app_attendue = _extraire_nom_application(titre_attendu) or titre_attendu corps = ( f"J'attends « {app_attendue} » mais c'est « {app_actuelle} » " f"qui est affiché. Peux-tu ouvrir la bonne fenêtre ?" ) return MessageUtilisateur( niveau=NiveauMessage.BLOCAGE, titre="Léa attend une fenêtre", corps=corps, duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE], persistent=True, ) def formatter_ecran_inchange(action_type: str = "") -> MessageUtilisateur: """Message quand l'action n'a pas eu d'effet visible. Exemple avant : Ecran inchange apres l'action Exemple après : Léa vérifie Mon clic n'a pas eu l'air de marcher. Je vais réessayer ou te rendre la main si ça ne passe pas. """ actions_fr = { "click": "Mon clic", "type": "Ma saisie", "key_combo": "Mon raccourci clavier", "scroll": "Mon défilement", } quoi = actions_fr.get(action_type, "Mon action") corps = ( f"{quoi} n'a pas eu l'air de marcher. Je vais réessayer, " f"ou te rendre la main si ça ne passe pas." ) return MessageUtilisateur( niveau=NiveauMessage.ATTENTION, titre="Léa vérifie", corps=corps, duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION], ) def formatter_mode_apprentissage( raison: str = "", description_cible: str = "", titre_fenetre: Optional[str] = None, ) -> MessageUtilisateur: """Message quand Léa passe en mode apprentissage (pause supervisée). L'utilisateur doit comprendre : 1. Léa est bloquée et a besoin d'aide 2. L'utilisateur doit prendre la main et montrer comment faire 3. Ctrl+Shift+L pour signaler qu'il a fini Le ton est humble, clair, actionnable. Pas technique. Exemple : Léa a besoin d'aide Je n'y arrive pas, montrez-moi comment faire. Quand vous avez fini, appuyez sur Ctrl+Shift+L. """ cible = _nettoyer_description_cible(description_cible) if description_cible else "" app = _extraire_nom_application(titre_fenetre or "") if titre_fenetre else "" # Construire un contexte court si disponible contexte = "" if cible and app: contexte = f" (« {cible} » dans {app})" elif cible: contexte = f" (« {cible} »)" corps = ( f"Je n'y arrive pas{contexte}, montrez-moi comment faire. " f"Quand vous avez fini, appuyez sur Ctrl+Shift+L." ) return MessageUtilisateur( niveau=NiveauMessage.BLOCAGE, titre="Léa a besoin d'aide", corps=corps, duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE], persistent=True, ) def formatter_connexion_perdue(hote_serveur: str = "") -> MessageUtilisateur: """Message quand la connexion avec le serveur est perdue. Rassurant : on dit qu'on va réessayer automatiquement. """ corps = ( "J'ai perdu le lien avec le serveur. Je retente automatiquement, " "pas besoin d'intervenir." ) return MessageUtilisateur( niveau=NiveauMessage.ATTENTION, titre="Léa est déconnectée", corps=corps, duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION], ) def formatter_connexion_retablie() -> MessageUtilisateur: """Message quand la connexion serveur est rétablie.""" return MessageUtilisateur( niveau=NiveauMessage.INFO, titre="Léa", corps="C'est bon, la connexion est revenue. Je continue.", duree_s=DUREE_PAR_NIVEAU[NiveauMessage.INFO], ) def formatter_debut_workflow(nom_workflow: str, nb_etapes: int = 0) -> MessageUtilisateur: """Message au démarrage d'un workflow de replay.""" if nb_etapes > 0: corps = ( f"Je démarre « {nom_workflow} » ({nb_etapes} étapes). " f"Je t'indique mon avancement." ) else: corps = f"Je démarre « {nom_workflow} ». Je t'indique mon avancement." return MessageUtilisateur( niveau=NiveauMessage.INFO, titre="Léa démarre", corps=corps, duree_s=DUREE_PAR_NIVEAU[NiveauMessage.INFO], ) def formatter_etape_workflow( etape_actuelle: int, nb_etapes: int, description: str = "", ) -> MessageUtilisateur: """Message pour la progression d'une étape.""" if description: desc = _nettoyer_description_cible(description) corps = f"Étape {etape_actuelle}/{nb_etapes} — {desc}" else: corps = f"Étape {etape_actuelle}/{nb_etapes}" return MessageUtilisateur( niveau=NiveauMessage.INFO, titre="Léa avance", corps=corps, duree_s=3, ) def formatter_retry(action_type: str = "", tentative: int = 2) -> MessageUtilisateur: """Message quand Léa retente une action.""" corps = ( f"Je retente (tentative {tentative}). Ça arrive parfois, " f"l'écran était peut-être en cours de chargement." ) return MessageUtilisateur( niveau=NiveauMessage.ATTENTION, titre="Léa retente", corps=corps, duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION], ) def formatter_ralentissement() -> MessageUtilisateur: """Message quand Léa prend plus de temps que prévu.""" return MessageUtilisateur( niveau=NiveauMessage.ATTENTION, titre="Léa prend son temps", corps="Je vais plus lentement que prévu. L'écran met du temps à répondre.", duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION], ) def formatter_fin_workflow( succes: bool, 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. 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 = ( f"C'est fait ! « {nom_workflow} » est terminé " f"({nb_etapes} étapes en {int(duree_s)}s)." ) else: corps = "C'est fait ! Tout s'est bien passé." return MessageUtilisateur( niveau=NiveauMessage.INFO, titre="Léa a terminé", corps=corps, duree_s=6, ) else: corps = ( "Je n'ai pas pu terminer. Je te rends la main, " "tu peux continuer à partir de là où je me suis arrêtée." ) return MessageUtilisateur( niveau=NiveauMessage.BLOCAGE, titre="Léa s'arrête", corps=corps, duree_s=DUREE_PAR_NIVEAU[NiveauMessage.BLOCAGE], persistent=True, ) 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( niveau=NiveauMessage.ATTENTION, titre="Léa", corps="J'ai rencontré un petit souci. Je continue.", duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION], ) msg_lower = message_technique.lower() # target_not_found[:...] if "target_not_found" in msg_lower: # 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, 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: # Extraire actuel et attendu m_actuel = re.search(r"[:,]\s*['\"]([^'\"]+)['\"]", message_technique) m_attendu = re.search(r"attendu[:\s]*['\"]([^'\"]+)['\"]", message_technique) actuel = m_actuel.group(1) if m_actuel else "" attendu = m_attendu.group(1) if m_attendu else "" return formatter_fenetre_incorrecte(actuel, attendu) # Ecran inchangé if "inchang" in msg_lower or "no_screen_change" in msg_lower: return formatter_ecran_inchange() # Policy abort / supervise if "policy_abort" in msg_lower or "visual_resolve_failed" in msg_lower: return formatter_cible_non_trouvee( message_technique, domain_id=domain_id, params=params ) # Fallback : message technique tronqué msg_tronque = message_technique.strip() if len(msg_tronque) > 120: msg_tronque = msg_tronque[:117] + "..." return MessageUtilisateur( niveau=NiveauMessage.ATTENTION, titre="Léa", corps=f"J'ai rencontré un souci : {msg_tronque}", duree_s=DUREE_PAR_NIVEAU[NiveauMessage.ATTENTION], ) # ============================================================================ # Détection fenêtre Léa (utilisé par l'executor pour ignorer sa propre UI) # ============================================================================ # Motifs qui identifient une fenêtre appartenant à Léa (l'agent lui-même). # On utilise des regex avec \b pour éviter les faux positifs sur des noms # contenant "lea" (ex: "cléa.txt", "leapfrog", "replay"). _MOTIFS_FENETRE_LEA_REGEX = ( r"\bléa\b", r"\blea\b(?!p)", # "lea" mot entier, pas "leapfrog" r"lea\s*[—–\-:]", # "Lea —", "Lea -", "Lea :" r"léa\s*[—–\-:]", r"\bassistante ia\b", r"\bléa ia\b", r"\blea ia\b", ) def est_fenetre_lea(titre_fenetre: str) -> bool: """Détecter si un titre de fenêtre appartient à l'agent Léa lui-même. Utilisé pour éviter que Léa ne se considère comme une fenêtre intrusive dans ses propres pré-vérifications. Utilise des regex avec des word boundaries pour éviter les faux positifs sur des noms de fichiers contenant "lea" (ex: "cléa.txt", "replay.log"). """ if not titre_fenetre: return False titre_lower = titre_fenetre.lower().strip() return any(re.search(motif, titre_lower) for motif in _MOTIFS_FENETRE_LEA_REGEX) # Fenêtres parasites Windows à ignorer dans les pré-vérifications. # Ce ne sont pas des fenêtres applicatives — c'est du bruit système # qui prend le focus de manière imprévisible. _FENETRES_BRUIT_SYSTEME = ( "fenêtre de dépassement de capacité", "overflow", # version anglaise systray "program manager", "barre des tâches", "task bar", "cortana", "action center", "centre de notifications", ) def est_fenetre_bruit(titre_fenetre: str) -> bool: """Détecter si un titre de fenêtre est du bruit système Windows. Ces fenêtres prennent le focus de manière imprévisible (systray overflow, taskbar, Program Manager) et ne sont jamais la cible d'une action utilisateur. """ if not titre_fenetre: return True # pas de titre = bruit titre_lower = titre_fenetre.lower().strip() if titre_lower == "unknown_window": return True return any(p in titre_lower for p in _FENETRES_BRUIT_SYSTEME) # Conservé pour rétro-compatibilité avec le code qui listait MOTIFS_FENETRE_LEA MOTIFS_FENETRE_LEA = ( "léa", "lea —", "léa —", "lea -", "léa -", "lea assistante", "léa assistante", "lea : ", "léa : ", "assistante ia", )