feat(server): R1 — import auto du workflow appris vers la DB VWB (gated)

finalize_session appelle _maybe_import_to_vwb : si RPA_R1_AUTO_IMPORT (OFF par
défaut), le workflow appris est assaini (sanitize_workflow_dict) puis importé en
DB VWB rejouable via le pont idempotent (import_core_workflow_to_db), dans un
app-context VWB lazy mutualisé (vwb_db). NON bloquant : un échec n'interrompt
jamais la finalisation. Rend l'appris rejouable sans geste manuel (R1).
Tests : câblage du seam + gating du flag + non-régression.

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

agent_v0/agent_v1/config.py

@@ -27,7 +27,7 @@ if platform.system() == "Windows":
         except Exception:
             pass
 
-AGENT_VERSION = "1.0.1"
+AGENT_VERSION = os.environ.get("RPA_AGENT_VERSION", "1.0.1")
 
 # Identifiant unique de la machine (utilisé pour le multi-machine)
 # Configurable via variable d'environnement, sinon auto-généré depuis hostname + OS

agent_v0/agent_v1/core/captor.py

@@ -32,6 +32,7 @@ from pynput.keyboard import Key, KeyCode
 # Importation relative pour rester dans le module v1
 from ..vision.capturer import VisionCapturer
 from ..vision.system_info import get_screen_metadata
+from .log_safe import _sanitize_metadata
 # from ..monitoring.system import SystemMonitor
 
 logger = logging.getLogger(__name__)
@@ -676,7 +677,7 @@ class EventCaptorV1:
             metadata = get_screen_metadata()
             with self._screen_metadata_lock:
                 self._screen_metadata = metadata
-            logger.debug(f"Métadonnées système rafraîchies : {metadata}")
+            logger.debug(f"Métadonnées système rafraîchies : {_sanitize_metadata(metadata)}")
         except Exception as e:
             logger.error(f"Erreur refresh métadonnées système : {e}")
 

agent_v0/agent_v1/core/executor.py

@@ -26,6 +26,7 @@ from typing import Any, Dict, Optional
 # DPI awareness est configure (SetProcessDpiAwareness(2) sur Windows).
 # Sans cela, pynput et mss utilisent des coordonnees logiques (virtualisees).
 from ..config import MACHINE_ID as _  # noqa: F401 — side-effect import
+from .log_safe import _title_hash
 
 import mss
 from pynput.mouse import Button, Controller as MouseController
@@ -862,7 +863,7 @@ class ActionExecutorV1:
                     )
                     if handled:
                         logger.info(
-                            f"[RUNTIME-DIALOG] '{current_title}' gere via serveur "
+                            f"[RUNTIME-DIALOG] [title_hash={_title_hash(current_title)}] gere via serveur "
                             f"fenetre -> bouton '{button_text}' "
                             f"[{resolved.get('method', 'server')}]"
                         )
@@ -890,7 +891,7 @@ class ActionExecutorV1:
                     )
                     if handled:
                         logger.info(
-                            f"[RUNTIME-DIALOG] '{current_title}' gere localement "
+                            f"[RUNTIME-DIALOG] [title_hash={_title_hash(current_title)}] gere localement "
                             f"fenetre -> bouton '{button_text}' [dialog_window_text_template]"
                         )
                         return handled
@@ -917,7 +918,7 @@ class ActionExecutorV1:
                 )
                 if handled:
                     logger.info(
-                        f"[RUNTIME-DIALOG] '{current_title}' gere par geometrie "
+                        f"[RUNTIME-DIALOG] [title_hash={_title_hash(current_title)}] gere par geometrie "
                         f"fenetre -> bouton '{button_text}'"
                     )
                     return handled
@@ -967,7 +968,7 @@ class ActionExecutorV1:
                     if not handled:
                         continue
                     logger.info(
-                        f"[RUNTIME-DIALOG] '{current_title}' gere via serveur "
+                        f"[RUNTIME-DIALOG] [title_hash={_title_hash(current_title)}] gere via serveur "
                         f"-> bouton '{button_text}' [{resolved.get('method', 'server')}]"
                     )
                     return handled
@@ -992,13 +993,13 @@ class ActionExecutorV1:
                 if not handled:
                     continue
                 logger.info(
-                    f"[RUNTIME-DIALOG] '{current_title}' gere localement "
+                    f"[RUNTIME-DIALOG] [title_hash={_title_hash(current_title)}] gere localement "
                     f"-> bouton '{button_text}' [dialog_text_template]"
                 )
                 return handled
 
         logger.info(
-            f"[RUNTIME-DIALOG] Aucun bouton resolu pour '{current_title}'"
+            f"[RUNTIME-DIALOG] Aucun bouton resolu pour [title_hash={_title_hash(current_title)}]"
         )
         return None
 
@@ -1258,7 +1259,7 @@ class ActionExecutorV1:
 
         if dialog_spec.get("skip_current_action_after_handle", False):
             logger.info(
-                f"[RUNTIME-DIALOG] Dialogue '{current_title}' gere -> "
+                f"[RUNTIME-DIALOG] Dialogue [title_hash={_title_hash(current_title)}] gere -> "
                 f"action {action.get('action_id', 'unknown')} skippée"
             )
             return {
@@ -1587,7 +1588,7 @@ class ActionExecutorV1:
             ]
             for pattern in popup_patterns:
                 if pattern in current_title:
-                    logger.info(f"Observer : popup détectée par titre — '{current_title}'")
+                    logger.info(f"Observer : popup détectée par titre — [title_hash={_title_hash(current_title)}]")
                     # On ne peut pas résoudre les coords juste par le titre
                     # → retourner popup sans coords, le caller fera handle_popup_vlm()
                     return {
@@ -1874,8 +1875,8 @@ class ActionExecutorV1:
                             )
                         else:
                             logger.warning(
-                                f"[LEA] Fenêtre incorrecte : attendu '{expected_title}', "
-                                f"actuel '{current_title}'"
+                                f"[LEA] Fenêtre incorrecte : attendu [title_hash={_title_hash(expected_title)}], "
+                                f"actuel [title_hash={_title_hash(current_title)}]"
                             )
                             auto_result = self._maybe_handle_runtime_dialog_before_pause(
                                 action=action,
@@ -1888,8 +1889,8 @@ class ActionExecutorV1:
                             if auto_result is not None:
                                 return auto_result
                             print(
-                                f"    [PRÉ-VÉRIF] Fenêtre '{current_title}' ≠ "
-                                f"attendu '{expected_title}' → mode apprentissage"
+                                f"    [PRÉ-VÉRIF] Fenêtre [title_hash={_title_hash(current_title)}] ≠ "
+                                f"attendu [title_hash={_title_hash(expected_title)}] → mode apprentissage"
                             )
                             try:
                                 self.notifier.replay_learning_mode(
@@ -1936,8 +1937,8 @@ class ActionExecutorV1:
                                 # des coordonnées devenues invalides.
                                 result["success"] = False
                                 result["error"] = (
-                                    f"Fenêtre incorrecte : attendu '{expected_title}', "
-                                    f"actuel '{current_title}'"
+                                    f"Fenêtre incorrecte : attendu [title_hash={_title_hash(expected_title)}], "
+                                    f"actuel [title_hash={_title_hash(current_title)}]"
                                 )
                                 result["warning"] = "wrong_window"
                                 result["target_description"] = expected_title
@@ -1945,11 +1946,11 @@ class ActionExecutorV1:
                                 result["screenshot"] = self._capture_screenshot_b64()
                                 logger.warning(
                                     f"[LEA] Wrong window sans correction → pause "
-                                    f"(attendu '{expected_title}', actuel '{current_title}')"
+                                    f"(attendu [title_hash={_title_hash(expected_title)}], actuel [title_hash={_title_hash(current_title)}])"
                                 )
                             return result
                     else:
-                        logger.info(f"[LEA] Pré-vérif OK : '{current_title}'")
+                        logger.info(f"[LEA] Pré-vérif OK : [title_hash={_title_hash(current_title)}]")
 
             # ── OBSERVER : pré-analyse écran avant résolution ──
             # Détecte popups, dialogues, états inattendus AVANT de chercher la cible.
@@ -1964,8 +1965,8 @@ class ActionExecutorV1:
                         # Popup détectée AVANT la résolution — la fermer
                         popup_label = observation.get("popup_label", "popup")
                         popup_coords = observation.get("popup_coords")
-                        print(f"    [OBSERVER] Popup détectée : '{popup_label}' — fermeture")
-                        logger.info(f"Observer : popup '{popup_label}' détectée avant résolution")
+                        print(f"    [OBSERVER] Popup détectée : [title_hash={_title_hash(popup_label)}] — fermeture")
+                        logger.info(f"Observer : popup [title_hash={_title_hash(popup_label)}] détectée avant résolution")
 
                         # ── SÉCURITÉ : refuser de cliquer sur un dialogue système ──
                         # Avant de suivre les coordonnées du serveur (VLM-based,
@@ -2365,8 +2366,8 @@ class ActionExecutorV1:
                         recheck_title = recheck_info.get("title", "")
                         if not _matches_expected_window(recheck_title):
                             logger.warning(
-                                f"P0.9 transition instable : matched '{post_title}' "
-                                f"puis '{recheck_title}' à T+0.5s ≠ '{expected_after}'"
+                                f"P0.9 transition instable : matched [title_hash={_title_hash(post_title)}] "
+                                f"puis [title_hash={_title_hash(recheck_title)}] à T+0.5s ≠ [title_hash={_title_hash(expected_after)}]"
                             )
                             matched = False
                             post_title = recheck_title
@@ -2376,19 +2377,19 @@ class ActionExecutorV1:
                             result["runtime_dialog"] = runtime_dialog_handled
                             print(
                                 f"    [POST-VÉRIF] Dialogue runtime géré "
-                                f"→ retour '{post_title}'"
+                                f"→ retour [title_hash={_title_hash(post_title)}]"
                             )
                             logger.info(
-                                "POST-VÉRIF runtime dialog géré : '%s' -> '%s'",
-                                runtime_dialog_handled.get("dialog_title", ""),
-                                post_title,
+                                "POST-VÉRIF runtime dialog géré : [title_hash=%s] -> [title_hash=%s]",
+                                _title_hash(runtime_dialog_handled.get("dialog_title", "")),
+                                _title_hash(post_title),
                             )
                         else:
-                            print(f"    [POST-VÉRIF] OK en {elapsed_wait:.1f}s — '{post_title}'")
-                            logger.info(f"POST-VÉRIF OK en {elapsed_wait:.1f}s : '{post_title}'")
+                            print(f"    [POST-VÉRIF] OK en {elapsed_wait:.1f}s — [title_hash={_title_hash(post_title)}]")
+                            logger.info(f"POST-VÉRIF OK en {elapsed_wait:.1f}s : [title_hash={_title_hash(post_title)}]")
                     else:
-                        print(f"    [POST-VÉRIF] TIMEOUT {max_wait}s — '{post_title}' ≠ '{expected_after}'")
-                        logger.warning(f"POST-VÉRIF TIMEOUT : '{post_title}' ≠ '{expected_after}'")
+                        print(f"    [POST-VÉRIF] TIMEOUT {max_wait}s — [title_hash={_title_hash(post_title)}] ≠ [title_hash={_title_hash(expected_after)}]")
+                        logger.warning(f"POST-VÉRIF TIMEOUT : [title_hash={_title_hash(post_title)}] ≠ [title_hash={_title_hash(expected_after)}]")
                         if runtime_dialog_handled:
                             result["warning"] = (
                                 f"runtime_dialog_handled_post_verify:{post_title}"
@@ -2396,9 +2397,9 @@ class ActionExecutorV1:
                             result["runtime_dialog"] = runtime_dialog_handled
                             logger.warning(
                                 "POST-VÉRIF runtime dialog géré mais "
-                                "fenêtre finale inattendue : '%s' ≠ '%s'",
-                                post_title,
-                                expected_after,
+                                "fenêtre finale inattendue : [title_hash=%s] ≠ [title_hash=%s]",
+                                _title_hash(post_title),
+                                _title_hash(expected_after),
                             )
                         # Contrôle strict : si success_strict, on STOP.
                         # On durcit aussi les vrais changements de fenêtre
@@ -2416,8 +2417,8 @@ class ActionExecutorV1:
                             if bool(action.get("success_strict")) or requires_transition:
                                 result["success"] = False
                                 result["error"] = (
-                                    f"Post-vérif échouée : fenêtre '{post_title}' "
-                                    f"au lieu de '{expected_after}'"
+                                    f"Post-vérif échouée : fenêtre [title_hash={_title_hash(post_title)}] "
+                                    f"au lieu de [title_hash={_title_hash(expected_after)}]"
                                 )
                                 result["warning"] = "wrong_window"
                                 result["needs_human"] = True
@@ -2458,7 +2459,7 @@ class ActionExecutorV1:
                     # paste=True (opt-in via action.paste) → clipboard + Ctrl+V (non-Citrix)
                     self._type_text(text, paste=bool(action.get("paste", False)))
                 print(f"    [TYPE] Termine.")
-                logger.info(f"Replay type : '{text[:30]}...' ({len(text)} chars, raw_keys={'oui' if raw_keys else 'non'})")
+                logger.info(f"Replay type : [{len(text)} chars] (raw_keys={'oui' if raw_keys else 'non'})")
 
             elif action_type == "key_combo":
                 keys = action.get("keys", [])
@@ -2524,12 +2525,12 @@ class ActionExecutorV1:
                     if not self._window_title_matches_any(current_title, patterns):
                         logger.warning(
                             "[LEA] verify_screen garde KO : attendu un titre "
-                            "contenant %s, actuel '%s'",
-                            patterns, current_title,
+                            "contenant %s, actuel [title_hash=%s]",
+                            patterns, _title_hash(current_title),
                         )
                         print(
                             f"    [VERIFY] Garde titre KO "
-                            f"(patterns={patterns}, actuel='{current_title}') "
+                            f"(patterns={patterns}, actuel=[title_hash={_title_hash(current_title)}]) "
                             "→ apprentissage humain"
                         )
                         try:
@@ -2557,15 +2558,15 @@ class ActionExecutorV1:
                             result["error"] = (
                                 f"verify_screen titre fenêtre KO : attendu "
                                 f"un titre contenant {patterns}, "
-                                f"actuel '{current_title}'"
+                                f"actuel [title_hash={_title_hash(current_title)}]"
                             )
                             result["warning"] = "setup_guard_window_mismatch"
                             result["needs_human"] = True
                             result["screenshot"] = self._capture_screenshot_b64()
                         return result
                     logger.info(
-                        "[LEA] verify_screen garde OK : '%s' matche %s",
-                        current_title, patterns,
+                        "[LEA] verify_screen garde OK : [title_hash=%s] matche %s",
+                        _title_hash(current_title), patterns,
                     )
 
                 print(f"    [VERIFY] Termine (verification deferred au serveur).")
@@ -3736,8 +3737,8 @@ Example: x_pct=0.50, y_pct=0.30"""
                 real_x = int(x_pct * sw)
                 real_y = int(y_pct * sh)
                 label = server_result.get("matched_element", {}).get("label", "popup")
-                print(f"    [POPUP-SERVER] Popup détectée ! Clic sur '{label}' → ({real_x}, {real_y})")
-                logger.info(f"[POPUP-SERVER] Clic popup '{label}' à ({real_x}, {real_y})")
+                print(f"    [POPUP-SERVER] Popup détectée ! Clic sur [title_hash={_title_hash(label)}] → ({real_x}, {real_y})")
+                logger.info(f"[POPUP-SERVER] Clic popup [title_hash={_title_hash(label)}] à ({real_x}, {real_y})")
                 self._click((real_x, real_y), "left")
                 time.sleep(1.0)
                 return True
@@ -3856,8 +3857,8 @@ Example: x_pct=0.50, y_pct=0.30"""
 
             raw_content = resp.json().get("message", {}).get("content", "")
             full_response = prefill + raw_content
-            print(f"    [POPUP-VLM] Réponse en {elapsed:.1f}s : {full_response.strip()}")
-            logger.info(f"[POPUP-VLM] Réponse VLM ({elapsed:.1f}s) : {full_response.strip()}")
+            print(f"    [POPUP-VLM] Réponse en {elapsed:.1f}s : [len={len(full_response)}, has_target={'target' in full_response}]")
+            logger.info(f"[POPUP-VLM] Réponse VLM ({elapsed:.1f}s) : [len={len(full_response)}, has_target={'target' in full_response}]")
 
             # Extraire le texte du bouton depuis la réponse
             button_text = raw_content.strip().strip('"').strip("'").strip(".")
@@ -4172,7 +4173,7 @@ Example: x_pct=0.50, y_pct=0.30"""
                 try:
                     self.keyboard.type(char)
                 except Exception as e:
-                    logger.debug(f"Impossible de taper '{char}': {e}")
+                    logger.debug(f"Impossible de taper [1 char typed]: {e}")
             # Délai humain entre les frappes (40-120ms)
             time.sleep(random.uniform(0.04, 0.12))
 

agent_v0/agent_v1/core/log_safe.py

@@ -0,0 +1,48 @@
+"""Helpers de logging PII-safe pour le client Léa (agent_v1).
+
+Convention : ne jamais logger le contenu brut d'une variable utilisateur
+(texte tapé, titre de fenêtre, nom de workflow, réponse VLM, chemin fichier).
+Le remplacer par :
+  - une longueur ou un hash court (corrélation de diagnostic sans révéler) ;
+  - un dict de métadonnées filtré (sans titre / fenêtre active).
+
+À importer dans tout module d'agent_v1 qui logge une donnée potentiellement
+sensible. Branche feat/push-log-dgx — DETTE-020 (assainissement à la source).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+
+
+def _title_hash(title: str) -> str:
+    """Hash SHA1 tronqué (8 hex) d'un titre.
+
+    Corrélation stable (même titre → même hash → « même popup re-détectée »)
+    sans exposer le contenu. `errors="replace"` pour ne jamais lever sur un
+    encodage exotique (titres Windows multi-langues).
+    """
+    return hashlib.sha1((title or "").encode("utf-8", errors="replace")).hexdigest()[:8]
+
+
+# Clés de métadonnées susceptibles de contenir du contenu utilisateur (PII).
+_PII_METADATA_KEYS = ("title", "active_window", "window_title")
+
+
+def _sanitize_metadata(metadata: dict) -> dict:
+    """Copie d'un dict de métadonnées sans les clés porteuses de PII.
+
+    Garde les champs techniques (resolution, dpi, theme, langue…), retire
+    titre / fenêtre active. Ne mute pas le dict d'origine.
+    """
+    return {k: v for k, v in metadata.items() if k not in _PII_METADATA_KEYS}
+
+
+def _path_ext(path: str) -> str:
+    """Extension seule d'un chemin (ex. « .png »), sans nom ni dossier.
+
+    Un chemin peut nommer un patient ; l'extension suffit au diagnostic.
+    Chaîne vide si pas de chemin ou pas d'extension.
+    """
+    return os.path.splitext(path)[1] if path else ""

agent_v0/agent_v1/core/recovery.py

@@ -24,6 +24,8 @@ from dataclasses import dataclass
 from enum import Enum
 from typing import Any, Dict, List, Optional
 
+from .log_safe import _title_hash
+
 logger = logging.getLogger(__name__)
 
 
@@ -168,8 +170,8 @@ class RecoveryEngine:
                     from ..window_info_crossplatform import get_active_window_info
                     active = get_active_window_info()
                     active_title = active.get("title", "")
-                    logger.info(f"Recovery : Alt+F4 sur '{active_title}'")
-                    print(f"    [RECOVERY] Alt+F4 — fermeture de '{active_title}'")
+                    logger.info(f"Recovery : Alt+F4 sur [title_hash={_title_hash(active_title)}]")
+                    print(f"    [RECOVERY] Alt+F4 — fermeture de [title_hash={_title_hash(active_title)}]")
                 except Exception:
                     logger.info("Recovery : Alt+F4 (fenêtre active inconnue)")
                     print("    [RECOVERY] Alt+F4 — fermeture fenêtre indésirable")
@@ -182,7 +184,7 @@ class RecoveryEngine:
                 return RecoveryResult(
                     action_taken=RecoveryAction.CLOSE_WINDOW,
                     success=True,
-                    detail=f"Alt+F4 exécuté sur '{active_title if 'active_title' in dir() else '?'}'",
+                    detail=f"Alt+F4 exécuté sur [title_hash={_title_hash(active_title) if 'active_title' in dir() else '?'}]",
                 )
 
             elif strategy == RecoveryAction.CLICK_AWAY:

agent_v0/agent_v1/logging_setup.py

@@ -0,0 +1,56 @@
+"""Journalisation client Léa — DETTE-021.
+
+Branche un handler **fichier** (`TimedRotatingFileHandler`) sur le logger racine,
+en plus de la console. Sans cela, sous `pythonw.exe` (pas de console), les logs
+partent sur stderr et sont **perdus** — diagnostic terrain impossible.
+
+Rotation quotidienne + rétention `retention_days` (Règlement IA Art. 12 :
+journalisation automatique + conservation minimum 180 j).
+"""
+import logging
+from logging.handlers import TimedRotatingFileHandler
+from pathlib import Path
+
+_FMT = "%(asctime)s %(levelname)-7s %(name)-25s %(message)s"
+
+
+def setup_logging(log_file, level=logging.INFO, retention_days=180):
+    """Configure le logging racine : fichier (rotation quotidienne, `retention_days`
+    fichiers conservés) + console. **Idempotent** : ne réempile pas nos handlers.
+
+    Args:
+        log_file: chemin du fichier de log (`config.LOG_FILE` en prod).
+        level: niveau racine (INFO par défaut ; DEBUG géré par l'appelant).
+        retention_days: nb de fichiers quotidiens conservés (180 = Règlement IA Art. 12).
+
+    Returns:
+        Le `TimedRotatingFileHandler` créé.
+    """
+    log_file = Path(log_file)
+    log_file.parent.mkdir(parents=True, exist_ok=True)
+
+    root = logging.getLogger()
+    root.setLevel(level)
+
+    # Idempotence : retirer nos propres handlers posés par un appel précédent.
+    for h in list(root.handlers):
+        if getattr(h, "_lea_managed", False):
+            h.close()
+            root.removeHandler(h)
+
+    file_handler = TimedRotatingFileHandler(
+        str(log_file), when="midnight", backupCount=retention_days, encoding="utf-8"
+    )
+    file_handler.setFormatter(logging.Formatter(_FMT, datefmt="%Y-%m-%d %H:%M:%S"))
+    file_handler.setLevel(level)
+    file_handler._lea_managed = True
+    root.addHandler(file_handler)
+
+    # Console conservée (utile en dev / si lancé avec une console).
+    console = logging.StreamHandler()
+    console.setFormatter(logging.Formatter(_FMT, datefmt="%H:%M:%S"))
+    console.setLevel(level)
+    console._lea_managed = True
+    root.addHandler(console)
+
+    return file_handler

agent_v0/agent_v1/main.py

@@ -15,7 +15,7 @@ import time
 import logging
 import threading
 from .config import (
-    SESSIONS_ROOT, AGENT_VERSION, SERVER_URL, MACHINE_ID, LOG_RETENTION_DAYS,
+    SESSIONS_ROOT, AGENT_VERSION, SERVER_URL, MACHINE_ID, LOG_RETENTION_DAYS, LOG_FILE,
     SCREEN_RESOLUTION, DPI_SCALE, OS_THEME, API_TOKEN, MAX_SESSION_DURATION_S,
     STREAMING_ENDPOINT,
 )
@@ -29,6 +29,7 @@ from .ui.capture_server import CaptureServer
 from .session.storage import SessionStorage
 from .vision.capturer import VisionCapturer
 from .finalize_contract import dispatch_finalize_result
+from .core.log_safe import _title_hash
 
 # Import optionnel du client serveur (pour le chat et les workflows)
 # Deux chemins : relatif (depuis agent_v0.agent_v1) ou absolu (depuis C:\rpa_vision\agent_v1)
@@ -43,11 +44,19 @@ except (ImportError, ValueError):
 # Configuration du logging — format structuré et lisible pour un TIM
 # Niveau de détail : INFO par défaut, DEBUG si RPA_AGENT_DEBUG=1
 _log_level = logging.DEBUG if os.environ.get("RPA_AGENT_DEBUG") == "1" else logging.INFO
-logging.basicConfig(
+# DETTE-021 : journaliser dans un FICHIER (rotation quotidienne + rétention 180 j,
+# Règlement IA Art. 12). Sous `pythonw.exe` (sans console), un basicConfig→stderr
+# serait perdu. Fallback console si le fichier est indisponible — ne JAMAIS
+# empêcher Léa de démarrer pour un problème de log.
+try:
+    from .logging_setup import setup_logging
+    setup_logging(LOG_FILE, level=_log_level, retention_days=LOG_RETENTION_DAYS)
+except Exception:
+    logging.basicConfig(
         level=_log_level,
         format="%(asctime)s %(levelname)-7s %(name)-25s %(message)s",
         datefmt="%H:%M:%S",
-)
+    )
 
 # Réduire le bruit de certaines libs
 for _noisy in ("urllib3", "requests.packages.urllib3", "PIL", "mss"):
@@ -253,7 +262,7 @@ class AgentV1:
         # Ne PAS en relancer une ici — deux threads poll simultanés causent
         # une race condition où les actions sont consommées mais pas exécutées.
 
-        logger.info(f"Session {self.session_id} ({workflow_name}) sur machine {self.machine_id} en cours...")
+        logger.info(f"Session {self.session_id} [wf_hash={_title_hash(workflow_name)}] sur machine {self.machine_id} en cours...")
 
     def _command_watchdog_loop(self):
         """Surveille un fichier de commande pour executer des ordres visuels (legacy)."""

agent_v0/agent_v1/network/streamer.py

@@ -36,6 +36,7 @@ import requests
 from PIL import Image
 
 from ..config import API_TOKEN, BASE_DIR, STREAMING_ENDPOINT
+from ..core.log_safe import _title_hash
 from .persistent_buffer import MAX_ATTEMPTS, PersistentBuffer
 
 
@@ -138,7 +139,7 @@ class TraceStreamer:
             target=self._buffer_drain_loop, daemon=True
         )
         self._drain_thread.start()
-        logger.info(f"Streamer pour {self.session_id} démarré")
+        logger.info(f"Streamer démarré")
 
     def stop(self):
         """Arrêter le streaming et finaliser la session côté serveur.
@@ -166,7 +167,7 @@ class TraceStreamer:
             self._drain_thread.join(timeout=2.0)
 
         self._finalize_session()
-        logger.info(f"Streamer pour {self.session_id} arrêté")
+        logger.info(f"Streamer arrêté")
 
     def push_event(self, event_data: dict):
         """Enfile un événement pour envoi immédiat.
@@ -632,7 +633,7 @@ class TraceStreamer:
             self._check_redirect(resp, url)
             if resp.ok:
                 result = resp.json()
-                logger.info(f"Session finalisée: {result}")
+                logger.info(f"Session finalisée [status={result.get('status')}, wf_hash={_title_hash(result.get('workflow_name',''))}]")
                 if self._on_finalize_result is not None:
                     try:
                         self._on_finalize_result(result)

agent_v0/agent_v1/ui/activity_panel.py

@@ -29,6 +29,8 @@ from dataclasses import dataclass, field
 from enum import Enum
 from typing import Optional
 
+from ..core.log_safe import _title_hash
+
 logger = logging.getLogger(__name__)
 
 
@@ -132,7 +134,7 @@ class ActivityPanel:
             )
         self._notifier_changement()
         self._rafraichir_ui()
-        logger.info(f"[ACTIVITY] Workflow démarré : {nom} ({nb_etapes} étapes)")
+        logger.info(f"[ACTIVITY] Workflow démarré : [wf_hash={_title_hash(nom)}] ({nb_etapes} étapes)")
 
     def mettre_a_jour(
         self,

agent_v0/agent_v1/ui/capture_server.py

@@ -27,6 +27,8 @@ import os
 import time
 from http.server import HTTPServer, BaseHTTPRequestHandler
 
+from ..core.log_safe import _path_ext
+
 logger = logging.getLogger(__name__)
 
 CAPTURE_PORT = int(os.environ.get("RPA_CAPTURE_PORT", "5006"))
@@ -312,7 +314,7 @@ class _FileActionHandlerLocal:
                 })
                 extensions[ext] = extensions.get(ext, 0) + 1
 
-        logger.info(f"Liste dossier '{path_str}' : {len(files)} fichiers")
+        logger.info(f"Liste dossier [ext={_path_ext(path_str)}] : {len(files)} fichiers")
         return {"files": files, "count": len(files), "extensions": extensions, "path": path_str}
 
     def _create_dir(self, params: dict) -> dict:
@@ -328,7 +330,7 @@ class _FileActionHandlerLocal:
         target = _Path(path_str)
         existed = target.exists()
         target.mkdir(parents=True, exist_ok=True)
-        logger.info(f"Dossier '{path_str}' {'existait deja' if existed else 'cree'}")
+        logger.info(f"Dossier [ext={_path_ext(path_str)}] {'existait deja' if existed else 'cree'}")
         return {"created": not existed, "path": path_str, "already_existed": existed}
 
     def _move_file(self, params: dict) -> dict:
@@ -350,7 +352,7 @@ class _FileActionHandlerLocal:
 
         _Path(dst).parent.mkdir(parents=True, exist_ok=True)
         _shutil.move(src, dst)
-        logger.info(f"Fichier deplace : '{src}' -> '{dst}'")
+        logger.info(f"Fichier deplace : [ext={_path_ext(src)}] -> [ext={_path_ext(dst)}]")
         return {"moved": True, "source": src, "destination": dst}
 
     def _copy_file(self, params: dict) -> dict:
@@ -376,7 +378,7 @@ class _FileActionHandlerLocal:
             _shutil.copytree(src, dst)
         else:
             _shutil.copy2(src, dst)
-        logger.info(f"Fichier copie : '{src}' -> '{dst}'")
+        logger.info(f"Fichier copie : [ext={_path_ext(src)}] -> [ext={_path_ext(dst)}]")
         return {"copied": True, "source": src, "destination": dst}
 
     def _sort_by_extension(self, params: dict) -> dict:
@@ -425,7 +427,7 @@ class _FileActionHandlerLocal:
                 extensions[ext] = extensions.get(ext, 0) + 1
 
         logger.info(
-            f"Classement par extension '{source_dir_str}' : {len(moved)} fichiers"
+            f"Classement par extension [ext={_path_ext(source_dir_str)}] : {len(moved)} fichiers"
         )
         return {
             "moved": moved,

agent_v0/agent_v1/window_info_crossplatform.py

@@ -19,6 +19,8 @@ import platform
 import subprocess
 from typing import Any, Dict, Optional
 
+from .core.log_safe import _title_hash
+
 
 def _run_cmd(cmd: list[str]) -> Optional[str]:
     """Exécute une commande et renvoie la sortie texte (strippée), ou None en cas d'erreur."""
@@ -372,7 +374,7 @@ if __name__ == "__main__":
     for i in range(5):
         info = get_active_window_info()
         rect = get_active_window_rect()
-        print(f"[{i+1}] App: {info['app_name']:20s} | Title: {info['title']}")
+        print(f"[{i+1}] App: {info['app_name']:20s} | Title: [title_hash={_title_hash(info['title'])}]")
         if rect:
             print(f"      Rect: {rect['rect']} | Size: {rect['size']}")
         else:

agent_v0/server_v1/agent_logs_store.py

@@ -0,0 +1,77 @@
+"""Store des logs poussés par les clients Léa (push-log-DGX).
+
+Persiste les logs reçus du client, rangés par `machine_id`, pour consultation
+au dashboard (diagnostic des postes sans AnyDesk). Stockage fichier JSONL
+(un fichier par jour et par machine_id), rétention configurable.
+
+DETTE-020/021 (observabilité). Branche feat/push-log-dgx.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+# machine_id = entrée réseau → neutraliser tout caractère hors liste blanche
+# (anti path-traversal : '/', '\\', '..' ne doivent pas s'échapper du base_dir).
+_SAFE_MACHINE_ID_RE = re.compile(r"[^A-Za-z0-9._-]")
+
+
+class AgentLogsStore:
+    """Persiste et relit les logs clients rangés par machine_id (JSONL)."""
+
+    def __init__(self, base_dir: str | Path = "data/agent_logs"):
+        self.base_dir = Path(base_dir)
+        self.base_dir.mkdir(parents=True, exist_ok=True)
+
+    def _machine_dir(self, machine_id: str) -> Path:
+        safe = _SAFE_MACHINE_ID_RE.sub("_", machine_id or "").strip("._") or "unknown"
+        d = self.base_dir / safe
+        d.mkdir(parents=True, exist_ok=True)
+        return d
+
+    def append(self, machine_id: str, entries: list[dict]) -> int:
+        """Ajoute un batch de logs pour un poste. Retourne le nb de lignes écrites."""
+        if not entries:
+            return 0
+        now = datetime.now(timezone.utc)
+        day_file = self._machine_dir(machine_id) / f"{now.date().isoformat()}.jsonl"
+        with day_file.open("a", encoding="utf-8") as f:
+            for entry in entries:
+                record = dict(entry)
+                record.setdefault("received_at", now.isoformat())
+                f.write(json.dumps(record, ensure_ascii=False) + "\n")
+        return len(entries)
+
+    def read(self, machine_id: str) -> list[dict]:
+        """Relit toutes les entrées d'un poste, triées par fichier (date) puis ordre d'écriture."""
+        d = self._machine_dir(machine_id)
+        out: list[dict] = []
+        for jsonl in sorted(d.glob("*.jsonl")):
+            with jsonl.open(encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if line:
+                        out.append(json.loads(line))
+        return out
+
+    def purge_old(self, retention_days: int = 30, now: datetime | None = None) -> int:
+        """Supprime les fichiers-jour antérieurs à la rétention. Retourne le nb supprimé.
+
+        Rétention basée sur la date encodée dans le nom du fichier (`YYYY-MM-DD.jsonl`),
+        pas sur le mtime (déterministe, non altérable). `now` injectable pour les tests.
+        """
+        now = now or datetime.now(timezone.utc)
+        cutoff = (now - timedelta(days=retention_days)).date()
+        removed = 0
+        for jsonl in self.base_dir.rglob("*.jsonl"):
+            try:
+                file_date = datetime.strptime(jsonl.stem, "%Y-%m-%d").date()
+            except ValueError:
+                continue  # nom inattendu → on ne touche pas
+            if file_date < cutoff:
+                jsonl.unlink()
+                removed += 1
+        return removed

agent_v0/server_v1/api_stream.py

@@ -27,6 +27,7 @@ from fastapi import BackgroundTasks, Depends, FastAPI, File, HTTPException, Requ
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 
+from .pii_sanitizer import sanitize_event
 from .replay_failure_logger import log_replay_failure
 from .replay_verifier import ReplayVerifier, VerificationResult
 from .replay_learner import ReplayLearner
@@ -583,6 +584,17 @@ _AGENTS_DB_PATH = os.environ.get(
 )
 agent_registry = AgentRegistry(db_path=_AGENTS_DB_PATH)
 
+# push-log-DGX : store des logs poussés par les clients, rangés par machine_id
+# (observabilité des postes sans AnyDesk — DETTE-020/021).
+from .agent_logs_store import AgentLogsStore  # noqa: E402
+
+_AGENT_LOGS_DIR = os.environ.get(
+    "RPA_AGENT_LOGS_DIR", str(ROOT_DIR / "data" / "agent_logs")
+)
+# Garde-fou anti-flood (G3) : nb max d'entrées acceptées par batch.
+_AGENT_LOGS_MAX_BATCH = int(os.environ.get("RPA_AGENT_LOGS_MAX_BATCH", "1000"))
+agent_logs_store = AgentLogsStore(base_dir=_AGENT_LOGS_DIR)
+
 
 def _agent_registry_has_entries() -> bool:
     try:
@@ -1562,6 +1574,16 @@ class AgentUninstallRequest(BaseModel):
     reason: Optional[str] = None
 
 
+class AgentLogsRequest(BaseModel):
+    """Batch de logs poussé par un client Léa (push-log-DGX).
+
+    `logs` = liste d'entrées {ts, level, logger, message} (format libre côté
+    serveur ; le client garantit le PII-safe avant push).
+    """
+    machine_id: str
+    logs: list[dict] = []
+
+
 # Thread de nettoyage périodique des replays terminés et sessions expirées
 _cleanup_thread: Optional[threading.Thread] = None
 _cleanup_running = False
@@ -1901,6 +1923,11 @@ async def stream_event(data: StreamEvent):
     # Auto-enregistrer la session si inconnue (robustesse au redémarrage serveur)
     _ensure_session_registered(session_id, machine_id=machine_id)
 
+    # ── Assainissement PII : sanitize une fois, les 3 chemins reçoivent la copie ──
+    sanitized_event = sanitize_event(
+        data.event, mapping=_session_pii_mapping[session_id]
+    )
+
     # Persister sur disque (journal JSONL, dans un sous-dossier par machine si multi-machine)
     if machine_id and machine_id != "default":
         session_path = LIVE_SESSIONS_DIR / machine_id / session_id
@@ -1909,21 +1936,26 @@ async def stream_event(data: StreamEvent):
     session_path.mkdir(parents=True, exist_ok=True)
     event_file = session_path / "live_events.jsonl"
     with open(event_file, "a", encoding="utf-8") as f:
-        f.write(json.dumps(data.dict()) + "\n")
+        f.write(json.dumps({
+            "session_id": data.session_id,
+            "timestamp": data.timestamp,
+            "event": sanitized_event,
+            "machine_id": machine_id,
+        }) + "\n")
 
     # Traitement direct via StreamProcessor
-    result = worker.process_event_direct(session_id, data.event)
+    result = worker.process_event_direct(session_id, sanitized_event)
 
     # ── Observation Shadow (si mode Shadow activé pour cette session) ──
     # L'appel est protégé et non bloquant : si l'observer n'est pas
     # actif, ou s'il lève, la capture continue normalement.
-    shadow_observe_event(session_id, data.event)
+    shadow_observe_event(session_id, sanitized_event)
 
     # ── Enrichissement SomEngine temps réel pour les mouse_click ──
     # Après l'enregistrement de l'event, tenter l'enrichissement si le
     # screenshot est déjà arrivé. Sinon, l'event est mis en attente et
     # sera enrichi quand le screenshot arrivera (voir stream_image).
-    event = data.event
+    event = sanitized_event
     if event.get("type") == "mouse_click" and event.get("screenshot_id"):
         session = processor.session_manager.get_session(session_id)
         if session:
@@ -1941,6 +1973,9 @@ async def stream_event(data: StreamEvent):
 # =========================================================================
 
 # Ensemble des screenshots déjà analysés (évite les doublons de retry)
+# Mapping PII par session — tokens cohérents intra-session (même patient → même [NOM_1])
+_session_pii_mapping: Dict[str, Dict] = defaultdict(dict)
+
 _analyzed_shots: Dict[str, set] = defaultdict(set)
 
 # Hash du dernier screenshot analysé par session (déduplication par similarité)
@@ -2337,9 +2372,12 @@ async def stream_image(
     # Le fichier brut (shot_XXXX_full.png) reste intact pour le replay,
     # le grounding VLM et l'entraînement. La version floutée est écrite en
     # parallèle sous shot_XXXX_full_blurred.png.
+    # focus_* : plein écran avec PII dans les titres (blind spot Qwen 28/06,
+    # 1440 fichiers/350 Mo non floutés) — désormais inclus dans le blur.
     if _PII_BLUR_ENABLED and _blur_pii_on_image is not None and (
         ("_full" in shot_id and shot_id.startswith("shot_"))
         or shot_id.startswith("heartbeat_")
+        or shot_id.startswith("focus_")
     ):
         _pii_blur_executor.submit(_produce_blurred_version, file_path_str, shot_id)
 
@@ -7200,6 +7238,59 @@ async def agents_fleet():
     }
 
 
+@app.post("/api/v1/agents/logs")
+async def agents_logs(request: AgentLogsRequest):
+    """Réception des logs poussés par un client Léa (push-log-DGX).
+
+    Range les logs par machine_id (AgentLogsStore) pour consultation au
+    dashboard — diagnostic des postes sans AnyDesk. Mêmes garde-fous fleet
+    que stream/poll : un poste révoqué/inconnu est refusé (403).
+    """
+    machine_id = (request.machine_id or "").strip()
+    if not machine_id:
+        raise HTTPException(status_code=400, detail="machine_id est obligatoire")
+
+    if len(request.logs) > _AGENT_LOGS_MAX_BATCH:
+        raise HTTPException(
+            status_code=413,
+            detail={
+                "error": "batch_too_large",
+                "max_batch": _AGENT_LOGS_MAX_BATCH,
+                "received": len(request.logs),
+            },
+        )
+
+    # Bloque les postes révoqués/désinstallés + met à jour last_seen_at.
+    _guard_agent_registry_access(machine_id, endpoint="agents/logs")
+
+    received = agent_logs_store.append(machine_id, request.logs)
+    return {"status": "ok", "received": received, "machine_id": machine_id}
+
+
+@app.get("/api/v1/agents/logs/{machine_id}")
+async def get_agents_logs(machine_id: str, limit: int = 1000):
+    """Lecture des logs poussés par un poste (push-log-DGX, brique 3).
+
+    Route de diagnostic dashboard : restitue les logs rangés par machine_id
+    (poste sans AnyDesk). Lecture admin read-only — volontairement SANS garde
+    fleet : on doit pouvoir consulter un poste révoqué ou en panne. Seul le
+    Bearer (dépendance globale `_verify_token`) protège l'accès.
+
+    `limit` borne la réponse aux N entrées les plus récentes (tail) pour éviter
+    de renvoyer plusieurs jours de logs d'un coup.
+    """
+    entries = agent_logs_store.read(machine_id)
+    total = len(entries)
+    if limit and limit > 0:
+        entries = entries[-limit:]
+    return {
+        "machine_id": machine_id,
+        "total": total,
+        "count": len(entries),
+        "logs": entries,
+    }
+
+
 # =========================================================================
 # R2 MVP P0 — DialogResolver (catalogue centralisé des modaux runtime)
 # Flag OFF par défaut. Activer en posant RPA_DIALOG_RESOLVER_ENABLED=true.

agent_v0/server_v1/pii_sanitizer.py

@@ -0,0 +1,239 @@
+"""Assainissement PII des données capturées (titres de fenêtre, texte saisi, OCR).
+
+Côté serveur. Remplace la PII par des **tokens typés et cohérents**
+(`[IPP_1]`, `[AGE_1]`, `[NOM_1]`…) : on protège la donnée **et** on garde la
+structure (champ de type NOM/IPP) utile à l'apprentissage des variables.
+
+Couche 1 (ce module, sans modèle) : filet **regex** sur la PII structurée
+(IPP, NIR, téléphone, email, âge) + règles **structurelles** des titres
+cliniques (`NOM (NAISSANCE) Prénom`, `[Nom Prénom]` des fenêtres PACS). Regex
+réutilisées du projet `anonymisation`.
+Couche 2 (à venir) : NER CamemBERT-bio (ONNX) pour les noms libres que la
+couche 1 ne capte pas — branchée plus tard, ce module marche sans.
+
+Branche feat/push-log-dgx — assainissement PII clinique.
+"""
+
+from __future__ import annotations
+
+import copy
+import re
+from typing import Dict, List, Optional, Tuple
+
+# --- Filet regex (réutilisé de anonymisation/anonymizer_core_refactored_onnx.py) ---
+RE_IPP = re.compile(r"\b(?:I\.?P\.?P\.?|IPP|N°\s*Ipp)\s*[:\-]?\s*([A-Za-z0-9]{6,})\b", re.IGNORECASE)
+RE_NIR = re.compile(r"(?<!\d)[12]\s?\d{2}\s?\d{2}\s?\d{2}\s?\d{3}\s?\d{3}\s?\d{2}(?!\d)")
+RE_EMAIL = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
+RE_TEL = re.compile(r"(?<!\d)(?:\+33\s?|0)\d(?:[ .\-]?\d){8}(?!\d)")
+# Âge format « titre » (« 90 ans »), plus large que le regex prose de anonymisation.
+RE_AGE = re.compile(r"\b(\d{1,3})\s*ans\b", re.IGNORECASE)
+
+_MAJ = r"A-ZÉÈÀÂÊÎÔÛÄËÏÖÜÇ"
+_MIN = r"a-zàâäéèêëïîôöùûüç"
+# Format clinique « NOM (NOM_NAISSANCE) Prénom » (ex. « ROSSIGNOL (SOUBIE) Pierrette »).
+RE_NOM_NAISSANCE = re.compile(
+    rf"\b[{_MAJ}][{_MAJ}\-']+\s+\([{_MAJ}][{_MAJ}\-']+\)\s+[{_MAJ}][{_MIN}\-']+\b"
+)
+# Patient entre crochets des fenêtres PACS (ex. « [DATTIN Alix] »), ≥ 2 tokens capitalisés.
+RE_NOM_BRACKET = re.compile(
+    rf"\[((?:[{_MAJ}][\w{_MIN}'\-]*\s+){{1,3}}[{_MAJ}][\w{_MIN}'\-]*)\]"
+)
+# « Prénom NOM » inversé, sans parenthèses ni crochets (ex. « Alix DATTIN »).
+# 2e mot tout en MAJUSCULES → faible risque de FP (« Mozilla Firefox » ne matche pas).
+RE_PRENOM_NOM = re.compile(rf"\b[{_MAJ}][{_MIN}]+\s+[{_MAJ}][{_MAJ}\-']+\b")
+
+# GXD5 Diagnostics : numéro de dossier + nom patient tout-majuscules.
+# Format réel : « GXD5 Diagnostics - 128008 - BENVENISTE MARIE-LAURENCE »
+# Le numéro (128008) = ID dossier patient (PII). Le nom = PII.
+# 2 groupes de capture : (1)=numéro, (2)=nom complet.
+RE_GXD5_DIAG = re.compile(
+    rf"GXD5\s+Diagnostics\s*-\s*(\d+)\s*-\s*([{_MAJ}][{_MAJ}\-' ]+)"
+)
+
+# Ordre = priorité ; group = portion à remplacer (0 = match entier).
+_DETECTORS: List[Tuple[re.Pattern, str, int]] = [
+    (RE_NOM_NAISSANCE, "NOM", 0),
+    (RE_NOM_BRACKET, "NOM", 0),
+    (RE_GXD5_DIAG, "DOSSIER", 1),   # numéro de dossier
+    (RE_PRENOM_NOM, "NOM", 0),
+    (RE_EMAIL, "EMAIL", 0),
+    (RE_NIR, "NIR", 0),
+    (RE_IPP, "IPP", 1),
+    (RE_TEL, "TEL", 0),
+    (RE_AGE, "AGE", 0),
+]
+# GXD5 nom (groupe 2) traité séparément — même regex, priorité juste après.
+_DETECTORS.append((RE_GXD5_DIAG, "NOM", 2))
+
+# Anti-faux-positifs : termes logiciels/UI à ne jamais prendre pour un nom.
+# (Sous-ensemble inline ; les gazetteers complets arrivent avec la couche NER.)
+_SOFTWARE_BLACKLIST = {
+    "FIREFOX", "MOZILLA", "CHROME", "EDGE", "EXPERT", "SANTE", "SANTÉ", "PACS",
+    "CIM", "ARES", "EASILY", "CONSULTATION", "URGENCES", "SAISIE", "COURRIER",
+    "DOSSIER", "PATIENT", "FENETRE", "FENÊTRE", "GXD", "WINDOWS", "CITRIX",
+}
+
+
+def _normalize(etype: str, value: str) -> str:
+    """Clé de cohérence : même entité -> même token."""
+    if etype in ("IPP", "NIR", "TEL"):
+        return re.sub(r"\s+", "", value)
+    if etype == "EMAIL":
+        return value.lower()
+    return re.sub(r"\s+", " ", value).strip().upper()
+
+
+def _is_blacklisted_name(value: str) -> bool:
+    toks = [t for t in re.split(r"[^\wÀ-ÿ]+", value) if t]
+    return bool(toks) and all(t.upper() in _SOFTWARE_BLACKLIST for t in toks)
+
+
+def _assign_token(mapping: Dict, etype: str, norm: str) -> str:
+    key = (etype, norm)
+    if key in mapping:
+        return mapping[key]
+    n = 1 + sum(1 for k in mapping if isinstance(k, tuple) and k[0] == etype)
+    token = f"[{etype}_{n}]"
+    mapping[key] = token
+    return token
+
+
+def anonymize_text(
+    text: str, *, mapping: Optional[Dict] = None
+) -> Tuple[str, List[Dict]]:
+    """Remplace la PII de `text` par des tokens typés cohérents.
+
+    `mapping` : table de cohérence partagée (ex. à l'échelle d'une session) —
+    la même valeur PII reçoit le même token d'un appel à l'autre. Mutée en place ;
+    si None, une table locale est utilisée.
+
+    Retourne `(texte_assaini, entités)` où chaque entité =
+    `{"type", "original", "token", "start", "end"}` (positions dans le texte source).
+    """
+    if not text:
+        return text, []
+    if mapping is None:
+        mapping = {}
+
+    # 1) collecte des candidats (start, end, type, valeur)
+    spans: List[Tuple[int, int, str, str]] = []
+    for pattern, etype, group in _DETECTORS:
+        for m in pattern.finditer(text):
+            start, end = m.span(group)
+            if start == end:
+                continue
+            value = m.group(group)
+            if etype == "NOM" and _is_blacklisted_name(value):
+                continue
+            spans.append((start, end, etype, value))
+
+    # 2) résolution des chevauchements (priorité = rang détecteur, puis -longueur)
+    # _DETECTORS est ordonné par priorité ; le rang dans cette liste détermine
+    # qui gagne quand deux patterns chevauchent. Plus prioritaire + plus long
+    # = accepté en premier, les plus courts/moins prioritaires sont éliminés.
+    # Fix FN « Dossier VIOLA (VIOLA) Liliane » : RE_PRENOM_NOM captait
+    # « Dossier VIOLA » (rang 2) et bloquait RE_NOM_NAISSANCE « VIOLA (VIOLA)
+    # Liliane » (rang 0, plus prioritaire et plus long).
+    det_rank = {p: i for i, (p, _, _) in enumerate(_DETECTORS)}
+    spans.sort(key=lambda s: (det_rank.get(s[2], 999), -(s[1] - s[0]), s[0]))
+    occupied: List[Tuple[int, int]] = []
+    accepted: List[Tuple[int, int, str, str]] = []
+    for start, end, etype, value in spans:
+        if all(start >= oe or end <= os for os, oe in occupied):
+            accepted.append((start, end, etype, value))
+            occupied.append((start, end))
+
+    # 3) substitution (de droite à gauche pour préserver les indices)
+    entities: List[Dict] = []
+    out = text
+    for start, end, etype, value in sorted(accepted, key=lambda s: s[0], reverse=True):
+        token = _assign_token(mapping, etype, _normalize(etype, value))
+        out = out[:start] + token + out[end:]
+        entities.append(
+            {"type": etype, "original": value, "token": token, "start": start, "end": end}
+        )
+    entities.reverse()
+    return out, entities
+
+
+# Clés portant un titre de fenêtre, où qu'elles soient imbriquées dans l'event
+# (top-level `active_window_title`, `window/to/from.title`, et surtout
+# `vision_info.window_capture.window_title` — blind spot signalé par Qwen).
+_TITLE_KEYS = ("title", "window_title", "active_window_title")
+_PLACEHOLDER_SAISIE = "[SAISIE]"
+
+
+def _walk_titles(obj, mapping: Dict) -> None:
+    """Parcourt récursivement l'event et assainit toute valeur de titre de fenêtre."""
+    if isinstance(obj, dict):
+        for k, v in obj.items():
+            if k in _TITLE_KEYS and isinstance(v, str):
+                obj[k] = anonymize_text(v, mapping=mapping)[0]
+            else:
+                _walk_titles(v, mapping)
+    elif isinstance(obj, list):
+        for item in obj:
+            _walk_titles(item, mapping)
+
+
+def sanitize_event(event: Dict, *, mapping: Optional[Dict] = None) -> Dict:
+    """Assainit un event capturé avant persistance (copie, ne mute pas l'original).
+
+    Principe « Léa apprend l'interface, pas la donnée » (décision Dom 28/06) :
+    - `text_input` : le **contenu tapé** (`text`, `raw_keys`) = donnée de santé →
+      remplacé par `[SAISIE]` (on garde le champ, pas la valeur — option b) ;
+    - **titres de fenêtre** (`active_window_title`, et `title` dans `window`/`to`/
+      `from`) : l'**identité patient** est tokenisée, l'app/écran est gardé
+      (contexte d'apprentissage), via `anonymize_text` + `mapping` partagé (cohérence).
+    """
+    if mapping is None:
+        mapping = {}
+    ev = copy.deepcopy(event)
+
+    # text_input : on ne garde pas le contenu
+    if ev.get("type") == "text_input":
+        for k in ("text", "raw_keys"):
+            if ev.get(k) not in (None, ""):
+                ev[k] = _PLACEHOLDER_SAISIE
+
+    # tous les titres de fenêtre, où qu'ils soient imbriqués
+    # (active_window_title, window/to/from.title, vision_info.window_capture.window_title…)
+    _walk_titles(ev, mapping)
+
+    return ev
+
+
+# Clés d'un workflow core portant du texte potentiellement PII : cible OCR
+# (`by_text`), noms d'écrans/labels dérivés des titres. Le contenu saisi est
+# déjà neutralisé à la source (sanitize_event → [SAISIE]).
+_WORKFLOW_TEXT_KEYS = ("by_text", "name", "label")
+
+
+def _walk_workflow_text(obj, mapping: Dict) -> None:
+    """Parcourt un workflow core et tokenise la PII des champs texte (cibles, noms)."""
+    if isinstance(obj, dict):
+        for k, v in obj.items():
+            if k in _WORKFLOW_TEXT_KEYS and isinstance(v, str) and v:
+                obj[k] = anonymize_text(v, mapping=mapping)[0]
+            else:
+                _walk_workflow_text(v, mapping)
+    elif isinstance(obj, list):
+        for item in obj:
+            _walk_workflow_text(item, mapping)
+
+
+def sanitize_workflow_dict(workflow_dict: Dict, *, mapping: Optional[Dict] = None) -> Dict:
+    """Assainit un workflow core (JSON appris) avant import/persistance en DB VWB.
+
+    Tokenise la PII des champs texte (cible OCR `by_text`, noms d'écrans, labels)
+    via `anonymize_text`, en gardant l'interface intacte (« Léa apprend
+    l'interface, pas la donnée »). Copie — l'original n'est pas muté.
+
+    Limite (couche 1) : ne capte que la PII structurée (IPP, NOM clinique…) ;
+    les noms libres relèvent de la couche 2 NER.
+    """
+    if mapping is None:
+        mapping = {}
+    wf = copy.deepcopy(workflow_dict)
+    _walk_workflow_text(wf, mapping)
+    return wf

agent_v0/server_v1/replay_engine.py

@@ -40,6 +40,7 @@ _ALLOWED_ACTION_TYPES = {
     "pause_for_human",   # Pause supervisée explicite (interceptée par /replay/next)
     "extract_text",      # OCR serveur sur dernier heartbeat → variable workflow
     "extract_table",     # OCR serveur + filtre regex → liste structurée (boucle)
+    "extract_dossier",   # OCR grille structurée → dossier patient persisté (brique 3)
     "extract_text_scroll", # Marker côté graphe — expansé en sous-actions par _edge_to_normalized_actions
     "_concat_text_vars",   # Action serveur interne (générée par expansion extract_text_scroll)
     "t2a_decision",      # Analyse LLM facturation T2A → variable workflow
@@ -53,6 +54,7 @@ _ALLOWED_ACTION_TYPES = {
 _SERVER_SIDE_ACTION_TYPES = {
     "extract_text",
     "extract_table",
+    "extract_dossier",
     "t2a_decision",
     "llm_generate",
     "_concat_text_vars",
@@ -2216,6 +2218,146 @@ def _handle_extract_table_action(
     return bool(rows)
 
 
+def _resolve_screenshot_path(replay_state: Dict[str, Any]) -> Optional[str]:
+    """Résout le chemin du dernier screenshot (path disque ou base64 → temp).
+
+    Calque la source utilisée par extract_text/extract_table : priorité au
+    ``last_screenshot`` (path ou data-URI base64). Retourne None si absent.
+    """
+    raw_screenshot = replay_state.get("last_screenshot") or ""
+    if not raw_screenshot:
+        return None
+    if raw_screenshot.startswith("data:"):
+        try:
+            import base64 as _b64, tempfile
+            header, b64data = raw_screenshot.split(",", 1)
+            suffix = ".jpg" if "jpeg" in header else ".png"
+            tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
+            tmp.write(_b64.b64decode(b64data))
+            tmp.close()
+            return tmp.name
+        except Exception as e:
+            logger.warning("extract_dossier: décodage base64 screenshot échoué: %s", e)
+            return None
+    if os.path.isfile(raw_screenshot):
+        return raw_screenshot
+    return None
+
+
+def _gate_dossier_quality(
+    grid: List[List[Dict[str, Any]]],
+    *,
+    min_confidence: float,
+    expected_cols: Optional[int],
+) -> str:
+    """Gate qualité simple → 'complete' ou 'needs_review'.
+
+    'complete' SSI : grille non vide ET confiance médiane ≥ seuil ET (si
+    expected_cols fourni) au moins une ligne avec ce nombre de colonnes.
+    Sinon 'needs_review'. Volontairement conservatrice (default-review).
+    """
+    confs = [
+        cell.get("confidence")
+        for row in grid for cell in row
+        if isinstance(cell.get("confidence"), (int, float))
+    ]
+    if not confs:
+        return "needs_review"
+    confs.sort()
+    median = confs[len(confs) // 2]
+    if median < min_confidence:
+        return "needs_review"
+    if expected_cols is not None:
+        if not any(len(row) == expected_cols for row in grid):
+            return "needs_review"
+    return "complete"
+
+
+def _handle_extract_dossier_action(
+    action: Dict[str, Any],
+    replay_state: Dict[str, Any],
+    session_id: str,
+) -> bool:
+    """Traite une action extract_dossier côté serveur (brique 3).
+
+    Lit le dernier screenshot, extrait une grille structurée via
+    ``extract_grid_from_image``, applique une gate qualité, puis PERSISTE un
+    « dossier patient extrait » (Job/Table/Field) dans la DB VWB avec preuve
+    (screenshot_ref + screen_bbox + confidences). Le job_id est stocké dans
+    ``replay_state["variables"][output_var]``.
+
+    Paramètres reconnus (action.parameters) :
+        output_var     : nom de variable runtime (default "extracted_dossier")
+        patient_ref    : référence patient EN CLAIR (volontaire) — non tokenisée
+        region         : (x, y, w, h) px pour cropper avant OCR (None = plein)
+        min_confidence : seuil de confiance médiane pour 'complete' (default 0.6)
+        expected_cols  : nb de colonnes attendu (optionnel) pour la gate
+
+    N'ÉCHOUE JAMAIS le replay : toute erreur → log + needs_review.
+    Retourne True SSI le dossier est persisté avec statut 'complete'.
+    """
+    params = action.get("parameters") or {}
+    output_var = (params.get("output_var") or params.get("variable_name") or "extracted_dossier").strip()
+    patient_ref = params.get("patient_ref")
+    region = params.get("region") or None
+    try:
+        min_confidence = float(params.get("min_confidence", 0.6))
+    except (TypeError, ValueError):
+        min_confidence = 0.6
+    expected_cols = params.get("expected_cols")
+    if isinstance(expected_cols, str):
+        try:
+            expected_cols = int(expected_cols)
+        except ValueError:
+            expected_cols = None
+
+    job_id = ""
+    status = "needs_review"
+    try:
+        path = _resolve_screenshot_path(replay_state)
+        grid: List[List[Dict[str, Any]]] = []
+        if path:
+            from core.llm import extract_grid_from_image
+            grid = extract_grid_from_image(
+                path, region=tuple(region) if region else None
+            )
+        else:
+            logger.warning(
+                "extract_dossier : pas de screenshot pour session %s — needs_review",
+                session_id,
+            )
+
+        status = _gate_dossier_quality(
+            grid, min_confidence=min_confidence, expected_cols=expected_cols
+        )
+
+        from . import vwb_db
+        with vwb_db.vwb_app_context():
+            job_id = vwb_db.persist_extracted_dossier(
+                grid,
+                patient_ref=patient_ref,
+                source_session_id=session_id,
+                screenshot_ref=path,
+                screen_bbox=({"x": region[0], "y": region[1], "width": region[2], "height": region[3]}
+                             if region and len(region) == 4 else None),
+                status=status,
+            )
+    except Exception as e:
+        # Ne JAMAIS échouer le replay : on log, on marque needs_review.
+        logger.warning(
+            "extract_dossier : échec persistance (%s) — needs_review, replay %s",
+            e, replay_state.get("replay_id", "?"),
+        )
+        status = "needs_review"
+
+    replay_state.setdefault("variables", {})[output_var] = job_id
+    logger.info(
+        "extract_dossier → variable '%s' job=%s statut=%s replay %s",
+        output_var, job_id or "?", status, replay_state.get("replay_id", "?"),
+    )
+    return status == "complete"
+
+
 def _handle_t2a_decision_action(
     action: Dict[str, Any],
     replay_state: Dict[str, Any],

agent_v0/server_v1/stream_processor.py

@@ -3066,6 +3066,8 @@ class StreamProcessor:
             saved_path = self._persist_workflow(workflow, session_id, machine_id=machine_id)
             # Stocker le machine_id dans le workflow pour le filtrage
             workflow._machine_id = machine_id
+            # R1 : import auto en DB VWB (rejouable) — gated RPA_R1_AUTO_IMPORT, non bloquant.
+            self._maybe_import_to_vwb(workflow, session_id, machine_id)
 
             # Récupérer les métadonnées applicatives de la session
             session_state = self.session_manager.get_session(session_id)
@@ -4444,6 +4446,45 @@ class StreamProcessor:
             logger.error(f"Erreur sauvegarde workflow {session_id}: {e}")
             return None
 
+    def _import_workflow_to_vwb(self, workflow, session_id: str, machine_id: str) -> Dict[str, Any]:
+        """Importer le workflow appris dans la DB VWB rejouable (Maillon A / R1).
+
+        Rend l'appris rejouable sans geste manuel, de façon idempotente (fusion
+        par signature de trajectoire). Suppose un app-context VWB actif fournissant
+        ``db.session`` (créé par l'appelant côté worker).
+        """
+        from .pii_sanitizer import sanitize_workflow_dict
+        from services.learned_workflow_bridge import import_core_workflow_to_db
+        from db.models import db
+        # Assainir la PII (cibles OCR `by_text`, noms) avant dépôt en DB VWB.
+        core_dict = sanitize_workflow_dict(workflow.to_dict())
+        return import_core_workflow_to_db(
+            core_dict,
+            machine_id=machine_id,
+            source_session_id=session_id,
+            db_session=db.session,
+        )
+
+    def _vwb_app_context(self):
+        """Couplage worker→DB VWB mutualisé (un seul pont, cf. vwb_db).
+
+        Délègue au helper module ``vwb_db.vwb_app_context`` partagé entre R1 et
+        l'extraction métier — pas de duplication de l'app Flask/init_app.
+        """
+        from .vwb_db import vwb_app_context
+        return vwb_app_context()
+
+    def _maybe_import_to_vwb(self, workflow, session_id: str, machine_id: str) -> None:
+        """Import auto de l'appris en DB VWB, gated par RPA_R1_AUTO_IMPORT (OFF
+        par défaut) et NON bloquant : un échec ne casse jamais la finalisation."""
+        if os.environ.get("RPA_R1_AUTO_IMPORT", "false").lower() not in ("true", "1", "yes"):
+            return
+        try:
+            with self._vwb_app_context():
+                self._import_workflow_to_vwb(workflow, session_id, machine_id)
+        except Exception as e:
+            logger.warning("[R1] import VWB auto échoué (non bloquant): %s", e)
+
     def _build_raw_session_fallback(self, session, raw_dict):
         """Construire un RawSession manuellement si from_dict échoue."""
         from core.models.raw_session import RawSession, Event, Screenshot, RawWindowContext

agent_v0/server_v1/vwb_db.py

@@ -0,0 +1,106 @@
+"""Couplage worker → DB VWB (mutualisé) + persistance « dossier patient extrait ».
+
+Le worker/serveur streaming est un process distinct du backend VWB : il n'a
+pas d'app Flask en mémoire. Ce module fournit :
+
+- ``vwb_app_context()`` : un app-context Flask lazy (singleton module) lié au
+  fichier SQLite VWB ``visual_workflow_builder/backend/instance/workflows.db``,
+  avec ``db.init_app`` (db de ``db.models``). Réutilisable par tout module
+  serveur qui doit écrire dans la DB VWB (R1, extraction métier, …).
+
+- ``persist_extracted_dossier(...)`` : depuis une grille OCR
+  (``List[List[cell]]``), crée ExtractionJob → ExtractedTable → ExtractedField
+  et commit. Suppose un app-context actif (comme le pont R1 existant).
+
+⚠️ CANAL EXTRACTION = données patient EN CLAIR (volontaire) : aucune
+tokenisation/assainissement PII ici (cf. note dans db/models.py).
+"""
+
+import sys
+import uuid
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+# Ajout du backend VWB au sys.path à l'import → rend ``db.models`` importable
+# (couplage worker→DB VWB mutualisé ; identique au pattern stream_processor).
+_VWB_BACKEND = Path(__file__).resolve().parents[2] / "visual_workflow_builder" / "backend"
+if str(_VWB_BACKEND) not in sys.path:
+    sys.path.insert(0, str(_VWB_BACKEND))
+
+# App Flask lazy (singleton module) — un seul db.init_app pour tout le process.
+_vwb_app = None
+
+
+@contextmanager
+def vwb_app_context():
+    """App-context Flask VWB (lazy singleton) sur instance/workflows.db.
+
+    À utiliser via ``with vwb_app_context(): ...`` autour des appels qui
+    nécessitent ``db.session`` (ex. persist_extracted_dossier).
+    """
+    global _vwb_app
+    if _vwb_app is None:
+        from flask import Flask
+        from db.models import db
+
+        db_path = _VWB_BACKEND / "instance" / "workflows.db"
+        app = Flask("worker_vwb")
+        app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}"
+        app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
+        db.init_app(app)
+        _vwb_app = app
+    with _vwb_app.app_context():
+        yield
+
+
+def persist_extracted_dossier(
+    grid: List[List[Dict[str, Any]]],
+    *,
+    patient_ref: Optional[str],
+    source_session_id: Optional[str],
+    screenshot_ref: Optional[str],
+    screen_bbox: Optional[Dict[str, Any]],
+    status: str,
+) -> str:
+    """Persiste un « dossier patient extrait » et retourne le job_id.
+
+    Crée 1 ExtractionJob → 1 ExtractedTable → N ExtractedField (une par
+    cellule de la grille), puis commit. Suppose un app-context VWB actif
+    (fourni par ``vwb_app_context()`` ou par l'appelant, comme le pont R1).
+
+    ⚠️ ``patient_ref`` et ``cell["text"]`` sont stockés EN CLAIR (volontaire) :
+    le but est de constituer le dossier, pas d'anonymiser.
+    """
+    from db.models import db, ExtractionJob, ExtractedTable, ExtractedField
+
+    job = ExtractionJob(
+        id=uuid.uuid4().hex,
+        patient_ref=patient_ref,
+        source_session_id=source_session_id,
+        status=status,
+    )
+    db.session.add(job)
+
+    table = ExtractedTable(
+        id=uuid.uuid4().hex,
+        job_id=job.id,
+        screen_bbox=screen_bbox,
+        screenshot_ref=screenshot_ref,
+    )
+    db.session.add(table)
+
+    for row in grid or []:
+        for cell in row or []:
+            db.session.add(ExtractedField(
+                id=uuid.uuid4().hex,
+                table_id=table.id,
+                row=cell.get("row"),
+                col=cell.get("col"),
+                value=cell.get("text"),
+                bbox=cell.get("bbox"),
+                confidence=cell.get("confidence"),
+            ))
+
+    db.session.commit()
+    return job.id

core/execution/trajectory_signature.py

@@ -0,0 +1,156 @@
+"""Signature de trajectoire — identité stable d'un parcours appris (décision F1).
+
+Une trajectoire = séquence ordonnée d'actions sur des cibles stables. La signature
+hashe uniquement `(action_type, target)` de chaque étape, dans l'ordre, en **ignorant
+les champs session-spécifiques** (IDs de nœuds, timestamps, coordonnées). Deux
+apprentissages du même parcours produisent donc la même signature → create-or-update.
+
+Primitive partagée (Phase 0) : consommée par SP-4 (dédup/persist), SP-2 (rejeu) et le
+cycle compétences (dédup des skills). Pour composer avec un descripteur d'écran stable,
+passer `core.execution.screen_signature.screen_signature(...)` comme valeur de `target`.
+"""
+
+import hashlib
+import re
+import unicodedata
+from typing import Any, Iterable, Mapping
+
+_FIELD_SEP = "\x1f"  # sépare action_type et target dans une étape
+_STEP_SEP = "\x1e"   # sépare les étapes
+
+# --- Cible stable : anonymisation PII + normalisation déterministes ----------
+# Verdict QG Qwen (2026-06-25) : regex DÉDIÉES à la signature (PAS `pii_blur`,
+# qui protège les dates alors qu'ici on les NEUTRALISE), PAS de NER (un hash
+# d'identité doit être déterministe et identique labo↔DGX, donc indépendant
+# d'un modèle versionné). Les noms propres sans titre ne sont pas neutralisés
+# ici (stratégie « (b) » : impact 0 sur l'audit labo ; gate = audit agrégat
+# `by_text` DGX avant prod, ajouter une regex ciblée si des noms apparaissent).
+_WS_RE = re.compile(r"\s+")
+# Ordre d'application : motifs structurés d'abord, identifiant numérique long
+# en dernier (sinon il mangerait des fragments de date/téléphone).
+_RE_EMAIL = re.compile(r"\b[\w.%+-]+@[\w.-]+\.[A-Za-z]{2,}\b")
+_RE_DATE = re.compile(r"\b\d{1,4}[/.\-]\d{1,2}[/.\-]\d{1,4}\b")
+_RE_PHONE = re.compile(r"\b(?:\+?33|0)\s?[1-9](?:[\s.\-]?\d{2}){4}\b")
+_RE_LONGNUM = re.compile(r"\d{6,}")  # IPP / NIR collé / autre identifiant long
+
+
+def _anonymize_pii(text: str) -> str:
+    """Neutralise la PII structurée par des tokens stables : deux sessions sur le
+    même champ (patients/dates différents) → même texte cible → même signature."""
+    text = _RE_EMAIL.sub("[email]", text)
+    text = _RE_DATE.sub("[date]", text)
+    text = _RE_PHONE.sub("[tel]", text)
+    text = _RE_LONGNUM.sub("[ipp]", text)
+    return text
+
+
+def _norm_text(text: str) -> str:
+    """Normalisation déterministe (même logique que `action_executor._norm_text`,
+    redéfinie ici pour garder ce module léger et sans effet de bord d'import) :
+    minuscules, suppression des accents (NFKD), espaces normalisés."""
+    if not text:
+        return ""
+    text = text.replace(" ", " ").strip().lower()
+    text = unicodedata.normalize("NFKD", text)
+    text = "".join(ch for ch in text if not unicodedata.combining(ch))
+    return _WS_RE.sub(" ", text).strip()
+
+
+def _normalize_target(target: str) -> str:
+    """Cible stable : PII neutralisée PUIS normalisée (casse/accents/espaces)."""
+    return _norm_text(_anonymize_pii(target))
+
+
+def _normalize_step(step: Mapping[str, Any]) -> str:
+    action_type = str(step.get("action_type", "unknown")).strip().lower()
+    target = _normalize_target(str(step.get("target", "")))
+    return f"{action_type}{_FIELD_SEP}{target}"
+
+
+def trajectory_signature(steps: Iterable[Mapping[str, Any]]) -> str:
+    """Retourne la signature SHA-256 (hex, 64 car.) d'une séquence d'étapes.
+
+    Chaque étape est un mapping ; seuls `action_type` et `target` sont pris en compte.
+    Tous les autres champs (node_id, timestamp, coordonnées…) sont ignorés afin de
+    garantir la stabilité de la signature entre deux sessions du même parcours.
+    """
+    canonical = _STEP_SEP.join(_normalize_step(step) for step in steps)
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+# ---------------------------------------------------------------------------
+# Adaptateur : workflow core (dict) → signature de trajectoire
+# ---------------------------------------------------------------------------
+
+def _stable_target(target: Any) -> str:
+    """Descripteur de cible **stable** entre sessions.
+
+    S'appuie sur le texte sémantique de la cible (`by_text`), volontairement
+    indépendant du moteur de grounding : `by_role` peut valoir 'yolo'/'ocr'/'vlm'
+    (méthode de détection, instable entre sessions) et n'entre donc PAS dans la
+    signature. Fallback quand `by_text` est absent : titre de fenêtre / description VLM.
+    """
+    if not isinstance(target, Mapping):
+        return ""
+    by_text = str(target.get("by_text") or "").strip()
+    if by_text:
+        return by_text
+    hints = target.get("context_hints")
+    if isinstance(hints, Mapping):
+        return str(hints.get("window_title") or hints.get("vlm_description") or "").strip()
+    return ""
+
+
+def _ordered_edges(workflow: Mapping[str, Any]) -> list:
+    """Edges dans l'ordre du parcours (BFS depuis entry_nodes), comme le bridge d'import."""
+    edges = list(workflow.get("edges") or [])
+    if not edges:
+        return []
+    by_from: dict = {}
+    for edge in edges:
+        by_from.setdefault((edge or {}).get("from_node"), []).append(edge)
+    entry = list(workflow.get("entry_nodes") or [])
+    nodes = workflow.get("nodes") or []
+    if not entry and nodes:
+        entry = [(nodes[0] or {}).get("node_id")]
+    if not entry:
+        return edges  # pas de point d'entrée : ordre brut de la liste
+    ordered: list = []
+    seen_edges: set = set()
+    visited: set = set()
+    queue = list(entry)
+    while queue:
+        node = queue.pop(0)
+        if node in visited:
+            continue
+        visited.add(node)
+        for edge in by_from.get(node, []):
+            key = id(edge)
+            if key in seen_edges:
+                continue
+            seen_edges.add(key)
+            ordered.append(edge)
+            to_node = (edge or {}).get("to_node")
+            if to_node and to_node not in visited:
+                queue.append(to_node)
+    for edge in edges:  # edges non atteints : ajout déterministe en fin
+        if id(edge) not in seen_edges:
+            ordered.append(edge)
+    return ordered
+
+
+def workflow_step_descriptors(workflow: Mapping[str, Any]) -> list:
+    """Séquence ordonnée de descripteurs `(action_type, target stable)` d'un workflow core."""
+    descriptors: list = []
+    for edge in _ordered_edges(workflow):
+        action = (edge or {}).get("action") or {}
+        descriptors.append({
+            "action_type": action.get("type", "unknown"),
+            "target": _stable_target(action.get("target")),
+        })
+    return descriptors
+
+
+def workflow_trajectory_signature(workflow: Mapping[str, Any]) -> str:
+    """Signature de trajectoire d'un workflow core (dict). Cf. `trajectory_signature`."""
+    return trajectory_signature(workflow_step_descriptors(workflow))

core/llm/__init__.py

@@ -8,6 +8,7 @@ from .t2a_decision import (
 )
 from .ocr_extractor import (
     extract_digits_tesseract_from_image,
+    extract_grid_from_image,
     extract_table_from_image,
     extract_text_from_image,
 )
@@ -19,5 +20,6 @@ __all__ = [
     "build_dpi_enriched",
     "extract_text_from_image",
     "extract_table_from_image",
+    "extract_grid_from_image",
     "extract_digits_tesseract_from_image",
 ]

core/llm/ocr_extractor.py

@@ -243,3 +243,107 @@ def extract_table_from_image(
     except Exception as e:
         logger.warning("extract_table échoué sur %s : %s", image_path, e)
         return []
+
+
+def _cluster_1d(centers: List[float], tol: float) -> List[Tuple[float, int]]:
+    """Regroupe des positions 1D par proximité (centres triés, gap > tol = nouveau cluster).
+
+    Retourne, pour chaque centre d'entrée (ordre d'origine), un couple
+    (centre_du_cluster, index_du_cluster), les clusters étant indexés dans
+    l'ordre croissant. Permet de mapper lignes (y) et colonnes (x).
+    """
+    order = sorted(range(len(centers)), key=lambda i: centers[i])
+    cluster_of = [0] * len(centers)
+    cluster_centers: List[List[float]] = []
+    prev = None
+    idx = -1
+    for i in order:
+        c = centers[i]
+        if prev is None or (c - prev) > tol:
+            idx += 1
+            cluster_centers.append([])
+        cluster_centers[idx].append(c)
+        cluster_of[i] = idx
+        prev = c
+    means = [sum(g) / len(g) for g in cluster_centers]
+    return [(means[cluster_of[i]], cluster_of[i]) for i in range(len(centers))]
+
+
+def extract_grid_from_image(
+    image_path: str,
+    region: Optional[Tuple[int, int, int, int]] = None,
+    row_tol: float = 12.0,
+    col_tol: float = 25.0,
+) -> List[List[dict]]:
+    """Extrait un tableau STRUCTURÉ (lignes ET colonnes) via OCR EasyOCR.
+
+    Contrairement à `extract_table_from_image` (liste plate triée par y, x jeté),
+    on conserve la coordonnée x pour reconstruire une grille. Clustering :
+    lignes par proximité du centre y, colonnes par proximité du centre x.
+
+    Args:
+        image_path: chemin du PNG sur disque.
+        region: (x, y, w, h) pour cropper avant OCR. None = image entière.
+        row_tol: écart vertical max (px) entre 2 tokens d'une même ligne.
+        col_tol: écart horizontal max (px) entre 2 tokens d'une même colonne.
+
+    Returns:
+        Grille `List[List[cell]]`, lignes top→bottom, colonnes left→right.
+        `cell = {"text", "bbox", "confidence", "row", "col"}`.
+        En cas d'erreur ou d'absence de tokens, retourne [].
+    """
+    path = Path(image_path)
+    if not path.exists():
+        logger.warning("extract_grid: fichier introuvable %s", image_path)
+        return []
+
+    try:
+        from PIL import Image
+        import numpy as np
+
+        img = Image.open(path)
+        if region:
+            x, y, w, h = region
+            img = img.crop((x, y, x + w, y + h))
+
+        reader = _get_reader()
+        results = reader.readtext(np.array(img), detail=1, paragraph=False)
+
+        toks = []
+        for bbox, text, conf in results:
+            t = str(text).strip()
+            if not t:
+                continue
+            xs = [p[0] for p in bbox]
+            ys = [p[1] for p in bbox]
+            toks.append({
+                "text": t,
+                "bbox": bbox,
+                "confidence": conf,
+                "xc": sum(xs) / len(xs),
+                "yc": sum(ys) / len(ys),
+            })
+        if not toks:
+            return []
+
+        rows_cl = _cluster_1d([tk["yc"] for tk in toks], row_tol)
+        cols_cl = _cluster_1d([tk["xc"] for tk in toks], col_tol)
+        for tk, (_yc, r), (_xc, c) in zip(toks, rows_cl, cols_cl):
+            tk["row"], tk["col"] = r, c
+
+        n_rows = max(tk["row"] for tk in toks) + 1
+        grid: List[List[dict]] = [[] for _ in range(n_rows)]
+        for tk in toks:
+            grid[tk["row"]].append({
+                "text": tk["text"],
+                "bbox": tk["bbox"],
+                "confidence": tk["confidence"],
+                "row": tk["row"],
+                "col": tk["col"],
+            })
+        for row in grid:
+            row.sort(key=lambda cell: cell["col"])
+        return grid
+    except Exception as e:
+        logger.warning("extract_grid échoué sur %s : %s", image_path, e)
+        return []

core/models/workflow_graph.py

@@ -1250,12 +1250,16 @@ class Workflow:
         }
         if self.chain_config:
             result["chain_config"] = self.chain_config.to_dict() if hasattr(self.chain_config, 'to_dict') else self.chain_config
+        # machine_id : attribut d'instance posé au runtime (pas un champ dataclass)
+        machine_id = getattr(self, "_machine_id", None)
+        if machine_id:
+            result["machine_id"] = machine_id
         return result
     
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> 'Workflow':
         """Désérialiser depuis JSON"""
-        return cls(
+        wf = cls(
             workflow_id=data["workflow_id"],
             name=data.get("name", data["workflow_id"]),
             description=data.get("description", ""),
@@ -1277,6 +1281,12 @@ class Workflow:
             references=data.get("references", []),
             chain_config=data.get("chain_config")
         )
+        # Reposer machine_id (attribut d'instance) : priorité au champ explicite,
+        # sinon depuis metadata['machine_id'] (rétrocompat des workflows déjà sur disque)
+        machine_id = data.get("machine_id") or (wf.metadata or {}).get("machine_id")
+        if machine_id:
+            wf._machine_id = machine_id
+        return wf
 
     def to_json(self) -> str:
         """Sérialiser en JSON string"""

docs/DETTE_TECHNIQUE.md

@@ -35,6 +35,9 @@ P0 / P1 / P2 / P3 (alignées sur convention handoffs)
 | DETTE-017 | 2026-06-12 | 2026-06-12 | P0 | OPEN | Auth Bearer **désactivée** (`RPA_AUTH_DISABLED=true`) sur streaming `5005` ET agent-chat `5004` du DGX, appliquée comme « fix » heartbeat B3 (rustine). Démontré inutile : les 3 tokens (DGX proc, DGX `.env.local`, Windows `.env`) sont identiques (SHA256 `43749362b1`, len 43) → l'auth peut être réactivée sans casser le heartbeat. Exposition `0.0.0.0:5004/5005` restreinte par iptables au seul poste `192.168.1.11` ; dashboard `5001` conserve son auth. **Exception temporaire validée par Dom (2026-06-12 09:35) pour test M2 local sur données factices.** ROLLBACK OBLIGATOIRE avant toute sortie clinique / données patient : `RPA_AUTH_DISABLED=false` dans `.env.local` DGX + `sudo systemctl restart rpa-streaming.service rpa-agent-chat.service` puis vérif (401 sans token / 200 avec / heartbeat maintenu). | docs/coordination/active/2026-06-12_0935_decision-dom-auth-off-exception-m2.md + alerte 2026-06-11_1535 |
 | DETTE-018 | 2026-06-13 | 2026-06-27 | P2 | OPEN | Garde-seuil inopérant sur le chemin grounding **legacy** : `_resolve_by_grounding` retourne `method="grounding_vlm"` (resolve_engine.py:1121, mode `RPA_GROUNDING_ENGINE` OFF), clé absente de `_RESOLUTION_MIN_SCORES` qui ne traite en **préfixe** que `memory_` (toutes les autres clés = match exact) → le Check-1 du validateur (seuil min de confiance) ne s'applique jamais à ce chemin. Le mode `qwen3vl_vllm` est lui correctement gardé (`method="grounding"`, clé exacte, seuil 0.60). Aligner le legacy (clé gardée ou renommage) tant que le mode legacy reste activable. | Découvert au câblage qwen3vl (commit 5c5ce747b) + validation E2E 2026-06-13 |
 | DETTE-019 | 2026-06-13 | 2026-06-27 | P2 | OPEN | Confiance grounding **figée à `0.85` en dur** dans le `return` de `_resolve_by_grounding` (resolve_engine.py:1128-1130 : `matched_element.confidence` et `score`), pour les DEUX modes (legacy et qwen3vl). Le garde-seuil (0.60) reçoit donc toujours 0.85 quel que soit le grounding réel → le filtre ne discrimine jamais la vraie qualité de localisation. Propager une confiance réelle (signal modèle/cascade) pour rendre le seuil opérant. | Découvert au câblage qwen3vl (commit 5c5ce747b) + validation E2E 2026-06-13 |
+| DETTE-020 | 2026-06-25 | 2026-07-09 | P1 | OPEN | **Incidents silencieux — aucune détection/alerte des composants critiques d'inférence.** Un composant critique peut tomber sans alerte : `rpa-vllm-grounder.service` (grounder Qwen3-VL/vLLM) trouvé en **crash-loop (auto-restart, restart counter ×3960)** → le runtime a basculé **silencieusement** sur le fallback `qwen2.5vl:7b-rpa` (Ollama, ~×7 plus lent), avec une latence/contention accrue mais **aucune remontée visible** (ni dashboard, ni log d'alerte). Découvert uniquement par vérif manuelle au runtime (session 2026-06-25). La cause de CE crash (SSL HuggingFace au boot vs cache local — manque `HF_HUB_OFFLINE`) se corrige à part ; la dette ici = **le mode dégradé est silencieux**. Cible : health-check + supervision des composants critiques (grounder vLLM, Ollama, services `rpa-*`) avec **remontée VISIBLE** (dashboard 5001 / log d'alerte / notification) → une bascule en mode dégradé ne doit jamais passer inaperçue. ⚠️ Vérifier d'abord l'existant (module monitoring `:5003`) avant de construire. | session vérif runtime DGX clinique 2026-06-25 |
+| DETTE-021 | 2026-06-25 | 2026-07-09 | P1 | OPEN | **Journalisation client Léa non effective.** `LOG_FILE` (`agent_v0/agent_v1/config.py:88` → `<install>/logs/agent_v1.log`) est défini mais **jamais branché** : aucun `FileHandler`/`addHandler` dans tout le client. Seul logging actif = `basicConfig` (`main.py:46`) → **stderr**, perdu car Léa tourne en `pythonw.exe` (sans console). Dossier `logs/` vide. Conséquences : (1) **diagnostic terrain aveugle** — impossible de tracer pourquoi Léa « disparaît » côté poste ; (2) **non-conformité Règlement IA Art. 12** (journalisation + conservation 180 j — citée dans le code mais non effective ; `LOG_RETENTION_DAYS` ne couvre que les *sessions*). Cible : brancher un `RotatingFileHandler`/`TimedRotating` vers `LOG_FILE` (rotation + purge 180 j, niveau INFO). ⚠️ modif client → **redéploiement** (cf. DETTE-022). Pendant client du DETTE-020 (observabilité serveur). | session diagnostic « disparition » Léa poste Émilie 2026-06-25 |
+| DETTE-022 | 2026-06-25 | 2026-07-09 | P1 | OPEN | **Pas de mise à jour automatique du client Léa.** Toute modif du client (`agent_v0/agent_v1/**`) impose un **redéploiement manuel poste par poste** (Léa « gelée »). En clinique (5 postes, croissant), intervenir sur chaque poste à chaque correctif (ex. fix logging DETTE-021) **dérange les TIM et décourage l'adoption** (constat Dom). Cible : mécanisme de **MAJ auto / en tâche de fond** (auto-update silencieux, versionné, piloté serveur/dashboard, avec rollback), **zéro intervention sur le poste**. ⚠️ Vérifier d'abord l'existant côté enrôlement Fleet (dashboard build ZIP + token) avant de construire. | décision Dom 2026-06-25 (« on ne peut pas intervenir constamment sur les postes, on va décourager ») |
 
 ## Convention de référencement
 

tests/integration/test_agent_logs_api.py

@@ -0,0 +1,192 @@
+"""Tests d'intégration de l'endpoint POST /api/v1/agents/logs (push-log-DGX).
+
+Le client Léa pousse ses logs (batch JSON) vers le DGX ; le serveur les range
+par machine_id (AgentLogsStore) pour consultation au dashboard — diagnostic des
+postes sans AnyDesk. Mêmes garde-fous fleet que stream/poll (agent actif).
+
+Branche feat/push-log-dgx — DETTE-020/021.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+_ROOT = str(Path(__file__).resolve().parents[2])
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+
+_TEST_API_TOKEN = "test_token_logs_endpoint_0123456789abcdef"
+
+
+@pytest.fixture
+def logs_client(monkeypatch, tmp_path):
+    """Client FastAPI de test avec registre ET store de logs isolés sur disque."""
+    monkeypatch.setenv("RPA_API_TOKEN", _TEST_API_TOKEN)
+    monkeypatch.setenv("RPA_AGENTS_DB_PATH", str(tmp_path / "test_agents.db"))
+
+    from fastapi.testclient import TestClient
+    from agent_v0.server_v1 import api_stream
+    from agent_v0.server_v1.agent_registry import AgentRegistry
+    from agent_v0.server_v1.agent_logs_store import AgentLogsStore
+
+    monkeypatch.setattr(api_stream, "API_TOKEN", _TEST_API_TOKEN)
+    test_registry = AgentRegistry(db_path=str(tmp_path / "test_agents.db"))
+    monkeypatch.setattr(api_stream, "agent_registry", test_registry)
+    test_store = AgentLogsStore(base_dir=tmp_path / "agent_logs")
+    monkeypatch.setattr(api_stream, "agent_logs_store", test_store, raising=False)
+
+    client = TestClient(api_stream.app, raise_server_exceptions=False)
+    yield client, _TEST_API_TOKEN, test_store
+
+
+def _auth_headers(token: str) -> dict:
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _enroll(client, token, machine_id):
+    return client.post(
+        "/api/v1/agents/enroll",
+        json={"machine_id": machine_id, "user_name": machine_id},
+        headers=_auth_headers(token),
+    )
+
+
+def test_post_logs_persists_for_active_agent(logs_client):
+    client, token, store = logs_client
+    _enroll(client, token, "lea-emilie-001")
+
+    payload = {
+        "machine_id": "lea-emilie-001",
+        "logs": [
+            {"ts": "2026-06-26T16:00:00", "level": "WARNING",
+             "logger": "agent_v1.core.executor", "message": "popup detectee"},
+        ],
+    }
+    resp = client.post(
+        "/api/v1/agents/logs", json=payload, headers=_auth_headers(token)
+    )
+
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["received"] == 1
+    stored = store.read("lea-emilie-001")
+    assert len(stored) == 1
+    assert stored[0]["message"] == "popup detectee"
+    assert stored[0]["level"] == "WARNING"
+
+
+def test_post_logs_without_token_returns_401(logs_client):
+    client, _, _ = logs_client
+    resp = client.post(
+        "/api/v1/agents/logs", json={"machine_id": "lea-001", "logs": []}
+    )
+    assert resp.status_code == 401
+
+
+def test_post_logs_rejected_for_revoked_agent(logs_client):
+    """Un poste révoqué ne peut plus pousser de logs (même garde-fou que stream/poll)."""
+    client, token, store = logs_client
+    _enroll(client, token, "lea-revoked")
+    client.post(
+        "/api/v1/agents/uninstall",
+        json={"machine_id": "lea-revoked", "reason": "admin_revoke"},
+        headers=_auth_headers(token),
+    )
+
+    resp = client.post(
+        "/api/v1/agents/logs",
+        json={"machine_id": "lea-revoked", "logs": [{"message": "x"}]},
+        headers=_auth_headers(token),
+    )
+
+    assert resp.status_code == 403, resp.text
+    assert resp.json()["detail"]["error"] == "agent_not_active"
+    assert store.read("lea-revoked") == []  # rien persisté
+
+
+def test_post_logs_rejects_oversized_batch(logs_client):
+    """Anti-flood (G3) : un batch trop volumineux est rejeté (413), rien persisté."""
+    client, token, store = logs_client
+    _enroll(client, token, "lea-flood")
+    big = [{"level": "INFO", "message": f"l{i}"} for i in range(1001)]
+
+    resp = client.post(
+        "/api/v1/agents/logs",
+        json={"machine_id": "lea-flood", "logs": big},
+        headers=_auth_headers(token),
+    )
+
+    assert resp.status_code == 413, resp.text
+    assert store.read("lea-flood") == []
+
+
+# ---------------------------------------------------------------------------
+# Brique 3 — lecture des logs par machine_id (route dashboard, read-only).
+# Lecture admin/diagnostic : PAS de garde fleet (on veut justement pouvoir
+# consulter un poste révoqué ou en panne) ; seul le Bearer protège.
+# ---------------------------------------------------------------------------
+
+
+def test_get_logs_returns_persisted_for_machine(logs_client):
+    """GET /agents/logs/{machine_id} restitue les logs stockés, dans l'ordre."""
+    client, token, store = logs_client
+    store.append(
+        "lea-emilie-001",
+        [
+            {"ts": "2026-06-26T16:00:00", "level": "INFO", "message": "demarrage"},
+            {"ts": "2026-06-26T16:00:01", "level": "WARNING", "message": "popup"},
+        ],
+    )
+
+    resp = client.get(
+        "/api/v1/agents/logs/lea-emilie-001", headers=_auth_headers(token)
+    )
+
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["machine_id"] == "lea-emilie-001"
+    assert body["count"] == 2
+    assert body["total"] == 2
+    assert body["logs"][0]["message"] == "demarrage"
+    assert body["logs"][1]["level"] == "WARNING"
+
+
+def test_get_logs_without_token_returns_401(logs_client):
+    client, _, _ = logs_client
+    resp = client.get("/api/v1/agents/logs/lea-emilie-001")
+    assert resp.status_code == 401
+
+
+def test_get_logs_empty_for_unknown_machine(logs_client):
+    """Un poste sans log remonte une liste vide (200), pas une erreur."""
+    client, token, _ = logs_client
+    resp = client.get(
+        "/api/v1/agents/logs/lea-inconnu", headers=_auth_headers(token)
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["count"] == 0
+    assert body["total"] == 0
+    assert body["logs"] == []
+
+
+def test_get_logs_limit_returns_tail(logs_client):
+    """`limit` borne la réponse aux N entrées les plus récentes (tail)."""
+    client, token, store = logs_client
+    store.append(
+        "lea-tail",
+        [{"level": "INFO", "message": f"m{i}"} for i in range(5)],
+    )
+
+    resp = client.get(
+        "/api/v1/agents/logs/lea-tail?limit=2", headers=_auth_headers(token)
+    )
+
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["total"] == 5
+    assert body["count"] == 2
+    assert [e["message"] for e in body["logs"]] == ["m3", "m4"]

tests/integration/test_worker_imports_learned_workflow_to_vwb.py

@@ -0,0 +1,215 @@
+#!/usr/bin/env python3
+"""Test RED — Maillon A (R1) : câblage worker → DB VWB rejouable.
+
+Invariant ciblé (le VRAI trou du chantier apprentissage) :
+    quand le worker `finalize_session` produit un workflow appris, ce workflow
+    doit devenir **rejouable** en atterrissant dans la DB VWB, **sans geste
+    manuel** — et un 2e passage de la MÊME trajectoire ne crée PAS de doublon.
+
+État vérifié au moment d'écrire ce test :
+- le pont `import_core_workflow_to_db` (services.learned_workflow_bridge) EXISTE
+  et est vert en isolation (idempotence par signature de trajectoire) ;
+- MAIS le worker (`agent_v0/server_v1/stream_processor.py`) ne l'appelle JAMAIS :
+  `_persist_workflow` écrit le JSON sur disque, puis rien ne l'importe en DB VWB.
+  → les deux mondes (JSON appris ↔ DB VWB rejouable) restent disjoints.
+
+Ce test cible le **seam de câblage** manquant côté worker, sans exécuter le
+chemin lourd de `finalize_session` (GraphBuilder / CLIP) : il appelle la méthode
+de pont attendue `StreamProcessor._import_workflow_to_vwb(workflow, session_id,
+machine_id)`. Cette méthode N'EXISTE PAS encore → le test échoue (RED) pour la
+bonne raison : le câblage worker→VWB est absent.
+
+Câblage minimal proposé (NON appliqué ici) :
+    dans `finalize_session`, juste après `_persist_workflow` (≈ ligne 3066), ajouter
+        self._import_workflow_to_vwb(workflow, session_id, machine_id)
+    où `_import_workflow_to_vwb` :
+      1. sérialise `workflow.to_dict()` ;
+      2. ouvre un app-context VWB (db.session) ;
+      3. délègue à `import_core_workflow_to_db(core_dict, machine_id=...,
+         source_session_id=..., db_session=db.session)`.
+"""
+
+import sys
+from pathlib import Path
+
+import pytest
+from flask import Flask
+
+# --- Chemins : racine projet (core.*, agent_v0.*) + backend VWB (db.models, services.*) ---
+_ROOT = Path(__file__).resolve().parents[2]                 # .../rpa_vision_v3
+_BACKEND = _ROOT / "visual_workflow_builder" / "backend"
+for _p in (str(_ROOT), str(_BACKEND)):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+from db.models import db, Workflow  # noqa: E402  (modèles ORM VWB)
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def vwb_db_app():
+    """App Flask minimale liée à une SQLite VWB en mémoire (schéma créé)."""
+    app = Flask("test_worker_import_to_vwb")
+    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()
+
+
+class _FakeCoreWorkflow:
+    """Stub léger d'un workflow core produit par le worker.
+
+    Seul le **contrat** importe ici : le worker détient un objet exposant
+    `workflow_id` et `to_dict()` (cf. `core.models.workflow_graph.Workflow`,
+    déjà sérialisé par `_persist_workflow` via `save_to_file`). On reproduit ce
+    contrat sans dépendre du constructeur dataclass core (constraints/
+    post_conditions obligatoires) — la cible du test est le câblage, pas la
+    construction d'objet. Le dict renvoyé est exactement la forme que le pont
+    `convert_learned_to_vwb_steps` consomme (validé en isolation).
+    """
+
+    def __init__(self):
+        self.workflow_id = "wf_sess_bloc_notes_worker"
+
+    def to_dict(self):
+        return {
+            "workflow_id": self.workflow_id,
+            # Nom porteur de PII clinique : l'import en DB VWB doit l'assainir
+            # (logiciel métier réel en préfixe, nom clinique structuré ensuite).
+            "name": "Gxd5diag - VIOLA (VIOLA) Liliane",
+            "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"},
+                    },
+                },
+            ],
+        }
+
+
+def _build_core_workflow():
+    """Workflow core tel que vu par le worker (contrat `workflow_id` + `to_dict`)."""
+    return _FakeCoreWorkflow()
+
+
+def _make_processor():
+    """Instancie un StreamProcessor sans déclencher l'init lourde (CLIP/FAISS).
+
+    On crée l'objet via __new__ : le test n'exerce QUE la méthode de câblage,
+    pas le pipeline complet.
+    """
+    from agent_v0.server_v1.stream_processor import StreamProcessor
+    return StreamProcessor.__new__(StreamProcessor)
+
+
+# ---------------------------------------------------------------------------
+# Test RED — le câblage worker→VWB
+# ---------------------------------------------------------------------------
+
+def test_finalized_workflow_becomes_replayable_in_vwb_db(vwb_db_app):
+    """Un workflow appris par le worker devient rejouable en DB VWB,
+    et un 2e import de la même trajectoire ne crée pas de doublon (idempotence)."""
+    processor = _make_processor()
+    workflow = _build_core_workflow()
+
+    # --- Seam de câblage attendu (à implémenter côté worker) ---
+    # _import_workflow_to_vwb(workflow, session_id, machine_id) doit :
+    #   - sérialiser workflow.to_dict()
+    #   - importer en DB VWB via import_core_workflow_to_db (idempotent)
+    assert hasattr(processor, "_import_workflow_to_vwb"), (
+        "Câblage R1 absent : StreamProcessor n'expose pas de pont vers la DB VWB. "
+        "Le workflow appris reste sur disque (JSON) et n'est jamais rejouable."
+    )
+
+    with vwb_db_app.app_context():
+        first = processor._import_workflow_to_vwb(
+            workflow,
+            session_id="sess_bloc_notes_worker",
+            machine_id="DESKTOP-TEST_windows",
+        )
+        # 1er import → workflow rejouable créé en DB VWB
+        assert Workflow.query.count() == 1
+        created = Workflow.query.first()
+        assert created.source == "learned_import"
+        assert created.review_status == "pending_review"
+        assert (first or {}).get("created") is True
+        # PII : le nom patient ne doit jamais atterrir en clair dans la DB VWB
+        assert "VIOLA" not in created.name, created.name
+
+        # 2e import de la MÊME trajectoire → pas de doublon (idempotence)
+        second = processor._import_workflow_to_vwb(
+            workflow,
+            session_id="sess_bloc_notes_worker_rerun",
+            machine_id="DESKTOP-TEST_windows",
+        )
+        assert Workflow.query.count() == 1, "ré-import du même parcours = pas de doublon"
+        assert (second or {}).get("created") is False
+        assert (first or {}).get("workflow_id") == (second or {}).get("workflow_id")
+
+
+# ---------------------------------------------------------------------------
+# Activation prod (couplage worker→DB VWB) : gating par feature-flag
+# ---------------------------------------------------------------------------
+
+def test_maybe_import_gated_off_par_defaut(monkeypatch):
+    """Sans RPA_R1_AUTO_IMPORT, l'import auto NE doit PAS se déclencher
+    (R1 reste inactif tant que le sanitizer n'est pas validé / GO Dom)."""
+    monkeypatch.delenv("RPA_R1_AUTO_IMPORT", raising=False)
+    processor = _make_processor()
+    appels = []
+    monkeypatch.setattr(processor, "_import_workflow_to_vwb",
+                        lambda *a, **k: appels.append(a), raising=False)
+
+    processor._maybe_import_to_vwb(_build_core_workflow(), "sess", "machine")
+
+    assert appels == []  # gated OFF : aucun import
+
+
+def test_maybe_import_actif_si_flag(monkeypatch):
+    """Avec RPA_R1_AUTO_IMPORT=true, l'import est appelé dans l'app-context VWB."""
+    import contextlib
+    monkeypatch.setenv("RPA_R1_AUTO_IMPORT", "true")
+    processor = _make_processor()
+    appels = []
+    monkeypatch.setattr(processor, "_import_workflow_to_vwb",
+                        lambda w, s, m: appels.append((s, m)), raising=False)
+    # neutralise la création réelle de l'app-context (testée au runtime)
+    monkeypatch.setattr(processor, "_vwb_app_context",
+                        lambda: contextlib.nullcontext(), raising=False)
+
+    processor._maybe_import_to_vwb(_build_core_workflow(), "sess-x", "machine-y")
+
+    assert appels == [("sess-x", "machine-y")]
+
+
+def test_maybe_import_ne_casse_pas_la_finalisation(monkeypatch):
+    """Un échec d'import VWB ne doit JAMAIS faire échouer la finalisation worker."""
+    import contextlib
+    monkeypatch.setenv("RPA_R1_AUTO_IMPORT", "true")
+    processor = _make_processor()
+    monkeypatch.setattr(processor, "_vwb_app_context",
+                        lambda: contextlib.nullcontext(), raising=False)
+
+    def _boom(*a, **k):
+        raise RuntimeError("DB VWB indisponible")
+    monkeypatch.setattr(processor, "_import_workflow_to_vwb", _boom, raising=False)
+
+    # ne doit pas lever
+    processor._maybe_import_to_vwb(_build_core_workflow(), "sess", "machine")

tests/unit/test_agent_logs_store.py

@@ -0,0 +1,78 @@
+"""Tests unitaires du store de logs poussés par les clients Léa (push-log-DGX).
+
+Le store persiste les logs reçus du client, rangés par `machine_id`, pour
+consultation au dashboard (diagnostic des postes sans AnyDesk). Stockage
+fichier (JSONL par machine_id), rétention configurable.
+
+Branche : feat/push-log-dgx — DETTE-020/021 (observabilité).
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Racine projet pour les imports locaux (meme pattern que tests/integration)
+_ROOT = str(Path(__file__).resolve().parents[2])
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+
+def test_append_then_read_roundtrip(tmp_path):
+    """append() persiste un batch ; read() le restitue dans l'ordre."""
+    from agent_v0.server_v1.agent_logs_store import AgentLogsStore
+
+    store = AgentLogsStore(base_dir=tmp_path / "agent_logs")
+    entries = [
+        {"ts": "2026-06-26T16:00:00", "level": "INFO",
+         "logger": "agent_v1.main", "message": "demarrage"},
+        {"ts": "2026-06-26T16:00:01", "level": "WARNING",
+         "logger": "agent_v1.core.executor", "message": "popup detectee"},
+    ]
+
+    store.append("lea-emilie-001", entries)
+    got = store.read("lea-emilie-001")
+
+    assert len(got) == 2
+    assert got[0]["message"] == "demarrage"
+    assert got[0]["level"] == "INFO"
+    assert got[1]["level"] == "WARNING"
+    assert got[1]["logger"] == "agent_v1.core.executor"
+
+
+def test_machine_id_path_traversal_stays_within_base(tmp_path):
+    """Un machine_id malveillant (entrée réseau) ne doit jamais écrire hors du base_dir."""
+    from agent_v0.server_v1.agent_logs_store import AgentLogsStore
+
+    base = (tmp_path / "agent_logs").resolve()
+    store = AgentLogsStore(base_dir=base)
+
+    store.append("../../../evil", [{"message": "pwn"}])
+
+    written = list(base.rglob("*.jsonl"))
+    assert written, "le batch doit être persisté SOUS base (pas d'évasion ni perte)"
+    for p in written:
+        assert base in p.resolve().parents, f"{p} échappe à {base}"
+    # Aucune fuite hors de base
+    assert not list(tmp_path.glob("evil*"))
+
+
+def test_purge_old_removes_files_older_than_retention(tmp_path):
+    """purge_old() supprime les fichiers-jour antérieurs à la rétention (G4 Qwen)."""
+    from datetime import datetime, timezone
+
+    from agent_v0.server_v1.agent_logs_store import AgentLogsStore
+
+    base = tmp_path / "agent_logs"
+    store = AgentLogsStore(base_dir=base)
+    mdir = base / "lea-001"
+    mdir.mkdir(parents=True)
+    (mdir / "2026-05-01.jsonl").write_text('{"message": "vieux"}\n', encoding="utf-8")
+    (mdir / "2026-06-26.jsonl").write_text('{"message": "recent"}\n', encoding="utf-8")
+
+    now = datetime(2026, 6, 26, tzinfo=timezone.utc)
+    removed = store.purge_old(retention_days=30, now=now)
+
+    remaining = {p.name for p in mdir.glob("*.jsonl")}
+    assert remaining == {"2026-06-26.jsonl"}
+    assert removed == 1

tests/unit/test_agent_v1_logging.py

@@ -0,0 +1,74 @@
+"""TDD — DETTE-021 : journalisation client Léa effective (vers fichier).
+
+Aujourd'hui `LOG_FILE` est défini (`agent_v0/agent_v1/config.py`) mais jamais
+branché ; `basicConfig` écrit sur stderr — perdu car Léa tourne en `pythonw.exe`
+(sans console). On veut une fonction `setup_logging()` qui branche un handler
+FICHIER avec rotation quotidienne + rétention (Règlement IA Art. 12, 180 j).
+
+Le module est chargé par chemin (importlib) pour ne dépendre d'aucun import
+lourd du package client (cf. DETTE-011/013).
+"""
+import importlib.util
+import logging
+from logging.handlers import TimedRotatingFileHandler
+from pathlib import Path
+
+_MOD_PATH = Path(__file__).resolve().parents[2] / "agent_v0" / "agent_v1" / "logging_setup.py"
+
+
+def _load_setup_logging():
+    spec = importlib.util.spec_from_file_location("lea_logging_setup", _MOD_PATH)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod.setup_logging
+
+
+def _cleanup_root():
+    root = logging.getLogger()
+    for h in list(root.handlers):
+        if getattr(h, "_lea_managed", False):
+            h.close()
+            root.removeHandler(h)
+
+
+def test_setup_logging_ecrit_dans_le_fichier(tmp_path):
+    """Les logs doivent atterrir dans LOG_FILE (et plus seulement sur stderr)."""
+    log_file = tmp_path / "agent_v1.log"
+    setup_logging = _load_setup_logging()
+    try:
+        setup_logging(log_file=log_file, level=logging.INFO)
+        logging.getLogger("lea.test").info("message de diagnostic")
+        for h in logging.getLogger().handlers:
+            h.flush()
+        assert log_file.exists(), "le fichier de log doit être créé"
+        assert "message de diagnostic" in log_file.read_text(encoding="utf-8")
+    finally:
+        _cleanup_root()
+
+
+def test_setup_logging_rotation_et_retention(tmp_path):
+    """Rotation quotidienne + rétention configurable (180 j par défaut — Art. 12)."""
+    log_file = tmp_path / "agent_v1.log"
+    setup_logging = _load_setup_logging()
+    try:
+        setup_logging(log_file=log_file, retention_days=180)
+        handlers = [h for h in logging.getLogger().handlers
+                    if isinstance(h, TimedRotatingFileHandler)]
+        assert handlers, "un TimedRotatingFileHandler doit être branché"
+        assert handlers[0].backupCount == 180
+    finally:
+        _cleanup_root()
+
+
+def test_setup_logging_idempotent(tmp_path):
+    """Appels répétés n'empilent pas les handlers fichier (pas de doublon)."""
+    log_file = tmp_path / "agent_v1.log"
+    setup_logging = _load_setup_logging()
+    try:
+        setup_logging(log_file=log_file)
+        setup_logging(log_file=log_file)
+        file_handlers = [h for h in logging.getLogger().handlers
+                         if isinstance(h, TimedRotatingFileHandler)]
+        assert len(file_handlers) == 1, "pas de handler fichier en double"
+    finally:
+        _cleanup_root()

tests/unit/test_extract_dossier.py

@@ -0,0 +1,219 @@
+"""Tests TDD — Extraction « dossier patient » (brique 3).
+
+Deux couches testées :
+
+1. ``vwb_db.persist_extracted_dossier`` : depuis une grille OCR
+   (List[List[cell]]), crée ExtractionJob → ExtractedTable → ExtractedField
+   et commit. Testé sur SQLite mémoire via un app-context Flask jetable
+   (PAS la vraie DB VWB — isolation).
+
+2. ``replay_engine._handle_extract_dossier_action`` : lit last_screenshot,
+   appelle ``extract_grid_from_image`` (mocké), applique la gate qualité
+   (complete / needs_review), persiste via vwb_db et n'échoue JAMAIS le
+   replay (grille vide → needs_review, sans lever).
+
+⚠️ Canal extraction = données patient EN CLAIR (volontaire) : on vérifie
+que les valeurs sont persistées telles quelles, sans tokenisation.
+"""
+import pytest
+from flask import Flask
+
+# vwb_db ajoute visual_workflow_builder/backend au sys.path à l'import →
+# doit précéder l'import de db.models (couplage worker→DB VWB mutualisé).
+import agent_v0.server_v1.vwb_db as vwb_db
+import agent_v0.server_v1.replay_engine as replay_engine
+
+from db.models import db, ExtractionJob, ExtractedTable, ExtractedField
+
+
+# ---------------------------------------------------------------------------
+# Fixtures : app Flask jetable sur SQLite mémoire (isolation totale)
+# ---------------------------------------------------------------------------
+@pytest.fixture
+def mem_app():
+    """App Flask minimale liée à une DB SQLite en mémoire."""
+    app = Flask("test_extract_dossier")
+    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
+
+
+def _grid_2x2():
+    """Grille connue 2×2 (confiances hautes)."""
+    return [
+        [
+            {"text": "Nom", "bbox": [[0, 0], [1, 0], [1, 1], [0, 1]], "confidence": 0.95, "row": 0, "col": 0},
+            {"text": "MOREL", "bbox": [[2, 0], [3, 0], [3, 1], [2, 1]], "confidence": 0.92, "row": 0, "col": 1},
+        ],
+        [
+            {"text": "IPP", "bbox": [[0, 2], [1, 2], [1, 3], [0, 3]], "confidence": 0.90, "row": 1, "col": 0},
+            {"text": "25123456", "bbox": [[2, 2], [3, 2], [3, 3], [2, 3]], "confidence": 0.88, "row": 1, "col": 1},
+        ],
+    ]
+
+
+# ---------------------------------------------------------------------------
+# 1) persist_extracted_dossier
+# ---------------------------------------------------------------------------
+@pytest.mark.unit
+def test_persist_extracted_dossier_creates_job_table_fields(mem_app):
+    job_id = vwb_db.persist_extracted_dossier(
+        _grid_2x2(),
+        patient_ref="MOREL Catherine",
+        source_session_id="sess-42",
+        screenshot_ref="/captures/last.png",
+        screen_bbox={"x": 0, "y": 0, "width": 800, "height": 600},
+        status="complete",
+    )
+
+    assert isinstance(job_id, str) and job_id
+
+    job = db.session.get(ExtractionJob, job_id)
+    assert job is not None
+    assert job.status == "complete"
+    assert job.patient_ref == "MOREL Catherine"  # EN CLAIR, non tokenisé
+    assert job.source_session_id == "sess-42"
+
+    tables = ExtractedTable.query.filter_by(job_id=job_id).all()
+    assert len(tables) == 1
+    assert tables[0].screenshot_ref == "/captures/last.png"
+    assert tables[0].screen_bbox == {"x": 0, "y": 0, "width": 800, "height": 600}
+
+    fields = ExtractedField.query.filter_by(table_id=tables[0].id).all()
+    assert len(fields) == 4  # 2×2 cellules
+    values = {(f.row, f.col): f.value for f in fields}
+    assert values[(0, 1)] == "MOREL"      # valeur patient EN CLAIR conservée
+    assert values[(1, 1)] == "25123456"
+    confs = {(f.row, f.col): f.confidence for f in fields}
+    assert confs[(0, 0)] == pytest.approx(0.95)
+
+
+@pytest.mark.unit
+def test_persist_extracted_dossier_empty_grid_still_creates_job(mem_app):
+    """Grille vide → Job + Table sans Field (statut transmis tel quel)."""
+    job_id = vwb_db.persist_extracted_dossier(
+        [],
+        patient_ref=None,
+        source_session_id="sess-empty",
+        screenshot_ref="/captures/empty.png",
+        screen_bbox=None,
+        status="needs_review",
+    )
+    job = db.session.get(ExtractionJob, job_id)
+    assert job is not None and job.status == "needs_review"
+    tables = ExtractedTable.query.filter_by(job_id=job_id).all()
+    assert len(tables) == 1
+    assert ExtractedField.query.filter_by(table_id=tables[0].id).count() == 0
+
+
+# ---------------------------------------------------------------------------
+# 2) _handle_extract_dossier_action
+# ---------------------------------------------------------------------------
+@pytest.mark.unit
+def test_handle_extract_dossier_complete(mem_app, monkeypatch, tmp_path):
+    # screenshot bidon sur disque (le mock OCR ignore le contenu)
+    shot = tmp_path / "shot.png"
+    shot.write_bytes(b"\x89PNG")
+
+    # extract_grid_from_image mocké → grille 2×2 de confiance haute
+    monkeypatch.setattr(
+        "core.llm.extract_grid_from_image",
+        lambda *a, **k: _grid_2x2(),
+    )
+    # vwb_app_context pointé sur l'app mémoire de la fixture
+    monkeypatch.setattr(vwb_db, "vwb_app_context", lambda: mem_app.app_context())
+    monkeypatch.setattr(replay_engine, "vwb_db", vwb_db, raising=False)
+
+    replay_state = {
+        "last_screenshot": str(shot),
+        "variables": {},
+        "replay_id": "rep-1",
+    }
+    action = {
+        "type": "extract_dossier",
+        "parameters": {
+            "output_var": "dossier_id",
+            "patient_ref": "MOREL Catherine",
+            "expected_cols": 2,
+            "min_confidence": 0.5,
+        },
+    }
+
+    ok = replay_engine._handle_extract_dossier_action(action, replay_state, "sess-42")
+    assert ok is True
+
+    job_id = replay_state["variables"]["dossier_id"]
+    assert isinstance(job_id, str) and job_id
+    with mem_app.app_context():
+        job = db.session.get(ExtractionJob, job_id)
+        assert job is not None
+        assert job.status == "complete"  # gate OK : non vide, conf ok, 2 cols
+
+
+@pytest.mark.unit
+def test_handle_extract_dossier_low_confidence_needs_review(mem_app, monkeypatch, tmp_path):
+    shot = tmp_path / "shot.png"
+    shot.write_bytes(b"\x89PNG")
+
+    low_grid = [
+        [{"text": "x", "bbox": [], "confidence": 0.10, "row": 0, "col": 0}],
+    ]
+    monkeypatch.setattr("core.llm.extract_grid_from_image", lambda *a, **k: low_grid)
+    monkeypatch.setattr(vwb_db, "vwb_app_context", lambda: mem_app.app_context())
+
+    replay_state = {"last_screenshot": str(shot), "variables": {}, "replay_id": "rep-2"}
+    action = {"type": "extract_dossier", "parameters": {"min_confidence": 0.5}}
+
+    ok = replay_engine._handle_extract_dossier_action(action, replay_state, "sess-low")
+    assert ok is False  # gate a basculé en needs_review
+    job_id = replay_state["variables"]["extracted_dossier"]
+    with mem_app.app_context():
+        assert db.session.get(ExtractionJob, job_id).status == "needs_review"
+
+
+@pytest.mark.unit
+def test_handle_extract_dossier_empty_grid_no_raise(mem_app, monkeypatch, tmp_path):
+    shot = tmp_path / "shot.png"
+    shot.write_bytes(b"\x89PNG")
+
+    monkeypatch.setattr("core.llm.extract_grid_from_image", lambda *a, **k: [])
+    monkeypatch.setattr(vwb_db, "vwb_app_context", lambda: mem_app.app_context())
+
+    replay_state = {"last_screenshot": str(shot), "variables": {}, "replay_id": "rep-3"}
+    action = {"type": "extract_dossier", "parameters": {}}
+
+    # Ne lève jamais ; grille vide → needs_review
+    ok = replay_engine._handle_extract_dossier_action(action, replay_state, "sess-empty")
+    assert ok is False
+    job_id = replay_state["variables"]["extracted_dossier"]
+    with mem_app.app_context():
+        assert db.session.get(ExtractionJob, job_id).status == "needs_review"
+
+
+@pytest.mark.unit
+def test_handle_extract_dossier_persist_failure_no_raise(mem_app, monkeypatch, tmp_path):
+    """Si la persistance lève, le handler log et n'échoue PAS le replay."""
+    shot = tmp_path / "shot.png"
+    shot.write_bytes(b"\x89PNG")
+
+    monkeypatch.setattr("core.llm.extract_grid_from_image", lambda *a, **k: _grid_2x2())
+    monkeypatch.setattr(vwb_db, "vwb_app_context", lambda: mem_app.app_context())
+
+    def _boom(*a, **k):
+        raise RuntimeError("DB down")
+    monkeypatch.setattr(vwb_db, "persist_extracted_dossier", _boom)
+
+    replay_state = {"last_screenshot": str(shot), "variables": {}, "replay_id": "rep-4"}
+    action = {"type": "extract_dossier", "parameters": {}}
+
+    ok = replay_engine._handle_extract_dossier_action(action, replay_state, "sess-boom")
+    assert ok is False  # jamais de raise
+
+
+@pytest.mark.unit
+def test_extract_dossier_declared_in_action_type_sets():
+    assert "extract_dossier" in replay_engine._ALLOWED_ACTION_TYPES
+    assert "extract_dossier" in replay_engine._SERVER_SIDE_ACTION_TYPES

tests/unit/test_extract_grid.py

@@ -0,0 +1,79 @@
+"""Tests pour extract_grid_from_image — lecture de tableau STRUCTURÉE.
+
+Contrairement à extract_table_from_image (qui jette x et retourne une liste
+plate triée par y), extract_grid_from_image reconstruit une vraie grille
+List[List[cell]] : clustering des lignes par proximité y, des colonnes par
+proximité x. bbox + confiance conservées par cellule.
+
+Les tokens OCR sont injectés (mock du reader EasyOCR) → pas de PNG réel,
+pas de GPU.
+"""
+from pathlib import Path
+from types import SimpleNamespace
+
+from PIL import Image
+
+import core.llm.ocr_extractor as ocr_extractor
+
+
+def _blank_png(path: Path) -> None:
+    Image.new("RGB", (300, 120), "white").save(path)
+
+
+def _bbox(x0: float, y0: float, x1: float, y1: float):
+    """bbox EasyOCR = 4 points [tl, tr, br, bl], chaque point (x, y)."""
+    return [[x0, y0], [x1, y0], [x1, y1], [x0, y1]]
+
+
+def _fake_reader(tokens):
+    """Reader factice : readtext() renvoie la liste (bbox, text, conf) fournie."""
+    return SimpleNamespace(readtext=lambda *a, **k: tokens)
+
+
+def test_extract_grid_2x3(tmp_path, monkeypatch):
+    image_path = tmp_path / "table.png"
+    _blank_png(image_path)
+
+    # 2 lignes (y≈10 et y≈60) × 3 colonnes (x≈10, x≈110, x≈210).
+    # Volontairement mélangées dans l'ordre OCR pour vérifier le tri.
+    tokens = [
+        (_bbox(110, 58, 160, 78), "B2", 0.97),
+        (_bbox(10, 10, 60, 30), "A1", 0.91),
+        (_bbox(210, 12, 260, 32), "C1", 0.88),
+        (_bbox(210, 60, 260, 80), "C2", 0.95),
+        (_bbox(10, 60, 60, 80), "A2", 0.90),
+        (_bbox(110, 8, 160, 28), "B1", 0.93),
+    ]
+    monkeypatch.setattr(ocr_extractor, "_get_reader", lambda: _fake_reader(tokens))
+
+    grid = ocr_extractor.extract_grid_from_image(str(image_path))
+
+    # Grille 2×3 ordonnée
+    assert len(grid) == 2, "doit détecter 2 lignes"
+    assert all(len(row) == 3 for row in grid), "chaque ligne doit avoir 3 colonnes"
+
+    texts = [[cell["text"] for cell in row] for row in grid]
+    assert texts == [["A1", "B1", "C1"], ["A2", "B2", "C2"]]
+
+    # Métadonnées conservées + indices row/col cohérents
+    cell = grid[0][2]
+    assert cell["text"] == "C1"
+    assert cell["confidence"] == 0.88
+    assert cell["bbox"] == _bbox(210, 12, 260, 32)
+    assert cell["row"] == 0
+    assert cell["col"] == 2
+    assert grid[1][0]["row"] == 1 and grid[1][0]["col"] == 0
+
+
+def test_extract_grid_empty_when_no_tokens(tmp_path, monkeypatch):
+    image_path = tmp_path / "blank.png"
+    _blank_png(image_path)
+    monkeypatch.setattr(ocr_extractor, "_get_reader", lambda: _fake_reader([]))
+
+    grid = ocr_extractor.extract_grid_from_image(str(image_path))
+    assert grid == []
+
+
+def test_extract_grid_missing_file_returns_empty():
+    grid = ocr_extractor.extract_grid_from_image("/no/such/file.png")
+    assert grid == []

tests/unit/test_log_safe.py

@@ -0,0 +1,73 @@
+"""Tests unitaires des helpers de logging PII-safe du client Léa (agent_v1).
+
+Assainissement des logs à la source : on ne logge jamais le contenu brut
+(titres de fenêtre, noms de workflow, chemins, métadonnées sensibles). On le
+remplace par un hash court stable, une longueur, ou un dict filtré.
+
+Branche feat/push-log-dgx — DETTE-020 (assainissement PII des logs, brique 4).
+"""
+
+from __future__ import annotations
+
+import re
+import sys
+from pathlib import Path
+
+_ROOT = str(Path(__file__).resolve().parents[2])
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+_HEX8 = re.compile(r"^[0-9a-f]{8}$")
+
+
+def test_title_hash_is_short_stable_hex():
+    from agent_v0.agent_v1.core.log_safe import _title_hash
+
+    h = _title_hash("Dossier MOREL Catherine")
+    assert _HEX8.match(h), f"attendu 8 hex, obtenu {h!r}"
+    assert h == _title_hash("Dossier MOREL Catherine")  # déterministe
+
+
+def test_title_hash_never_reveals_raw_title():
+    """Propriété PII centrale : le hash ne contient jamais le contenu brut."""
+    from agent_v0.agent_v1.core.log_safe import _title_hash
+
+    title = "Dossier MOREL Catherine"
+    h = _title_hash(title)
+    assert title not in h
+    assert "MOREL" not in h
+
+
+def test_title_hash_distinguishes_different_titles():
+    from agent_v0.agent_v1.core.log_safe import _title_hash
+
+    assert _title_hash("popup A") != _title_hash("popup B")
+
+
+def test_title_hash_handles_empty_and_non_ascii():
+    from agent_v0.agent_v1.core.log_safe import _title_hash
+
+    assert _HEX8.match(_title_hash(""))
+    assert _HEX8.match(_title_hash("Éléonore — café ☕"))
+
+
+def test_sanitize_metadata_drops_pii_keys_keeps_technical():
+    from agent_v0.agent_v1.core.log_safe import _sanitize_metadata
+
+    meta = {
+        "resolution": "1920x1080", "dpi": 96, "theme": "dark",
+        "title": "Dossier Dupont", "active_window": "Medicare", "window_title": "x",
+    }
+    safe = _sanitize_metadata(meta)
+
+    assert safe == {"resolution": "1920x1080", "dpi": 96, "theme": "dark"}
+    assert meta.get("title") == "Dossier Dupont"  # original non muté
+
+
+def test_path_ext_returns_extension_only():
+    from agent_v0.agent_v1.core.log_safe import _path_ext
+
+    assert _path_ext("/home/tim/Dossier Dupont 1980.png") == ".png"
+    assert "Dupont" not in _path_ext("/x/Dupont.png")
+    assert _path_ext("") == ""
+    assert _path_ext("/no/ext/here") == ""

tests/unit/test_pii_sanitizer.py

@@ -0,0 +1,236 @@
+"""Tests de l'assainissement PII des données capturées (titres, texte, OCR).
+
+Couche 1 (sans modèle) : filet regex sur la PII structurée (IPP, NIR, TEL,
+EMAIL, AGE) + règles structurelles cliniques (NOM (NAISSANCE) Prénom ;
+[Nom Prénom] des fenêtres PACS), avec tokens TYPÉS et COHÉRENTS ([IPP_1]…).
+
+Réutilise l'approche du projet `anonymisation` (placeholders + regex). La
+couche NER (noms libres) viendra en complément. Cas réels remontés en clinique
+le 28/06 (anonymisés ici par construction). Branche feat/push-log-dgx.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+_ROOT = str(Path(__file__).resolve().parents[2])
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+
+def test_ipp_et_age_tokenises():
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    titre = "VIOLA (VIOLA) Liliane 90 ans - IPP: 168246 - Expert Sante - Mozilla Firefox"
+    out, ents = anonymize_text(titre)
+
+    assert "168246" not in out, out          # IPP retiré
+    assert "[IPP_1]" in out
+    assert "90 ans" not in out               # âge retiré
+    assert "[AGE_1]" in out
+    # le nom format clinique « NOM (NAISSANCE) Prénom » est tokenisé
+    assert "VIOLA" not in out and "Liliane" not in out, out
+    assert "[NOM_1]" in out
+    # le logiciel n'est pas pris pour de la PII
+    assert "Firefox" in out and "Expert Sante" in out
+    types = {e["type"] for e in ents}
+    assert {"IPP", "AGE", "NOM"} <= types
+
+
+def test_nom_entre_crochets_pacs():
+    """Le PACS met le patient entre crochets : `[DATTIN Alix]`."""
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    titre = "GXD5 Pacs 4.0.4.307 CIM ARES - [DATTIN Alix] - Mozilla Firefox"
+    out, _ = anonymize_text(titre)
+
+    assert "DATTIN" not in out and "Alix" not in out, out
+    assert "[NOM_1]" in out
+    assert "Pacs" in out and "Firefox" in out  # contexte logiciel préservé
+
+
+def test_coherence_meme_ipp_meme_token():
+    """Même valeur PII -> même token (sur un mapping partagé de session)."""
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    mapping: dict = {}
+    o1, _ = anonymize_text("IPP: 168246 ouvert", mapping=mapping)
+    o2, _ = anonymize_text("dossier IPP: 168246 fermé", mapping=mapping)
+    o3, _ = anonymize_text("IPP: 270020 autre", mapping=mapping)
+
+    assert "[IPP_1]" in o1 and "[IPP_1]" in o2      # même patient -> même token
+    assert "[IPP_2]" in o3                           # patient différent -> token différent
+    assert "270020" not in o3
+
+
+def test_email_et_telephone():
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    out, _ = anonymize_text("contact j.dupont@chu.fr / 06 12 34 56 78")
+    assert "@chu.fr" not in out and "[EMAIL_1]" in out
+    assert "06 12 34 56 78" not in out and "[TEL_1]" in out
+
+
+def test_texte_sans_pii_inchange():
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    t = "Expert Sante - Consultation - Mozilla Firefox"
+    out, ents = anonymize_text(t)
+    assert out == t
+    assert ents == []
+
+
+# --- sanitize_event : assainissement au niveau event (option b pour text_input) ---
+
+def test_sanitize_text_input_remplace_contenu_par_saisie():
+    """Option b (Dom) : le contenu tapé n'est pas gardé -> [SAISIE]."""
+    from agent_v0.server_v1.pii_sanitizer import sanitize_event
+
+    ev = {
+        "type": "text_input",
+        "text": "hemorragie post-operatoire saignement",  # contenu médical
+        "raw_keys": ["h", "e", "m"],
+        "window": {"title": "VIOLA (VIOLA) Liliane 90 ans - IPP: 168246 - Firefox",
+                   "app_name": "firefox.exe"},
+    }
+    out = sanitize_event(ev)
+
+    assert out["text"] == "[SAISIE]"
+    assert out["raw_keys"] == "[SAISIE]"
+    # le titre de la fenêtre est assaini (identité tokenisée, app gardée)
+    assert "168246" not in out["window"]["title"]
+    assert "VIOLA" not in out["window"]["title"]
+    assert "[IPP_1]" in out["window"]["title"] and "Firefox" in out["window"]["title"]
+    # l'event d'origine n'est PAS muté
+    assert ev["text"].startswith("hemorragie")
+
+
+def test_sanitize_heartbeat_titre_direct():
+    from agent_v0.server_v1.pii_sanitizer import sanitize_event
+
+    ev = {"type": "heartbeat",
+          "active_window_title": "GXD5 Pacs CIM ARES - [DATTIN Alix] - Firefox"}
+    out = sanitize_event(ev)
+    assert "DATTIN" not in out["active_window_title"]
+    assert "[NOM_1]" in out["active_window_title"] and "Pacs" in out["active_window_title"]
+
+
+def test_sanitize_focus_change_to_from_window():
+    from agent_v0.server_v1.pii_sanitizer import sanitize_event
+
+    ev = {"type": "window_focus_change",
+          "from": None,
+          "to": {"title": "LAVAL (BARTHELEMY) Nicole 86 ans - Expert Sante", "app_name": "firefox.exe"},
+          "window": {"title": "LAVAL (BARTHELEMY) Nicole 86 ans - Expert Sante"}}
+    out = sanitize_event(ev)
+    assert out["from"] is None                       # null géré
+    assert "LAVAL" not in out["to"]["title"]
+    assert "[NOM_1]" in out["to"]["title"]
+    # cohérence : même patient dans to et window -> même token
+    assert out["window"]["title"] == out["to"]["title"]
+
+
+def test_sanitize_action_result_inchange():
+    from agent_v0.server_v1.pii_sanitizer import sanitize_event
+
+    ev = {"type": "action_result", "base_shot_id": "shot_0003", "image": "x.png"}
+    assert sanitize_event(ev) == ev
+
+
+def test_prenom_nom_inverse():
+    """FN-1/2/3 (Qwen) : « Prénom NOM » inversé (sans parens/crochets)."""
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    m: dict = {}
+    for s, leak in [("Alix DATTIN - Mozilla Firefox", "DATTIN"),
+                    ("Agathe RONDOT - PACS CIM ARES", "RONDOT"),
+                    ("Marie FLANDINETTE - Mozilla Firefox", "FLANDINETTE")]:
+        out, _ = anonymize_text(s, mapping=m)
+        assert leak not in out, out
+        assert "[NOM_" in out
+    # pas de faux positif sur les logiciels (2e mot non capitalisé tout en majuscules)
+    out, ents = anonymize_text("Mozilla Firefox - Expert Sante - Consultation")
+    assert out == "Mozilla Firefox - Expert Sante - Consultation"
+    assert ents == []
+
+
+def test_sanitize_event_titre_imbrique_vision_info():
+    """FN-4 (Qwen) : titre PII imbriqué dans vision_info.window_capture (228 events)."""
+    from agent_v0.server_v1.pii_sanitizer import sanitize_event
+
+    titre = "VIOLA (VIOLA) Liliane 90 ans - IPP: 168246 - Firefox"
+    ev = {
+        "type": "mouse_click",
+        "window": {"title": titre, "app_name": "firefox.exe"},
+        "vision_info": {"window_capture": {"window_title": titre, "app_name": "firefox.exe"}},
+    }
+    out = sanitize_event(ev)
+
+    wc = out["vision_info"]["window_capture"]["window_title"]
+    assert "168246" not in wc and "VIOLA" not in wc, wc
+    assert "[IPP_1]" in wc
+    # cohérence : même titre dans window et vision_info -> même token
+    assert out["window"]["title"] == wc
+
+
+def test_sanitize_workflow_dict_tokenise_by_text_garde_ui():
+    """R1/PII : un workflow appris ne doit pas porter de PII brute dans ses cibles
+    (by_text) ni ses noms avant import en DB VWB ; l'interface est préservée."""
+    import json
+    from agent_v0.server_v1.pii_sanitizer import sanitize_workflow_dict
+
+    wf = {
+        "name": "Dossier patient",
+        "nodes": [{"node_id": "n1", "name": "VIOLA (VIOLA) Liliane 90 ans"}],
+        "edges": [{
+            "edge_id": "e1",
+            "action": {
+                "type": "mouse_click",
+                "target": {"by_text": "Valider", "by_role": "ocr"},
+            },
+        }],
+    }
+    out = sanitize_workflow_dict(wf)
+    s = json.dumps(out, ensure_ascii=False)
+    assert "VIOLA" not in s           # nom clinique tokenisé (dans un node name)
+    assert "[NOM_1]" in s
+    assert "90 ans" not in s          # âge tokenisé
+    assert "Valider" in s             # cible UI préservée (by_text)
+    assert "VIOLA" in json.dumps(wf, ensure_ascii=False)  # original non muté
+
+
+def test_chevauchement_prefix_capitalise():
+    """FN bloquant (Claude R1) : mot capitalisé avant NOM (NAISSANCE) Prénom
+    -> RE_PRENOM_NOM captait « Dossier VIOLA » et bloquait RE_NOM_NAISSANCE
+    « VIOLA (VIOLA) Liliane ». Fix : résolution par priorité détecteur + longueur."""
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    m: dict = {}
+    for titre, leak in [("Dossier VIOLA (VIOLA) Liliane", "VIOLA"),
+                        ("Patient ROSSIGNOL (SOUBIE) Pierrette", "ROSSIGNOL"),
+                        ("Fenetre LAVAL (BARTHELEMY) Nicole", "LAVAL")]:
+        out, _ = anonymize_text(titre, mapping=m)
+        assert leak not in out, f"FN: {leak} still visible in '{out}'"
+
+    # contrôle : sans préfixe, toujours OK
+    out, _ = anonymize_text("VIOLA (VIOLA) Liliane", mapping=m)
+    assert "VIOLA" not in out
+
+
+def test_gxd5_diagnostics_numero_et_nom():
+    """GXD5 Diagnostics — numéro de dossier + nom tout-majuscules (3 patients prod)."""
+    from agent_v0.server_v1.pii_sanitizer import anonymize_text
+
+    m: dict = {}
+    for titre, num_leak, nom_leak in [
+        ("GXD5 Diagnostics - 128008 - BENVENISTE MARIE-LAURENCE", "128008", "BENVENISTE"),
+        ("GXD5 Diagnostics - 272223 - LEMOINE ERIC", "272223", "LEMOINE"),
+        ("GXD5 Diagnostics - 153442 - ROSELIER MATHEO", "153442", "ROSELIER"),
+    ]:
+        out, ents = anonymize_text(titre, mapping=m)
+        assert num_leak not in out, f"FN: numéro {num_leak} visible dans '{out}'"
+        assert nom_leak not in out, f"FN: nom {nom_leak} visible dans '{out}'"
+        types = {e["type"] for e in ents}
+        assert "DOSSIER" in types, f"Pas de token DOSSIER dans {ents}"
+        assert "NOM" in types, f"Pas de token NOM dans {ents}"

tests/unit/test_stream_event_pii_wiring.py

@@ -0,0 +1,68 @@
+"""Non-régression sécurité : câblage PII au chokepoint ``stream_event``.
+
+Invariant : un event contenant de la PII patient (titre de fenêtre + contenu
+saisi) passé à ``stream_event`` ne doit JAMAIS écrire la PII brute dans le
+journal ``live_events.jsonl``, ni la propager au worker ou au shadow observer.
+L'assainissement a lieu une seule fois, en amont des chemins de
+persistance/traitement (``api_stream.py``, hook ``sanitize_event``).
+"""
+import asyncio
+import json
+import os
+
+# Le module serveur refuse de se charger sans token (sécurité prod) ;
+# en test unitaire on désactive l'auth pour pouvoir importer le module.
+os.environ.setdefault("RPA_AUTH_DISABLED", "true")
+
+import agent_v0.server_v1.api_stream as api
+
+
+def _event_avec_pii():
+    # PII captée par la couche 1 : IPP (structurel) + contenu saisi.
+    # Contexte = logiciel métier réel du POC (pas la maquette Easily abandonnée).
+    # (Les noms libres sans marqueur relèvent de la couche 2 NER — hors scope ici.)
+    return {
+        "type": "text_input",
+        "text": "anticoagulant 75mg matin",
+        "active_window_title": "Gxd5diag - Recherche dossier (IPP: 123456)",
+    }
+
+
+def test_stream_event_assainit_et_propage_sur_les_chemins(tmp_path, monkeypatch):
+    """Le chokepoint applique sanitize_event UNE fois et tous les chemins
+    (jsonl, worker, shadow) reçoivent la copie assainie — pas la valeur brute."""
+    captured = {}
+    monkeypatch.setattr(api, "_ensure_session_registered", lambda *a, **k: None)
+    monkeypatch.setattr(
+        api.worker,
+        "process_event_direct",
+        lambda sid, ev: (captured.__setitem__("worker", ev), {})[1],
+    )
+    monkeypatch.setattr(
+        api, "shadow_observe_event", lambda sid, ev: captured.__setitem__("shadow", ev)
+    )
+    monkeypatch.setattr(api, "LIVE_SESSIONS_DIR", tmp_path)
+    api._session_pii_mapping.pop("sess_pii", None)
+
+    se = api.StreamEvent(
+        session_id="sess_pii",
+        machine_id="lea-test",
+        timestamp=1000.0,
+        event=_event_avec_pii(),
+    )
+
+    asyncio.run(api.stream_event(se))
+
+    # 1. le journal sur disque ne contient ni l'IPP brut ni le contenu saisi
+    jsonl = (tmp_path / "lea-test" / "sess_pii" / "live_events.jsonl").read_text(
+        encoding="utf-8"
+    )
+    assert "123456" not in jsonl
+    assert "anticoagulant 75mg" not in jsonl
+    # 2. contenu saisi masqué + IPP tokenisé (preuve que le titre est traité)
+    assert "[SAISIE]" in jsonl
+    assert "[IPP_1]" in jsonl
+    # 3. worker et shadow reçoivent l'event assaini, pas la valeur brute
+    assert captured["worker"]["text"] == "[SAISIE]"
+    assert "123456" not in json.dumps(captured["worker"], ensure_ascii=False)
+    assert "123456" not in json.dumps(captured["shadow"], ensure_ascii=False)

tests/unit/test_trajectory_signature.py

@@ -0,0 +1,101 @@
+"""TDD — signature de trajectoire (Phase 0 ; primitive partagée SP-4 / SP-2 / compétences).
+
+Propriété centrale : la signature identifie une TRAJECTOIRE (séquence d'actions sur des
+cibles stables). Elle doit être **stable entre sessions** — donc indépendante des champs
+session-spécifiques (IDs de nœuds, timestamps, coordonnées). C'est ce qui rend le
+create-or-update (décision F1) possible : deux apprentissages du même parcours = même id.
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from core.execution.trajectory_signature import trajectory_signature
+
+
+def test_deterministic_same_sequence():
+    steps = [
+        {"action_type": "mouse_click", "target": "menu Fichier"},
+        {"action_type": "text_input", "target": "champ recherche"},
+    ]
+    assert trajectory_signature(steps) == trajectory_signature(steps)
+
+
+def test_ignores_session_specific_fields():
+    """Deux sessions du MÊME parcours (mêmes action_type+target) mais IDs de nœuds /
+    timestamps / coords différents → MÊME signature."""
+    session_a = [
+        {"action_type": "mouse_click", "target": "menu Fichier",
+         "node_id": "n_abc", "timestamp": 1000, "x": 12, "y": 34},
+        {"action_type": "text_input", "target": "champ recherche",
+         "node_id": "n_def", "timestamp": 1100, "x": 50, "y": 60},
+    ]
+    session_b = [
+        {"action_type": "mouse_click", "target": "menu Fichier",
+         "node_id": "n_zzz", "timestamp": 9000, "x": 99, "y": 88},
+        {"action_type": "text_input", "target": "champ recherche",
+         "node_id": "n_yyy", "timestamp": 9100, "x": 11, "y": 22},
+    ]
+    assert trajectory_signature(session_a) == trajectory_signature(session_b)
+
+
+def test_order_sensitive():
+    a = [{"action_type": "mouse_click", "target": "A"},
+         {"action_type": "text_input", "target": "B"}]
+    b = list(reversed(a))
+    assert trajectory_signature(a) != trajectory_signature(b)
+
+
+def test_target_discriminates():
+    a = [{"action_type": "mouse_click", "target": "bouton Valider"}]
+    b = [{"action_type": "mouse_click", "target": "bouton Annuler"}]
+    assert trajectory_signature(a) != trajectory_signature(b)
+
+
+def test_returns_sha256_hex():
+    sig = trajectory_signature([{"action_type": "mouse_click", "target": "x"}])
+    assert len(sig) == 64
+    assert all(c in "0123456789abcdef" for c in sig)
+
+
+# ---------------------------------------------------------------------------
+# R1/R2 amendés — verdict Qwen 2026-06-25 : normalisation déterministe + PII
+# neutralisée par regex DÉDIÉES (pas de pii_blur, pas de NER). Stabilité
+# labo/DGX = portabilité de la signature. Noms sans titre : stratégie (b)
+# (impact 0 en labo, gate = audit agrégat DGX avant prod).
+# ---------------------------------------------------------------------------
+
+def test_target_normalized_case_and_accents():
+    """Q2 : casse et accents ne changent pas la signature (même cible sémantique)."""
+    a = [{"action_type": "mouse_click", "target": "Valider"}]
+    b = [{"action_type": "mouse_click", "target": "VALIDER"}]
+    c = [{"action_type": "mouse_click", "target": "validér"}]
+    assert trajectory_signature(a) == trajectory_signature(b) == trajectory_signature(c)
+
+
+def test_pii_ipp_neutralized():
+    """R1 : deux IPP différents sur le même champ → MÊME signature (PII neutralisée).
+    Et une cible sans identifiant reste discriminée."""
+    a = [{"action_type": "mouse_click", "target": "Patient IPP 25012257"}]
+    b = [{"action_type": "mouse_click", "target": "Patient IPP 30045678"}]
+    assert trajectory_signature(a) == trajectory_signature(b)
+    c = [{"action_type": "mouse_click", "target": "Patient liste"}]
+    assert trajectory_signature(a) != trajectory_signature(c)
+
+
+def test_pii_date_neutralized():
+    """R1 : deux dates différentes → MÊME signature."""
+    a = [{"action_type": "mouse_click", "target": "RDV du 12/05/2026"}]
+    b = [{"action_type": "mouse_click", "target": "RDV du 03/11/2025"}]
+    assert trajectory_signature(a) == trajectory_signature(b)
+
+
+def test_pii_phone_and_email_neutralized():
+    """R1 : téléphone (FR) et email neutralisés (deux valeurs distinctes → même sig)."""
+    tel_a = [{"action_type": "text_input", "target": "tel 06 12 34 56 78"}]
+    tel_b = [{"action_type": "text_input", "target": "tel 07 98 76 54 32"}]
+    assert trajectory_signature(tel_a) == trajectory_signature(tel_b)
+    mail_a = [{"action_type": "text_input", "target": "mail jean.dupont@chu.fr"}]
+    mail_b = [{"action_type": "text_input", "target": "mail m.martin@chu.fr"}]
+    assert trajectory_signature(mail_a) == trajectory_signature(mail_b)

tests/unit/test_workflow_graph_machine_id.py

@@ -0,0 +1,44 @@
+"""
+Test de non-régression : conservation du machine_id au round-trip to_dict/from_dict.
+
+Bug : les workflows listés via /api/v1/traces/stream/workflows étaient tous
+attribués à machine_id="default" alors que les sessions portaient le bon
+machine_id (lea-*). Cause : to_dict ne sérialisait pas l'attribut d'instance
+`_machine_id` et from_dict ne le reposait pas (il dormait dans
+metadata['machine_id']). list_workflows tombait alors sur le fallback "default".
+"""
+
+from datetime import datetime
+
+from core.models.workflow_graph import Workflow
+
+
+def _make_minimal_workflow(machine_id: str) -> Workflow:
+    """Construit un workflow minimal portant un machine_id dans ses métadonnées."""
+    now = datetime.now().isoformat()
+    return Workflow.from_dict({
+        "workflow_id": "wf-test",
+        "name": "wf-test",
+        "nodes": [],
+        "edges": [],
+        "safety_rules": {},
+        "stats": {},
+        "learning": {},
+        "entry_nodes": [],
+        "end_nodes": [],
+        "created_at": now,
+        "updated_at": now,
+        "metadata": {"machine_id": machine_id},
+    })
+
+
+def test_machine_id_preserved_after_to_dict_from_dict_round_trip():
+    """Un workflow doit conserver son machine_id après un round-trip de (dé)sérialisation."""
+    wf = _make_minimal_workflow("lea-poste-3")
+    # Simule l'étiquetage runtime fait par le stream_processor
+    wf._machine_id = "lea-poste-3"
+
+    restored = Workflow.from_dict(wf.to_dict())
+
+    # Invariant : le machine_id survit au round-trip (comme le fait list_workflows)
+    assert getattr(restored, "_machine_id", "default") == "lea-poste-3"

tests/unit/test_workflow_trajectory_signature.py

@@ -0,0 +1,86 @@
+"""TDD — adaptateur Workflow → signature de trajectoire (Phase 0, lot 2).
+
+Branche la primitive `trajectory_signature` sur un vrai workflow core (dict).
+Doit : traverser les edges dans l'ordre du parcours (BFS depuis entry_nodes), et
+n'extraire que des descripteurs de cible **stables** (by_role/by_text/window),
+en ignorant coords (`by_position`) et IDs de nœuds session-spécifiques.
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from core.execution.trajectory_signature import workflow_trajectory_signature
+
+
+def _edge(from_node, to_node, action_type, *, by_role="", by_text="", by_position=None):
+    target = {"by_role": by_role, "by_text": by_text}
+    if by_position is not None:
+        target["by_position"] = by_position
+    return {
+        "from_node": from_node,
+        "to_node": to_node,
+        "action": {"type": action_type, "target": target},
+    }
+
+
+def test_signature_stable_across_sessions():
+    """Même parcours, IDs de nœuds + coords différents → même signature."""
+    session_a = {
+        "entry_nodes": ["n1"],
+        "nodes": [{"node_id": "n1"}, {"node_id": "n2"}, {"node_id": "n3"}],
+        "edges": [
+            _edge("n1", "n2", "mouse_click", by_role="button", by_text="Fichier", by_position=[0.1, 0.2]),
+            _edge("n2", "n3", "text_input", by_text="recherche", by_position=[0.5, 0.6]),
+        ],
+    }
+    session_b = {
+        "entry_nodes": ["a1"],
+        "nodes": [{"node_id": "a1"}, {"node_id": "a2"}, {"node_id": "a3"}],
+        "edges": [
+            _edge("a1", "a2", "mouse_click", by_role="button", by_text="Fichier", by_position=[0.9, 0.8]),
+            _edge("a2", "a3", "text_input", by_text="recherche", by_position=[0.05, 0.04]),
+        ],
+    }
+    assert workflow_trajectory_signature(session_a) == workflow_trajectory_signature(session_b)
+
+
+def test_signature_differs_on_different_target():
+    base = {
+        "entry_nodes": ["n1"],
+        "nodes": [{"node_id": "n1"}, {"node_id": "n2"}],
+        "edges": [_edge("n1", "n2", "mouse_click", by_role="button", by_text="Valider")],
+    }
+    other = {
+        "entry_nodes": ["n1"],
+        "nodes": [{"node_id": "n1"}, {"node_id": "n2"}],
+        "edges": [_edge("n1", "n2", "mouse_click", by_role="button", by_text="Annuler")],
+    }
+    assert workflow_trajectory_signature(base) != workflow_trajectory_signature(other)
+
+
+def test_signature_follows_edge_chain_not_list_order():
+    """L'ordre vient de la chaîne from→to (BFS), pas de l'ordre brut de la liste."""
+    e1 = _edge("n1", "n2", "mouse_click", by_text="A")
+    e2 = _edge("n2", "n3", "text_input", by_text="B")
+    ordered = {"entry_nodes": ["n1"], "nodes": [{"node_id": "n1"}, {"node_id": "n2"}, {"node_id": "n3"}],
+               "edges": [e1, e2]}
+    scrambled = {"entry_nodes": ["n1"], "nodes": [{"node_id": "n1"}, {"node_id": "n2"}, {"node_id": "n3"}],
+                 "edges": [e2, e1]}  # liste inversée, même chaîne
+    assert workflow_trajectory_signature(ordered) == workflow_trajectory_signature(scrambled)
+
+
+def test_signature_stable_despite_grounding_role_difference():
+    """`by_role` peut porter le moteur de grounding (yolo/ocr/vlm) — instable entre
+    sessions. La signature doit rester identique si seul `by_role` change → elle
+    s'appuie sur le texte sémantique `by_text`, pas sur la méthode de détection."""
+    wf_yolo = {
+        "entry_nodes": ["n1"], "nodes": [{"node_id": "n1"}, {"node_id": "n2"}],
+        "edges": [_edge("n1", "n2", "mouse_click", by_role="yolo", by_text="Fichier")],
+    }
+    wf_ocr = {
+        "entry_nodes": ["n1"], "nodes": [{"node_id": "n1"}, {"node_id": "n2"}],
+        "edges": [_edge("n1", "n2", "mouse_click", by_role="ocr", by_text="Fichier")],
+    }
+    assert workflow_trajectory_signature(wf_yolo) == workflow_trajectory_signature(wf_ocr)

visual_workflow_builder/backend/db/models.py

@@ -321,6 +321,70 @@ class ExecutionStep(db.Model):
         }
 
 
+# ---------------------------------------------------------------------------
+# Extraction — « dossier patient extrait » (brique 2)
+#
+# ⚠️ CANAL EXTRACTION ≠ canal apprentissage. Ces tables conservent les
+# VRAIES données patient (patient_ref, ExtractedField.value) : c'est le but,
+# constituer le dossier. Elles NE doivent PAS être anonymisées/tokenisées
+# (à l'inverse du canal apprentissage, cf. pii_sanitizer). Aucun appel
+# d'assainissement PII ne doit cibler ces colonnes.
+#
+# Sémantique de preuve réutilisée de contracts/evidence.py (VWBEvidence) :
+# screenshot_ref ≈ screenshot, screen_bbox/bbox ≈ highlight_box, confidence
+# ≈ confidence_score, created_at ≈ timestamp.
+# ---------------------------------------------------------------------------
+
+class ExtractionJob(db.Model):
+    """Dossier patient extrait — racine d'une session d'extraction."""
+    __tablename__ = 'extraction_jobs'
+
+    id = db.Column(db.String(64), primary_key=True)
+    patient_ref = db.Column(db.String(255), nullable=True)  # donnée patient EN CLAIR (volontaire)
+    source_session_id = db.Column(db.String(64), nullable=True)
+    created_at = db.Column(db.DateTime, default=datetime.utcnow)
+    # status: 'needs_review' (revue humaine requise) | 'complete' (validé)
+    status = db.Column(db.String(32), default='needs_review')
+
+    tables = db.relationship('ExtractedTable', backref='job', lazy='dynamic',
+                             cascade='all, delete-orphan')
+
+    def __repr__(self):
+        return f'<ExtractionJob {self.id}: {self.status}>'
+
+
+class ExtractedTable(db.Model):
+    """Tableau extrait d'un écran (preuve : screenshot_ref + screen_bbox)."""
+    __tablename__ = 'extracted_tables'
+
+    id = db.Column(db.String(64), primary_key=True)
+    job_id = db.Column(db.String(64), db.ForeignKey('extraction_jobs.id'), nullable=False)
+    screen_bbox = db.Column(db.JSON, nullable=True)  # {x, y, width, height}
+    screenshot_ref = db.Column(db.String(512), nullable=True)
+
+    fields = db.relationship('ExtractedField', backref='table', lazy='dynamic',
+                             cascade='all, delete-orphan')
+
+    def __repr__(self):
+        return f'<ExtractedTable {self.id}>'
+
+
+class ExtractedField(db.Model):
+    """Cellule extraite (donnée patient EN CLAIR) + preuve bbox/confidence."""
+    __tablename__ = 'extracted_fields'
+
+    id = db.Column(db.String(64), primary_key=True)
+    table_id = db.Column(db.String(64), db.ForeignKey('extracted_tables.id'), nullable=False)
+    row = db.Column(db.Integer, nullable=True)
+    col = db.Column(db.Integer, nullable=True)
+    value = db.Column(db.Text, nullable=True)  # valeur patient EN CLAIR (volontaire)
+    bbox = db.Column(db.JSON, nullable=True)   # {x, y, width, height}
+    confidence = db.Column(db.Float, nullable=True)
+
+    def __repr__(self):
+        return f'<ExtractedField {self.id}: r{self.row}c{self.col}>'
+
+
 # Session active (en mémoire, pas en DB)
 class SessionState:
     """État de la session utilisateur (en mémoire)"""

visual_workflow_builder/backend/tests/unit/test_extraction_models.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""
+Test TDD — Extraction (brique 2) : modèle « dossier patient extrait ».
+
+Objectif : valider les 3 modèles métier d'extraction (absents avant cette brique) :
+    ExtractionJob → ExtractedTable → ExtractedField
+avec leurs relations, cascade, et le `status` ∈ {complete, needs_review}.
+
+⚠️ CANAL EXTRACTION ≠ canal apprentissage : ici on conserve les **vraies
+données patient** (le but est de constituer le dossier). Pas d'anonymisation.
+Le test pose donc une valeur patient en clair et vérifie qu'elle est restituée
+telle quelle.
+
+Isolation (même pattern que test_import_core_workflow_to_db.py) :
+- pas d'app Flask complète (`app.py`), pas de socketio/blueprints ;
+- `db` partagé (`db.models.db`) lié à une SQLite **en mémoire**.
+"""
+
+import sys
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+from flask import Flask
+
+_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  # noqa: E402
+
+
+@pytest.fixture
+def db_app():
+    """App Flask minimale liée à une SQLite en mémoire, schéma créé."""
+    app = Flask("test_extraction_models")
+    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()
+
+
+def test_extraction_job_table_field_chain(db_app):
+    """Chaîne complète Job → Table → Field, relations + status par défaut."""
+    from db.models import ExtractionJob, ExtractedTable, ExtractedField
+
+    with db_app.app_context():
+        job = ExtractionJob(
+            id="job_001",
+            patient_ref="MOREL Catherine",  # donnée patient EN CLAIR (canal extraction)
+            source_session_id="sess_extract_001",
+        )
+
+        table = ExtractedTable(
+            id="tbl_001",
+            job=job,
+            screen_bbox={"x": 10, "y": 20, "width": 300, "height": 120},
+            screenshot_ref="data/extract/sess_extract_001/screen_0.png",
+        )
+        field = ExtractedField(
+            id="fld_001",
+            table=table,
+            row=0,
+            col=1,
+            value="1975-04-12",
+            bbox={"x": 110, "y": 22, "width": 80, "height": 18},
+            confidence=0.94,
+        )
+
+        db.session.add(job)
+        db.session.commit()
+
+        # status par défaut appliqué à l'INSERT = needs_review (revue humaine requise)
+        assert job.status == "needs_review"
+
+        # Relations descendantes
+        assert job.tables.count() == 1
+        assert job.tables.first().fields.count() == 1
+
+        # Relations remontantes
+        f = ExtractedField.query.get("fld_001")
+        assert f.table.job.patient_ref == "MOREL Catherine"  # patient conservé en clair
+        assert f.value == "1975-04-12"
+        assert f.bbox["width"] == 80
+        assert f.confidence == pytest.approx(0.94)
+        assert f.table.screen_bbox["height"] == 120
+
+
+def test_status_complete_is_accepted(db_app):
+    """`status` accepte 'complete' (extraction validée)."""
+    from db.models import ExtractionJob
+
+    with db_app.app_context():
+        job = ExtractionJob(id="job_ok", patient_ref="DUPONT Jean", status="complete")
+        db.session.add(job)
+        db.session.commit()
+        assert ExtractionJob.query.get("job_ok").status == "complete"
+        assert job.created_at is not None and isinstance(job.created_at, datetime)
+
+
+def test_cascade_delete_removes_children(db_app):
+    """Supprimer le Job supprime tables + fields (cascade, pas d'orphelins)."""
+    from db.models import ExtractionJob, ExtractedTable, ExtractedField
+
+    with db_app.app_context():
+        job = ExtractionJob(id="job_del", patient_ref="X")
+        table = ExtractedTable(id="tbl_del", job=job, screen_bbox={}, screenshot_ref="s.png")
+        ExtractedField(id="fld_del", table=table, row=0, col=0, value="v",
+                       bbox={}, confidence=0.5)
+        db.session.add(job)
+        db.session.commit()
+
+        db.session.delete(job)
+        db.session.commit()
+
+        assert ExtractionJob.query.count() == 0
+        assert ExtractedTable.query.count() == 0
+        assert ExtractedField.query.count() == 0