fix: contrôle strict des étapes + routage par machine_id

Corrections critiques après test E2E qui montrait des clics au mauvais endroit :

1. Routage par machine_id (api_stream.py)
   Quand 2 machines partagent le même session_id (agent_demo_user),
   les actions d'un replay pour la VM ne doivent PLUS être distribuées
   au PC physique. Vérification que le replay_state appartient bien à
   la machine qui poll avant de consommer la queue.

2. IRBuilder extrait expected_window_before/after (ir_builder.py)
   Pour chaque action click/type/key_combo, stocke le titre de la fenêtre
   au moment du clic (before) et le titre du prochain événement (after).
   Ces champs alimentent le contrôle strict au runtime.

3. ExecutionCompiler crée SuccessCondition title_match (execution_compiler.py)
   Quand expected_window_after est défini, crée une condition de succès
   STRICTE avec method="title_match" et expected_title. Plus de simple
   "l'écran a changé" — on vérifie la fenêtre résultante.

4. Runner propage expected_window_before et success_strict
   Le flag success_strict indique à l'agent que le contrôle post-action
   DOIT être strict (STOP sur mismatch au lieu de warning).

5. UIA strict sur parent_path (executor.py)
   _resolve_via_uia_local REJETTE un match si l'élément trouvé n'est pas
   dans la bonne fenêtre parente (évite ex: "Rechercher" taskbar confondu
   avec "Rechercher" explorateur).

6. Pré/post vérif stricte et bloquante (executor.py)
   - expected_window_before lu en priorité depuis l'action (plan V4)
   - Post-vérif : si success_strict=True et timeout, result.success=False
     → le replay s'arrête au lieu de continuer avec des warnings.

Validé sur la VM :
- Le replay s'arrête proprement quand l'étape 2 aboutit dans "Propriétés de
  Internet" au lieu de "blocnote.txt - Bloc-notes"
- Plus de clics en aveugle / saisie au mauvais endroit

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dom
2026-04-10 14:05:23 +02:00
parent 56e3cc052a
commit cecdf417b7
7 changed files with 213 additions and 18 deletions

View File

@@ -277,8 +277,12 @@ class ActionExecutorV1:
On appelle le helper Rust qui interroge UIAutomationCore.dll et
retourne les coordonnées pixel-perfect de l'élément.
Retourne (x_pct, y_pct) si trouvé, None sinon.
Le fallback vers le grounding serveur est géré par l'appelant.
STRICT : si l'élément trouvé n'appartient pas à la bonne fenêtre
parente (comparaison du parent_path), on REFUSE — sinon on clique
au mauvais endroit (ex: 'Rechercher' de la taskbar au lieu de
l'explorateur).
Retourne (x_pct, y_pct) si trouvé ET validé, None sinon.
"""
try:
from .uia_helper import get_shared_helper
@@ -289,6 +293,7 @@ class ActionExecutorV1:
name = uia_target.get("name", "")
control_type = uia_target.get("control_type", "") or None
automation_id = uia_target.get("automation_id", "") or None
expected_parent_path = uia_target.get("parent_path", []) or []
if not name:
return None
@@ -300,8 +305,41 @@ class ActionExecutorV1:
timeout_ms=1500,
)
if element is None or not element.is_clickable():
logger.debug(f"UIA: '{name}' non trouvé ou non cliquable")
return None
# ── VÉRIFICATION STRICTE du parent_path ──
# Si l'élément a été enregistré dans une fenêtre spécifique,
# il doit être trouvé dans la MÊME fenêtre au replay.
# Sinon on clique sur un homonyme dans une autre app.
if expected_parent_path:
expected_root = None
for p in expected_parent_path:
if p.get("control_type", "").lower() in ("fenêtre", "window"):
expected_root = p.get("name", "").strip()
break
if expected_root:
found_root = None
for p in element.parent_path:
if p.get("control_type", "").lower() in ("fenêtre", "window"):
found_root = p.get("name", "").strip()
break
if found_root and expected_root != found_root:
# Match souple : une sous-partie commune (ex: "Bloc-notes")
if (expected_root.lower() not in found_root.lower()
and found_root.lower() not in expected_root.lower()):
logger.warning(
f"UIA REJET : '{name}' trouvé dans '{found_root}' "
f"mais attendu dans '{expected_root}'"
)
print(
f" [UIA] REJET — '{name}' trouvé dans mauvaise fenêtre "
f"({found_root}{expected_root})"
)
return None
cx, cy = element.center()
if screen_width <= 0 or screen_height <= 0:
return None
@@ -479,7 +517,13 @@ class ActionExecutorV1:
# ── Pré-vérification : titre fenêtre ──
# Vérifier que l'écran est dans l'état attendu AVANT de cliquer.
if visual_mode and target_spec:
expected_title = target_spec.get("window_title", "")
# Le champ explicite `expected_window_before` a priorité
# (il vient du plan V4 et indique la fenêtre STRICTEMENT
# attendue avant l'action). Sinon fallback sur target_spec.
expected_title = (
action.get("expected_window_before", "")
or target_spec.get("window_title", "")
)
if expected_title and expected_title != "unknown_window":
from ..window_info_crossplatform import get_active_window_info
current_info = get_active_window_info()
@@ -728,7 +772,28 @@ class ActionExecutorV1:
else:
print(f" [POST-VÉRIF] TIMEOUT {max_wait}s — '{post_title}''{expected_after}'")
logger.warning(f"POST-VÉRIF TIMEOUT : '{post_title}''{expected_after}'")
result["warning"] = f"post_verif_timeout:{post_title}"
# Contrôle strict : si success_strict, on STOP.
# Sinon on continue avec un warning (legacy).
is_strict = bool(action.get("success_strict"))
if is_strict:
result["success"] = False
result["error"] = (
f"Post-vérif échouée : fenêtre '{post_title}' "
f"au lieu de '{expected_after}'"
)
print(
f" [POST-VÉRIF] STOP STRICT — l'étape ne s'est "
f"pas déroulée comme prévu, arrêt du replay"
)
try:
self.notifier.replay_wrong_window(
post_title, expected_after,
)
except Exception:
pass
return result
else:
result["warning"] = f"post_verif_timeout:{post_title}"
else:
print(f" [CLICK] Terminé.")