# AXE D2 — Deep dive « chaîne popup plus propre » (impl. production-ready) **Date :** 2026-05-24 **Auteur :** agent recherche (dispatch Claude Opus 4.7 1M) **Périmètre :** Compléter `AXE_D2_DIALOG_POPUP.md` par une **implémentation prête à coller** du package `core/dialog/`, le câblage exact, la décision orphelin vs actif, la coordination avec le Validator B2, et l'outillage test offline. **Statut :** brief de recherche. Lecture seule sur le code existant. Aucune modification proposée à committer. **Prérequis lecture :** `AXE_D2_DIALOG_POPUP.md` (matrice modal→action et taxonomie déjà couvertes), `AXE_B2_VALIDATOR_PATTERN.md` (interface Verdict/FailureCategory), `LESSONS_LEARNED_GHT_2026-05.md` §🔴. --- ## 1. TL;DR + recommandation immédiate **Le code existant fait déjà 70 % du travail, mais éparpillé en 4 endroits :** - `agent_v0/agent_v1/core/system_dialog_guard.py` — détection multi-signal (ClassName UIA, processus, titre) UAC/CredUI/SmartScreen. **Excellent**. Fail-closed implémenté. - `core/grounding/dialog_handler.py` — `KNOWN_DIALOGS` métier + `_click_via_infigui` + `_click_via_ocr`. **Réutilisable tel quel**. - `core/grounding/title_verifier.py` — OCR titre 45 px, ~120 ms. **Réutilisable tel quel**. - `agent_v0/agent_v1/core/executor.py` — `_handle_popup_vlm` (actif, 4 sites d'appel), `_handle_possible_popup` (orphelin clavier, 0 site), `_KNOWN_RUNTIME_DIALOGS` (côté client, 1 entrée). **Recommandation immédiatement actionnable (1 jour) :** 1. **Créer un nouveau package `core/dialog/`** (côté serveur, pas client — pour mutualiser avec `dialog_handler.py` déjà serveur-side) qui **wrappe** les composants existants derrière 3 classes : `ChangeDetector`, `DialogClassifier`, `DialogResolver`. 2. **Garder `_handle_popup_vlm` côté client** (Léa Windows) mais le faire **déléguer** la décision politique au serveur via un endpoint `POST /api/v1/dialog/resolve`. Le client devient un exécuteur (capture + click), le serveur orchestre la cascade détection→classif→politique. 3. **Supprimer `_handle_possible_popup`** (orphelin, antipattern Tab+Enter+Esc aveugle qui viole `feedback_100pct_visual.md`). Référencé 0 fois, code mort. 4. **Câbler 3 sites d'appel** dans `executor.py:1108` (Observer pre-resolve, **déjà câblé** via `_handle_popup_vlm`), `executor.py:1262` (Policy post-grounding failed, **déjà câblé**), et 1 nouveau site **post-action click** (entre L:2270 et L:2475) — actuellement aucune vérif modal après un click réussi côté visuel mais ayant ouvert un dialog métier. 5. **Coordination Validator B2** : `Validator.validate()` retourne `Verdict.TERMINATE` + `FailureCategory.UNEXPECTED_DIALOG` → `api_stream` appelle `DialogResolver.resolve()` qui retourne soit `auto_dismissed` (replay continue) soit `pause_supervised` (replay stop avec event structuré). **Couverture estimée :** ~85 % des modaux courants traités sans intervention humaine (métier sauvegarde/écrasement + permission navigateur déclarative + OK trivial). Les 15 % restants (UAC, Hello, SmartScreen, INCONNU) = pause supervisée **par design healthtech**. **Effort :** 1 j pour MVP (DialogResolver côté serveur + 1 endpoint + suppression orphelin + 1 site câblé) → 1 sem pour matrice complète + tests offline → 1 mois pour bench injection + apprentissage catalogue. --- ## 2. Architecture finale du package `core/dialog/` ### 2.1. Arborescence ``` core/dialog/ ├── __init__.py # Exports publics (DialogResolver, DialogEvent, Verdict) ├── signatures.py # KNOWN_DIALOGS étendu fr+en + signatures par catégorie ├── change_detector.py # ChangeDetector léger (~80 LOC) ├── classifier.py # DialogClassifier OCR + VLM fallback (~150 LOC) ├── resolver.py # DialogResolver routing par politique (~200 LOC) └── events.py # DialogEvent dataclass + persistance audit ``` ### 2.2. Responsabilités (séparation stricte) | Module | Entrée | Sortie | Coût | |---|---|---|---| | `ChangeDetector` | screenshot_after PIL | `ChangeSignal` (bool is_modal + diagnostic) | ~50 ms | | `DialogClassifier` | screenshot PIL + ocr_text optionnel | `DialogType` enum | 0-1.7 s | | `DialogResolver` | screenshot + DialogType + workflow_ctx | `DialogEvent` (verdict + action) | 0-3 s | **Principe clé** : chaque composant est appelable **isolément**, sans dépendance circulaire. Tests unitaires triviaux. ### 2.3. Interface publique (`__init__.py`) ```python """core.dialog — Chaîne de gestion modaux & popups inattendus. Stack en 3 couches : ChangeDetector → DialogClassifier → DialogResolver Politique healthtech (immutable) : - JAMAIS d'auto-accept système (UAC/Hello/SmartScreen). - JAMAIS de raccourci système inventé (pas de Tab+Enter+Esc aveugle). - Catalogue déclaratif métier (KNOWN_DIALOGS) auto-dismiss explicite. - Tout dialog inconnu → pause supervisée. """ from core.dialog.change_detector import ChangeDetector, ChangeSignal from core.dialog.classifier import DialogClassifier, DialogType from core.dialog.resolver import DialogResolver, Policy from core.dialog.events import DialogEvent __all__ = [ "ChangeDetector", "ChangeSignal", "DialogClassifier", "DialogType", "DialogResolver", "Policy", "DialogEvent", ] ``` --- ## 3. Code complet (copy-paste-ready, testé syntaxiquement) ### 3.1. `signatures.py` — catalogue étendu (~120 LOC) ```python """core/dialog/signatures.py — Catalogue exhaustif des signatures dialog. Étend `core/grounding/dialog_handler.KNOWN_DIALOGS` avec les catégories système (UAC/Hello/SmartScreen) et navigateur (permissions). Source de vérité unique pour la classification. Ordre du dict = priorité de matching (popups modaux AVANT fenêtres parents, voir commentaire `dialog_handler.KNOWN_DIALOGS`). Toutes les signatures sont en MINUSCULES, l'OCR text doit être .lower() avant matching. Caractères accentués préservés (EasyOCR fr les conserve). """ from __future__ import annotations from enum import Enum from typing import Dict, List, Tuple class DialogType(str, Enum): # Catégorie SYSTÈME — pause supervisée obligatoire (healthtech). UAC = "uac" HELLO = "windows_hello" SMARTSCREEN = "defender_smartscreen" DEFENDER = "windows_defender" DRIVER = "driver_install" CREDUI = "credential_prompt" # Catégorie NAVIGATEUR — déclaratif workflow OU pause. BROWSER_PERMISSION = "browser_permission" BROWSER_SAVE_PASSWORD = "browser_save_password" BROWSER_BLOCKED_PAGE = "browser_blocked_page" # Catégorie MÉTIER — auto-dismiss déterministe via KNOWN_DIALOGS. METIER_SAVE = "metier_save" METIER_CONFIRM = "metier_confirm" METIER_OVERWRITE = "metier_overwrite" METIER_OK_TRIVIAL = "ok_trivial" METIER_OK_SUSPECT = "ok_suspect" # mots-clés "supprimé/perdu" → pause INCONNU = "inconnu" # Politique par catégorie (immutable healthtech). class Policy(str, Enum): AUTO_DISMISS = "auto_dismiss" # OK trivial seulement DECLARATIVE = "declarative" # catalog match → click ASK_HUMAN = "ask_human" # pause supervisée ESCALATE_SECURITY = "escalate_security" # pause + audit log full POLICY_BY_TYPE: Dict[DialogType, Policy] = { DialogType.UAC: Policy.ESCALATE_SECURITY, DialogType.HELLO: Policy.ESCALATE_SECURITY, DialogType.SMARTSCREEN: Policy.ESCALATE_SECURITY, DialogType.DEFENDER: Policy.ESCALATE_SECURITY, DialogType.DRIVER: Policy.ESCALATE_SECURITY, DialogType.CREDUI: Policy.ESCALATE_SECURITY, DialogType.BROWSER_PERMISSION: Policy.ASK_HUMAN, # sauf si déclaré workflow DialogType.BROWSER_SAVE_PASSWORD: Policy.ASK_HUMAN, DialogType.BROWSER_BLOCKED_PAGE: Policy.ASK_HUMAN, DialogType.METIER_SAVE: Policy.DECLARATIVE, DialogType.METIER_CONFIRM: Policy.DECLARATIVE, DialogType.METIER_OVERWRITE: Policy.DECLARATIVE, DialogType.METIER_OK_TRIVIAL: Policy.AUTO_DISMISS, DialogType.METIER_OK_SUSPECT: Policy.ASK_HUMAN, DialogType.INCONNU: Policy.ASK_HUMAN, } # Signatures texte (lowercase) → DialogType. Listes ordonnées car évaluées # séquentiellement (premier match gagne). Les signatures les plus spécifiques # en premier dans chaque catégorie. SIGNATURES_BY_TYPE: Dict[DialogType, List[str]] = { # ── SYSTÈME ───────────────────────────────────────────────────────── DialogType.UAC: [ "contrôle de compte d'utilisateur", "contrôle de compte dutilisateur", # OCR sans apostrophe "user account control", "voulez-vous autoriser cette application", "do you want to allow this app", "do you want to allow the following", ], DialogType.HELLO: [ "windows hello", "saisissez votre code pin", "saisir votre code pin", "enter your pin", "touchez le capteur d'empreintes", "touchez le lecteur", "use your fingerprint", "vérification de votre identité", # cf. feedback_auth_dialogs_runtime "analysez votre doigt", ], DialogType.SMARTSCREEN: [ "windows a protégé votre pc", "windows a protégé votre ordinateur", "windows protected your pc", "defender smartscreen", "smartscreen a empêché", "informations complémentaires", # accompagne SmartScreen "exécuter quand même", "run anyway", "éditeur inconnu", "unknown publisher", ], DialogType.DEFENDER: [ "windows defender", "menace détectée", "threat detected", "virus detected", ], DialogType.DRIVER: [ "installer ce pilote", "install this driver", "signature numérique du pilote", ], DialogType.CREDUI: [ "sécurité windows", "windows security", "entrer les informations d'identification", "enter your credentials", "connectez-vous à votre compte", "sign in to your account", ], # ── NAVIGATEUR ────────────────────────────────────────────────────── DialogType.BROWSER_PERMISSION: [ "souhaite utiliser votre microphone", "souhaite utiliser votre caméra", "souhaite utiliser votre micro", "souhaite afficher des notifications", "souhaite connaître votre position", "wants to use your microphone", "wants to use your camera", "wants to show notifications", "wants to know your location", "autoriser l'utilisation", "allow microphone", "allow camera", "autoriser les notifications", ], DialogType.BROWSER_SAVE_PASSWORD: [ "voulez-vous enregistrer ce mot de passe", "save password", "enregistrer le mot de passe", "voulez-vous que google chrome enregistre", ], DialogType.BROWSER_BLOCKED_PAGE: [ "cette page web n'a pas répondu", "cette page web ne répond pas", "page unresponsive", "this page isn't responding", "tuer les pages", "kill pages", ], # ── MÉTIER ────────────────────────────────────────────────────────── # Déjà couvert par core/grounding/dialog_handler.KNOWN_DIALOGS, # importé dynamiquement par DialogClassifier (single source of truth). } # Blocklist pour OK trivial : si ces mots apparaissent, on REFUSE l'auto-dismiss # et on escalade à l'humain. Action irréversible présumée. SUSPECT_TOKENS_BLOCKLIST: Tuple[str, ...] = ( "supprimer définitivement", "delete permanently", "perdu", "perdues", "lost", "irréversible", "irreversible", "cannot be undone", "vider la corbeille", "empty trash", "formater", "format", "effacer toutes", "erase all", ) def is_suspect_ok(ocr_text: str) -> bool: """Retourne True si l'OCR contient un mot-clé bloquant l'auto-dismiss.""" text_lower = ocr_text.lower() return any(token in text_lower for token in SUSPECT_TOKENS_BLOCKLIST) ``` ### 3.2. `change_detector.py` — détecteur léger (~100 LOC) ```python """core/dialog/change_detector.py — Détection rapide d'apparition de modal. Cible : < 50 ms par appel. Combine 3 signaux composables : 1. Foreground window changed (Windows API, ~1 ms) — signal complémentaire. 2. Screenshot diff zone centrale vs périphérie (~10 ms numpy). 3. Secure desktop detection (~1 ms, écran ~noir UAC). JAMAIS source unique. La décision finale repose sur composition (au moins 2 signaux concordants OU un signal très fort comme secure desktop). cf. feedback_popup_vlm.md : GetForegroundWindow seul n'est pas fiable (retourne 0 en SSH, popups modernes partagent hwnd parent). """ from __future__ import annotations import logging import time from dataclasses import dataclass from typing import Optional logger = logging.getLogger(__name__) @dataclass class ChangeSignal: is_modal: bool # Verdict composite (décision finale) foreground_changed: bool # Signal Windows API diff_ratio_global: float # 0.0-1.0 sur tout l'écran diff_ratio_central: float # 0.0-1.0 sur zone centrale secure_desktop: bool # Écran type UAC (très assombri) elapsed_ms: float class ChangeDetector: """Détecte qu'un modal vient d'apparaître sans appeler le VLM.""" # Seuils empiriques. À calibrer post-bench (cf. plan §11). DIFF_CENTRAL_THRESHOLD = 0.10 # >10% pixels modifiés zone centrale DIFF_GLOBAL_MAX_FOR_MODAL = 0.40 # un modal = changement local, pas global LUMINANCE_SECURE_DESKTOP = 50 # pixels < 50 / 255 = très sombre SECURE_DESKTOP_RATIO = 0.60 # > 60 % écran assombri = UAC probable def __init__(self): self._last_screenshot = None self._last_hwnd: Optional[int] = None def detect(self, screenshot_pil) -> ChangeSignal: """Analyser le screenshot courant vs précédent. Idempotent.""" import numpy as np t0 = time.time() arr = np.asarray(screenshot_pil.convert("L")) # Signal 1 : foreground window change fg_changed = self._check_foreground_changed() # Signal 2 : diff zoné central vs global diff_global = 0.0 diff_central = 0.0 if self._last_screenshot is not None: prev = np.asarray(self._last_screenshot.convert("L")) if prev.shape == arr.shape: diff = np.abs(prev.astype(int) - arr.astype(int)) diff_global = float((diff > 25).mean()) h, w = arr.shape cy0, cy1 = h // 4, 3 * h // 4 cx0, cx1 = w // 4, 3 * w // 4 diff_central = float((diff[cy0:cy1, cx0:cx1] > 25).mean()) # Signal 3 : secure desktop UAC (écran très assombri global) secure_desktop = ( float((arr < self.LUMINANCE_SECURE_DESKTOP).mean()) > self.SECURE_DESKTOP_RATIO ) # Décision composite. Au moins 2 signaux OU secure desktop seul. is_modal = secure_desktop or ( diff_central > self.DIFF_CENTRAL_THRESHOLD and diff_global < self.DIFF_GLOBAL_MAX_FOR_MODAL ) or (fg_changed and diff_central > 0.05) self._last_screenshot = screenshot_pil elapsed_ms = (time.time() - t0) * 1000 signal = ChangeSignal( is_modal=is_modal, foreground_changed=fg_changed, diff_ratio_global=diff_global, diff_ratio_central=diff_central, secure_desktop=secure_desktop, elapsed_ms=elapsed_ms, ) if is_modal: logger.info( "[CHANGE-DET] modal probable : fg=%s, diff_c=%.2f, " "diff_g=%.2f, secure=%s (%.0fms)", fg_changed, diff_central, diff_global, secure_desktop, elapsed_ms, ) return signal def reset(self) -> None: """Réinitialise l'état (utile entre 2 sessions de replay).""" self._last_screenshot = None self._last_hwnd = None def _check_foreground_changed(self) -> bool: """Windows-only. Renvoie False ailleurs ou en cas d'erreur.""" try: import ctypes hwnd = int(ctypes.windll.user32.GetForegroundWindow()) except Exception: return False if hwnd == 0: # SSH/Léa sans desktop accessible — signal inutilisable. return False changed = (self._last_hwnd is not None) and (hwnd != self._last_hwnd) self._last_hwnd = hwnd return changed ``` ### 3.3. `classifier.py` — classification cascade (~170 LOC) ```python """core/dialog/classifier.py — Classification d'un dialogue détecté. Stratégie cascade : 1. OCR full-screen (EasyOCR fr+en, ~150 ms via singleton partagé). 2. Match signatures texte (signatures.SIGNATURES_BY_TYPE + KNOWN_DIALOGS). 3. Fallback VLM compact (qwen3-vl:8b via Ollama LAN, ~1.7 s). 4. Si toujours rien → DialogType.INCONNU (politique = ASK_HUMAN). Le VLM compact est appelé UNIQUEMENT si signatures texte échouent ET ChangeDetector a confirmé is_modal. Évite de bloquer la boucle replay. """ from __future__ import annotations import logging import re import time from dataclasses import dataclass from typing import Optional from core.dialog.signatures import ( SIGNATURES_BY_TYPE, DialogType, is_suspect_ok, ) logger = logging.getLogger(__name__) @dataclass class ClassificationResult: dialog_type: DialogType confidence: float method: str # "signature" | "known_dialogs" | "vlm" | "fallback" ocr_text: str # texte brut OCR (pour audit) elapsed_ms: float class DialogClassifier: """Classifie un dialogue en type connu.""" # Modèle VLM compact — court output (max 50 tokens). VLM_MODEL_DEFAULT = "qwen3-vl:8b" VLM_TIMEOUT_S = 5.0 def __init__(self, ocr_fn=None, ollama_host: Optional[str] = None): """ ocr_fn: callable(PIL.Image) -> str. Si None, lazy-load EasyOCR fr+en. ollama_host: hôte Ollama pour fallback VLM. Défaut env RPA_OLLAMA_HOST ou "localhost". """ self._ocr = ocr_fn self._easyocr_reader = None import os self._ollama_host = ollama_host or os.environ.get("RPA_OLLAMA_HOST", "localhost") self._vlm_model = os.environ.get( "RPA_DIALOG_CLASSIFIER_MODEL", self.VLM_MODEL_DEFAULT ) def classify(self, screenshot_pil) -> ClassificationResult: t0 = time.time() # Étape 1 : OCR plein écran ocr_text = self._read_ocr(screenshot_pil) # Étape 2 : match signatures fr+en dtype = self._match_signatures(ocr_text) if dtype: elapsed = (time.time() - t0) * 1000 return ClassificationResult( dialog_type=dtype, confidence=0.9, method="signature", ocr_text=ocr_text[:300], elapsed_ms=elapsed, ) # Étape 3 : match catalogue métier existant (single source of truth) dtype = self._match_known_dialogs(ocr_text) if dtype: elapsed = (time.time() - t0) * 1000 return ClassificationResult( dialog_type=dtype, confidence=0.85, method="known_dialogs", ocr_text=ocr_text[:300], elapsed_ms=elapsed, ) # Étape 4 : OK trivial vs suspect (heuristique sans signature) if self._looks_like_ok_trivial(ocr_text): dtype = (DialogType.METIER_OK_SUSPECT if is_suspect_ok(ocr_text) else DialogType.METIER_OK_TRIVIAL) elapsed = (time.time() - t0) * 1000 return ClassificationResult( dialog_type=dtype, confidence=0.6, method="heuristic", ocr_text=ocr_text[:300], elapsed_ms=elapsed, ) # Étape 5 : fallback VLM compact dtype = self._classify_via_vlm(screenshot_pil) elapsed = (time.time() - t0) * 1000 if dtype: return ClassificationResult( dialog_type=dtype, confidence=0.7, method="vlm", ocr_text=ocr_text[:300], elapsed_ms=elapsed, ) # Fallback ultime : INCONNU → ASK_HUMAN return ClassificationResult( dialog_type=DialogType.INCONNU, confidence=0.0, method="fallback", ocr_text=ocr_text[:300], elapsed_ms=elapsed, ) # ── Implémentations ──────────────────────────────────────────────── def _read_ocr(self, screenshot_pil) -> str: if self._ocr is not None: return self._ocr(screenshot_pil) or "" # Lazy-load EasyOCR try: import numpy as np if self._easyocr_reader is None: import easyocr self._easyocr_reader = easyocr.Reader( ['fr', 'en'], gpu=True, verbose=False, ) results = self._easyocr_reader.readtext(np.array(screenshot_pil)) return ' '.join(r[1] for r in results if r[1].strip()) except Exception as e: logger.warning("[CLASSIFIER] OCR failed: %s", e) return "" @staticmethod def _match_signatures(ocr_text: str) -> Optional[DialogType]: text_lower = ocr_text.lower() if not text_lower: return None for dtype, signatures in SIGNATURES_BY_TYPE.items(): for sig in signatures: if sig in text_lower: logger.info("[CLASSIFIER] signature match '%s' → %s", sig, dtype) return dtype return None @staticmethod def _match_known_dialogs(ocr_text: str) -> Optional[DialogType]: """Réutilise core/grounding/dialog_handler.KNOWN_DIALOGS (source unique).""" try: from core.grounding.dialog_handler import KNOWN_DIALOGS except Exception: return None text_lower = ocr_text.lower() for key, info in KNOWN_DIALOGS.items(): if key in text_lower: # Heuristique : target=Oui → confirm, target=Enregistrer → save target = info.get("target", "").lower() if target in ("oui", "yes"): if "remplac" in key or "replace" in key or "écraser" in key: return DialogType.METIER_OVERWRITE return DialogType.METIER_CONFIRM elif target in ("enregistrer", "save"): return DialogType.METIER_SAVE return DialogType.METIER_OK_TRIVIAL return None @staticmethod def _looks_like_ok_trivial(ocr_text: str) -> bool: """Heuristique : 1 mot 'OK' isolé + court contexte = OK trivial.""" text_lower = ocr_text.lower() if not re.search(r"\b(ok|fermer|close|ferme)\b", text_lower): return False # Si trop de texte, ce n'est probablement pas un simple OK return len(ocr_text) < 400 def _classify_via_vlm(self, screenshot_pil) -> Optional[DialogType]: """Appel Ollama qwen3-vl:8b avec prompt français court. Latence cible < 2 s. Sortie attendue : un mot parmi la liste enum. Si format JSON requis, qwen3-vl:8b ignore parfois `format=json` (cf. BENCH_SAFETY_CHECKS_2026-05-06 §résultats). On parse en regex. """ try: import base64 import io import requests buf = io.BytesIO() screenshot_pil.convert("RGB").save(buf, format="JPEG", quality=75) img_b64 = base64.b64encode(buf.getvalue()).decode("ascii") prompt = ( "Cette capture montre un dialogue/popup. Classe-le en UN SEUL mot " "parmi : uac, hello, smartscreen, browser_permission, metier_save, " "metier_confirm, ok_trivial, inconnu. Réponds UNIQUEMENT le mot." ) payload = { "model": self._vlm_model, "messages": [ {"role": "system", "content": "Tu classes des dialogues Windows. Réponds en un mot."}, {"role": "user", "content": prompt, "images": [img_b64]}, ], "stream": False, "options": {"num_predict": 20, "temperature": 0.0}, } r = requests.post( f"http://{self._ollama_host}:11434/api/chat", json=payload, timeout=self.VLM_TIMEOUT_S, ) r.raise_for_status() response = r.json().get("message", {}).get("content", "").strip().lower() # Parse : chercher un mot enum dans la réponse for word in re.findall(r"[a-z_]+", response): try: return DialogType(word) except ValueError: continue return None except Exception as e: logger.warning("[CLASSIFIER] VLM fallback failed: %s", e) return None ``` ### 3.4. `resolver.py` — politique et action (~180 LOC) ```python """core/dialog/resolver.py — Application de la politique par catégorie. Routing : - SYSTÈME (UAC/Hello/SmartScreen/...) → escalation_pause_supervised() - MÉTIER déclaré → résolution via dialog_handler existant (InfiGUI + OCR) - INCONNU → pause par défaut, JAMAIS auto-dismiss Cf. AXE_D2_DIALOG_POPUP.md §5 matrice modal → action (autoritative). """ from __future__ import annotations import logging import time from typing import Any, Callable, Dict, Optional from core.dialog.classifier import ClassificationResult, DialogClassifier from core.dialog.change_detector import ChangeDetector, ChangeSignal from core.dialog.events import DialogEvent from core.dialog.signatures import ( POLICY_BY_TYPE, DialogType, Policy, ) logger = logging.getLogger(__name__) # Politique callbacks signature : (event, screenshot, workflow_ctx) -> bool resolved PolicyCallback = Callable[[DialogEvent, Any, Dict], bool] class DialogResolver: """Orchestre la chaîne ChangeDetector → Classifier → Politique → Action.""" def __init__( self, change_detector: Optional[ChangeDetector] = None, classifier: Optional[DialogClassifier] = None, on_pause_supervised: Optional[PolicyCallback] = None, on_auto_dismiss: Optional[PolicyCallback] = None, workflow_declared_handlers: Optional[Dict[DialogType, PolicyCallback]] = None, ): """ on_pause_supervised: callback appelé en cas d'ASK_HUMAN/ESCALATE. Signature : (event, screenshot, ctx) -> bool. Doit déclencher la pause dans api_stream/replay_engine. on_auto_dismiss: callback pour OK trivial. Doit cliquer le bouton OK via dialog_handler.handle_if_dialog OU pyautogui. workflow_declared_handlers: si un workflow déclare anticiper un type (ex. browser_permission "autoriser micro"), on appelle ce handler en priorité. """ self._change_detector = change_detector or ChangeDetector() self._classifier = classifier or DialogClassifier() self._on_pause = on_pause_supervised or self._default_pause self._on_dismiss = on_auto_dismiss or self._default_dismiss self._workflow_handlers = workflow_declared_handlers or {} def check_and_resolve( self, screenshot_pil, workflow_context: Optional[Dict[str, Any]] = None, force_classify: bool = False, ) -> Optional[DialogEvent]: """Point d'entrée principal. Args: screenshot_pil: capture courante (post-action ou pre-tick). workflow_context: ctx avec step_idx, action_id, declared_dialogs, etc. force_classify: bypass le ChangeDetector (utile pour B2 Validator qui sait déjà que quelque chose cloche). Returns: DialogEvent si un dialog a été détecté et traité, None sinon. """ workflow_context = workflow_context or {} t0 = time.time() # Étape 1 : détection rapide (sauf bypass) signal: Optional[ChangeSignal] = None if not force_classify: signal = self._change_detector.detect(screenshot_pil) if not signal.is_modal: return None # Étape 2 : classification classif = self._classifier.classify(screenshot_pil) # Étape 3 : politique policy = POLICY_BY_TYPE.get(classif.dialog_type, Policy.ASK_HUMAN) # Étape 3bis : exception workflow déclaratif (ex. permission micro attendue) declared = (workflow_context.get("declared_dialogs") or {}).get( classif.dialog_type.value ) if declared: policy = Policy.DECLARATIVE logger.info( "[RESOLVER] %s : politique forcée DECLARATIVE (workflow ctx)", classif.dialog_type, ) event = DialogEvent( dialog_type=classif.dialog_type, policy_applied=policy, confidence=classif.confidence, classification_method=classif.method, ocr_text=classif.ocr_text, change_signal=signal, workflow_step=workflow_context.get("step_idx"), workflow_action_id=workflow_context.get("action_id"), elapsed_ms=(time.time() - t0) * 1000, ) # Étape 4 : action selon politique resolved = self._apply_policy(event, screenshot_pil, workflow_context) event.action_taken = "resolved" if resolved else "paused" return event # ── Politiques ───────────────────────────────────────────────────── def _apply_policy( self, event: DialogEvent, screenshot, ctx: Dict[str, Any], ) -> bool: if event.policy_applied in (Policy.ESCALATE_SECURITY, Policy.ASK_HUMAN): return self._on_pause(event, screenshot, ctx) if event.policy_applied == Policy.AUTO_DISMISS: return self._on_dismiss(event, screenshot, ctx) if event.policy_applied == Policy.DECLARATIVE: # Workflow-declared handler en priorité handler = self._workflow_handlers.get(event.dialog_type) if handler: return handler(event, screenshot, ctx) # Fallback : utiliser le dialog_handler existant (InfiGUI + OCR) return self._default_declarative(event, screenshot, ctx) # Default safe : pause return self._on_pause(event, screenshot, ctx) @staticmethod def _default_pause(event: DialogEvent, screenshot, ctx: Dict) -> bool: logger.warning( "[RESOLVER] PAUSE SUPERVISÉE : %s (policy=%s, conf=%.2f) — " "aucun callback on_pause fourni, écho info uniquement.", event.dialog_type, event.policy_applied, event.confidence, ) return False @staticmethod def _default_dismiss(event: DialogEvent, screenshot, ctx: Dict) -> bool: """Auto-dismiss OK trivial : utilise dialog_handler existant.""" try: from core.grounding.dialog_handler import DialogHandler handler = DialogHandler() result = handler.handle_if_dialog(screenshot) return bool(result.get("handled")) except Exception as e: logger.warning("[RESOLVER] auto_dismiss failed: %s", e) return False @staticmethod def _default_declarative(event: DialogEvent, screenshot, ctx: Dict) -> bool: """Métier déclaratif : délégue au dialog_handler.KNOWN_DIALOGS.""" try: from core.grounding.dialog_handler import DialogHandler handler = DialogHandler() result = handler.handle_if_dialog(screenshot) if result.get("handled"): logger.info( "[RESOLVER] declarative resolved via dialog_handler: %s → %s", result.get("title", "?"), result.get("action", "?"), ) return True return False except Exception as e: logger.warning("[RESOLVER] declarative failed: %s", e) return False ``` ### 3.5. `events.py` — audit (~40 LOC) ```python """core/dialog/events.py — Event structuré pour audit + dashboard.""" from __future__ import annotations from dataclasses import asdict, dataclass, field from typing import Any, Dict, Optional from core.dialog.signatures import DialogType, Policy @dataclass class DialogEvent: dialog_type: DialogType policy_applied: Policy confidence: float classification_method: str # "signature" | "vlm" | "known_dialogs" | ... ocr_text: str change_signal: Optional[Any] = None # ChangeSignal dataclass workflow_step: Optional[int] = None workflow_action_id: Optional[str] = None action_taken: str = "unknown" elapsed_ms: float = 0.0 screenshot_path: Optional[str] = None extra: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: d = asdict(self) d["dialog_type"] = self.dialog_type.value d["policy_applied"] = self.policy_applied.value if self.change_signal is not None: d["change_signal"] = asdict(self.change_signal) return d ``` --- ## 4. Matrice modal → action finale (complète celle de `AXE_D2_DIALOG_POPUP.md §5`, focalisée sur l'**action concrète** par le `DialogResolver`) | `DialogType` | Politique | Signature match | Action `DialogResolver` | Latence cible | |---|---|---|---|---| | `UAC` | ESCALATE_SECURITY | "contrôle de compte", "user account control" | `on_pause(event)` → `replay_status=paused_need_help`, audit critical, screenshot full | 50ms détect + 150ms OCR | | `HELLO` | ESCALATE_SECURITY | "windows hello", "code pin", "vérification de votre identité" | identique UAC + tip pré-démo dans event.extra | 200ms | | `SMARTSCREEN` | ESCALATE_SECURITY | "windows a protégé", "smartscreen" | identique + ref `project_code_signing.md` dans event | 200ms | | `DEFENDER` | ESCALATE_SECURITY | "menace détectée", "threat detected" | identique | 200ms | | `DRIVER` | ESCALATE_SECURITY | "installer ce pilote", "signature pilote" | identique | 200ms | | `CREDUI` | ESCALATE_SECURITY | "sécurité windows", "entrer informations identification" | identique, vault Léa = orthogonal (gestion long terme) | 200ms | | `BROWSER_PERMISSION` | DECLARATIVE (si workflow) sinon ASK_HUMAN | "souhaite utiliser microphone/caméra/...", "autoriser/bloquer" | si `declared_dialogs[browser_permission]` → workflow_handler ; sinon `on_pause` | 200ms + 50ms click | | `BROWSER_SAVE_PASSWORD` | ASK_HUMAN | "enregistrer ce mot de passe" | `on_pause` (audit security) | 200ms | | `BROWSER_BLOCKED_PAGE` | ASK_HUMAN | "page n'a pas répondu" | `on_pause` (réseau ou crash app) | 200ms | | `METIER_SAVE` | DECLARATIVE | KNOWN_DIALOGS["voulez-vous enregistrer"] | `_default_declarative` → `DialogHandler.handle_if_dialog` → InfiGUI click "Enregistrer" | 200ms + 3s InfiGUI | | `METIER_CONFIRM` | DECLARATIVE | KNOWN_DIALOGS["confirmer"] | identique, click "Oui" | 200ms + 3s | | `METIER_OVERWRITE` | DECLARATIVE | KNOWN_DIALOGS["remplacer/écraser/already exists"] | identique, click "Oui" / "Yes" | 200ms + 3s | | `METIER_OK_TRIVIAL` | AUTO_DISMISS | heuristique `_looks_like_ok_trivial` + pas suspect | `_default_dismiss` → click "OK" via InfiGUI | 200ms + 3s | | `METIER_OK_SUSPECT` | ASK_HUMAN | mots-clés `SUSPECT_TOKENS_BLOCKLIST` ("supprimé", "perdu", ...) | `on_pause` (audit + screenshot full) | 200ms | | `INCONNU` | ASK_HUMAN | aucun match signature ni VLM | `on_pause` (capture VLM pour enrichir catalogue post-démo) | 50ms + 150ms OCR + 1.7s VLM | --- ## 5. Wiring : sites d'appel exacts ### 5.1. Côté serveur — `agent_v0/server_v1/api_stream.py` **Initialisation (boot)** : ```python # api_stream.py — au démarrage du module, à côté des autres init engines from core.dialog import DialogResolver, ChangeDetector, DialogClassifier from core.dialog.signatures import DialogType _DIALOG_RESOLVER = None def _get_dialog_resolver(): global _DIALOG_RESOLVER if _DIALOG_RESOLVER is None: _DIALOG_RESOLVER = DialogResolver( change_detector=ChangeDetector(), classifier=DialogClassifier(), on_pause_supervised=_dialog_pause_supervised, on_auto_dismiss=_dialog_auto_dismiss_via_lea, ) return _DIALOG_RESOLVER ``` **Endpoint nouveau** : ```python # api_stream.py — nouveau endpoint qui sert le client Léa @app.route("/api/v1/dialog/resolve", methods=["POST"]) def dialog_resolve(): """Le client Léa nous envoie un screenshot suspect, on classifie et retourne. Permet au client de ne pas dupliquer la logique de signatures. Le client reste responsable de l'action click (résolution coords). """ payload = request.get_json(force=True) screenshot_b64 = payload.get("screenshot") workflow_ctx = payload.get("workflow_context", {}) img = _b64_to_pil(screenshot_b64) resolver = _get_dialog_resolver() event = resolver.check_and_resolve(img, workflow_ctx, force_classify=True) if event is None: return jsonify({"dialog_detected": False}) return jsonify({ "dialog_detected": True, "event": event.to_dict(), # Si DECLARATIVE : indiquer au client le bouton à cliquer "click_target": _suggest_click_target_for(event), }) ``` **Site 1 : post-REPORT action** (le client a rapporté son résultat, on vérifie modal en sortie) : ```python # api_stream.py:report_action_result, ajouter APRÈS le pixel-diff actuel # (juste avant le `return jsonify({"replay_status": ...})`) if RPA_DIALOG_RESOLVER_ENABLED and screenshot_after_b64: img_after = _b64_to_pil(screenshot_after_b64) resolver = _get_dialog_resolver() event = resolver.check_and_resolve( img_after, workflow_context={ "step_idx": current_step_idx, "action_id": action_id, "declared_dialogs": current_workflow.get("declared_dialogs", {}), }, ) if event and event.policy_applied in (Policy.ASK_HUMAN, Policy.ESCALATE_SECURITY): _set_replay_status_paused( replay_id, reason=f"dialog:{event.dialog_type.value}", evidence=event.to_dict(), ) return jsonify({"replay_status": "paused_need_help", "dialog_event": event.to_dict()}) ``` **Site 2 : coordination Validator B2** (quand B2 retourne `TERMINATE` + `FailureCategory.UNEXPECTED_DIALOG`) : ```python # api_stream.py, après l'appel Validator.validate() (cf. AXE_B2 §6.4) if val.verdict == Verdict.TERMINATE and val.failure_category == FailureCategory.UNEXPECTED_DIALOG: # Le Validator a détecté qu'un modal bloque, mais ne sait pas quoi en faire. # Déléguer au DialogResolver pour classification + politique. resolver = _get_dialog_resolver() event = resolver.check_and_resolve( img_after, workflow_context={"step_idx": step_idx, "action_id": action_id, "declared_dialogs": current_workflow.get("declared_dialogs", {})}, force_classify=True, # B2 sait déjà qu'il y a un modal ) if event and event.policy_applied == Policy.DECLARATIVE and event.action_taken == "resolved": # Modal métier auto-résolu (ex. "Enregistrer ?") → reprendre normalement return jsonify({"replay_status": "in_progress", "dialog_event": event.to_dict()}) else: # Pause supervisée _set_replay_status_paused(replay_id, reason=val.reasoning, evidence=event.to_dict() if event else val.to_dict()) return jsonify({"replay_status": "paused_need_help"}) ``` ### 5.2. Côté client — `agent_v0/agent_v1/core/executor.py` **Décision** : NE PAS dupliquer la classification côté client. Simplifier `_handle_popup_vlm` pour qu'il appelle `/api/v1/dialog/resolve` du serveur, puis exécute le `click_target` retourné. **Diff conceptuel** (à valider Dom, lecture seule pour ce doc) : ```diff # agent_v0/agent_v1/core/executor.py — _handle_popup_vlm def _handle_popup_vlm(self) -> bool: # ── SÉCURITÉ inchangée : refus absolu sur dialogue système ── if self._check_and_pause_on_system_dialog(context="handle_popup_vlm"): return False screenshot_b64 = self._capture_screenshot_b64(max_width=0, quality=75) if not screenshot_b64: return False - # Essayer la détection popup via le serveur d'abord - from ..config import SERVER_URL, API_TOKEN - if SERVER_URL: - monitor = self.sct.monitors[1] - sw, sh = monitor["width"], monitor["height"] - server_result = self._server_resolve_target( - SERVER_URL, screenshot_b64, - {"vlm_description": "popup, dialog box, confirmation..."}, - 0.5, 0.5, sw, sh, - ) - if server_result and server_result.get("resolved"): - ... + # Nouveau : déléguer à DialogResolver serveur (single source of truth) + from ..config import SERVER_URL + if SERVER_URL: + try: + r = requests.post( + f"{SERVER_URL}/api/v1/dialog/resolve", + json={"screenshot": screenshot_b64, + "workflow_context": { + "action_id": self._current_action_id, + "declared_dialogs": self._current_declared_dialogs, + }}, + headers=self._auth_headers(), + timeout=8.0, + ) + if r.ok: + data = r.json() + if not data.get("dialog_detected"): + return False + event = data["event"] + if event["policy_applied"] in ("escalate_security", "ask_human"): + # Le serveur a déjà tracé l'event, on positionne juste + # le flag pause locale pour que le caller remonte. + self._system_dialog_pause = { + "category": event["dialog_type"], + "matched_signal": event["classification_method"], + "matched_value": event.get("ocr_text", "")[:80], + "reason": f"DialogResolver: {event['dialog_type']}", + "context": "handle_popup_vlm", + } + return False + # DECLARATIVE/AUTO_DISMISS → on a un click_target + target = data.get("click_target") + if target: + real_x = int(target["x_pct"] * sw) + real_y = int(target["y_pct"] * sh) + self._click((real_x, real_y), "left") + time.sleep(1.0) + return True + except Exception as e: + logger.warning("[POPUP-VLM] DialogResolver server failed: %s", e) - # Fallback : VLM local identifie le bouton à cliquer - button_text = self._vlm_identify_popup_button(screenshot_b64) + # Fallback (serveur indisponible) : VLM local existant + button_text = self._vlm_identify_popup_button(screenshot_b64) ... ``` --- ## 6. Activation `_handle_possible_popup` orphelin — décision finale **Verdict : SUPPRIMER.** 5 raisons : 1. **0 site d'appel** confirmé par grep — code mort (`executor.py:2960`). 2. **Antipattern strict `feedback_100pct_visual.md`** : tente Enter → Escape → Tab+Enter de manière aveugle. Or Échap dans un formulaire métier peut purger des données saisies ; Tab+Enter sur "Confirmer l'enregistrement" peut cliquer "Non" si le focus est dessus. 3. **Antipattern `feedback_lea_reflexes_catalog.md`** : si on veut composer Enter/Escape, ça passe par `gesture_catalog.py`, pas par un handler ad hoc. 4. **Couvert par `_handle_popup_vlm` actif** (4 sites d'appel) qui suit la cascade autorisée VLM → InfiGUI → OCR → click ciblé. 5. **Confusion code** : 2 méthodes avec noms similaires (`_handle_possible_popup` vs `_handle_popup_vlm`) → dette technique. Nettoyage = 1 modification triviale. **Patch suggéré** (lecture seule, à proposer à Dom) : ```diff # agent_v0/agent_v1/core/executor.py @@ -2956,73 +2956,8 @@ # ========================================================================= - # Gestion automatique des popups imprevues (legacy clavier) + # NOTE: _handle_possible_popup supprimé (orphelin, antipattern Tab+Enter aveugle). + # Cf. AXE_D2_DEEP_POPUP_CHAIN §6. Remplacé par chaîne DialogResolver serveur. # ========================================================================= - - def _handle_possible_popup(self) -> bool: - """Tenter de gerer une popup imprevue. - ... - """ - hash_before = self._quick_screenshot_hash() - ... - return False - - def _press_key(self, key): - ... - - def _press_tab_enter(self): - ... ``` Vérifier après suppression que `_press_key` / `_press_tab_enter` ne sont pas appelés ailleurs (probable, à grep avant patch). **Note méthode** : créer une **DETTE-XXX** dans `DETTE_TECHNIQUE.md` pour tracer la suppression et le replacement par DialogResolver. Cohérent avec `feedback_no_rustine.md` (corriger la cause = absence de chaîne unifiée, pas le symptôme). --- ## 7. Coordination Validator B2 ↔ DialogResolver — pseudo-code **Principe** : le Validator détecte qu'un check post-action **échoue** ; il route vers DialogResolver pour comprendre si la cause est un modal et le résoudre. ```python # agent_v0/server_v1/api_stream.py — handler de REPORT from agent_v0.server_v1.validator import Validator, Verdict, FailureCategory from core.dialog import DialogResolver from core.dialog.signatures import Policy async def report_action_result(payload): ... # Phase 1 : Validator B2 (matrice par action_type) val = _validator.validate( action=action, result=result, screenshot_before=before, screenshot_after=after, context=ctx, ) # Phase 2 : si Validator suspecte un modal OU verdict TERMINATE → DialogResolver needs_dialog_check = ( val.verdict == Verdict.TERMINATE or val.failure_category in ( FailureCategory.UNEXPECTED_DIALOG, FailureCategory.NO_VISUAL_CHANGE, FailureCategory.WRONG_APPLICATION, # cas bug step 10 démo GHT ) ) if needs_dialog_check: resolver = _get_dialog_resolver() event = resolver.check_and_resolve( after_pil, workflow_context={ "step_idx": ctx["step_idx"], "action_id": action["action_id"], "declared_dialogs": ctx.get("declared_dialogs", {}), }, force_classify=(val.failure_category == FailureCategory.UNEXPECTED_DIALOG), ) if event and event.policy_applied == Policy.DECLARATIVE and event.action_taken == "resolved": # Modal métier auto-résolu → on RÉ-EXÉCUTE l'action originale logger.info("[B2+D2] Dialog résolu (%s) → re-tentative action %s", event.dialog_type, action["action_id"]) return jsonify({"replay_status": "retry_after_dialog", "dialog_event": event.to_dict()}) if event and event.policy_applied in (Policy.ESCALATE_SECURITY, Policy.ASK_HUMAN): return jsonify({"replay_status": "paused_need_help", "dialog_event": event.to_dict(), "validator_evidence": val.to_dict()}) # Phase 3 : Validator dit COMPLETE → continuer if val.verdict == Verdict.COMPLETE: return jsonify({"replay_status": "in_progress"}) # Phase 4 : Validator dit CONTINUE (effet pas encore visible) → re-vérifier return jsonify({"replay_status": "wait_recheck", "recheck_ms": 1500}) ``` **Interface contractuelle** : - B2 produit `ValidationResult.failure_category` typé (cf. AXE_B2 §6.1). - D2 consomme cette `failure_category` pour décider de lancer/skipper le ChangeDetector. - Boucle : D2 résout métier → renvoie `retry_after_dialog` → B2 re-valide après retry. --- ## 8. Heartbeat & state machine ### 8.1. Rythme d'invocation | Phase Léa | Rythme `DialogResolver.check_and_resolve` | Justification | |---|---|---| | **`exec_action`** (action en cours côté client) | jamais (le client ne capture pas) | Pas pertinent, pas de signal | | **`post_action_report`** (REPORT serveur) | **SYSTÉMATIQUE** (1 fois par action) | Site primaire — cf. §5.1 Site 1 | | **`validator_b2`** (Validator post-action déclenché) | **CONDITIONNEL** (si TERMINATE/NO_VISUAL_CHANGE/UNEXPECTED_DIALOG) | Délégation §7 | | **`heartbeat_observe`** (tick `observe_reason_act` côté serveur) | **OPTIONNEL** (toutes les 5 s pendant un wait long) | Capture un modal apparu pendant attente t2a/extract_text | | **`paused_state`** (Léa en pause manuelle) | jamais | Inutile, humain a la main | **Total coût démo (40 steps, 2 min cible)** : - Post-REPORT : 40 × 50ms (ChangeDetector seul si pas de modal) = **2 s** - Si 5 modaux détectés sur la démo : +5 × 200ms (OCR+classif) = **+1 s** - Si 1 modal INCONNU appelle VLM fallback : +1 × 1.7s = **+1.7 s** - **Total : ~5 s sur démo de 120 s = 4 % overhead**. Acceptable. ### 8.2. State machine Léa simplifiée ``` ┌──────────────┐ │ IDLE │ └──────┬───────┘ │ start_action ▼ ┌──────────────┐ │ EXEC_ACTION │ (client capture+click) └──────┬───────┘ │ report ▼ ┌──────────────┐ ChangeDetector + Classifier │ POST_REPORT │──────► + Validator B2 └──────┬───────┘ │ ┌────────────┼────────────┬─────────────┐ ▼ ▼ ▼ ▼ no_dialog metier_auto system_dialog unknown continue resolved pause_super pause_super │ │ │ │ └────────────┴───────┬────┴─────────────┘ ▼ ┌──────────────┐ │ NEXT_ACTION │ │ or PAUSED │ └──────────────┘ ``` --- ## 9. Test offline pytest — snippet complet ### 9.1. `tests/unit/test_dialog_chain.py` ```python """tests/unit/test_dialog_chain.py — Tests offline chaîne DialogResolver. Charge des screenshots fixture, vérifie la cascade complète : ChangeDetector → DialogClassifier → DialogResolver → action. """ from __future__ import annotations from pathlib import Path from typing import Dict, List import pytest from PIL import Image from core.dialog import ChangeDetector, DialogClassifier, DialogResolver from core.dialog.signatures import DialogType, Policy FIXTURE_DIR = Path(__file__).parent.parent / "fixtures" / "dialogs" def _load(name: str) -> Image.Image: path = FIXTURE_DIR / name if not path.exists(): pytest.skip(f"Fixture absente : {path}") return Image.open(path).convert("RGB") def _fake_ocr(text: str): """Helper : retourne une fonction OCR qui renvoie toujours `text`.""" return lambda img: text # ── Tests ChangeDetector ─────────────────────────────────────────────── def test_change_detector_first_call_returns_no_change(): """Premier appel : pas de référence précédente → is_modal=False.""" det = ChangeDetector() img = Image.new("RGB", (1920, 1080), color=(128, 128, 128)) signal = det.detect(img) assert signal.is_modal is False assert signal.elapsed_ms < 100 # cible < 50ms, marge def test_change_detector_detects_central_change(): """Centre change beaucoup, périphérie stable → is_modal=True.""" det = ChangeDetector() img_before = Image.new("RGB", (1920, 1080), color=(200, 200, 200)) det.detect(img_before) # 1er appel img_after = img_before.copy() # Modal centré 800×500 sombre from PIL import ImageDraw draw = ImageDraw.Draw(img_after) draw.rectangle((560, 290, 1360, 790), fill=(50, 50, 50)) signal = det.detect(img_after) assert signal.is_modal is True assert signal.diff_ratio_central > 0.1 def test_change_detector_secure_desktop(): """Écran majoritairement très sombre → UAC secure desktop.""" det = ChangeDetector() img_dark = Image.new("RGB", (1920, 1080), color=(30, 30, 30)) signal = det.detect(img_dark) assert signal.secure_desktop is True assert signal.is_modal is True # ── Tests DialogClassifier ───────────────────────────────────────────── @pytest.mark.parametrize("ocr_text,expected", [ ("Contrôle de compte d'utilisateur Voulez-vous autoriser", DialogType.UAC), ("User Account Control", DialogType.UAC), ("Windows Hello Saisissez votre code PIN", DialogType.HELLO), ("Touchez le capteur d'empreintes digitales", DialogType.HELLO), ("Windows a protégé votre PC", DialogType.SMARTSCREEN), ("Defender SmartScreen a empêché", DialogType.SMARTSCREEN), ("Souhaite utiliser votre microphone Autoriser Bloquer", DialogType.BROWSER_PERMISSION), ("Voulez-vous enregistrer ce mot de passe", DialogType.BROWSER_SAVE_PASSWORD), ]) def test_classifier_signature_match(ocr_text, expected): classifier = DialogClassifier(ocr_fn=_fake_ocr(ocr_text)) img = Image.new("RGB", (100, 100)) result = classifier.classify(img) assert result.dialog_type == expected assert result.method == "signature" def test_classifier_known_dialogs_fallback(): """KNOWN_DIALOGS catalogue métier (single source of truth).""" classifier = DialogClassifier(ocr_fn=_fake_ocr("Voulez-vous remplacer le fichier ?")) img = Image.new("RGB", (100, 100)) result = classifier.classify(img) assert result.dialog_type == DialogType.METIER_OVERWRITE def test_classifier_unknown_no_vlm(): """Texte non match + VLM absent (timeout) → INCONNU.""" classifier = DialogClassifier(ocr_fn=_fake_ocr("Zorglub flubbergrabben")) classifier._ollama_host = "127.0.0.1:9999" # port impossible img = Image.new("RGB", (100, 100)) result = classifier.classify(img) assert result.dialog_type == DialogType.INCONNU # ── Tests DialogResolver ─────────────────────────────────────────────── def test_resolver_uac_triggers_pause(): """UAC → ESCALATE_SECURITY → on_pause callback appelé.""" calls = [] def on_pause(event, screenshot, ctx): calls.append((event.dialog_type, event.policy_applied)) return False img = Image.new("RGB", (1920, 1080), color=(30, 30, 30)) # dark = secure desktop classifier = DialogClassifier(ocr_fn=_fake_ocr("Contrôle de compte d'utilisateur")) resolver = DialogResolver( change_detector=ChangeDetector(), classifier=classifier, on_pause_supervised=on_pause, ) event = resolver.check_and_resolve(img) assert event is not None assert event.dialog_type == DialogType.UAC assert event.policy_applied == Policy.ESCALATE_SECURITY assert calls == [(DialogType.UAC, Policy.ESCALATE_SECURITY)] def test_resolver_workflow_declared_browser_permission(): """Permission micro DÉCLARÉE dans workflow → DECLARATIVE handler appelé.""" declared_calls = [] def declared_handler(event, screenshot, ctx): declared_calls.append(event.dialog_type) return True img = Image.new("RGB", (1920, 1080)) classifier = DialogClassifier(ocr_fn=_fake_ocr("Souhaite utiliser votre microphone")) resolver = DialogResolver( change_detector=ChangeDetector(), classifier=classifier, workflow_declared_handlers={ DialogType.BROWSER_PERMISSION: declared_handler, }, ) # Forcer is_modal pour éviter dépendance ChangeDetector event = resolver.check_and_resolve( img, workflow_context={ "declared_dialogs": {"browser_permission": {"action": "allow"}}, }, force_classify=True, ) assert event.policy_applied == Policy.DECLARATIVE assert event.action_taken == "resolved" assert declared_calls == [DialogType.BROWSER_PERMISSION] def test_resolver_suspect_ok_escalates(): """Mots-clés 'supprimer définitivement' → ASK_HUMAN même si OK trivial.""" img = Image.new("RGB", (1920, 1080)) classifier = DialogClassifier( ocr_fn=_fake_ocr("OK pour supprimer définitivement ce fichier") ) resolver = DialogResolver(classifier=classifier) event = resolver.check_and_resolve(img, force_classify=True) assert event.dialog_type == DialogType.METIER_OK_SUSPECT assert event.policy_applied == Policy.ASK_HUMAN ``` ### 9.2. Comment produire les fixtures `tests/fixtures/dialogs/` PowerShell snippets pour générer chaque type de dialog sur le PC Windows Léa : ```powershell # fixture_uac.png — déclencher UAC via runas Start-Process powershell -Verb RunAs # (cliquer Non → screenshot pendant prompt) # fixture_hello.png — déclencher Hello via gestionnaire identifiants control.exe /name Microsoft.CredentialManager # (cliquer "Ajouter un identifiant Windows" puis screenshot) # fixture_smartscreen.png — télécharger un .exe non signé Invoke-WebRequest -Uri "https://example.com/test.exe" -OutFile "$env:TEMP\test.exe" & "$env:TEMP\test.exe" # (SmartScreen popup → screenshot) # fixture_browser_permission.png — page test microphone Start-Process msedge "https://webcamtests.com" # (autoriser micro → screenshot) # fixture_metier_save.png — Bloc-notes non sauvé Start-Process notepad # (taper texte, Ctrl+W → "Voulez-vous enregistrer ?") ``` **Sources publiques de captures** (si Dom n'a pas d'accès Windows live pour générer) : - [Windows 11 UAC screenshots galerie Microsoft Learn](https://learn.microsoft.com/en-us/windows/security/application-security/application-control/user-account-control/how-it-works) — fixture statiques officielles. - [Defender SmartScreen UI samples — Microsoft GitHub docs](https://github.com/MicrosoftDocs/windows-itpro-docs/tree/public/windows/security/threat-protection/microsoft-defender-smartscreen). - Notre `data/runner_captures/` contient déjà des screenshots de replays — `grep`er ceux qui ont émis un event "popup_handled" pour réutilisation. **Tests sans fixture (CI sans Windows)** : les `_fake_ocr(...)` permettent de tester la cascade Signature → Classifier → Resolver sans aucun screenshot réel. Couvre 80 % de la logique. Les fixtures restent utiles pour `ChangeDetector` (qui dépend du diff pixel réel). --- ## 10. Patterns externes 2026 (compléments à AXE_D2 §3) ### 10.1. Skyvern dialog handling (vérif source mai 2026) **Issue #69 ouverte sept. 2024, toujours active** : « Unable to interact with popup modals on costcotravel.com ». Skyvern délègue tout à son Validator post-action (cf. `complete_verify` analysé dans AXE_B2 §2.1). Pas de DialogResolver dédié → c'est leur **point faible**. [Source : Skyvern Issue #69](https://github.com/Skyvern-AI/skyvern/issues/69), [Prompting Guide](https://docs.skyvern.com/getting-started/prompting-guide) — la doc officielle invite à **décrire le popup attendu dans le prompt** (« déclaratif workflow ») exactement comme notre `declared_dialogs[...]`. ### 10.2. browser-use Issue #1996 (juin 2025, closed) [Source : browser-use Issue #1996](https://github.com/browser-use/browser-use/issues/1996) — issue **fermée sans fix** (« handling left to the LLM in the prompt »). Confirme que **l'écosystème open source n'a pas de standard** sur ce sujet. Notre approche `signatures + VLM fallback` est en avance. ### 10.3. Anthropic Computer Use 2026 — politique dialogs [Source : Computer Use API Docs](https://docs.anthropic.com/en/docs/build-with-claude/computer-use), [Claude Opus 4.6 system card fév. 2026](https://www-cdn.anthropic.com/0dd865075ad3132672ee0ab40b05a53f14cf5288.pdf) : - **Permission-first** : Claude demande confirmation avant tout nouvel app. - **Classifiers anti-prompt-injection** : screenshots suspects → demande confirmation user. - **Pas de UAC/Hello handling spécifique** documenté. Anthropic se repose sur le fait que Claude **identifie** un dialog système et **refuse implicitement** d'y cliquer si pas explicitement instruit. Risqué : aucune garantie. Notre approche `system_dialog_guard.py` (multi-signal ClassName UIA + processus + titre) est **plus robuste** que la self-reflection LLM. ### 10.4. OpenAI Operator / ChatGPT Agent — handover explicite [Source : ChatGPT Agent help](https://help.openai.com/en/articles/11752874-chatgpt-agent), [Operator system card](https://openai.com/index/operator-system-card/) : - Sur CAPTCHA, login, paiement → **« proactively asks the user to take over »**. - Pendant le takeover : **screenshots OFF** (protection credentials). - Modèle CUA 2026 : OSWorld 45 % (vs 38 % preview). **Pattern transposable** : notre `pause_supervised` doit ressembler à ce handover. Le dashboard VWB devrait afficher le screenshot **figé au moment du modal** (pas live) puis reprendre la capture après resolution humaine. Cohérent avec `feedback_failure_is_learning.md`. ### 10.5. Cradle (BAAI) — Self-Reflection module [Source : Cradle GitHub](https://github.com/BAAI-Agents/Cradle), [arXiv 2403.03186](https://arxiv.org/pdf/2403.03186) — agent jeu vidéo avec module Self-Reflection (+20.41 pts). N'a **pas** de DialogResolver spécifique car le contexte (jeu) n'a pas de modaux système Windows. Non transposable directement. ### 10.6. UAC secure desktop detection (Win32) [Source : Microsoft GetSystemMetrics docs](https://learn.microsoft.com/en-us/windows/win32/api/winuser/nf-winuser-getsystemmetrics), [PyAutoGUI _pyautogui_win.py](https://github.com/asweigart/pyautogui/blob/master/pyautogui/_pyautogui_win.py), [Sigma rule UAC secure desktop disabled](https://detection.fyi/sigmahq/sigma/windows/registry/registry_set/registry_set_uac_disable_secure_desktop_prompt/) : - `ctypes.windll.user32.GetSystemMetrics(SM_REMOTESESSION=0x1000)` indique session distante (RDP/Citrix), pas secure desktop directement. - **Pas d'API publique** Win32 pour détecter le secure desktop UAC. Solution : screenshot luminance (déjà dans `ChangeDetector._check_foreground_changed`). - Registry `PromptOnSecureDesktop=0` désactive le secure desktop (option config Windows, à anticiper en démo). --- ## 11. Plan d'intégration gradué ### 11.1. Court terme — 1 jour (P0, avant prochaine démo) **But** : MVP fonctionnel sans casser l'existant, kill-switch off par défaut. 1. Créer `core/dialog/` avec les 5 fichiers (§3) — **3 h**. 2. Endpoint serveur `POST /api/v1/dialog/resolve` (§5.1) — **1 h**. 3. Câbler Site 1 (post-REPORT) avec env var `RPA_DIALOG_RESOLVER_ENABLED=false` par défaut — **1 h**. 4. Tests unit (`test_dialog_chain.py` §9) sans fixtures réelles, juste `_fake_ocr` — **2 h**. 5. Smoke test : démarrer Léa + serveur + workflow Demo_urgence_3_db avec flag `=true`, mesurer latence (cible < 50 ms par check sans modal) — **1 h**. **Livrable** : DialogResolver disponible derrière flag. Démo inchangée si flag off. ### 11.2. Moyen terme — 1 semaine (P1) 1. Migration `_handle_popup_vlm` côté client vers délégation serveur (§5.2 diff) — **3 h**. 2. Suppression `_handle_possible_popup` orphelin + grep cleanup `_press_key/_press_tab_enter` — **1 h**. 3. Capture des 5 fixtures Windows (UAC/Hello/SmartScreen/permission/métier) via PowerShell (§9.2) — **2 h**. 4. Tests d'intégration avec fixtures réelles — **3 h**. 5. Coordination Validator B2 (§7) — **3 h**. 6. Dashboard VWB : panneau "Dialog events" par session (count par type + dernier ocr_text) — **1 j**. ### 11.3. Long terme — 1 mois (P2) 1. **Bench injection** : harness qui injecte UAC simulé / popup métier / SmartScreen pendant replay test, mesure detect→classify→resolve, taux pause vs auto-dismiss — **3 j**. 2. **Apprentissage catalogue** : chaque `DialogType.INCONNU` enregistré dans BDD → revue Dom toutes les semaines → enrichit `SIGNATURES_BY_TYPE` (pattern OpenAdapt Evaluation-Driven Feedback) — **continu**. 3. **Win32 UIA hook** : `SetWindowsHookEx(WH_CBT)` pour détecter `HCBT_CREATEWND` d'une fenêtre modale → signal complémentaire au screenshot diff. Pertinence Citrix douteuse (UIA aveugle), à benchmarker — **2 j R&D**. 4. **DialogResolver pour Citrix** : adapter détection (secure desktop UAC d'un client Citrix passe dans le framebuffer hôte) — **3 j**. 5. **Synergie AXE_A5 (tokenisation écran)** : si parser UI produit liste éléments interactifs, classification devient déterministe (matche label bouton) — **dépend roadmap A5**. --- ## 12. Sources (liens cliquables, dates 2025-2026) ### Frameworks externes - [Skyvern Issue #69 — Unable to interact with popup modals](https://github.com/Skyvern-AI/skyvern/issues/69) (sept. 2024, toujours actif) - [Skyvern Prompting Guide — handle modals declaratively](https://docs.skyvern.com/getting-started/prompting-guide) - [Skyvern Blog Index](https://www.skyvern.com/blog/) - [browser-use Issue #1996 — Need Robust Strategy for Handling Dynamic Popups](https://github.com/browser-use/browser-use/issues/1996) (juin 2025, fermée sans fix) - [Anthropic Computer Use API docs](https://docs.anthropic.com/en/docs/build-with-claude/computer-use) - [Claude Opus 4.6 system card (fév. 2026)](https://www-cdn.anthropic.com/0dd865075ad3132672ee0ab40b05a53f14cf5288.pdf) - [OpenAI Operator system card](https://openai.com/index/operator-system-card/) - [OpenAI ChatGPT Agent — takeover sur login/CAPTCHA](https://help.openai.com/en/articles/11752874-chatgpt-agent) - [Anthropic Claude Computer Use on Windows (avril 2026)](https://www.thurrott.com/a-i/anthropic/334498/anthropic-brings-claude-computer-use-to-windows) - [Cradle GitHub (BAAI)](https://github.com/BAAI-Agents/Cradle), [Cradle paper arXiv 2403.03186](https://arxiv.org/pdf/2403.03186) ### Windows 11 / UAC / Hello / SmartScreen / Secure Desktop - [Microsoft UAC architecture](https://learn.microsoft.com/en-us/windows/security/application-security/application-control/user-account-control/architecture) - [Microsoft GetSystemMetrics function](https://learn.microsoft.com/en-us/windows/win32/api/winuser/nf-winuser-getsystemmetrics) - [Sigma rule UAC Secure Desktop Prompt Disabled](https://detection.fyi/sigmahq/sigma/windows/registry/registry_set/registry_set_uac_disable_secure_desktop_prompt/) - [PyAutoGUI _pyautogui_win.py ctypes patterns](https://github.com/asweigart/pyautogui/blob/master/pyautogui/_pyautogui_win.py) - [ctypes user32 reference jerblack gist](https://gist.github.com/jerblack/2b294916bd46eac13da7d8da48fcf4ab) ### CVE & menaces 2026 - [CVE-2026-0628 Chrome Gemini Live extension takeover](https://news.corksafetyalerts.com/chrome-flaw-allowed-extensions-to-hijack-googles-ai-assistant-camera-and-microphone/) - [AI-powered phishing leveraging hardware access (2026)](https://www.scworld.com/brief/ai-powered-phishing-campaign-leverages-hardware-access-for-data-theft) ### Documents internes rpa_vision_v3 - `docs/recherche/AXE_D2_DIALOG_POPUP.md` (parent, matrice §5 autoritative) - `docs/recherche/AXE_B2_VALIDATOR_PATTERN.md` (interface Verdict, FailureCategory) - `docs/LESSONS_LEARNED_GHT_2026-05.md` §🔴 (bugs P0) - `core/grounding/dialog_handler.py` (KNOWN_DIALOGS réutilisé) - `core/grounding/title_verifier.py` (OCR titre 45px) - `agent_v0/agent_v1/core/system_dialog_guard.py` (multi-signal système) - `agent_v0/agent_v1/core/executor.py` (sites d'appel `_handle_popup_vlm`) - `agent_chat/gesture_catalog.py` (seul "réflexe système" autorisé) - `memory/feedback_popup_vlm.md`, `feedback_100pct_visual.md`, `feedback_lea_reflexes_catalog.md`, `feedback_auth_dialogs_runtime.md`, `feedback_phash_vs_dialog_in_vm.md` --- ## 13. Hors-périmètre — questions à valider Dom avant action 1. **Décision suppression `_handle_possible_popup`** : confirmer (grep + retrait, créer DETTE pour traçabilité). 2. **Choix modèle VLM fallback** : `qwen3-vl:8b` retenu (cohérent §2.4 synthèse) mais à benchmarker sur 10 captures de dialogs (fixture). 3. **Politique rétention RGPD/HDS screenshots `DialogEvent`** : par défaut `data/runner_captures/dialogs//.png`, purge après ACK serveur ou TTL 30 j ? Aligner avec `feedback_capture_purge_policy.md`. 4. **Workflow `declared_dialogs`** : extension VWB pour permettre au designer de déclarer "à cette étape, autoriser le micro". Format JSON suggéré : `{"browser_permission": {"action": "allow", "label": "Autoriser"}}`. À spécifier avec frontend VWB. 5. **Synergie AXE_B1 (watchdog transport)** : si une action est en `_retry_pending` côté serveur et qu'un modal apparaît côté Léa pendant l'attente, le watchdog doit-il propager l'event ? Couplage à clarifier. 6. **Bench latence empirique** : valider `ChangeDetector < 50 ms` sur capture réelle 2560×1600 (Demo_urgence_3_db). Si dépassement, downscale 1/4 avant diff numpy. --- *Document de recherche. Lecture seule sur code existant. Suite = décision Dom + chirurgie itérative supervisée (CLAUDE.md projet).*