fix(replay): cure régression b584bbabc — fallback recorded_coords aveugle

Trois changements complémentaires dans la cascade de résolution serveur,
finis ce soir 7 mai pour la démo GHT 8 mai. Restaure le comportement strict
d'avril 2026 (workflow qui passait 20 fois d'affilée sans incident).

1. resolve_engine.py — _validate_resolution_quality (lignes 2255-2289) :
   Le commit b584bbabc du 1er mai 2026 ("fix(stream): démo UHCD") avait
   transformé le rejet strict (resolved=False, method="rejected_drift_*")
   en fallback aveugle (resolved=True, method="fallback_recorded_coords",
   coords du record). Symptôme observé : Léa cliquait sur "Dossier en
   cours" du menu au lieu de "Synthèse Urgences" du tab — le VLM Quick
   Find Ollama hallucinait à (0.526, 0.918), drift dépassé, fallback
   ratait. Restauré : resolved=False explicite, le client passe en
   pause supervisée comme prévu (philosophie échec = apprentissage).

2. resolve_engine.py — exemption high-confidence élargie :
   L'exemption drift>0.20 IGNORÉ ne couvrait que template_matching ≥ 0.95
   (commit 35b27ae49 du 2 mai). Étendue à hybrid_text_direct ≥ 0.80 :
   un OCR direct qui trouve le texte cible exact à score 0.80+ est aussi
   sûr qu'un template à 0.95 — la position est sémantiquement vraie,
   le drift reflète juste un changement de layout (résolution écran,
   refonte UI, scroll), pas une erreur de résolution.

3. resolve_engine.py + api_stream.py — pré-check OCR sémantique :
   Nouvelle fonction _validate_text_at_position (singleton EasyOCR fr+en,
   crop 200px autour de la coord résolue, fuzzy match 60% des tokens
   ≥3 caractères de l'expected_text). Câblée dans api_stream.py juste
   après _validate_resolution_quality. Si le by_text attendu n'est PAS
   présent dans la zone autour de la coord résolue → resolved=False
   method="rejected_text_mismatch" → pause supervisée.

Pattern Verification-Aware Planning (state of the art 2026 — voir
recommandations agent archéologue + agent SOTA review) : le serveur
ne renvoie une coord que s'il est sémantiquement sûr du résultat.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

agent_v0/server_v1/api_stream.py

@@ -4503,6 +4503,44 @@ async def resolve_target(request: ResolveTargetRequest):
             request.fallback_y_pct,
         )
 
+        # Pré-check sémantique post-cascade : OCR sur une zone autour de la
+        # coordonnée résolue pour vérifier que le by_text attendu y est bien
+        # présent. Attrape les cas où la cascade rend des coords plausibles
+        # mais pointant sur un autre élément (ex : clic sur "Dossier en cours"
+        # du menu au lieu de "Synthèse Urgences" du tab plus bas).
+        if result and result.get("resolved"):
+            _by_text = (request.target_spec.get("by_text") or "").strip()
+            if _by_text:
+                from agent_v0.server_v1.resolve_engine import _validate_text_at_position
+                _is_valid, _observed, _ocr_ms = _validate_text_at_position(
+                    tmp_path,
+                    float(result.get("x_pct", 0) or 0),
+                    float(result.get("y_pct", 0) or 0),
+                    _by_text,
+                    effective_w,
+                    effective_h,
+                )
+                if not _is_valid:
+                    logger.warning(
+                        "[REPLAY] Pre-check OCR REJET : '%s' attendu @ (%.4f, %.4f) "
+                        "via %s mais OCR voit '%s' (%.0fms)",
+                        _by_text[:40],
+                        float(result.get("x_pct", 0) or 0),
+                        float(result.get("y_pct", 0) or 0),
+                        result.get("method", "?"),
+                        _observed[:80],
+                        _ocr_ms,
+                    )
+                    result = {
+                        "resolved": False,
+                        "method": "rejected_text_mismatch",
+                        "reason": f"expected='{_by_text[:40]}' observed='{_observed[:60]}'",
+                        "original_method": result.get("method"),
+                        "original_score": result.get("score"),
+                        "x_pct": None,
+                        "y_pct": None,
+                    }
+
         # [REPLAY] log structuré de sortie résolution (après validation)
         logger.info(
             f"[REPLAY] RESOLVE_EXIT session={request.session_id} "

agent_v0/server_v1/resolve_engine.py

@@ -2160,6 +2160,135 @@ _RESOLUTION_MIN_SCORES: Dict[str, float] = {
 _RESOLUTION_MAX_DRIFT: float = 0.20
 
 
+# ===========================================================================
+# Pré-check sémantique : OCR de validation de position
+# ===========================================================================
+# Avant de dispatcher un clic, on vérifie que le texte attendu (by_text) est
+# bien présent dans une fenêtre OCR autour de la coordonnée résolue. Cela
+# attrape les cas où la cascade renvoie une coordonnée plausible mais qui
+# pointe en réalité sur un autre élément (ex: clic sur "Dossier en cours" du
+# menu au lieu de "Synthèse Urgences" du tab plus bas).
+# ===========================================================================
+
+_VALIDATION_OCR_READER = None
+_VALIDATION_OCR_LOCK = threading.Lock()
+_VALIDATION_OCR_FAILED = False
+
+
+def _get_validation_ocr_reader():
+    """Singleton EasyOCR partagé pour la validation post-cascade.
+
+    Chargement paresseux à la première requête. En cas d'échec, on cache
+    le statut FAILED pour ne pas retenter à chaque appel et bloquer le flux.
+    """
+    global _VALIDATION_OCR_READER, _VALIDATION_OCR_FAILED
+    if _VALIDATION_OCR_FAILED:
+        return None
+    with _VALIDATION_OCR_LOCK:
+        if _VALIDATION_OCR_READER is None and not _VALIDATION_OCR_FAILED:
+            try:
+                import easyocr  # type: ignore
+                _VALIDATION_OCR_READER = easyocr.Reader(
+                    ['fr', 'en'], gpu=True, verbose=False
+                )
+                logger.info("[REPLAY] EasyOCR validator chargé (fr+en, GPU)")
+            except Exception as e:
+                logger.warning("[REPLAY] EasyOCR validator indisponible (%s) — pré-check désactivé", e)
+                _VALIDATION_OCR_FAILED = True
+                return None
+    return _VALIDATION_OCR_READER
+
+
+def _normalize_for_match(s: str) -> str:
+    """Normalisation pour comparaison textuelle robuste : lowercase, sans
+    accents, ponctuation → espace, espaces multiples écrasés.
+    """
+    import unicodedata
+    decomposed = unicodedata.normalize('NFD', s.lower())
+    no_accents = ''.join(c for c in decomposed if unicodedata.category(c) != 'Mn')
+    cleaned = ''.join(c if c.isalnum() or c.isspace() else ' ' for c in no_accents)
+    return ' '.join(cleaned.split())
+
+
+def _text_match_fuzzy(expected: str, observed: str, min_token_ratio: float = 0.60) -> bool:
+    """Match tolérant aux imperfections OCR.
+
+    1. Substring exacte → match.
+    2. Sinon : split en tokens ≥3 caractères, retourne True si au moins
+       `min_token_ratio` des tokens attendus apparaissent dans observed.
+       Ex : "Coller ou saisir le dossier patient" → tokens
+       ['coller', 'saisir', 'dossier', 'patient'] ; si OCR voit "u saisir
+       le dossier patient" → 3/4 = 75% présents → match accepté.
+
+    Cible le compromis entre strict (faux négatifs sur erreurs OCR) et
+    permissif (faux positifs sur textes voisins).
+    """
+    nexp = _normalize_for_match(expected)
+    nobs = _normalize_for_match(observed)
+    if not nexp:
+        return True
+    if nexp in nobs:
+        return True
+    tokens = [t for t in nexp.split() if len(t) >= 3]
+    if not tokens:
+        return False
+    matched = sum(1 for t in tokens if t in nobs)
+    return matched / len(tokens) >= min_token_ratio
+
+
+def _validate_text_at_position(
+    screenshot_path: str,
+    x_pct: float,
+    y_pct: float,
+    expected_text: str,
+    screen_width: int,
+    screen_height: int,
+    radius_px: int = 200,
+) -> tuple:
+    """Pré-check sémantique : OCR sur une zone autour de (x_pct, y_pct) et
+    vérifie que `expected_text` y est présent (substring ou fuzzy 60%).
+
+    Retourne (is_valid: bool, observed_text: str, elapsed_ms: float).
+
+    Politique en cas d'échec OCR (lib absente, exception) : retourne
+    (True, "", 0.0) pour ne pas bloquer le flux. Mieux vaut un faux positif
+    rare qu'une régression bloquante introduite par la validation elle-même.
+    """
+    reader = _get_validation_ocr_reader()
+    if reader is None:
+        return True, "", 0.0
+    if not expected_text or not expected_text.strip():
+        return True, "", 0.0
+    try:
+        from PIL import Image
+        import numpy as np
+
+        t0 = time.time()
+        img = Image.open(screenshot_path).convert("RGB")
+        img_w, img_h = img.size
+        cx = int(x_pct * screen_width)
+        cy = int(y_pct * screen_height)
+        # Saturer dans les bornes de l'image (le screenshot peut être plus
+        # large que la fenêtre logique — utiliser min(img_*, screen_*) en sécurité).
+        max_x = min(img_w, screen_width)
+        max_y = min(img_h, screen_height)
+        x1 = max(0, cx - radius_px)
+        y1 = max(0, cy - radius_px)
+        x2 = min(max_x, cx + radius_px)
+        y2 = min(max_y, cy + radius_px)
+        if x2 - x1 < 10 or y2 - y1 < 10:
+            return True, "", 0.0
+        crop = img.crop((x1, y1, x2, y2))
+        results = reader.readtext(np.array(crop))
+        observed = " ".join(r[1] for r in results if r and len(r) >= 2)
+        elapsed_ms = (time.time() - t0) * 1000
+        is_valid = _text_match_fuzzy(expected_text, observed, min_token_ratio=0.60)
+        return is_valid, observed, elapsed_ms
+    except Exception as e:
+        logger.warning("[REPLAY] _validate_text_at_position erreur (%s) — pas de blocage", e)
+        return True, "", 0.0
+
+
 def _validate_resolution_quality(
     result: Optional[Dict[str, Any]],
     fallback_x_pct: float,
@@ -2236,33 +2365,50 @@ def _validate_resolution_quality(
         dx = abs(resolved_x - fallback_x_pct)
         dy = abs(resolved_y - fallback_y_pct)
         if dx > _RESOLUTION_MAX_DRIFT or dy > _RESOLUTION_MAX_DRIFT:
-            # Exception : si le template matching trouve l'image avec une
+            # Exception : pour les méthodes "haute confiance" qui ont
-            # similarité quasi parfaite, on fait confiance à la position
+            # identifié sémantiquement la cible (texte exact via OCR ou
-            # visuelle peu importe le drift. Une image retrouvée à >= 0.95
+            # image quasi parfaite via template), on fait confiance à la
-            # de score est SUR l'écran à l'endroit indiqué — le drift par
+            # position visuelle peu importe le drift. Le drift par rapport
-            # rapport à l'enregistrement ne reflète qu'un changement de
+            # à l'enregistrement ne reflète qu'un changement de layout
-            # layout (scroll, redimensionnement, F11, devtools), pas une
+            # (scroll, redimensionnement, F11, refonte UI, résolution
-            # erreur de résolution.
+            # différente), pas une erreur de résolution.
-            _HIGH_CONFIDENCE = 0.95
+            #
-            if score >= _HIGH_CONFIDENCE and method.startswith("template_matching"):
+            # - template_matching ≥ 0.95 : image retrouvée pixel-perfect
+            # - hybrid_text_direct ≥ 0.80 : texte exact reconnu par OCR
+            #   (0.80 est déjà le seuil d'acceptation côté _RESOLUTION_MIN_SCORES,
+            #   au-dessus on a un signal sémantique fiable).
+            _high_confidence_method = (
+                (method.startswith("template_matching") and score >= 0.95)
+                or (method == "hybrid_text_direct" and score >= 0.80)
+            )
+            if _high_confidence_method:
                 logger.info(
-                    "[REPLAY] Drift (%.3f, %.3f) > %.2f IGNORÉ : score=%.3f >= %.2f "
+                    "[REPLAY] Drift (%.3f, %.3f) > %.2f IGNORÉ : score=%.3f "
                     "sur %s — résultat visuel fiable, on l'utilise",
-                    dx, dy, _RESOLUTION_MAX_DRIFT, score, _HIGH_CONFIDENCE, method,
+                    dx, dy, _RESOLUTION_MAX_DRIFT, score, method,
                 )
                 return result
 
             logger.warning(
-                "[REPLAY] Drift trop grand (%.3f, %.3f) > %.2f — fallback coords enregistrées (%.3f, %.3f)",
+                "[REPLAY] Resolution REJETÉE (drift trop grand) : "
-                dx, dy, _RESOLUTION_MAX_DRIFT, fallback_x_pct, fallback_y_pct,
+                "method=%s resolved=(%.3f, %.3f) expected=(%.3f, %.3f) "
+                "drift=(%.3f, %.3f) max=%.2f",
+                method, resolved_x, resolved_y,
+                fallback_x_pct, fallback_y_pct,
+                dx, dy, _RESOLUTION_MAX_DRIFT,
             )
-            # Fallback : coordonnées enregistrées lors de la capture (écran identique = safe)
+            # 100% visuel : on ne clique JAMAIS aux coords enregistrées en aveugle.
+            # resolved=False → la couche supérieure tente la méthode suivante
+            # (VLM Quick Find, SoM, grounding) ; si toutes échouent, l'agent
+            # passe par "visual_resolve_failed" → Policy → pause supervisée.
             return {
-                "resolved": True,
+                "resolved": False,
-                "method": "fallback_recorded_coords",
+                "method": f"rejected_drift_{method}",
-                "reason": f"drift_dx{dx:.3f}_dy{dy:.3f}_using_recorded",
+                "reason": f"drift_dx{dx:.3f}_dy{dy:.3f}_max{_RESOLUTION_MAX_DRIFT:.2f}",
                 "original_method": method,
                 "original_score": score,
+                "drift_dx": round(dx, 3),
+                "drift_dy": round(dy, 3),
                 "x_pct": fallback_x_pct,
                 "y_pct": fallback_y_pct,
             }