Dom/rpa_vision_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

snapshot: WIP 5j replay reliability (B1 watchdog + dialog handlers + grounding drift)

Browse Source 
Snapshot avant correction du blocage relance Léa (3 incidents 24h: SSH refusé,
polls morts ×2). Point de rollback stable.

Contenu:
- agent_v1/core/executor.py: 5 patchs dialog handling (saveas drift, close_tab
  hotkey fallback, confirm_save Unicode apostrophe, foreground dialog
  recontextualization, runtime_dialog in-loop) + helpers normalize_window_hint,
  requires_post_verify_window_transition
- agent_v1/core/grounding.py: garde drift template fix (fallback_x/y plumbed)
- server_v1/replay_watchdog.py (NEW): orphan watchdog B1, scan 10s timeout 30s
- server_v1/api_stream.py: dispatched_action plumbing, watchdog lifespan,
  metrics endpoint
- server_v1/replay_engine.py: _schedule_retry préserve original_action +
  dispatched_action
- stream_processor.py: gardes _infer_tab_switch_target (no false switch_tab
  on save_as dialog open) + _attach_expected_window_before
- tests/integration: test_replay_watchdog.py (8 cas), test_stream_processor.py
- tests/unit: test_executor_verify_window_guard.py (start_button, close_tab,
  runtime_dialog, post_verify, transition fallbacks)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dom
2026-05-24 16:48:37 +02:00
parent 5ea4960e65
commit 7df51d2c79
47 changed files with 9811 additions and 451 deletions
Download Patch File Download Diff File Expand all files Collapse all files

363
tests/unit/test_policy_grounding_recovery_learning.py
View File

@@ -111,6 +111,310 @@ class TestGroundingEngine:
assert d["x_pct"] == 0.5
assert d["method"] == "som"
def test_start_button_uses_full_screen_instead_of_active_window(self):
"""Le bouton Démarrer doit être résolu sur l'écran entier."""
engine, executor = self._make_engine()
executor._server_resolve_target.return_value = {
"resolved": True,
"x_pct": 0.02,
"y_pct": 0.98,
"method": "som_text",
"score": 0.9,
"matched_element": {"label": "Démarrer"},
}
engine._capture_window_or_screen = MagicMock(return_value="fake_b64_data")
with patch(
"agent_v0.agent_v1.window_info_crossplatform.get_active_window_rect",
return_value={"rect": [100, 100, 1100, 900]},
):
result = engine.locate(
"http://server",
{"by_text": "Démarrer", "by_role": "start_button"},
0.02, 0.98, 1920, 1080,
)
assert result.found is True
engine._capture_window_or_screen.assert_called_once_with(None)
def test_regular_targets_stay_scoped_to_active_window(self):
"""Les cibles applicatives ordinaires restent bornées à la fenêtre active."""
engine, executor = self._make_engine()
executor._server_resolve_target.return_value = {
"resolved": True,
"x_pct": 0.5,
"y_pct": 0.25,
"method": "som_text",
"score": 0.9,
"matched_element": {"label": "Enregistrer"},
}
engine._capture_window_or_screen = MagicMock(return_value="fake_b64_data")
with patch(
"agent_v0.agent_v1.window_info_crossplatform.get_active_window_rect",
return_value={"rect": [100, 200, 1100, 1000]},
):
result = engine.locate(
"http://server",
{"by_text": "Enregistrer", "by_role": "button"},
0.5, 0.3, 1920, 1080,
)
assert result.found is True
engine._capture_window_or_screen.assert_called_once_with(
{"left": 100, "top": 200, "width": 1000, "height": 800}
)
def test_unknown_window_rect_falls_back_to_full_screen_on_visual_mismatch(self):
"""Un titre inconnu n'est accepté que si le crop est validé visuellement."""
engine, executor = self._make_engine()
executor._server_resolve_target.return_value = {
"resolved": True,
"x_pct": 0.5,
"y_pct": 0.25,
"method": "som_text",
"score": 0.9,
"matched_element": {"label": "Enregistrer"},
}
executor._find_text_on_screen.return_value = None
engine._capture_window_or_screen = MagicMock(
side_effect=["fake_window_b64", "fake_screen_b64"]
)
with patch(
"agent_v0.agent_v1.window_info_crossplatform.get_active_window_rect",
return_value={
"title": "unknown_window",
"rect": [100, 200, 1100, 1000],
},
):
result = engine.locate(
"http://server",
{"by_text": "Enregistrer", "by_role": "button"},
0.5, 0.3, 1920, 1080,
)
assert result.found is True
assert [c.args[0] for c in engine._capture_window_or_screen.call_args_list] == [
{"left": 100, "top": 200, "width": 1000, "height": 800},
None,
]
def test_taskbar_like_rect_falls_back_to_full_screen(self):
"""Une taskbar/systray ne doit jamais être utilisée comme fenêtre active."""
engine, executor = self._make_engine()
executor._server_resolve_target.return_value = {
"resolved": True,
"x_pct": 0.5,
"y_pct": 0.25,
"method": "som_text",
"score": 0.9,
"matched_element": {"label": "Enregistrer"},
}
engine._capture_window_or_screen = MagicMock(return_value="fake_b64_data")
with patch(
"agent_v0.agent_v1.window_info_crossplatform.get_active_window_rect",
return_value={
"title": "Fenêtre de dépassement de capacité de la barre d'état système",
"rect": [0, 1492, 2560, 1600],
},
):
result = engine.locate(
"http://server",
{"by_text": "Enregistrer", "by_role": "button"},
0.5, 0.3, 2560, 1600,
)
assert result.found is True
engine._capture_window_or_screen.assert_called_once_with(None)
def test_visually_mismatched_window_crop_falls_back_to_full_screen(self):
"""Un crop fenêtre plausible mais visuellement faux est rejeté."""
engine, executor = self._make_engine()
executor._server_resolve_target.return_value = {
"resolved": True,
"x_pct": 0.5,
"y_pct": 0.25,
"method": "som_text",
"score": 0.9,
"matched_element": {"label": "Enregistrer"},
}
executor._find_text_on_screen.return_value = None
engine._capture_window_or_screen = MagicMock(
side_effect=["fake_window_b64", "fake_screen_b64"]
)
with patch(
"agent_v0.agent_v1.window_info_crossplatform.get_active_window_rect",
return_value={
"title": "Enregistrer sous",
"rect": [100, 200, 1100, 1000],
},
):
result = engine.locate(
"http://server",
{
"by_text": "Enregistrer",
"by_role": "button",
"window_title": "Enregistrer sous",
},
0.5, 0.3, 1920, 1080,
)
assert result.found is True
assert [c.args[0] for c in engine._capture_window_or_screen.call_args_list] == [
{"left": 100, "top": 200, "width": 1000, "height": 800},
None,
]
executor._server_resolve_target.assert_called_once_with(
"http://server",
"fake_screen_b64",
{
"by_text": "Enregistrer",
"by_role": "button",
"window_title": "Enregistrer sous",
},
0.5,
0.3,
1920,
1080,
)
def test_visually_validated_window_crop_stays_scoped(self):
"""Un crop fenêtre plausible et validé visuellement reste autorisé."""
engine, executor = self._make_engine()
executor._server_resolve_target.return_value = {
"resolved": True,
"x_pct": 0.5,
"y_pct": 0.25,
"method": "som_text",
"score": 0.9,
"matched_element": {"label": "Enregistrer"},
}
executor._find_text_on_screen.return_value = (321, 222)
engine._capture_window_or_screen = MagicMock(return_value="fake_window_b64")
with patch(
"agent_v0.agent_v1.window_info_crossplatform.get_active_window_rect",
return_value={
"title": "Enregistrer sous",
"rect": [100, 200, 1100, 1000],
},
):
result = engine.locate(
"http://server",
{
"by_text": "Enregistrer",
"by_role": "button",
"window_title": "Enregistrer sous",
},
0.5, 0.3, 1920, 1080,
)
assert result.found is True
engine._capture_window_or_screen.assert_called_once_with(
{"left": 100, "top": 200, "width": 1000, "height": 800}
)
def test_lea_active_window_does_not_scope_external_target(self):
"""Une fenêtre Léa au premier plan ne doit jamais contraindre une cible externe."""
engine, executor = self._make_engine()
executor._server_resolve_target.return_value = {
"resolved": True,
"x_pct": 0.5,
"y_pct": 0.25,
"method": "som_text",
"score": 0.9,
"matched_element": {"label": "Bloc-notes"},
}
engine._capture_window_or_screen = MagicMock(return_value="fake_b64_data")
with patch(
"agent_v0.agent_v1.window_info_crossplatform.get_active_window_rect",
return_value={
"title": "Léa — Assistante",
"app_name": "pythonw.exe",
"rect": [1948, 750, 2570, 1606],
},
):
result = engine.locate(
"http://server",
{"by_text": "Bloc-notes", "by_role": "search_result"},
0.2, 0.5, 2560, 1600,
)
assert result.found is True
engine._capture_window_or_screen.assert_called_once_with(None)
executor._server_resolve_target.assert_called_once_with(
"http://server",
"fake_b64_data",
{"by_text": "Bloc-notes", "by_role": "search_result"},
0.2,
0.5,
2560,
1600,
)
def test_lea_active_window_stays_scoped_for_explicit_lea_target(self):
"""Si la cible mentionne explicitement Léa, le scope sur sa fenêtre reste autorisé."""
engine, executor = self._make_engine()
executor._server_resolve_target.return_value = {
"resolved": True,
"x_pct": 0.5,
"y_pct": 0.25,
"method": "som_text",
"score": 0.9,
"matched_element": {"label": "Continuer"},
}
engine._capture_window_or_screen = MagicMock(return_value="fake_b64_data")
with patch(
"agent_v0.agent_v1.window_info_crossplatform.get_active_window_rect",
return_value={
"title": "Léa — Assistante",
"app_name": "pythonw.exe",
"rect": [1948, 750, 2570, 1606],
},
):
result = engine.locate(
"http://server",
{
"by_text": "Continuer",
"by_role": "button",
"window_title": "Léa — Assistante",
},
0.5, 0.3, 3000, 2000,
)
assert result.found is True
engine._capture_window_or_screen.assert_called_once_with(
{"left": 1948, "top": 750, "width": 622, "height": 856}
)
def test_allow_position_fallback_returns_recorded_coords(self):
"""Quand autorisé, le grounding peut retomber sur la position enregistrée."""
engine, executor = self._make_engine()
executor._server_resolve_target.return_value = None
executor._template_match_anchor.return_value = None
executor._hybrid_vlm_resolve.return_value = None
result = engine.locate(
"http://server",
{
"by_role": "start_button",
"vlm_description": "icône Windows",
"screen_scope": "full_screen",
"allow_position_fallback": True,
},
0.387891, 0.974375, 1920, 1080,
)
assert result.found is True
assert result.method == "position_fallback"
assert result.x_pct == pytest.approx(0.387891)
assert result.y_pct == pytest.approx(0.974375)
# =========================================================================
# P2 : Policy — décisions quand grounding échoue
@@ -407,6 +711,65 @@ class TestReplayLearner:
assert "action_id" in data
assert "success" in data
def test_record_human_correction_persists_to_memory_helper(self, learner, monkeypatch):
"""Une correction humaine doit alimenter la mémoire persistante via replay_memory."""
captured = {}
def fake_memory_record_success(**kwargs):
captured.update(kwargs)
return True
monkeypatch.setattr(
"agent_v0.server_v1.replay_memory.memory_record_success",
fake_memory_record_success,
)
learner.record_human_correction(
session_id="s_corr",
action={
"action_id": "a_corr",
"target_spec": {"by_text": "Valider", "window_title": "Bloc-notes"},
},
correction={"x_pct": 0.42, "y_pct": 0.84},
)
loaded = learner.load_session("s_corr")
assert len(loaded) == 1
assert loaded[0].resolution_method == "human_supervised"
assert loaded[0].window_title == "Bloc-notes"
assert captured["window_title"] == "Bloc-notes"
assert captured["target_spec"]["by_text"] == "Valider"
assert captured["x_pct"] == 0.42
assert captured["y_pct"] == 0.84
assert captured["method"] == "human_supervised"
assert captured["confidence"] == 1.0
def test_record_human_correction_fallback_window_title_from_action(self, learner, monkeypatch):
"""Si target_spec.window_title est absent, on retombe sur action.window_title."""
captured = {}
def fake_memory_record_success(**kwargs):
captured.update(kwargs)
return True
monkeypatch.setattr(
"agent_v0.server_v1.replay_memory.memory_record_success",
fake_memory_record_success,
)
learner.record_human_correction(
session_id="s_corr2",
action={
"action_id": "a_corr2",
"window_title": "Fenêtre fallback",
"target_spec": {"by_text": "Enregistrer"},
},
correction={"x_pct": 0.1, "y_pct": 0.2},
)
assert captured["window_title"] == "Fenêtre fallback"
# =========================================================================
# Boucle d'apprentissage : consolidation cross-workflow