docs(design+audit): navigate coords consumption gaps + dead code C-MORT audit

DESIGN_NAVIGATE_COORDS_CONSUMPTION_2026-07-02.md: 3 structural gaps
with code evidence (Gap A/B/C), 2 resolution options with comparative
table, test rouge proposal.

AUDIT_CODE_MORT_2026-07-02.md: 8 C-MORT, 5 B-ORPHELIN, 4 duplicats,
3 QG-gated suppression lots (~1900 lines).

.gitignore

@@ -126,7 +126,14 @@ tools/codex_windows_correction_rapport.py
 docs/clients/
 
 .qw-baseline.log
+# Coordination ephemeral — inbox messages, active decisions, loop state
 docs/coordination/.loop_state/
+docs/coordination/.inbox_baseline.txt
+docs/coordination/.loop_log.txt
+docs/coordination/inbox_qwen/
+docs/coordination/inbox_codex/
+docs/coordination/inbox_claude/
+docs/coordination/active/
 
 # Runtime Python embedded pour l'installateur Inno Setup (local, ~11M, non versionné)
 deploy/installer/python-3.12-embed/

docs/AUDIT_CODE_MORT_2026-07-02.md

@@ -0,0 +1,106 @@
+# Audit Code Mort — Classification A/B/C — 2026-07-02
+
+**Auteur**: Qwen (vérifié par grep/glob/commandes réelles)  
+**Date**: 2026-07-02  
+**Méthode**: Parallel agent exploration + grep verification + graphify cross-check
+
+---
+
+## Méthodologie
+
+- **A (WIRED/ACTIF)** : Code importé et appelé dans le runtime de production
+- **B (ORPHAN/PROJECTION)** : Code avec lazy import ou projection future, pas appelé actuellement mais structuré pour activation
+- **C (MORT/CONFIRMÉ)** : Code zero imports, zero callers, zero runtime activation — candidat suppression
+
+**Règle**: C-MORT nécessite GO Dom avant suppression. B-ORPHELIN conserve. A-WIRED documenté.
+
+---
+
+## C-MORT Confirmé (8 items, ~843 lignes)
+
+| # | Fichier/Zone | Lignes | Preuve C-MORT | Risque suppression |
+|---|-------------|--------|---------------|-------------------|
+| C1 | `agent_v0/deploy_windows.py` | ~244 | Comment "OBSOLETE avril 2026" + zero imports | LOW — standalone script |
+| C2 | `core/config.py`: 7 deprecated config classes | ~160 | Zero prod imports, mirrorent SystemConfig | LOW — mais vérifier .env references |
+| C3 | `core/detection/owl_detector.py`: 4 methods | ~90 | Zero callers dans prod | LOW — vérifier examples/ |
+| C4 | `core/detection/ollama_client.py`: 5 old methods | ~150 | Remplacés par classify_element_complete() | LOW — vérifier examples/ |
+| C5 | `ollama_client.py:check_ollama_available()` standalone | ~15 | 8/9 callers in examples/, 1 in VWB (duplicat D2) | LOW — VWB a sa propre copie |
+| C6 | `agent_chat/app.py`: 2 Flask 410 endpoints | ~14 | Endpoints déprecated, retour 410 Gone | LOW — API contract check |
+| C7 | `core/grounding/smart_resize.py` (77 lines) | 77 | Zero prod callers, DETTE-007 triple impl | LOW — 2 autres impls existent |
+| C8 | PP-OCRv5 (paddleocr+paddlepaddle venv) | ~deps | 0 .py imports across entire project | LOW — venv deps uninstall |
+
+**Total C-MORT**: ~843 lignes code + venv deps
+
+---
+
+## B-ORPHELIN (5 items, ~537 lignes)
+
+| # | Fichier/Zone | Lignes | Preuve B | Action |
+|---|-------------|--------|----------|--------|
+| B1 | VWB ui_detection_service OmniParser path | ~70 | HARD-DISABILÉ `_omniparser_available = False # DÉSACTIVÉ` | Conserver, documenter activation condition |
+| B2 | `fusion_engine.py:_fuse_concat_projection()` | ~15 | Stub, prévu pour future fusion modes | Conserver, marque PROJECTION |
+| B3 | `omniparser_adapter.py` | ~429 | BRANCHABLE DORMANT, try/except import | Conserver, documenter activation condition |
+| B4 | `CorrectionStatus.DEPRECATED` enum value | ~3 | Enum value, pas supprimable sans break | Conserver, marque DEPRECATED |
+| B5 | `catalog_routes_v2_vlm.py:check_ollama_available()` | ~20 | Duplicat de ollama_client.py (D2) | DÉCISION Dom : unifier ou garder 2 impls |
+
+---
+
+## Duplicats Identifiés (4)
+
+| # | Item | Impl 1 | Impl 2 | Statut |
+|---|------|--------|--------|--------|
+| D1 | smart_resize | smart_resize.py (C7) | ui_detection_service.py resize | C-MORT vs WIRED |
+| D2 | check_ollama_available | ollama_client.py (C5) | catalog_routes_v2_vlm.py (B5) | C-MORT vs B-ORPHELIN |
+| D3 | ground_element | seeclick_adapter.py (B→provenance?) | ollama_client.py old method | B vs C4 |
+| D4 | 7 deprecated config classes | core/config.py (C2) | SystemConfig (WIRED) | C-MORT vs A-WIRED |
+
+---
+
+## Classification Updates (C→A upgrades confirmés)
+
+| Item | Prior Status | Current Status | Preuve upgrade |
+|------|-------------|---------------|---------------|
+| autonomous_planner.py | C | **A** | Migrated to agent_chat/, wired by app.py |
+| seeclick_adapter.py | C | **B** | Lazy re-export, `_seeclick_available` never consulted mais impl ground_element indépendante |
+| grounding/server.py | C | **A** | HTTP service port 8200, standalone Flask |
+| get_grounding_profile() | C | **A** | Wired via ollama_client.py:303-304 lazy import |
+
+---
+
+## OmniParser — Classification 7 Zones
+
+| # | Zone | Statut | Activation | Fallback |
+|---|------|---------|-----------|----------|
+| 1 | SoM engine (som_engine.py) | **A-WIRED** | YOLO weights direct | docTR OCR |
+| 2 | resolve_engine (_get_omniparser) | **B-DORMANT** | Lazy Optional[bool] | None → skipped |
+| 3 | phase25_analyzer (_OmniParserSafeWrapper) | **B-DORMANT** | Lazy import + healthcheck | docTR-only |
+| 4 | api_stream healthcheck | **A-WIRED** | Always 200 omniparser_available:bool | degraded:true |
+| 5 | omniparser_adapter.py | **B-DORMANT** | Import phase25 & resolve | empty list |
+| 6 | VWB ui_detection_service.py | **B-HARD-DISABILÉ** | `_omniparser_available = False # DÉSACTIVÉ` | ui-detr-1 only |
+| 7 | VWB catalog_routes_v2_vlm.py | **B-DORMANT** | try/except, flips True si installé | VLM fallback |
+
+---
+
+## QG-Gated Lots (proposé, nécessite GO Dom)
+
+### Lot 1 — C-MORT Low Risk (suppression directe après GO Dom)
+- C1 deploy_windows.py
+- C7 smart_resize.py
+- C6 agent_chat 410 endpoints
+- C8 PP-OCRv5 venv deps uninstall
+
+### Lot 2 — C-MORT Medium Risk (vérification examples/ avant suppression)
+- C2 7 deprecated config classes (vérifier .env)
+- C3 owl_detector 4 methods (vérifier examples/)
+- C4 ollama_client 5 old methods (vérifier examples/)
+- C5 check_ollama_available standalone (vérifier VWB duplicat)
+
+### Lot 3 — Duplicats Unification (décision Dom)
+- D1 smart_resize: unifier ou garder 2 impls
+- D2 check_ollama_available: unifier VWB vs core
+- D3 ground_element: unifier seeclick vs ollama
+- D4 config classes: supprimer deprecated vs garder compat
+
+---
+
+**Prochaine étape**: Dom review → GO/NOGO par lot → exécution séquentielle avec tests verification après chaque lot.

docs/DESIGN_NAVIGATE_COORDS_CONSUMPTION_2026-07-02.md

@@ -0,0 +1,218 @@
+# Design Note — NavigateCoords Consumption Gap (Write-Only)
+
+**Auteur**: Qwen  
+**Date**: 2026-07-02  
+**Statut**: DESIGN NOTE — pas de câblage sans GO Dom  
+**Référence**: `tests/unit/test_coords_consumption_gap.py` (10 tests PASSING documenting the gap)
+
+---
+
+## Executive Summary
+
+Le module navigation (`core/navigation`) produit des coords normalisés (`NavigateCoords`) via OCR/VLM, les stocke dans `replay_state["variables"]`, mais **aucun consommateur** dans le runtime n'utilise ces coords. Le résultat est un pattern **write-only** : coords générés mais jamais consommés par les actions suivantes (click/type).
+
+Trois gaps structurels confirmés par code lecture :
+
+---
+
+## Gap A — Compiler Produces Literals, Not Templates
+
+**Localisation**: `replay_engine.py:1832-1846` (`_edge_to_normalized_actions`)
+
+**Problème**: Pour `mouse_click`, le compiler bake `x_pct` et `y_pct` comme **floats littéraux** depuis `by_position` :
+
+```python
+# replay_engine.py:1843-1846
+normalized["type"] = "click"
+normalized["x_pct"] = x_pct   # float littéral (ex: 0.15)
+normalized["y_pct"] = y_pct   # float littéral (ex: 0.07)
+```
+
+Ces floats sont **hardcodés** dans le step definition. Il n'existe pas de mécanisme pour référencer les coords navigate via templates comme `{{navigate_login_coords.x_pct}}`.
+
+**La substitution existante ne couvre pas ce cas** :
+- `_substitute_variables()` → `${var}` → appliqué uniquement à `text_input.text`
+- `_RUNTIME_VAR_PATTERN` → `{{var.field}}` → compilé regex, **jamais appliqué à `x_pct/y_pct`**
+
+**Conséquence**: Un navigate step qui résolve coords login à (0.15, 0.07) ne peut PAS injecter ces coords dans un click step suivant, car le click step a ses propres `x_pct/y_pct` hardcodés.
+
+---
+
+## Gap B — Zero Consumers in Runtime
+
+**Localisation**: `core/navigation/__init__.py:43-113` (`_handle_navigate_action`)
+
+**Problème**: `_handle_navigate_action` stocke coords dans `replay_state["variables"]` :
+
+```python
+# core/navigation/__init__.py:100-105
+if result.login_coords:
+    variables[login_var] = result.login_coords.to_dict()
+# → {"x_pct": 0.15, "y_pct": 0.07, "method": "ocr_anchor"}
+```
+
+**Zéro consommateur** : aucun action handler (click, type, double_click, right_click) lit `variables["navigate_login_coords"]` pour résoudre ses propres coords. Chaque action utilise exclusivement `by_position` depuis son edge definition.
+
+**Preuve par grep** : `navigate_login_coords|navigate_password_coords|navigate_submit_coords` apparaît uniquement dans :
+- `core/navigation/__init__.py` (write)
+- `tests/unit/test_*.py` (test verification)
+- **0 occurrences** dans `replay_engine.py` action dispatch ou `api_stream.py` action handlers
+
+---
+
+## Gap C — Navigate Edge → Empty Actions List
+
+**Localisation**: `replay_engine.py:1806-1955` (`_edge_to_normalized_actions`)
+
+**Problème**: Le type `navigate` est dans `_ALLOWED_ACTION_TYPES` (ligne 44) et possède un handler câblé dans `api_stream.py` (ligne 4459-4463 via `_handle_navigate_action`). Mais `_edge_to_normalized_actions` **n'a pas de branche** pour `navigate` :
+
+```python
+# replay_engine.py:1954-1955 (else branch)
+else:
+    logger.warning(f"Type d'action inconnu : {action_type}")
+    return []
+```
+
+**Conséquence** : Quand le BFS traverse un edge navigate, `_edge_to_normalized_actions(edge, params)` retourne `[]`. L'action navigate est **skippée** dans le path. Le handler existe dans `api_stream.py` mais est **inaccessible** car le normalized action dict n'est jamais produit.
+
+**Paradoxe** : Le navigate handler est câblé et fonctionnel, mais le pipeline edge→action le bloque à l'entrée.
+
+---
+
+## Options de Résolution
+
+### Option 1 — Compiler Injection (modifier `_edge_to_normalized_actions`)
+
+**Approche**: Ajouter une branche `navigate` dans `_edge_to_normalized_actions` qui produit un normalized action dict. Modifier les actions click/type pour permettre des template refs `{{navigate_login_coords.x_pct}}` dans `x_pct/y_pct`, avec résolution runtime.
+
+```python
+# Option 1 — Branch navigate dans _edge_to_normalized_actions
+elif action_type == "navigate":
+    normalized["type"] = "navigate"
+    normalized["parameters"] = {
+        "action": action_params.get("action", "login"),
+        "login_coords_var": action_params.get("login_coords_var", "navigate_login_coords"),
+        "password_coords_var": action_params.get("password_coords_var", "navigate_password_coords"),
+        "submit_coords_var": action_params.get("submit_coords_var", "navigate_submit_coords"),
+    }
+    return [normalized]
+```
+
+**+ Avantages** :
+- Minimal change — 1 branche ajoutée + template resolution dans click/type
+- Compatible avec handler existant (`_handle_navigate_action`)
+- BFS path inclut navigate → handler appelé → coords stockés → consommés
+
+**– Risques** :
+- Template resolution dans `x_pct/y_pct` nécessite modification de click/type dispatch
+- Float vs string : `{{navigate_login_coords.x_pct}}` résout en `"0.15"` (string), pas `0.15` (float) — nécessite conversion
+- Ordonnancement : navigate doit s'exécuter AVANT les actions click/type qui consomment ses coords — scheduling implication
+
+### Option 2 — Declarative YAML Templates (step definitions avec coords_template)
+
+**Approche**: Ajouter un champ `coords_template` dans les step YAML definitions. Au runtime, le template est résolu par substitution des variables navigate.
+
+```yaml
+# Option 2 — YAML step definition avec coords_template
+steps:
+  - action: navigate
+    parameters:
+      action: login
+      login_coords_var: navigate_login_coords
+  - action: mouse_click
+    coords_template: "{{navigate_login_coords}}"
+    # Au runtime : x_pct/y_pct résolus depuis navigate_login_coords dict
+```
+
+**+ Avantages** :
+- Déclaratif — coords templates dans YAML, pas hardcoded
+- Séparation compiler/runtime : compiler produit templates, runtime résout
+- Extensible à autres types de coords (search, dossier)
+
+**– Risques** :
+- Plus de changement : schema YAML + template resolver + compiler modifications
+- Retro-compatibilité : workflows existants sans coords_template doivent continuer à fonctionner (fallback by_position)
+- Validation : templates malformés → runtime errors subtiles
+
+---
+
+## Table Comparative
+
+| Critère | Option 1 (Compiler Injection) | Option 2 (YAML Templates) |
+|---------|-------------------------------|---------------------------|
+| Changement code | Small — 1 branch + template resolve | Medium — schema + resolver + compiler |
+| Retro-compat | Full — by_position fallback intact | Full — fallback by_position si pas de template |
+| Ordonnancement | Navigate avant click (BFS order) | Navigate avant click (step order) |
+| Extensibilité | Navigate-specific | General — coords_template applicable à tout |
+| Risque runtime | Float/string conversion | Template validation errors |
+| Tests impact | 1-3 nouveaux tests | 5-8 nouveaux tests (schema + resolver) |
+| GO Dom needed | YES | YES |
+| Timeline | ~2h implementation | ~4h implementation + schema design |
+
+---
+
+## Test Rouge Proposal
+
+**Objectif**: Démontrer Gap C avec 1 test unitaire qui montre qu'un edge navigate produit une empty action list.
+
+```python
+# tests/unit/test_coords_consumption_gap.py — ajout proposé
+
+def test_gap_c_navigate_edge_produces_empty_actions():
+    """Gap C: _edge_to_normalized_actions returns [] for navigate edge.
+    
+    Prove: navigate is in _ALLOWED_ACTION_TYPES but has no branch
+    in _edge_to_normalized_actions → falls into else → empty list.
+    """
+    from agent_v0.server_v1.replay_engine import _edge_to_normalized_actions
+    
+    # Minimal mock edge with navigate action type
+    edge = MockEdge(
+        edge_id="e1",
+        from_node="start",
+        to_node="login",
+        action=MockAction(
+            type="navigate",
+            target=None,
+            parameters={"action": "login"},
+        ),
+    )
+    result = _edge_to_normalized_actions(edge, {})
+    
+    # GAP: navigate edge produces zero actions
+    assert result == [], f"Expected empty list, got {result}"
+    # This proves the handler in api_stream.py is unreachable
+```
+
+**Note**: Ce test est un **red flag** — il doit FAIL quand le gap est résolu (navigate branch ajoutée → result ≠ []). Il sert de guardrail : si quelqu'un câble navigate sans résoudre les gaps A+B, le test rouge continue à signaler le problème.
+
+---
+
+## Decision Required from Dom
+
+**⚠️ PAS DE CÂBLAGE SANS GO DOM**
+
+Ce design note documente les gaps et propose des options. La décision appartient à Dom :
+
+1. **Option préférée** : 1 (compiler injection) ou 2 (YAML templates) ?
+2. **Timeline** : implémenter maintenant (POC phase) ou post-POC ?
+3. **Scope** : navigate login only, ou general coords template system ?
+4. **Test rouge** : ajouter le test gap C maintenant (documentation) ou attendre GO ?
+
+---
+
+## Appendix — Code References
+
+| Fichier | Lignes | Rôle |
+|---------|--------|------|
+| `replay_engine.py:44` | `_ALLOWED_ACTION_TYPES` includes "navigate" | Allowlist |
+| `replay_engine.py:1806-1955` | `_edge_to_normalized_actions` — no navigate branch | Gap C |
+| `replay_engine.py:1843-1846` | mouse_click bakes literal x_pct/y_pct | Gap A |
+| `core/navigation/__init__.py:43-113` | `_handle_navigate_action` — writes coords to variables | Gap B (write) |
+| `core/navigation/action_resolver.py:47-62` | `NavigateCoords` dataclass definition | Data model |
+| `api_stream.py:4459-4463` | navigate handler dispatch | Wired but unreachable |
+| `tests/unit/test_coords_consumption_gap.py` | 10 tests documenting write-only gap | Evidence |
+
+---
+
+*Qwen — design note, pas wiring. GO Dom required.*

tests/test_image_chat_cli.py

@@ -0,0 +1,285 @@
+#!/usr/bin/env python3
+"""
+Chat interactif en ligne de commande avec gemma4:26b via Ollama.
+
+Usage interactif :
+    python tests/test_image_chat_cli.py
+    # puis taper des questions sur l'image fournie
+
+Usage one-shot :
+    python tests/test_image_chat_cli.py /chemin/vers/image.png "Que vois-tu ?"
+
+Usage avec modèle différent :
+    python tests/test_image_chat_cli.py --model qwen3-vl:8b image.png
+
+Le script utilise l'API Ollama directement (via la lib `ollama` du projet,
+`ollama==0.6.1` dans requirements.txt).
+"""
+
+import argparse
+import base64
+import sys
+from pathlib import Path
+
+try:
+    import ollama
+except ImportError:
+    print("ERREUR : la librairie 'ollama' n'est pas installée.")
+    print("Installez-la avec : pip install ollama")
+    sys.exit(1)
+
+
+DEFAULT_MODEL = "gemma4:26b"
+
+
+def encode_image(image_path: str) -> str:
+    """Encode une image en base64 pour l'API Ollama."""
+    path = Path(image_path)
+    if not path.exists():
+        print(f"ERREUR : le fichier '{image_path}' n'existe pas.")
+        sys.exit(1)
+    if not path.is_file():
+        print(f"ERREUR : '{image_path}' n'est pas un fichier.")
+        sys.exit(1)
+    with open(path, "rb") as f:
+        return base64.b64encode(f.read()).decode("utf-8")
+
+
+def get_client(host: str):
+    """Renvoie un client Ollama configuré pour l'hôte donné."""
+    return ollama.Client(host=host)
+
+
+def check_ollama_running(host: str = "http://localhost:11434") -> bool:
+    """Vérifie que le serveur Ollama est accessible."""
+    try:
+        client = get_client(host)
+        client.list()
+        return True
+    except Exception as e:
+        print(f"ERREUR : impossible de joindre Ollama sur {host}")
+        print(f"Détail : {e}")
+        print()
+        print("Assurez-vous qu'Ollama est lancé :")
+        print("  ollama serve")
+        return False
+
+
+def check_model_available(model: str, host: str = "http://localhost:11434") -> bool:
+    """Vérifie que le modèle est disponible dans Ollama."""
+    try:
+        client = get_client(host)
+        tags = client.list()
+        # ollama.list() retourne un ListResponse avec un attribut 'models'
+        models = getattr(tags, "models", [])
+
+        model_names = []
+        for m in models:
+            if isinstance(m, dict):
+                model_names.append(m.get("name", ""))
+            else:
+                model_names.append(getattr(m, "name", str(m)))
+
+        # Correspondance exacte ou préfixe
+        matched = [name for name in model_names if model in name]
+        if matched:
+            return True
+        else:
+            print(f"AVERTISSEMENT : modèle '{model}' non trouvé dans Ollama.")
+            print(f"Modèles disponibles : {', '.join(model_names) or '(aucun)'}")
+            print()
+            print(f"Pour le télécharger :")
+            print(f"  ollama pull {model}")
+            return False
+    except Exception as e:
+        print(f"ERREUR : impossible de lister les modèles : {e}")
+        return False
+
+
+def chat_with_image(image_path: str, model: str, host: str = "http://localhost:11434") -> None:
+    """Mode interactif : charge l'image une fois, puis pose des questions."""
+    client = get_client(host)
+    image_b64 = encode_image(image_path)
+    print(f"🖼️  Image chargée : {image_path}")
+    print(f"🤖 Modèle       : {model}")
+    print(f"🔗 Ollama       : {host}")
+    print()
+    print("Mode interactif — tapez vos questions (ou 'exit'/'quit' pour sortir)")
+    print("Tapez '/image /chemin/nouvelle.png' pour changer d'image")
+    print("-" * 60)
+
+    # Historique de conversation (sans l'image à chaque fois pour économiser la mémoire)
+    messages = []
+
+    while True:
+        try:
+            question = input("\nVous > ").strip()
+        except (EOFError, KeyboardInterrupt):
+            print("\n👋 Au revoir !")
+            break
+
+        if not question:
+            continue
+
+        if question.lower() in ("exit", "quit", "q"):
+            print("👋 Au revoir !")
+            break
+
+        # Changement d'image
+        if question.startswith("/image "):
+            new_path = question[len("/image "):].strip()
+            try:
+                image_b64 = encode_image(new_path)
+                image_path = new_path
+                # Réinitialiser l'historique car image différente
+                messages = []
+                print(f"🖼️  Nouvelle image : {new_path}")
+            except SystemExit:
+                pass
+            continue
+
+        # Construire le message user avec l'image au premier tour
+        # Ensuite, l'image n'est ré-envoyée que si l'historique est vide
+        has_image_in_context = any(
+            isinstance(m.get("images"), list) and len(m["images"]) > 0
+            for m in messages
+        )
+
+        user_msg = {"role": "user", "content": question}
+        if not has_image_in_context:
+            # Première question ou image changée — inclure l'image
+            user_msg["images"] = [image_b64]
+        messages.append(user_msg)
+
+        print(f"🤖 Réponse ({model})...", end=" ", flush=True)
+
+        try:
+            response = client.chat(
+                model=model,
+                messages=messages,
+                stream=True,
+                options={
+                    "temperature": 0.2,
+                    "num_predict": 2048,
+                },
+            )
+
+            full_response = ""
+            print()  # nouvelle ligne après le "..."
+            for chunk in response:
+                content = chunk.get("message", {}).get("content", "")
+                if content:
+                    print(content, end="", flush=True)
+                    full_response += content
+
+            print()  # retour à la ligne après la réponse
+            messages.append({"role": "assistant", "content": full_response})
+
+        except Exception as e:
+            print(f"\n❌ Erreur : {e}")
+            # Retirer le dernier message user en cas d'erreur
+            messages.pop()
+
+
+def one_shot(image_path: str, question: str, model: str, host: str = "http://localhost:11434") -> None:
+    """Mode one-shot : une question, une réponse."""
+    client = get_client(host)
+    image_b64 = encode_image(image_path)
+
+    messages = [
+        {"role": "user", "content": question, "images": [image_b64]},
+    ]
+
+    try:
+        response = client.chat(
+            model=model,
+            messages=messages,
+            stream=True,
+            options={
+                "temperature": 0.2,
+                "num_predict": 2048,
+            },
+        )
+
+        print(f"🤖 {model} — '{question}'\n")
+        for chunk in response:
+            content = chunk.get("message", {}).get("content", "")
+            if content:
+                print(content, end="", flush=True)
+        print()
+
+    except Exception as e:
+        print(f"❌ Erreur : {e}")
+        sys.exit(1)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Chat interactif avec une image via Ollama (gemma4:26b par défaut)",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+        epilog="""
+Exemples :
+  # Mode interactif avec une image
+  python tests/test_image_chat_cli.py screenshot.png
+
+  # Mode one-shot (question directe)
+  python tests/test_image_chat_cli.py screenshot.png "Quels boutons vois-tu ?"
+
+  # Avec un autre modèle
+  python tests/test_image_chat_cli.py --model qwen3-vl:8b screenshot.png
+
+  # Ollama sur une machine distante
+  python tests/test_image_chat_cli.py --host http://dgx:11434 screenshot.png
+        """,
+    )
+    parser.add_argument(
+        "image",
+        nargs="?",
+        help="Chemin vers l'image à analyser",
+    )
+    parser.add_argument(
+        "question",
+        nargs="?",
+        default=None,
+        help="Question one-shot (si absent → mode interactif)",
+    )
+    parser.add_argument(
+        "--model",
+        default=DEFAULT_MODEL,
+        help=f"Modèle Ollama à utiliser (défaut: {DEFAULT_MODEL})",
+    )
+    parser.add_argument(
+        "--host",
+        default="http://localhost:11434",
+        help="URL du serveur Ollama (défaut: http://localhost:11434)",
+    )
+
+    args = parser.parse_args()
+
+    # Vérifications préalables
+    if not check_ollama_running(args.host):
+        sys.exit(1)
+
+    if not check_model_available(args.model, args.host):
+        sys.exit(1)
+
+    if not args.image:
+        print("Utilisation interactive — veuillez fournir le chemin d'une image.")
+        print()
+        print("Usage :")
+        print(f"  python {sys.argv[0]} /chemin/vers/image.png")
+        print(f"  python {sys.argv[0]} /chemin/vers/image.png \"Votre question\"")
+        print()
+        parser.print_help()
+        sys.exit(1)
+
+    if args.question:
+        # Mode one-shot
+        one_shot(args.image, args.question, args.model, args.host)
+    else:
+        # Mode interactif
+        chat_with_image(args.image, args.model, args.host)
+
+
+if __name__ == "__main__":
+    main()

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)

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 == []

visual_workflow_builder/backend/services/learned_workflow_bridge.py

@@ -295,6 +295,175 @@ def convert_learned_to_vwb_steps(
     return workflow_meta, steps, warnings
 
 
+# ---------------------------------------------------------------------------
+# Pont R1 — import IDEMPOTENT d'un workflow core en DB VWB (create-or-update)
+# ---------------------------------------------------------------------------
+
+# Marqueur stable de signature de trajectoire embarqué dans `Workflow.description`.
+# Le modèle `Workflow` n'a PAS (encore) de colonne dédiée ; on réutilise donc le
+# même mécanisme que la route GET /learned-workflows existante, qui détecte les
+# imports via `description.contains(...)`. La clé d'idempotence est la SIGNATURE
+# DE TRAJECTOIRE (cf. core.execution.trajectory_signature), pas le workflow_id de
+# session (qui change à chaque ré-apprentissage du même parcours).
+_TRAJ_SIG_MARKER = "[traj_sig:"
+
+
+def _trajectory_signature_marker(signature: str) -> str:
+    """Marqueur texte stable à embarquer dans la description."""
+    return f"{_TRAJ_SIG_MARKER}{signature}]"
+
+
+def _find_existing_learned_workflow(db_session, signature: str):
+    """Cherche un Workflow `source='learned_import'` de MÊME signature de trajectoire.
+
+    Ne considère QUE les imports appris : les workflows `source='manual'`
+    (démo Urgence_aiva, etc.) sont volontairement exclus du filtre et donc
+    jamais candidats à la mise à jour.
+    """
+    from db.models import Workflow  # import paresseux (modèles liés au runtime VWB)
+
+    marker = _trajectory_signature_marker(signature)
+    return (
+        db_session.query(Workflow)
+        .filter(
+            Workflow.source == "learned_import",
+            Workflow.description.contains(marker),
+        )
+        .first()
+    )
+
+
+def import_core_workflow_to_db(
+    core_dict: Dict[str, Any],
+    *,
+    machine_id: str,
+    source_session_id: str,
+    db_session,
+) -> Dict[str, Any]:
+    """Importe un workflow core (JSON appris par Léa) en DB VWB, de façon IDEMPOTENTE.
+
+    Fusion par **signature de trajectoire** (décision produit Dom 23/06) :
+      1. calcule `sig = workflow_trajectory_signature(core_dict)` ;
+      2. cherche un `Workflow` `source='learned_import'` de même signature ;
+      3. si trouvé → **skip** (pas de doublon, le workflow existant fait foi) ;
+         sinon → crée `Workflow` + `Step`(s) via `convert_learned_to_vwb_steps`.
+
+    Le nouveau workflow est marqué `source='learned_import'`,
+    `review_status='pending_review'`. Les workflows `source='manual'` ne sont
+    JAMAIS touchés (cf. `_find_existing_learned_workflow`).
+
+    Args:
+        core_dict: workflow core (dict JSON) tel qu'appris/sauvegardé.
+        machine_id: poste d'origine (traçabilité, stocké en tag/description).
+        source_session_id: session ayant produit ce workflow (traçabilité).
+        db_session: session SQLAlchemy (l'app appelante détient le contexte).
+
+    Returns:
+        dict {created: bool, workflow_id: str, signature: str, warnings: list}.
+        `created=False` quand un workflow de même trajectoire existait déjà.
+
+    Note (non-wiring) : cette unité n'est PAS branchée au worker live ni à la
+    route HTTP existante ; voir le rapport de câblage R1.
+    """
+    # Imports paresseux : garde le module léger et évite un import core/DB au load.
+    from core.execution.trajectory_signature import workflow_trajectory_signature
+    from db.models import Workflow, Step
+
+    signature = workflow_trajectory_signature(core_dict)
+
+    # --- Idempotence : même trajectoire déjà importée ? → skip (pas de doublon) ---
+    existing = _find_existing_learned_workflow(db_session, signature)
+    if existing is not None:
+        logger.info(
+            "Workflow appris déjà présent (signature %s…) → import ignoré, "
+            "réutilisation de %s",
+            signature[:12],
+            existing.id,
+        )
+        return {
+            "created": False,
+            "workflow_id": existing.id,
+            "signature": signature,
+            "warnings": [],
+        }
+
+    # --- Création : conversion core → steps VWB, puis écriture DB ---
+    wf_meta, steps_list, warnings = convert_learned_to_vwb_steps(core_dict)
+
+    current_name = (wf_meta.get("name") or "").strip()
+    if current_name.lower() in {"", "unnamed workflow", "workflow importé"}:
+        # Réutilise la dérivation de nom de la route HTTP si disponible.
+        try:
+            from api_v3.learned_workflows import _derive_default_name
+            wf_meta["name"] = _derive_default_name(core_dict)
+        except Exception:  # pragma: no cover - fallback minimal
+            wf_meta["name"] = f"Léa import — {datetime.now():%Y-%m-%d %H:%M}"
+
+    wf_id = f"wf_{uuid.uuid4().hex[:12]}"
+
+    # La signature est embarquée dans la description (clé d'idempotence) + une
+    # ligne de traçabilité (workflow core d'origine).
+    base_desc = (wf_meta.get("description") or "").strip()
+    description = "\n\n".join(
+        part
+        for part in (
+            base_desc,
+            f"[Importé depuis workflow appris: {core_dict.get('workflow_id', '')}]",
+            _trajectory_signature_marker(signature),
+        )
+        if part
+    )
+
+    workflow = Workflow(
+        id=wf_id,
+        name=wf_meta["name"],
+        description=description,
+        source="learned_import",
+        review_status="pending_review",
+    )
+
+    # Tags : conserver ceux du workflow + traçabilité machine/session.
+    tags = list(wf_meta.get("tags") or [])
+    tags.extend([f"machine:{machine_id}", f"session:{source_session_id}"])
+    workflow.tags = tags
+
+    db_session.add(workflow)
+
+    for step_data in steps_list:
+        step = Step(
+            id=f"step_{uuid.uuid4().hex[:12]}",
+            workflow_id=wf_id,
+            action_type=step_data["action_type"],
+            order=step_data["order"],
+            position_x=step_data.get("position_x", 0),
+            position_y=step_data.get("position_y", 0),
+            label=step_data.get("label", step_data["action_type"]),
+        )
+        params = dict(step_data.get("parameters", {}))
+        # L'image d'ancre (_anchor_image_base64) est laissée dans params : la
+        # persistance d'ancre (VisualAnchor + fichier) reste pilotée par la route
+        # HTTP existante. Cette unité se concentre sur l'idempotence Workflow/Step.
+        step.parameters = params
+        db_session.add(step)
+
+    db_session.commit()
+
+    logger.info(
+        "Workflow appris importé (R1) : %s (signature %s…, %d étapes, machine %s)",
+        wf_id,
+        signature[:12],
+        len(steps_list),
+        machine_id,
+    )
+
+    return {
+        "created": True,
+        "workflow_id": wf_id,
+        "signature": signature,
+        "warnings": warnings,
+    }
+
+
 def _convert_compound_substep(
     sub_type: str, sub: Dict[str, Any], parent_target: Dict[str, Any]
 ) -> Tuple[str, Dict[str, Any]]:

visual_workflow_builder/backend/tests/unit/test_import_core_workflow_to_db.py

@@ -0,0 +1,226 @@
+#!/usr/bin/env python3
+"""
+Test TDD — pont R1 : `import_core_workflow_to_db` IDEMPOTENT.
+
+Objectif chantier R1 : une session auto-apprise (workflow core JSON) doit pouvoir
+être (ré)importée en DB VWB **sans créer de doublon**. La fusion se fait par
+**signature de trajectoire** (cf. `core.execution.trajectory_signature`) — décision
+produit Dom 23/06 : create-or-update, pas create-only.
+
+Coeur du test (b) : ré-importer le MÊME core_dict 2× → toujours UN seul workflow.
+
+Ce module est volontairement isolé du chemin live :
+- il ne démarre PAS l'app Flask complète (`app.py`) ;
+- il lie le `db` partagé (`db.models.db`) à une SQLite **en mémoire** via une
+  app Flask minimale, même pattern que `tests/conftest.py` mais sans dépendances
+  lourdes (pas de socketio, pas de blueprints).
+"""
+
+import sys
+from pathlib import Path
+
+import pytest
+from flask import Flask
+
+# --- Chemins : racine projet (pour core.*) + backend (pour db.models, services.*) ---
+_BACKEND = Path(__file__).resolve().parent.parent.parent  # .../visual_workflow_builder/backend
+_ROOT = _BACKEND.parent.parent                            # .../rpa_vision_v3
+for p in (str(_ROOT), str(_BACKEND)):
+    if p not in sys.path:
+        sys.path.insert(0, p)
+
+from db.models import db, Workflow, Step  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# Fixtures DB en mémoire (app Flask minimale, db partagé)
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def db_app():
+    """App Flask minimale liée à une SQLite en mémoire, schéma créé."""
+    app = Flask("test_import_core_workflow")
+    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
+    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
+    db.init_app(app)
+    with app.app_context():
+        db.create_all()
+        yield app
+        db.session.remove()
+        db.drop_all()
+
+
+# ---------------------------------------------------------------------------
+# Fixtures de workflows core (format JSON appris par Léa)
+# ---------------------------------------------------------------------------
+
+def _core_workflow_bloc_notes() -> dict:
+    """Workflow core minimal : ouvrir Bloc-notes et saisir du texte."""
+    return {
+        "workflow_id": "wf_sess_bloc_notes_001",
+        "name": "Léa Bloc-notes",
+        "entry_nodes": ["n1"],
+        "nodes": [
+            {"node_id": "n1", "name": "Bureau"},
+            {"node_id": "n2", "name": "Bloc-notes ouvert"},
+        ],
+        "edges": [
+            {
+                "edge_id": "e1",
+                "from_node": "n1",
+                "to_node": "n2",
+                "action": {
+                    "type": "mouse_click",
+                    "target": {"by_text": "Bloc-notes", "by_role": "ocr"},
+                    "parameters": {"button": "left"},
+                },
+            },
+            {
+                "edge_id": "e2",
+                "from_node": "n2",
+                "to_node": "n2",
+                "action": {
+                    "type": "text_input",
+                    "target": {"by_text": "zone de saisie"},
+                    "parameters": {"text": "bonjour"},
+                },
+            },
+        ],
+    }
+
+
+def _core_workflow_calculatrice() -> dict:
+    """Workflow core d'une trajectoire DIFFÉRENTE (calculatrice)."""
+    return {
+        "workflow_id": "wf_sess_calc_002",
+        "name": "Léa Calculatrice",
+        "entry_nodes": ["n1"],
+        "nodes": [
+            {"node_id": "n1", "name": "Bureau"},
+            {"node_id": "n2", "name": "Calculatrice"},
+        ],
+        "edges": [
+            {
+                "edge_id": "e1",
+                "from_node": "n1",
+                "to_node": "n2",
+                "action": {
+                    "type": "mouse_click",
+                    "target": {"by_text": "Calculatrice", "by_role": "ocr"},
+                    "parameters": {"button": "left"},
+                },
+            },
+        ],
+    }
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+def test_import_creates_workflow_with_steps(db_app):
+    """(a) Un core_dict → 1 workflow VWB créé, avec ses steps."""
+    from services.learned_workflow_bridge import import_core_workflow_to_db
+
+    with db_app.app_context():
+        result = import_core_workflow_to_db(
+            _core_workflow_bloc_notes(),
+            machine_id="DESKTOP-TEST_windows",
+            source_session_id="sess_bloc_notes_001",
+            db_session=db.session,
+        )
+
+        assert result["created"] is True
+        wf_id = result["workflow_id"]
+        assert wf_id
+
+        wf = Workflow.query.get(wf_id)
+        assert wf is not None
+        assert wf.source == "learned_import"
+        assert wf.review_status == "pending_review"
+
+        steps = Step.query.filter_by(workflow_id=wf_id).all()
+        assert len(steps) >= 1, "le workflow importé doit avoir au moins une étape"
+
+
+def test_reimport_same_workflow_is_idempotent(db_app):
+    """(b) COEUR — ré-importer le MÊME core_dict 2× → toujours 1 seul workflow."""
+    from services.learned_workflow_bridge import import_core_workflow_to_db
+
+    with db_app.app_context():
+        first = import_core_workflow_to_db(
+            _core_workflow_bloc_notes(),
+            machine_id="DESKTOP-TEST_windows",
+            source_session_id="sess_bloc_notes_001",
+            db_session=db.session,
+        )
+        second = import_core_workflow_to_db(
+            _core_workflow_bloc_notes(),
+            machine_id="DESKTOP-TEST_windows",
+            source_session_id="sess_bloc_notes_001_rerun",
+            db_session=db.session,
+        )
+
+        # UN seul workflow en DB malgré deux imports
+        assert Workflow.query.count() == 1, "ré-import du même parcours = pas de doublon"
+
+        # Le second pointe vers le même workflow, marqué non-créé
+        assert first["workflow_id"] == second["workflow_id"]
+        assert second["created"] is False
+
+
+def test_different_trajectories_create_two_workflows(db_app):
+    """(c) Deux trajectoires différentes → 2 workflows distincts."""
+    from services.learned_workflow_bridge import import_core_workflow_to_db
+
+    with db_app.app_context():
+        r1 = import_core_workflow_to_db(
+            _core_workflow_bloc_notes(),
+            machine_id="DESKTOP-TEST_windows",
+            source_session_id="sess_a",
+            db_session=db.session,
+        )
+        r2 = import_core_workflow_to_db(
+            _core_workflow_calculatrice(),
+            machine_id="DESKTOP-TEST_windows",
+            source_session_id="sess_b",
+            db_session=db.session,
+        )
+
+        assert Workflow.query.count() == 2
+        assert r1["workflow_id"] != r2["workflow_id"]
+
+
+def test_manual_workflow_is_never_touched(db_app):
+    """(d) Un workflow source='manual' préexistant n'est jamais modifié.
+
+    Même si, par construction, il partageait la signature d'un parcours importé,
+    la fonction ne doit cibler QUE les workflows source='learned_import'.
+    """
+    from services.learned_workflow_bridge import import_core_workflow_to_db
+
+    with db_app.app_context():
+        # Workflow manuel préexistant (démo Urgence_aiva) — intouchable
+        manual = Workflow(
+            id="wf_manual_demo",
+            name="Urgence_aiva_demo",
+            description="Démo manuelle critique",
+            source="manual",
+            review_status="approved",
+        )
+        db.session.add(manual)
+        db.session.commit()
+        manual_name_before = manual.name
+        manual_review_before = manual.review_status
+
+        import_core_workflow_to_db(
+            _core_workflow_bloc_notes(),
+            machine_id="DESKTOP-TEST_windows",
+            source_session_id="sess_x",
+            db_session=db.session,
+        )
+
+        manual_after = Workflow.query.get("wf_manual_demo")
+        assert manual_after.name == manual_name_before
+        assert manual_after.review_status == manual_review_before
+        assert manual_after.source == "manual"