feat(workflow): variables runtime + extract_text serveur + t2a_decision LLM

Pipeline streaming étendu pour supporter des actions exécutées entièrement
côté serveur (jamais transmises à l'Agent V1) qui produisent des variables
réutilisables dans les steps suivants via templating {{var}} ou {{var.field}}.

== Variables d'exécution ==
- replay_state["variables"] : Dict[str, Any] initialisé vide à la création
- _resolve_runtime_vars() : résout {{var}} et {{var.field}} récursivement
  dans str/dict/list. Variables absentes laissées intactes.
- /replay/next applique la résolution sur l'action AVANT toute interception
  ou envoi à l'Agent V1.

== Boucle d'exécution serveur ==
- _SERVER_SIDE_ACTION_TYPES = {"extract_text", "t2a_decision"}
- /replay/next pop+execute en boucle ces actions jusqu'à trouver une action
  visuelle (à transmettre Agent V1) ou un pause_for_human (qui bloque).
- Latence acceptable : t2a_decision = 5-10s côté serveur, l'Agent V1 attend
  la réponse HTTP.

== Action extract_text ==
- Handler côté serveur réutilisant le dernier heartbeat (max 5s d'âge)
- core/llm/ocr_extractor.py : EasyOCR fr+en singleton + extract_text_from_image
- Stockage dans replay_state["variables"][output_var]
- Robuste : pas de heartbeat → variable = "" + log warning, pipeline continue

== Action t2a_decision ==
- core/llm/t2a_decision.py : refactor de demo_app.py query_model en module
  importable. Prompt expert DIM T2A/PMSI, qwen2.5:7b par défaut (100% bench).
- Handler côté serveur appelle analyze_dpi(input_template_resolved)
- Stockage du JSON décision dans replay_state["variables"][output_var]
- Erreurs (Ollama down, parse) → variable = INDETERMINE + _error, pipeline continue

== VWB UI ==
- types.ts : nouveau type 't2a_decision' (icône 🧠 catégorie logic)
- extract_text refondu : needsAnchor=false, paramètre output_var (au lieu de
  variable_name legacy — bridge accepte les deux pour compat)
- Bridge VWB→core : passthrough des deux types + paramètres préservés

== Tests ==
- tests/integration/test_t2a_extract.py : 25 tests verts
  - templating runtime (8 tests)
  - handler extract_text (3 tests, OCR mocké)
  - handler t2a_decision (3 tests, analyze_dpi mocké)
  - edge → action normalisée (2 tests)
  - bridge VWB → core (5 tests)
  - workflow chain extract→t2a→pause→clic (1 test)

Total branche : 82/82 verts.

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

agent_v0/server_v1/api_stream.py

@@ -219,6 +219,10 @@ from .replay_engine import (
     _is_learned_workflow,
     _edge_to_normalized_actions,
     _substitute_variables,
+    _resolve_runtime_vars,
+    _SERVER_SIDE_ACTION_TYPES,
+    _handle_extract_text_action,
+    _handle_t2a_decision_action,
     _expand_compound_steps,
     _pre_check_screen_state as _pre_check_screen_state_impl,
     _detect_popup_hint as _detect_popup_hint_impl,
@@ -2850,37 +2854,68 @@ async def get_next_action(session_id: str, machine_id: str = "default"):
         if not queue:
             return {"action": None, "session_id": session_id, "machine_id": machine_id}
 
-        # Peek à la prochaine action SANS la retirer (pour le pre-check)
-        action = queue[0]
+        # ── Boucle de traitement : actions serveur (extract_text, t2a_decision)
+        # exécutées entièrement côté serveur jusqu'à trouver une action visuelle
+        # à transmettre à l'Agent V1 ou un pause_for_human qui bloque le replay.
+        action = None
+        while queue:
+            action = queue[0]
 
-        # ── pause_for_human : interception avant exécution ──
-        # Cette action n'est jamais transmise à l'Agent V1. Elle bascule
-        # le replay en paused_need_help avec le message custom, ce qui
-        # déclenche la bulle interactive ChatWindow (J3.5).
-        if action.get("type") == "pause_for_human" and owning_replay is not None:
-            params = action.get("parameters") or {}
-            message = params.get("message") or "Validation requise"
-            queue.pop(0)
-            _replay_queues[session_id] = queue
-            owning_replay["status"] = "paused_need_help"
-            owning_replay["pause_message"] = message
-            owning_replay["failed_action"] = {
-                "action_id": action.get("action_id", ""),
-                "type": "pause_for_human",
-                "reason": "user_request",
-            }
-            logger.info(
-                f"Replay {owning_replay['replay_id']} pause supervisée demandée "
-                f"par le workflow : {message[:80]}"
-            )
-            return {
-                "action": None,
-                "session_id": session_id,
-                "machine_id": machine_id,
-                "replay_paused": True,
-                "pause_message": message,
-                "replay_id": owning_replay["replay_id"],
-            }
+            # Résoudre les variables runtime ({{var}} et {{var.field}})
+            if owning_replay is not None:
+                runtime_vars = owning_replay.get("variables") or {}
+                if runtime_vars:
+                    action = _resolve_runtime_vars(action, runtime_vars)
+
+            type_ = action.get("type")
+
+            # pause_for_human : bascule en paused_need_help, return action=None
+            if type_ == "pause_for_human" and owning_replay is not None:
+                params = action.get("parameters") or {}
+                message = params.get("message") or "Validation requise"
+                queue.pop(0)
+                _replay_queues[session_id] = queue
+                owning_replay["status"] = "paused_need_help"
+                owning_replay["pause_message"] = message
+                owning_replay["failed_action"] = {
+                    "action_id": action.get("action_id", ""),
+                    "type": "pause_for_human",
+                    "reason": "user_request",
+                }
+                logger.info(
+                    f"Replay {owning_replay['replay_id']} pause supervisée demandée "
+                    f"par le workflow : {message[:80]}"
+                )
+                return {
+                    "action": None,
+                    "session_id": session_id,
+                    "machine_id": machine_id,
+                    "replay_paused": True,
+                    "pause_message": message,
+                    "replay_id": owning_replay["replay_id"],
+                }
+
+            # Actions serveur : exécuter, pop, continuer
+            if type_ in _SERVER_SIDE_ACTION_TYPES and owning_replay is not None:
+                try:
+                    if type_ == "extract_text":
+                        _handle_extract_text_action(
+                            action, owning_replay, session_id, _last_heartbeat
+                        )
+                    elif type_ == "t2a_decision":
+                        _handle_t2a_decision_action(action, owning_replay)
+                except Exception as e:
+                    logger.warning(f"Action serveur {type_} a levé : {e}")
+                queue.pop(0)
+                _replay_queues[session_id] = queue
+                continue  # action suivante
+
+            # Action visuelle : sortir de la boucle pour la transmettre à l'Agent V1
+            break
+
+        # Si la queue s'est vidée après les exécutions serveur, rien à transmettre
+        if not queue or action is None:
+            return {"action": None, "session_id": session_id, "machine_id": machine_id}
 
     # ---- Pre-check écran (optionnel, non bloquant) ----
     # Ne s'applique qu'aux actions qui ont un from_node (actions de workflow,

agent_v0/server_v1/replay_engine.py

@@ -34,7 +34,14 @@ _ALLOWED_ACTION_TYPES = {
     "double_click", "right_click", "drag",
     "verify_screen",     # Replay hybride : vérification visuelle entre groupes
     "pause_for_human",   # Pause supervisée explicite (interceptée par /replay/next)
+    "extract_text",      # OCR serveur sur dernier heartbeat → variable workflow
+    "t2a_decision",      # Analyse LLM facturation T2A → variable workflow
 }
+
+# Types d'actions exécutées CÔTÉ SERVEUR (jamais transmises à l'Agent V1).
+# Le pipeline /replay/next les traite en boucle interne et passe à l'action
+# suivante jusqu'à trouver une action visuelle (à transmettre au client).
+_SERVER_SIDE_ACTION_TYPES = {"extract_text", "t2a_decision"}
 _MAX_ACTION_TEXT_LENGTH = 10000
 _MAX_KEYS_PER_COMBO = 10
 # Touches autorisées dans les key_combo (modificateurs + touches spéciales + caractères simples)
@@ -860,6 +867,23 @@ def _edge_to_normalized_actions(edge, params: Dict[str, Any]) -> List[Dict[str,
         }
         return [normalized]  # pas de target/coords pour cette action logique
 
+    elif action_type == "extract_text":
+        normalized["type"] = "extract_text"
+        normalized["parameters"] = {
+            "output_var": action_params.get("output_var", "extracted_text"),
+            "paragraph": bool(action_params.get("paragraph", True)),
+        }
+        return [normalized]
+
+    elif action_type == "t2a_decision":
+        normalized["type"] = "t2a_decision"
+        normalized["parameters"] = {
+            "input_template": action_params.get("input_template", ""),
+            "output_var": action_params.get("output_var", "t2a_result"),
+            "model": action_params.get("model"),
+        }
+        return [normalized]
+
     else:
         logger.warning(f"Type d'action inconnu : {action_type}")
         return []
@@ -894,6 +918,143 @@ def _substitute_variables(text: str, params: Dict[str, Any], defaults: Dict[str,
     return re.sub(r'\$\{(\w+)\}', replacer, text)
 
 
+# Regex pour le templating runtime : {{var}} ou {{var.champ}} ou {{var.champ.sous}}
+_RUNTIME_VAR_PATTERN = re.compile(r'\{\{\s*(\w+)(?:\.([\w.]+))?\s*\}\}')
+
+
+def _resolve_runtime_vars_in_str(text: str, variables: Dict[str, Any]) -> str:
+    """Remplace {{var}} et {{var.field}} par leur valeur depuis le dict variables.
+
+    Variables/champs absents : laissés tels quels (ne casse pas le pipeline).
+    Pour les valeurs non-str (dict, list), str() est appelé.
+    """
+    def replacer(match):
+        var_name = match.group(1)
+        path = match.group(2)
+        if var_name not in variables:
+            return match.group(0)
+        value = variables[var_name]
+        if path:
+            for field in path.split('.'):
+                if isinstance(value, dict) and field in value:
+                    value = value[field]
+                else:
+                    return match.group(0)
+        return str(value)
+
+    return _RUNTIME_VAR_PATTERN.sub(replacer, text)
+
+
+def _resolve_runtime_vars(value: Any, variables: Dict[str, Any]) -> Any:
+    """Résout récursivement les {{var}} et {{var.field}} dans une valeur.
+
+    Supporte str, dict, list. Les autres types sont retournés tels quels.
+    Si variables est vide ou None, value est retournée inchangée.
+    """
+    if not variables:
+        return value
+    if isinstance(value, str):
+        return _resolve_runtime_vars_in_str(value, variables)
+    if isinstance(value, dict):
+        return {k: _resolve_runtime_vars(v, variables) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_resolve_runtime_vars(item, variables) for item in value]
+    return value
+
+
+# =========================================================================
+# Handlers pour les actions exécutées côté serveur (extract_text, t2a_decision)
+# =========================================================================
+
+def _handle_extract_text_action(
+    action: Dict[str, Any],
+    replay_state: Dict[str, Any],
+    session_id: str,
+    last_heartbeat: Dict[str, Dict[str, Any]],
+) -> bool:
+    """Traite une action extract_text côté serveur. Stocke le texte OCRisé dans
+    replay_state["variables"][output_var]. Retourne True si succès.
+
+    Robuste aux échecs : si pas de heartbeat ou OCR raté, stocke "" et retourne
+    False (le pipeline continue, pas de blocage).
+    """
+    params = action.get("parameters") or {}
+    output_var = (params.get("output_var") or "extracted_text").strip()
+    paragraph = bool(params.get("paragraph", True))
+
+    heartbeat = last_heartbeat.get(session_id) or {}
+    path = heartbeat.get("path")
+    text = ""
+
+    if path:
+        try:
+            from core.llm import extract_text_from_image
+            text = extract_text_from_image(path, paragraph=paragraph)
+        except Exception as e:
+            logger.warning("extract_text OCR échoué (%s) — variable '%s' = ''", e, output_var)
+    else:
+        logger.warning(
+            "extract_text : pas de heartbeat pour session %s — variable '%s' = ''",
+            session_id, output_var,
+        )
+
+    replay_state.setdefault("variables", {})[output_var] = text
+    logger.info(
+        "extract_text → variable '%s' (%d chars) replay %s",
+        output_var, len(text), replay_state.get("replay_id", "?"),
+    )
+    return bool(text)
+
+
+def _handle_t2a_decision_action(
+    action: Dict[str, Any],
+    replay_state: Dict[str, Any],
+) -> bool:
+    """Traite une action t2a_decision côté serveur. Stocke le résultat JSON
+    dans replay_state["variables"][output_var]. Retourne True si succès.
+
+    Le DPI à analyser vient de action.parameters.input_template (déjà résolu
+    par _resolve_runtime_vars donc les {{var}} sont remplis).
+    """
+    params = action.get("parameters") or {}
+    output_var = (params.get("output_var") or "t2a_result").strip()
+    dpi_text = (params.get("input_template") or params.get("dpi") or "").strip()
+    model = params.get("model") or None  # None → DEFAULT_MODEL
+
+    if not dpi_text:
+        logger.warning(
+            "t2a_decision : input vide — variable '%s' = {decision: 'INDETERMINE'}", output_var,
+        )
+        replay_state.setdefault("variables", {})[output_var] = {
+            "decision": "INDETERMINE",
+            "justification": "DPI vide ou non extrait",
+            "confiance": "faible",
+            "_error": "empty_input",
+        }
+        return False
+
+    try:
+        from core.llm import analyze_dpi, DEFAULT_MODEL
+        result = analyze_dpi(dpi_text, model=model or DEFAULT_MODEL)
+    except Exception as e:
+        logger.warning("t2a_decision : analyze_dpi exception %s", e)
+        result = {
+            "decision": "INDETERMINE",
+            "justification": f"Erreur analyse : {e}",
+            "confiance": "faible",
+            "_error": str(e),
+        }
+
+    replay_state.setdefault("variables", {})[output_var] = result
+    decision = result.get("decision", "?")
+    elapsed = result.get("_elapsed_s", "?")
+    logger.info(
+        "t2a_decision → variable '%s' decision=%s (%ss) replay %s",
+        output_var, decision, elapsed, replay_state.get("replay_id", "?"),
+    )
+    return "_error" not in result
+
+
 def _expand_compound_steps(
     steps: List[Dict[str, Any]], base: Dict[str, Any], params: Dict[str, Any]
 ) -> List[Dict[str, Any]]:
@@ -1216,6 +1377,10 @@ def _create_replay_state(
         # Champs pour pause supervisée (target_not_found)
         "failed_action": None,   # Contexte de l'action en echec (quand paused_need_help)
         "pause_message": None,   # Message a afficher a l'utilisateur
+        # Variables d'exécution produites en cours de workflow (extract_text,
+        # t2a_decision, etc.). Résolues via templating {{var}} ou {{var.field}}
+        # dans les paramètres des actions suivantes.
+        "variables": {},
     }