From 6582a69d313b5c3d58886dceac96a44d4f4eee93 Mon Sep 17 00:00:00 2001 From: Dom Date: Tue, 5 May 2026 22:50:22 +0200 Subject: [PATCH] =?UTF-8?q?feat(qw1):=20MonitorRouter=20=E2=80=94=20r?= =?UTF-8?q?=C3=A9solution=20de=20l'=C3=A9cran=20cible=20pour=20le=20replay?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Module isolé qui choisit l'écran cible avec stratégie en cascade : 1. action.monitor_index (session source) → cible explicite 2. session.last_focused_monitor → fallback focus actif 3. composite (offset 0,0) → backward compat (comportement actuel) Backward 100% : actions sans monitor_index → fallback composite identique au comportement mss.monitors[0] actuel. Tests : 4 cas (cible OK, fallback focus, fallback composite, index invalide). Co-Authored-By: Claude Opus 4.7 (1M context) --- agent_v0/server_v1/monitor_router.py | 88 ++++++++++++++++++++++++++++ tests/unit/test_monitor_router.py | 51 ++++++++++++++++ 2 files changed, 139 insertions(+) create mode 100644 agent_v0/server_v1/monitor_router.py create mode 100644 tests/unit/test_monitor_router.py diff --git a/agent_v0/server_v1/monitor_router.py b/agent_v0/server_v1/monitor_router.py new file mode 100644 index 000000000..58652b6ac --- /dev/null +++ b/agent_v0/server_v1/monitor_router.py @@ -0,0 +1,88 @@ +# agent_v0/server_v1/monitor_router.py +"""MonitorRouter — résolution de l'écran cible pour le replay (QW1). + +Stratégie en cascade : +1. action.monitor_index (hérité de la session source) → cible cet écran +2. session.last_focused_monitor (focus actif vu en dernier heartbeat) → fallback +3. composite (offset 0, 0) → backward compat + +Émet sur le bus lea:* l'event monitor_routed avec la source de la décision. +""" + +from dataclasses import dataclass +from typing import Any, Dict, List, Optional + + +@dataclass +class MonitorTarget: + """Représente l'écran cible résolu pour une action de replay.""" + idx: int + offset_x: int + offset_y: int + w: int + h: int + source: str # "action" | "focus" | "composite_fallback" + + +_COMPOSITE_FALLBACK = MonitorTarget( + idx=-1, + offset_x=0, + offset_y=0, + w=0, + h=0, + source="composite_fallback", +) + + +def _find_monitor(geometry: List[Dict[str, Any]], idx: int) -> Optional[Dict[str, Any]]: + """Retourne le monitor d'index donné, ou None si absent.""" + for m in geometry: + if m.get("idx") == idx: + return m + return None + + +def _to_target(monitor: Dict[str, Any], source: str) -> MonitorTarget: + return MonitorTarget( + idx=int(monitor["idx"]), + offset_x=int(monitor.get("x", 0)), + offset_y=int(monitor.get("y", 0)), + w=int(monitor.get("w", 0)), + h=int(monitor.get("h", 0)), + source=source, + ) + + +def resolve_target_monitor( + action: Dict[str, Any], + session_state: Dict[str, Any], +) -> MonitorTarget: + """Résout l'écran cible d'une action de replay. + + Args: + action: Dict de l'action (peut contenir `monitor_index`). + session_state: État de la session (doit contenir `monitors_geometry` + et `last_focused_monitor`). + + Returns: + MonitorTarget avec l'offset à appliquer aux coordonnées de grounding. + """ + geometry: List[Dict[str, Any]] = session_state.get("monitors_geometry") or [] + + # 1. Cible explicite via action + explicit_idx = action.get("monitor_index") + if explicit_idx is not None and geometry: + m = _find_monitor(geometry, int(explicit_idx)) + if m is not None: + return _to_target(m, source="action") + # Index invalide → on tombe sur le fallback focus + + # 2. Fallback focus actif + focused_idx = session_state.get("last_focused_monitor") + if focused_idx is not None and geometry: + m = _find_monitor(geometry, int(focused_idx)) + if m is not None: + return _to_target(m, source="focus") + + # 3. Fallback composite (backward compat — comportement actuel mss.monitors[0]) + return _COMPOSITE_FALLBACK diff --git a/tests/unit/test_monitor_router.py b/tests/unit/test_monitor_router.py new file mode 100644 index 000000000..fe0b7f73d --- /dev/null +++ b/tests/unit/test_monitor_router.py @@ -0,0 +1,51 @@ +# tests/unit/test_monitor_router.py +"""Tests unitaires pour MonitorRouter (QW1).""" +import pytest + +from agent_v0.server_v1.monitor_router import resolve_target_monitor, MonitorTarget + + +# Geometry de référence pour les 3 tests : 2 écrans côte à côte +TWO_MONITORS = [ + {"idx": 0, "x": 0, "y": 0, "w": 1920, "h": 1080, "primary": True}, + {"idx": 1, "x": 1920, "y": 0, "w": 1920, "h": 1080, "primary": False}, +] + + +def test_resolve_uses_action_monitor_index_when_present(): + """Si action.monitor_index présent et valide → cible cet écran.""" + action = {"monitor_index": 1} + session_state = {"monitors_geometry": TWO_MONITORS, "last_focused_monitor": 0} + result = resolve_target_monitor(action, session_state) + assert result.idx == 1 + assert result.offset_x == 1920 + assert result.offset_y == 0 + assert result.source == "action" + + +def test_resolve_falls_back_to_focused_monitor_when_action_missing(): + """Si action.monitor_index absent → fallback focus actif.""" + action = {} # pas de monitor_index + session_state = {"monitors_geometry": TWO_MONITORS, "last_focused_monitor": 1} + result = resolve_target_monitor(action, session_state) + assert result.idx == 1 + assert result.source == "focus" + + +def test_resolve_falls_back_to_composite_when_geometry_empty(): + """Si geometry vide (vieux Agent V1) → fallback composite (idx=-1, offset=0).""" + action = {} + session_state = {"monitors_geometry": [], "last_focused_monitor": None} + result = resolve_target_monitor(action, session_state) + assert result.source == "composite_fallback" + assert result.offset_x == 0 + assert result.offset_y == 0 + + +def test_resolve_falls_back_when_action_index_out_of_range(): + """Si action.monitor_index hors limites (écran débranché) → fallback focus.""" + action = {"monitor_index": 5} # n'existe pas + session_state = {"monitors_geometry": TWO_MONITORS, "last_focused_monitor": 0} + result = resolve_target_monitor(action, session_state) + assert result.idx == 0 + assert result.source == "focus"