feat(qw2): hook LoopDetector dans api_stream + extension replay_state

replay_state enrichi de _screenshot_history (5 dernières images PIL) et
_action_history (5 dernières signatures action).

report_action_result :
- met à jour les deux anneaux après chaque action
- évalue le LoopDetector (singleton lazy avec _clip_embedder serveur)
- si detected → bascule paused_need_help avec pause_reason="loop_detected"
  et bus event lea:loop_detected (signal + evidence)

Tous les chemins d'erreur (embedder absent, OOM, exception) loggent et
laissent le replay continuer — aucun blocage par la couche détection.

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

agent_v0/deploy/windows_client/agent_v1/core/executor.py

@@ -512,6 +512,21 @@ class ActionExecutorV1:
             x_pct = action.get("x_pct", 0.0)
             y_pct = action.get("y_pct", 0.0)
 
+            # QW1 — Si le serveur a résolu un monitor cible (idx >= 0),
+            # appliquer son offset aux coords absolues. Pour idx == -1
+            # (composite_fallback), aucun offset (backward compat).
+            # Le calcul des coords reste percent * (width/height) du monitor[1]
+            # côté client (x_pct est exprimé sur l'écran physique principal).
+            mon_res = action.get("monitor_resolution") or {}
+            mon_idx = mon_res.get("idx", -1)
+            mon_offset_x = mon_res.get("offset_x", 0) if mon_idx >= 0 else 0
+            mon_offset_y = mon_res.get("offset_y", 0) if mon_idx >= 0 else 0
+            if mon_idx >= 0 and (mon_offset_x or mon_offset_y):
+                logger.info(
+                    f"[REPLAY] QW1 monitor cible idx={mon_idx} source={mon_res.get('source')} "
+                    f"offset=({mon_offset_x},{mon_offset_y}) — appliqué aux coords"
+                )
+
             # ── Diagnostic résolution ──
             logger.info(
                 f"[REPLAY] Action {action_id} ({action_type}) — "
@@ -578,8 +593,8 @@ class ActionExecutorV1:
                         print(f"    [OBSERVER] Popup détectée : '{popup_label}' — fermeture")
                         logger.info(f"Observer : popup '{popup_label}' détectée avant résolution")
                         if popup_coords:
-                            real_x = int(popup_coords["x_pct"] * width)
-                            real_y = int(popup_coords["y_pct"] * height)
+                            real_x = int(popup_coords["x_pct"] * width) + mon_offset_x
+                            real_y = int(popup_coords["y_pct"] * height) + mon_offset_y
                             self._click((real_x, real_y), "left")
                             time.sleep(1.0)
                             print(f"    [OBSERVER] Popup fermée — reprise du flow normal")
@@ -718,8 +733,8 @@ class ActionExecutorV1:
                         self.notifier.replay_target_not_found(target_desc)
                         return result
 
-                real_x = int(x_pct * width)
-                real_y = int(y_pct * height)
+                real_x = int(x_pct * width) + mon_offset_x
+                real_y = int(y_pct * height) + mon_offset_y
                 button = action.get("button", "left")
                 mode = "VISUAL" if result.get("visual_resolved") else "COORD"
                 print(
@@ -781,8 +796,8 @@ class ActionExecutorV1:
                     print(f"    [TYPE] raw_keys disponibles ({len(raw_keys)} events) — replay exact")
                 # Cliquer sur le champ avant de taper (si coordonnees disponibles)
                 if x_pct > 0 and y_pct > 0:
-                    real_x = int(x_pct * width)
-                    real_y = int(y_pct * height)
+                    real_x = int(x_pct * width) + mon_offset_x
+                    real_y = int(y_pct * height) + mon_offset_y
                     print(f"    [TYPE] Clic prealable sur ({real_x}, {real_y})")
                     self._click((real_x, real_y), "left")
                     time.sleep(0.3)
@@ -808,8 +823,8 @@ class ActionExecutorV1:
                 logger.info(f"Replay key_combo : {keys} (raw_keys={'oui' if raw_keys else 'non'})")
 
             elif action_type == "scroll":
-                real_x = int(x_pct * width) if x_pct > 0 else int(0.5 * width)
-                real_y = int(y_pct * height) if y_pct > 0 else int(0.5 * height)
+                real_x = (int(x_pct * width) if x_pct > 0 else int(0.5 * width)) + mon_offset_x
+                real_y = (int(y_pct * height) if y_pct > 0 else int(0.5 * height)) + mon_offset_y
                 delta = action.get("delta", -3)
                 print(f"    [SCROLL] delta={delta} a ({real_x}, {real_y})")
                 self.mouse.position = (real_x, real_y)
@@ -1386,6 +1401,16 @@ Example: x_pct=0.50, y_pct=0.30"""
             data = resp.json()
             action = data.get("action")
             if action is None:
+                # pause_for_human : afficher le message de décision à l'utilisateur
+                if data.get("replay_paused") and data.get("pause_message"):
+                    msg = data["pause_message"]
+                    print(f"[PAUSE] {msg}")
+                    logger.info(f"Replay en pause — message : {msg}")
+                    self.notifier.notify(
+                        title="Léa — Validation requise",
+                        message=msg[:250],
+                        timeout=30,
+                    )
                 return False
 
         except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:

agent_v0/server_v1/api_stream.py

@@ -34,6 +34,7 @@ from .agent_registry import AgentRegistry, AgentAlreadyEnrolledError
 from .stream_processor import StreamProcessor, build_replay_from_raw_events, enrich_click_from_screenshot
 from .worker_stream import StreamWorker
 from .monitor_router import resolve_target_monitor  # QW1 — résolution écran cible
+from .loop_detector import LoopDetector  # QW2 — détection de boucle pendant replay
 from .execution_plan_runner import (
     execution_plan_to_actions,
     inject_plan_into_queue,
@@ -361,6 +362,18 @@ REPLAY_LOCK_FILE = _DATA_DIR / "_replay_active.lock"
 processor = StreamProcessor(data_dir=str(LIVE_SESSIONS_DIR))
 worker = StreamWorker(live_dir=str(LIVE_SESSIONS_DIR), processor=processor)
 
+# QW2 — LoopDetector singleton lazy (utilise le CLIP embedder du processor)
+_loop_detector: Optional["LoopDetector"] = None
+
+
+def _get_loop_detector() -> "LoopDetector":
+    """Singleton lazy — crée le LoopDetector avec le CLIP embedder du processor."""
+    global _loop_detector
+    if _loop_detector is None:
+        embedder = getattr(processor, "_clip_embedder", None)
+        _loop_detector = LoopDetector(clip_embedder=embedder)
+    return _loop_detector
+
 # Registre des postes Lea enroles (table enrolled_agents dans rpa_data.db)
 # Emplacement configurable via RPA_AGENTS_DB_PATH pour les tests.
 _AGENTS_DB_PATH = os.environ.get(
@@ -3166,6 +3179,28 @@ async def get_next_action(session_id: str, machine_id: str = "default"):
             "h": target.h,
             "source": target.source,
         }
+        # QW1 — Émission bus lea:monitor_routed (no-op si bus indisponible)
+        # Le serveur streaming n'a pas de SocketIO local : on logge en INFO
+        # bien lisible. Un consommateur (agent_chat / dashboard) peut tailer
+        # `journalctl -u rpa-streaming | grep '\[BUS\] lea:monitor_routed'`.
+        try:
+            _replay_id_bus = (
+                owning_replay.get("replay_id") if owning_replay else None
+            )
+            logger.info(
+                "[BUS] lea:monitor_routed replay=%s action=%s idx=%d source=%s "
+                "offset=(%d,%d) wh=(%d,%d)",
+                _replay_id_bus,
+                action.get("action_id"),
+                target.idx,
+                target.source,
+                target.offset_x,
+                target.offset_y,
+                target.w,
+                target.h,
+            )
+        except Exception as _e_bus:
+            logger.debug("emit lea:monitor_routed échec (non bloquant): %s", _e_bus)
     except Exception as e:
         logger.debug("QW1 monitor_resolution skip (%s)", e)
 
@@ -3907,6 +3942,82 @@ async def report_action_result(report: ReplayResultReport):
                 f"— worker VLM autorisé à reprendre"
             )
 
+        # ===================================================================
+        # QW2 — LoopDetector : alimentation des anneaux + évaluation
+        # ===================================================================
+        # On n'évalue que si le replay est encore "running" — inutile de
+        # pauser quelque chose de déjà completed/error/paused.
+        if replay_state["status"] == "running":
+            # Snapshot image (PIL) dans l'anneau
+            try:
+                from PIL import Image
+                ss_raw = screenshot_after or replay_state.get("last_screenshot")
+                img = None
+                if isinstance(ss_raw, str) and ss_raw:
+                    if os.path.isfile(ss_raw):
+                        img = Image.open(ss_raw).copy()  # détache du file handle
+                    else:
+                        # Possible base64 — décoder
+                        try:
+                            import base64
+                            import io as _io
+                            img_bytes = base64.b64decode(ss_raw, validate=False)
+                            img = Image.open(_io.BytesIO(img_bytes)).copy()
+                        except Exception:
+                            img = None
+                if img is not None:
+                    replay_state.setdefault("_screenshot_history", []).append(img)
+                    replay_state["_screenshot_history"] = replay_state["_screenshot_history"][-5:]
+            except Exception as e:
+                logger.debug("LoopDetector: snapshot historique échoué: %s", e)
+
+            # Snapshot signature de l'action courante
+            try:
+                _act_pos = report.actual_position or {}
+                action_sig = {
+                    "type": (original_action or {}).get("type")
+                            or replay_state.get("_last_action_type", ""),
+                    "x_pct": _act_pos.get("x_pct") if isinstance(_act_pos, dict)
+                             else (original_action or {}).get("x_pct"),
+                    "y_pct": _act_pos.get("y_pct") if isinstance(_act_pos, dict)
+                             else (original_action or {}).get("y_pct"),
+                }
+                replay_state.setdefault("_action_history", []).append(action_sig)
+                replay_state["_action_history"] = replay_state["_action_history"][-5:]
+            except Exception as e:
+                logger.debug("LoopDetector: snapshot action_sig échoué: %s", e)
+
+            # Évaluation (silencieux si rien)
+            try:
+                verdict = _get_loop_detector().evaluate(
+                    replay_state,
+                    screenshots=replay_state.get("_screenshot_history", []),
+                    actions=replay_state.get("_action_history", []),
+                )
+                if verdict.detected:
+                    replay_state["status"] = "paused_need_help"
+                    replay_state["pause_reason"] = "loop_detected"
+                    replay_state["pause_message"] = (
+                        f"Léa semble bloquée — {verdict.signal} "
+                        f"(détail: {verdict.evidence})"
+                    )
+                    logger.warning(
+                        "LoopDetector: replay %s mis en pause — signal=%s evidence=%s",
+                        replay_state["replay_id"], verdict.signal, verdict.evidence,
+                    )
+                    # Bus event d'observabilité (logger pattern QW1)
+                    try:
+                        logger.info(
+                            "[BUS] lea:loop_detected replay=%s signal=%s evidence=%s",
+                            replay_state["replay_id"],
+                            verdict.signal,
+                            verdict.evidence,
+                        )
+                    except Exception as _e_bus:
+                        logger.debug("emit lea:loop_detected échec: %s", _e_bus)
+            except Exception as e:
+                logger.warning("LoopDetector: évaluation échouée (non bloquant): %s", e)
+
     return {
         "status": "recorded",
         "action_id": action_id,

agent_v0/server_v1/loop_detector.py

@@ -0,0 +1,154 @@
+# agent_v0/server_v1/loop_detector.py
+"""LoopDetector composite — détection de stagnation de Léa pendant un replay (QW2).
+
+Trois signaux indépendants :
+- screen_static : N captures consécutives avec CLIP similarity > seuil
+- action_repeat : N actions consécutives identiques (type + coords)
+- retry_threshold : nombre de retries cumulés >= seuil
+
+Un seul signal positif → verdict.detected=True. Le serveur bascule alors le
+replay en paused_need_help avec pause_reason explicite.
+
+Désactivable via env var RPA_LOOP_DETECTOR_ENABLED=0.
+"""
+
+import logging
+import os
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class LoopVerdict:
+    detected: bool = False
+    reason: str = ""
+    signal: str = ""  # "screen_static" | "action_repeat" | "retry_threshold" | ""
+    evidence: Dict[str, Any] = field(default_factory=dict)
+
+
+def _env_int(name: str, default: int) -> int:
+    try:
+        return int(os.environ.get(name, default))
+    except (TypeError, ValueError):
+        return default
+
+
+def _env_float(name: str, default: float) -> float:
+    try:
+        return float(os.environ.get(name, default))
+    except (TypeError, ValueError):
+        return default
+
+
+def _env_bool_enabled(name: str) -> bool:
+    val = os.environ.get(name, "1").strip().lower()
+    return val not in ("0", "false", "no", "off", "")
+
+
+def _cosine_similarity(a, b) -> float:
+    """Similarité cosine entre deux vecteurs (listes ou np.array). Robuste vecteur nul."""
+    import numpy as np
+    av = np.asarray(a, dtype=np.float32).flatten()
+    bv = np.asarray(b, dtype=np.float32).flatten()
+    na, nb = float(np.linalg.norm(av)), float(np.linalg.norm(bv))
+    if na < 1e-8 or nb < 1e-8:
+        return 0.0
+    return float(np.dot(av, bv) / (na * nb))
+
+
+class LoopDetector:
+    def __init__(self, clip_embedder=None):
+        self.clip_embedder = clip_embedder
+
+    def evaluate(
+        self,
+        state: Dict[str, Any],
+        screenshots: List[Any],
+        actions: List[Dict[str, Any]],
+    ) -> LoopVerdict:
+        """Évalue les 3 signaux. Retourne le premier déclenché.
+
+        Args:
+            state: replay_state (utilisé pour retried_actions)
+            screenshots: anneau d'embeddings CLIP (les N derniers)
+            actions: anneau des N dernières actions exécutées
+        """
+        if not _env_bool_enabled("RPA_LOOP_DETECTOR_ENABLED"):
+            return LoopVerdict(detected=False)
+
+        # Signal A : screen_static
+        verdict = self._check_screen_static(screenshots)
+        if verdict.detected:
+            return verdict
+
+        # Signal B : action_repeat
+        verdict = self._check_action_repeat(actions)
+        if verdict.detected:
+            return verdict
+
+        # Signal C : retry_threshold
+        verdict = self._check_retry_threshold(state)
+        if verdict.detected:
+            return verdict
+
+        return LoopVerdict(detected=False)
+
+    def _check_screen_static(self, screenshots: List[Any]) -> LoopVerdict:
+        n_required = _env_int("RPA_LOOP_SCREEN_STATIC_N", 4)
+        threshold = _env_float("RPA_LOOP_SCREEN_STATIC_THRESHOLD", 0.99)
+
+        if self.clip_embedder is None or len(screenshots) < n_required:
+            return LoopVerdict()
+
+        try:
+            recent = screenshots[-n_required:]
+            # Embed chaque capture via le CLIP embedder (peut lever)
+            embeddings = [self.clip_embedder.embed_image(img) for img in recent]
+            sims = [_cosine_similarity(embeddings[i], embeddings[i + 1])
+                    for i in range(len(embeddings) - 1)]
+            min_sim = min(sims)
+            if min_sim > threshold:
+                return LoopVerdict(
+                    detected=True,
+                    reason="loop_detected",
+                    signal="screen_static",
+                    evidence={"min_similarity": round(min_sim, 4),
+                              "n_captures": n_required,
+                              "threshold": threshold},
+                )
+        except Exception as e:
+            logger.warning("LoopDetector signal_A erreur (%s) — signal inerte ce tick", e)
+        return LoopVerdict()
+
+    def _check_action_repeat(self, actions: List[Dict[str, Any]]) -> LoopVerdict:
+        n_required = _env_int("RPA_LOOP_ACTION_REPEAT_N", 3)
+        if len(actions) < n_required:
+            return LoopVerdict()
+        recent = actions[-n_required:]
+
+        def _signature(a: Dict[str, Any]) -> tuple:
+            return (a.get("type"), a.get("x_pct"), a.get("y_pct"))
+
+        sigs = [_signature(a) for a in recent]
+        if all(s == sigs[0] for s in sigs):
+            return LoopVerdict(
+                detected=True,
+                reason="loop_detected",
+                signal="action_repeat",
+                evidence={"signature": sigs[0], "count": n_required},
+            )
+        return LoopVerdict()
+
+    def _check_retry_threshold(self, state: Dict[str, Any]) -> LoopVerdict:
+        threshold = _env_int("RPA_LOOP_RETRY_THRESHOLD", 3)
+        retried = int(state.get("retried_actions", 0))
+        if retried >= threshold:
+            return LoopVerdict(
+                detected=True,
+                reason="loop_detected",
+                signal="retry_threshold",
+                evidence={"retried_actions": retried, "threshold": threshold},
+            )
+        return LoopVerdict()

agent_v0/server_v1/replay_engine.py

@@ -1381,6 +1381,9 @@ def _create_replay_state(
         # t2a_decision, etc.). Résolues via templating {{var}} ou {{var.field}}
         # dans les paramètres des actions suivantes.
         "variables": {},
+        # QW2 — Anneaux d'historique pour LoopDetector (5 derniers max)
+        "_screenshot_history": [],   # images PIL des N derniers heartbeats (LoopDetector embed à chaque tick)
+        "_action_history": [],       # N dernières actions exécutées (signature)
     }
 
 

tests/integration/test_loop_detector_replay.py

@@ -0,0 +1,61 @@
+# tests/integration/test_loop_detector_replay.py
+"""Tests intégration : un replay simulé qui boucle bascule en paused_need_help."""
+import pytest
+from unittest.mock import MagicMock
+
+from agent_v0.server_v1.loop_detector import LoopDetector
+
+
+def test_replay_state_transitions_to_paused_on_screen_static():
+    """Cas : 4 screenshots identiques → replay passe à paused_need_help."""
+    embedder = MagicMock()
+    embedder.embed_image.return_value = [1.0, 0.0, 0.0]  # constant
+    detector = LoopDetector(clip_embedder=embedder)
+
+    state = {
+        "replay_id": "r_test",
+        "status": "running",
+        "retried_actions": 0,
+        "_screenshot_history": ["img1", "img2", "img3", "img4"],  # 4 images factices
+        "_action_history": [
+            {"type": "click", "x_pct": 0.1, "y_pct": 0.1},
+            {"type": "type", "x_pct": 0.2, "y_pct": 0.2},
+        ],
+    }
+    verdict = detector.evaluate(state, state["_screenshot_history"], state["_action_history"])
+
+    # Simuler ce que ferait api_stream après verdict
+    if verdict.detected:
+        state["status"] = "paused_need_help"
+        state["pause_reason"] = verdict.reason
+        state["pause_message"] = f"signal={verdict.signal}"
+
+    assert state["status"] == "paused_need_help"
+    assert state["pause_reason"] == "loop_detected"
+    assert "screen_static" in state["pause_message"]
+
+
+def test_replay_state_transitions_on_action_repeat():
+    """Cas : 3 actions identiques → paused_need_help signal action_repeat."""
+    detector = LoopDetector(clip_embedder=None)
+    actions = [{"type": "click", "x_pct": 0.5, "y_pct": 0.5}] * 3
+    state = {"replay_id": "r2", "status": "running", "retried_actions": 0,
+             "_screenshot_history": [], "_action_history": actions}
+
+    verdict = detector.evaluate(state, [], actions)
+    assert verdict.detected and verdict.signal == "action_repeat"
+
+
+def test_kill_switch_keeps_replay_running(monkeypatch):
+    """Avec RPA_LOOP_DETECTOR_ENABLED=0 le replay continue même en boucle."""
+    monkeypatch.setenv("RPA_LOOP_DETECTOR_ENABLED", "0")
+    embedder = MagicMock()
+    embedder.embed_image.return_value = [1.0, 0.0, 0.0]
+    detector = LoopDetector(clip_embedder=embedder)
+
+    state = {"retried_actions": 10,
+             "_screenshot_history": ["img1"] * 10,
+             "_action_history": [{"type": "click", "x_pct": 0.5, "y_pct": 0.5}] * 10}
+
+    verdict = detector.evaluate(state, state["_screenshot_history"], state["_action_history"])
+    assert verdict.detected is False

tests/unit/test_loop_detector.py

@@ -0,0 +1,96 @@
+# tests/unit/test_loop_detector.py
+"""Tests unitaires pour LoopDetector composite (QW2)."""
+import os
+import pytest
+from unittest.mock import MagicMock
+
+from agent_v0.server_v1.loop_detector import LoopDetector, LoopVerdict
+
+
+@pytest.fixture
+def detector():
+    """LoopDetector avec embedder mocké (signal A toujours dispo)."""
+    embedder = MagicMock()
+    # Par défaut : 4 embeddings tous identiques → similarity 1.0
+    embedder.embed_image.return_value = [1.0, 0.0, 0.0]
+    return LoopDetector(clip_embedder=embedder)
+
+
+def _state(retried=0, n_screenshots=0, n_actions=0):
+    return {
+        "retried_actions": retried,
+        "_screenshot_history": [[1.0, 0.0, 0.0]] * n_screenshots,
+        "_action_history": [{"type": "click", "x_pct": 0.5, "y_pct": 0.5}] * n_actions,
+    }
+
+
+def test_screen_static_triggers_when_n_identical_embeddings(detector):
+    """Signal A : 4 captures identiques (similarity > 0.99) → detected."""
+    state = _state(n_screenshots=4)
+    verdict = detector.evaluate(state, screenshots=state["_screenshot_history"], actions=[])
+    assert verdict.detected is True
+    assert verdict.signal == "screen_static"
+
+
+def test_screen_static_skipped_when_history_too_short(detector):
+    """Signal A : moins de N captures → pas de détection."""
+    state = _state(n_screenshots=2)
+    verdict = detector.evaluate(state, screenshots=state["_screenshot_history"], actions=[])
+    # Si seul A pourrait déclencher mais skip, et B/C pas remplis : detected=False
+    assert verdict.detected is False
+
+
+def test_action_repeat_triggers_when_n_identical_actions(detector):
+    """Signal B : 3 actions consécutives identiques → detected."""
+    state = _state(n_actions=3)
+    verdict = detector.evaluate(state, screenshots=[], actions=state["_action_history"])
+    assert verdict.detected is True
+    assert verdict.signal == "action_repeat"
+
+
+def test_action_repeat_skipped_when_actions_differ(detector):
+    """Signal B : actions différentes → pas de détection."""
+    actions = [
+        {"type": "click", "x_pct": 0.1, "y_pct": 0.1},
+        {"type": "click", "x_pct": 0.2, "y_pct": 0.2},
+        {"type": "click", "x_pct": 0.3, "y_pct": 0.3},
+    ]
+    verdict = detector.evaluate(_state(), screenshots=[], actions=actions)
+    assert verdict.detected is False
+
+
+def test_retry_threshold_triggers_at_3(detector):
+    """Signal C : retried_actions >= 3 → detected."""
+    state = _state(retried=3)
+    verdict = detector.evaluate(state, screenshots=[], actions=[])
+    assert verdict.detected is True
+    assert verdict.signal == "retry_threshold"
+
+
+def test_kill_switch_disables_all_signals(monkeypatch, detector):
+    """Si RPA_LOOP_DETECTOR_ENABLED=0 → toujours detected=False."""
+    monkeypatch.setenv("RPA_LOOP_DETECTOR_ENABLED", "0")
+    state = _state(retried=10, n_screenshots=10, n_actions=10)
+    verdict = detector.evaluate(state, screenshots=state["_screenshot_history"],
+                                actions=state["_action_history"])
+    assert verdict.detected is False
+
+
+def test_embedder_unavailable_skips_signal_A_continues_others():
+    """Si CLIP embedder None → signal A skip, B et C continuent."""
+    detector = LoopDetector(clip_embedder=None)
+    # Trigger signal C
+    state = _state(retried=3)
+    verdict = detector.evaluate(state, screenshots=[], actions=[])
+    assert verdict.detected is True
+    assert verdict.signal == "retry_threshold"
+
+
+def test_embedder_exception_does_not_crash(detector):
+    """Si embed_image lève une exception → log + verdict detected=False."""
+    detector.clip_embedder.embed_image.side_effect = RuntimeError("CUDA OOM")
+    state = _state(n_screenshots=4)
+    # Ne doit PAS lever : signal A devient inerte
+    verdict = detector.evaluate(state, screenshots=state["_screenshot_history"], actions=[])
+    # Signal A inerte, B/C pas remplis → detected False
+    assert verdict.detected is False