# agent_v0/server_v1/replay_verifier.py """ ReplayVerifier — Vérification post-action pour le replay de workflows. Compare les screenshots avant/après une action pour détecter si elle a eu un effet visible. Utilisé par l'API de replay pour décider si une action a réussi ou si un retry est nécessaire. Stratégies de vérification : 1. Différence d'image globale (avant == après → probablement rien ne s'est passé) 2. Zone locale autour du clic (si l'action est un clic) 3. Détection de texte apparu (si l'action est une frappe) """ import logging from dataclasses import dataclass, field from typing import Any, Dict, Optional, Tuple logger = logging.getLogger(__name__) # Seuils de détection configurables DEFAULT_GLOBAL_CHANGE_THRESHOLD = 0.005 # 0.5% de pixels différents = changement détecté DEFAULT_LOCAL_CHANGE_THRESHOLD = 0.02 # 2% de la zone locale doit changer pour un clic DEFAULT_LOCAL_RADIUS_PCT = 0.05 # 5% de la taille d'image autour du point de clic DEFAULT_PIXEL_DIFF_THRESHOLD = 30 # Différence minimale par canal pour compter un pixel comme "changé" @dataclass class VerificationResult: """Résultat de vérification d'une action de replay.""" verified: bool # L'action semble avoir fonctionné confidence: float # 0.0-1.0 changes_detected: bool # Des pixels ont changé change_area_pct: float # % de l'image qui a changé (0.0-100.0) suggestion: str # "retry", "skip", "abort", "continue" detail: str = "" # Description humaine du résultat local_change_pct: float = 0.0 # % de changement dans la zone locale (si applicable) def to_dict(self) -> Dict[str, Any]: return { "verified": self.verified, "confidence": round(self.confidence, 3), "changes_detected": self.changes_detected, "change_area_pct": round(self.change_area_pct, 3), "suggestion": self.suggestion, "detail": self.detail, "local_change_pct": round(self.local_change_pct, 3), } class ReplayVerifier: """Vérifie que les actions de replay ont produit l'effet attendu.""" def __init__( self, global_change_threshold: float = DEFAULT_GLOBAL_CHANGE_THRESHOLD, local_change_threshold: float = DEFAULT_LOCAL_CHANGE_THRESHOLD, local_radius_pct: float = DEFAULT_LOCAL_RADIUS_PCT, pixel_diff_threshold: int = DEFAULT_PIXEL_DIFF_THRESHOLD, ): self.global_change_threshold = global_change_threshold self.local_change_threshold = local_change_threshold self.local_radius_pct = local_radius_pct self.pixel_diff_threshold = pixel_diff_threshold def verify_action( self, action: Dict[str, Any], result: Dict[str, Any], screenshot_before: Optional[str] = None, screenshot_after: Optional[str] = None, ) -> VerificationResult: """ Compare les screenshots avant/après pour détecter si l'action a eu un effet. Stratégies : 1. Différence d'image (si avant == après, l'action n'a probablement rien fait) 2. Si l'action est un clic, vérifier que la zone autour du clic a changé 3. Si l'action est une frappe, vérifier que du texte est apparu Args: action: L'action exécutée (type, x_pct, y_pct, text, etc.) result: Le résultat rapporté par l'Agent V1 (success, error, etc.) screenshot_before: Chemin du screenshot avant l'action (optionnel) screenshot_after: Chemin du screenshot après l'action (optionnel) Returns: VerificationResult avec la conclusion et la suggestion de suite """ # Si l'agent a rapporté une erreur explicite, pas besoin de vérifier visuellement if not result.get("success", True): return VerificationResult( verified=False, confidence=0.9, changes_detected=False, change_area_pct=0.0, suggestion="retry", detail=f"Action échouée: {result.get('error', 'erreur inconnue')}", ) # Si pas de screenshots, on ne peut pas vérifier if not screenshot_before or not screenshot_after: return VerificationResult( verified=True, confidence=0.3, changes_detected=True, # On ne sait pas, on assume que ça a marché change_area_pct=0.0, suggestion="continue", detail="Vérification impossible (pas de screenshots avant/après)", ) # Charger les images try: img_before, img_after = self._load_images(screenshot_before, screenshot_after) except Exception as e: logger.warning(f"Impossible de charger les screenshots: {e}") return VerificationResult( verified=True, confidence=0.2, changes_detected=True, change_area_pct=0.0, suggestion="continue", detail=f"Erreur chargement images: {e}", ) # Vérifier les dimensions if img_before.size != img_after.size: # Résolutions différentes = probablement un changement d'écran return VerificationResult( verified=True, confidence=0.7, changes_detected=True, change_area_pct=100.0, suggestion="continue", detail="Résolution d'écran modifiée (changement de contexte)", ) # 1. Calcul de la différence globale global_change_pct = self._compute_global_diff(img_before, img_after) # 2. Calcul de la différence locale (zone autour du clic si applicable) action_type = action.get("type", "") local_change_pct = 0.0 if action_type in ("click", "type") and "x_pct" in action and "y_pct" in action: local_change_pct = self._compute_local_diff( img_before, img_after, action["x_pct"], action["y_pct"], ) # 3. Décision return self._decide( action_type=action_type, global_change_pct=global_change_pct, local_change_pct=local_change_pct, ) def _load_images(self, path_before: str, path_after: str): """Charger deux images PIL depuis des chemins fichier ou base64.""" from PIL import Image img_before = self._load_single_image(path_before) img_after = self._load_single_image(path_after) return img_before, img_after def _load_single_image(self, source: str): """Charger une image depuis un chemin fichier ou une string base64.""" from PIL import Image # Détection base64 (commence par /9j pour JPEG ou iVBOR pour PNG en base64) if source.startswith(("/9j", "iVBOR", "data:image")): import base64 import io # Retirer le préfixe data:image/...;base64, si présent if source.startswith("data:image"): source = source.split(",", 1)[1] img_bytes = base64.b64decode(source) return Image.open(io.BytesIO(img_bytes)).convert("RGB") else: return Image.open(source).convert("RGB") def _compute_global_diff(self, img_before, img_after) -> float: """ Calculer le pourcentage de pixels qui ont changé significativement. Returns: Pourcentage de pixels changés (0.0-100.0) """ import numpy as np arr_before = np.array(img_before, dtype=np.int16) arr_after = np.array(img_after, dtype=np.int16) # Différence absolue par canal, puis max par pixel diff = np.abs(arr_after - arr_before) max_diff_per_pixel = diff.max(axis=2) # (H, W) # Compter les pixels dont la différence dépasse le seuil changed_pixels = (max_diff_per_pixel > self.pixel_diff_threshold).sum() total_pixels = max_diff_per_pixel.size return (changed_pixels / total_pixels) * 100.0 def _compute_local_diff( self, img_before, img_after, x_pct: float, y_pct: float, ) -> float: """ Calculer le pourcentage de changement dans une zone locale autour d'un point. Args: img_before, img_after: Images PIL (même taille) x_pct, y_pct: Coordonnées du point en pourcentage (0.0-1.0) Returns: Pourcentage de pixels changés dans la zone locale (0.0-100.0) """ import numpy as np w, h = img_before.size cx = int(x_pct * w) cy = int(y_pct * h) radius_x = int(self.local_radius_pct * w) radius_y = int(self.local_radius_pct * h) # Borner la zone au cadre de l'image x1 = max(0, cx - radius_x) y1 = max(0, cy - radius_y) x2 = min(w, cx + radius_x) y2 = min(h, cy + radius_y) if x2 <= x1 or y2 <= y1: return 0.0 # Extraire les zones locales crop_before = img_before.crop((x1, y1, x2, y2)) crop_after = img_after.crop((x1, y1, x2, y2)) arr_before = np.array(crop_before, dtype=np.int16) arr_after = np.array(crop_after, dtype=np.int16) diff = np.abs(arr_after - arr_before) max_diff = diff.max(axis=2) changed = (max_diff > self.pixel_diff_threshold).sum() total = max_diff.size return (changed / total) * 100.0 if total > 0 else 0.0 def _decide( self, action_type: str, global_change_pct: float, local_change_pct: float, ) -> VerificationResult: """ Prendre une décision basée sur les métriques de changement. Logique : - Changement global > seuil → action vérifiée (confiance haute) - Changement local > seuil (pour clic/frappe) → action vérifiée (confiance moyenne) - Aucun changement → action non vérifiée, suggestion retry - Changement massif (>50%) → possible popup/erreur, marquer pour attention """ global_threshold_pct = self.global_change_threshold * 100 local_threshold_pct = self.local_change_threshold * 100 has_global_change = global_change_pct > global_threshold_pct has_local_change = local_change_pct > local_threshold_pct # Cas 1 : Changement massif (possible popup/erreur/crash) if global_change_pct > 50.0: return VerificationResult( verified=True, confidence=0.6, changes_detected=True, change_area_pct=global_change_pct, local_change_pct=local_change_pct, suggestion="continue", detail=( f"Changement massif détecté ({global_change_pct:.1f}%) — " "possible changement de contexte (popup, nouvelle page)" ), ) # Cas 2 : Changement global détecté if has_global_change: confidence = min(0.9, 0.5 + global_change_pct / 100.0) return VerificationResult( verified=True, confidence=confidence, changes_detected=True, change_area_pct=global_change_pct, local_change_pct=local_change_pct, suggestion="continue", detail=f"Changement global détecté ({global_change_pct:.2f}%)", ) # Cas 3 : Pas de changement global, mais changement local (clic/frappe) if has_local_change and action_type in ("click", "type"): confidence = min(0.7, 0.3 + local_change_pct / 100.0) return VerificationResult( verified=True, confidence=confidence, changes_detected=True, change_area_pct=global_change_pct, local_change_pct=local_change_pct, suggestion="continue", detail=( f"Changement local détecté ({local_change_pct:.2f}%) " f"autour de ({action_type})" ), ) # Cas 4 : Pas de changement (key_combo, wait) # Pour les raccourcis clavier et attentes, l'absence de changement # n'est pas forcément un problème (ex: Ctrl+C ne change pas l'écran) if action_type in ("key_combo", "wait"): return VerificationResult( verified=True, confidence=0.4, changes_detected=False, change_area_pct=global_change_pct, local_change_pct=local_change_pct, suggestion="continue", detail=( f"Aucun changement visible pour {action_type} " "(normal pour ce type d'action)" ), ) # Cas 5 : Aucun changement détecté pour un clic/frappe → suspect return VerificationResult( verified=False, confidence=0.6, changes_detected=False, change_area_pct=global_change_pct, local_change_pct=local_change_pct, suggestion="retry", detail=( f"Aucun changement détecté après {action_type} " f"(global={global_change_pct:.3f}%, local={local_change_pct:.3f}%)" ), )