test(coords+capture): coords write-only gap (10 tests) + capture I/O + image_chat_cli

test_coords_consumption_gap.py documents 3 structural gaps where NavigateCoords are written but never consumed. test_capture_io.py and test_image_chat_cli.py cover capture and chat CLI paths.
2026-07-02 13:01:49 +02:00
parent ebed4d7546
commit cac965cef9
3 changed files with 642 additions and 0 deletions
--- a/tests/unit/test_capture_io.py
+++ b/tests/unit/test_capture_io.py
@@ -0,0 +1,155 @@
+"""Tests unitaires de la politique de sauvegarde des captures (agent_v1).
+
+Objectif : réduire le poids disque des captures (90 Go / 13 sessions = trop)
+sans casser la précision du grounding. La politique distingue le *type* de
+shot :
+
+- ``crop``      → PNG lossless (cible de grounding qwen3-vl, précision pixel) ;
+- ``full`` / ``window`` / ``context`` → JPEG ``optimize=True`` (vue humaine /
+  contexte, compression ~5-10x acceptable) ;
+- ``heartbeat`` → JPEG **downscalé** (liveness, pas de grounding → on peut
+  réduire la résolution).
+
+La fonction ``save_capture`` retourne le chemin RÉELLEMENT écrit (extension
+ajustée selon le format), pour que l'appelant streame le bon fichier.
+
+Branche feat/push-log-dgx — réduction du poids de capture (unité testée,
+non encore câblée dans capturer.py).
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+from PIL import Image
+
+_ROOT = str(Path(__file__).resolve().parents[2])
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+
+def _noisy_image(width: int, height: int) -> Image.Image:
+    """Image RGB avec du bruit réel.
+
+    Un aplat uni se compresse à quasi-zéro en PNG comme en JPEG : la
+    comparaison de poids serait truquée. On injecte du bruit pour que la
+    différence PNG/JPEG soit représentative d'un vrai screenshot.
+    """
+    return Image.frombytes("RGB", (width, height), os.urandom(width * height * 3))
+
+
+def test_crop_reste_png_et_dimensions_identiques(tmp_path):
+    """Un crop est sauvé en PNG lossless, dimensions inchangées."""
+    from agent_v0.agent_v1.vision.capture_io import save_capture
+
+    img = _noisy_image(80, 80)
+    base = str(tmp_path / "shot_0001_crop")
+
+    out_path = save_capture(img, base, kind="crop")
+
+    assert out_path.endswith(".png"), f"crop doit rester PNG, obtenu {out_path}"
+    assert os.path.exists(out_path)
+    reread = Image.open(out_path)
+    assert reread.size == (80, 80)
+    # PNG lossless : les pixels doivent être identiques au bruit d'origine.
+    assert list(reread.convert("RGB").getdata()) == list(img.getdata())
+
+
+def test_full_est_jpeg(tmp_path):
+    """Un full est sauvé en JPEG (.jpg)."""
+    from agent_v0.agent_v1.vision.capture_io import save_capture
+
+    img = _noisy_image(640, 480)
+    base = str(tmp_path / "shot_0001_full")
+
+    out_path = save_capture(img, base, kind="full")
+
+    assert out_path.endswith(".jpg"), f"full doit être JPEG, obtenu {out_path}"
+    assert os.path.exists(out_path)
+
+
+def test_full_jpeg_significativement_plus_leger_que_png(tmp_path):
+    """Le JPEG full doit peser nettement moins que le PNG équivalent.
+
+    On génère une image bruitée plein écran (2560×1600) et on compare le
+    poids du JPEG produit par la politique au poids d'un PNG lossless du
+    même contenu. Le gain doit être substantiel (au moins 2x plus léger).
+    """
+    from agent_v0.agent_v1.vision.capture_io import save_capture
+
+    img = _noisy_image(2560, 1600)
+
+    jpeg_path = save_capture(img, str(tmp_path / "full_jpeg"), kind="full")
+    png_ref = tmp_path / "full_ref.png"
+    img.save(png_ref, "PNG")
+
+    jpeg_size = os.path.getsize(jpeg_path)
+    png_size = os.path.getsize(png_ref)
+
+    assert jpeg_size < png_size / 2, (
+        f"JPEG ({jpeg_size}o) doit peser < moitié du PNG ({png_size}o)"
+    )
+
+
+def test_context_et_window_sont_jpeg(tmp_path):
+    """context et window suivent la même politique JPEG que full."""
+    from agent_v0.agent_v1.vision.capture_io import save_capture
+
+    img = _noisy_image(320, 240)
+    for kind in ("context", "window"):
+        out_path = save_capture(img, str(tmp_path / f"x_{kind}"), kind=kind)
+        assert out_path.endswith(".jpg"), f"{kind} doit être JPEG, obtenu {out_path}"
+        assert os.path.exists(out_path)
+
+
+def test_heartbeat_est_downscale(tmp_path):
+    """Un heartbeat est downscalé (largeur réduite) et reste JPEG."""
+    from agent_v0.agent_v1.vision.capture_io import save_capture
+
+    img = _noisy_image(2560, 1600)
+    out_path = save_capture(img, str(tmp_path / "heartbeat_1234"), kind="heartbeat")
+
+    assert out_path.endswith(".jpg"), f"heartbeat doit être JPEG, obtenu {out_path}"
+    reread = Image.open(out_path)
+    assert reread.width < 2560, "heartbeat doit être downscalé en largeur"
+    # Ratio préservé (16:10 → la hauteur doit suivre la largeur réduite).
+    ratio_src = 2560 / 1600
+    ratio_out = reread.width / reread.height
+    assert abs(ratio_src - ratio_out) < 0.02, "le ratio doit être préservé"
+
+
+def test_heartbeat_plus_leger_que_full_jpeg(tmp_path):
+    """Le downscale du heartbeat le rend plus léger que le full JPEG plein res."""
+    from agent_v0.agent_v1.vision.capture_io import save_capture
+
+    img = _noisy_image(2560, 1600)
+    hb = save_capture(img, str(tmp_path / "heartbeat_5678"), kind="heartbeat")
+    full = save_capture(img, str(tmp_path / "shot_9999_full"), kind="full")
+
+    assert os.path.getsize(hb) < os.path.getsize(full), (
+        "le heartbeat downscalé doit peser moins que le full JPEG plein res"
+    )
+
+
+def test_kind_inconnu_leve_erreur(tmp_path):
+    """Un kind non reconnu doit échouer explicitement (fail-closed)."""
+    from agent_v0.agent_v1.vision.capture_io import save_capture
+
+    img = _noisy_image(40, 40)
+    try:
+        save_capture(img, str(tmp_path / "x"), kind="inexistant")
+    except ValueError:
+        return
+    raise AssertionError("un kind inconnu doit lever ValueError")
+
+
+def test_rgba_converti_pour_jpeg(tmp_path):
+    """Une image RGBA doit être convertie avant l'encodage JPEG (pas d'alpha)."""
+    from agent_v0.agent_v1.vision.capture_io import save_capture
+
+    img = Image.new("RGBA", (64, 64), (10, 20, 30, 128))
+    out_path = save_capture(img, str(tmp_path / "shot_rgba_full"), kind="full")
+    assert out_path.endswith(".jpg")
+    assert os.path.exists(out_path)
--- a/tests/unit/test_coords_consumption_gap.py
+++ b/tests/unit/test_coords_consumption_gap.py
@@ -0,0 +1,202 @@
+"""Tests documenting the coords consumption gap: write-only navigate coords.
+
+Test 1 (POSITIVE): _resolve_runtime_vars mechanism works — template strings
+like {{navigate_login_coords.x_pct}} resolve correctly when variables dict
+contains the stored coords.
+
+Test 2 (NEGATIVE): _edge_to_normalized_actions bakes coords as literal floats,
+never producing template strings — so runtime variable resolution is never
+triggered for navigate coords, proving the write-only gap.
+
+These tests are evidence, not regression guards. Test 2 documents a known
+structural gap; when the gap is fixed, Test 2 should be updated to assert
+templates ARE produced.
+"""
+
+import os
+import re
+from types import SimpleNamespace
+
+os.environ.setdefault("RPA_AUTH_DISABLED", "true")
+
+from agent_v0.server_v1.replay_engine import (
+    _edge_to_normalized_actions,
+    _resolve_runtime_vars,
+    _resolve_runtime_vars_in_str,
+)
+
+
+# ── Fake fixtures (minimal, per test_visual_anchor_semantics.py pattern) ──
+
+
+class _FakeAction:
+    def __init__(self, type_, target=None, parameters=None):
+        self.type = type_
+        self.target = target
+        self.parameters = parameters or {}
+
+
+class _FakeEdge:
+    def __init__(self, action):
+        self.edge_id = "edge_coords_gap"
+        self.from_node = "node_src"
+        self.to_node = "node_dst"
+        self.action = action
+
+
+# ── Test 1: resolve mechanism is viable ──────────────────────────────────
+
+
+class TestResolveRuntimeVarsViable:
+    """Prove _resolve_runtime_vars infrastructure works with template strings."""
+
+    VARIABLES = {
+        "navigate_login_coords": {
+            "x_pct": 0.15,
+            "y_pct": 0.07,
+            "method": "ocr_anchor",
+        }
+    }
+
+    def test_resolve_in_str_dot_path(self):
+        """{{navigate_login_coords.x_pct}} → "0.15" (string, not float)."""
+        result = _resolve_runtime_vars_in_str(
+            "{{navigate_login_coords.x_pct}}", self.VARIABLES
+        )
+        assert result == "0.15"
+
+    def test_resolve_in_str_y_pct(self):
+        """{{navigate_login_coords.y_pct}} → "0.07"."""
+        result = _resolve_runtime_vars_in_str(
+            "{{navigate_login_coords.y_pct}}", self.VARIABLES
+        )
+        assert result == "0.07"
+
+    def test_resolve_dict_with_templates(self):
+        """_resolve_runtime_vars substitutes templates inside dict values."""
+        action = {
+            "type": "click",
+            "x_pct": "{{navigate_login_coords.x_pct}}",
+            "y_pct": "{{navigate_login_coords.y_pct}}",
+        }
+        resolved = _resolve_runtime_vars(action, self.VARIABLES)
+        assert resolved["x_pct"] == "0.15"
+        assert resolved["y_pct"] == "0.07"
+        assert resolved["type"] == "click"  # no-template strings unchanged
+
+    def test_resolve_nested_dict(self):
+        """_resolve_runtime_vars handles nested dicts with templates."""
+        action = {
+            "parameters": {
+                "coords": "{{navigate_login_coords.x_pct}}",
+            },
+        }
+        resolved = _resolve_runtime_vars(action, self.VARIABLES)
+        assert resolved["parameters"]["coords"] == "0.15"
+
+    def test_resolve_missing_var_leaves_template_intact(self):
+        """Missing variable: template string stays unchanged."""
+        result = _resolve_runtime_vars_in_str(
+            "{{navigate_password_coords.x_pct}}", self.VARIABLES
+        )
+        assert "{{navigate_password_coords.x_pct}}" in result
+
+    def test_resolve_float_passthrough(self):
+        """_resolve_runtime_vars returns non-str values unchanged — floats pass through."""
+        action = {"x_pct": 0.15, "y_pct": 0.07}
+        resolved = _resolve_runtime_vars(action, self.VARIABLES)
+        # Floats are NOT substituted — they're not strings containing {{...}}
+        assert resolved["x_pct"] == 0.15  # literal float, unchanged
+        assert resolved["y_pct"] == 0.07
+
+
+# ── Test 2: compiler gap — literals not templates ────────────────────────
+
+
+class TestCompilerGapLiteralFloats:
+    """Document that _edge_to_normalized_actions produces literal floats,
+    never template strings — so navigate coords are write-only.
+
+    This is the STRUCTURAL GAP: the compiler bakes coords as floats,
+    _resolve_runtime_vars only operates on strings, so stored navigate
+    variables are never consumed downstream.
+    """
+
+    def test_mouse_click_produces_literal_floats(self):
+        """mouse_click edge: x_pct/y_pct are literal floats, not templates."""
+        target = SimpleNamespace(
+            by_position=(0.15, 0.07),
+            by_role=None,
+            by_text=None,
+            context_hints={},
+        )
+        edge = _FakeEdge(
+            _FakeAction("mouse_click", target=target, parameters={"button": "left"})
+        )
+        actions = _edge_to_normalized_actions(edge, params={})
+        assert len(actions) == 1
+        action = actions[0]
+
+        # GAP: coords are literal floats, not template strings
+        assert isinstance(action["x_pct"], float)
+        assert isinstance(action["y_pct"], float)
+        assert action["x_pct"] == 0.15
+        assert action["y_pct"] == 0.07
+
+        # Proof: no template string is ever produced by the compiler
+        assert not isinstance(action["x_pct"], str)
+        assert not isinstance(action["y_pct"], str)
+
+    def test_literal_floats_not_resolved(self):
+        """Literal floats pass through _resolve_runtime_vars unchanged —
+        proving navigate coords stored in variables are NEVER consumed."""
+        target = SimpleNamespace(
+            by_position=(0.15, 0.07),
+            by_role=None,
+            by_text=None,
+            context_hints={},
+        )
+        edge = _FakeEdge(
+            _FakeAction("mouse_click", target=target, parameters={"button": "left"})
+        )
+        actions = _edge_to_normalized_actions(edge, params={})
+        action = actions[0]
+
+        # Simulate variables from a prior navigate_login step
+        different_coords = {
+            "navigate_login_coords": {"x_pct": 0.20, "y_pct": 0.10}
+        }
+        resolved = _resolve_runtime_vars(action, different_coords)
+
+        # Coords REMAIN the original literal floats — no substitution
+        assert resolved["x_pct"] == 0.15  # NOT 0.20 (no substitution)
+        assert resolved["y_pct"] == 0.07  # NOT 0.10 (no substitution)
+
+    def test_text_input_produces_literal_floats(self):
+        """text_input edge: same literal float pattern for click target."""
+        target = SimpleNamespace(
+            by_position=(0.30, 0.50),
+            by_role=None,
+            by_text=None,
+            context_hints={},
+        )
+        edge = _FakeEdge(
+            _FakeAction("text_input", target=target, parameters={"text": "admin"})
+        )
+        actions = _edge_to_normalized_actions(edge, params={})
+        assert len(actions) == 1
+        action = actions[0]
+
+        assert isinstance(action["x_pct"], float)
+        assert isinstance(action["y_pct"], float)
+        assert action["x_pct"] == 0.30
+        assert action["y_pct"] == 0.50
+
+    def test_navigate_action_type_unknown(self):
+        """navigate action type is NOT handled by _edge_to_normalized_actions —
+        falls into the else branch logging "Type d'action inconnu"."""
+        edge = _FakeEdge(_FakeAction("navigate", parameters={"target": "login"}))
+        actions = _edge_to_normalized_actions(edge, params={})
+
+        # navigate produces empty actions — not compiled at all
+        assert actions == []