feat: ExecutionCompiler — compile WorkflowIR en plan d'exécution borné

Pièce maîtresse de l'architecture V4 : - ExecutionPlan : nœuds avec stratégies de résolution pré-compilées - ExecutionCompiler : WorkflowIR → ExecutionPlan déterministe - Résolution : OCR (primaire, 100ms) > template > VLM (exception handler) - Chaque nœud : timeout, max_retries, recovery, condition de succès - Variables substituables, versionné, sérialisable JSON - 18 tests (compilation, stratégies, fallbacks, variables, roundtrip) "Le LLM compile. Le runtime exécute." Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-09 22:21:40 +02:00
parent cc673755f7
commit bffcfb2db3
3 changed files with 795 additions and 0 deletions
--- a/tests/unit/test_execution_compiler.py
+++ b/tests/unit/test_execution_compiler.py
@@ -0,0 +1,264 @@
+"""
+Tests de l'ExecutionCompiler et de l'ExecutionPlan.
+
+Vérifie que :
+- Le compilateur produit un plan déterministe depuis un WorkflowIR
+- Les stratégies de résolution sont correctement compilées (OCR > template > VLM)
+- Les timeouts, retries et recovery sont définis
+- Le plan est sérialisable et versionné
+"""
+
+import json
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+
+import pytest
+
+_ROOT = str(Path(__file__).resolve().parents[2])
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+from core.workflow.workflow_ir import WorkflowIR
+from core.workflow.execution_plan import ExecutionPlan, ExecutionNode, ResolutionStrategy, SuccessCondition
+from core.workflow.execution_compiler import ExecutionCompiler
+
+
+# =========================================================================
+# ExecutionPlan — format et sérialisation
+# =========================================================================
+
+
+class TestExecutionPlan:
+
+    def test_serialisation_roundtrip(self):
+        plan = ExecutionPlan(
+            plan_id="plan_test",
+            workflow_id="wf_123",
+            version=1,
+        )
+        plan.nodes.append(ExecutionNode(
+            node_id="n1",
+            action_type="click",
+            intent="Cliquer sur Enregistrer",
+            strategy_primary=ResolutionStrategy(method="ocr", target_text="Enregistrer"),
+            strategy_fallbacks=[ResolutionStrategy(method="vlm", vlm_description="bouton Enregistrer")],
+            success_condition=SuccessCondition(method="title_match", expected_title="Fichier sauvegardé"),
+        ))
+
+        json_str = plan.to_json()
+        plan2 = ExecutionPlan.from_json(json_str)
+
+        assert plan2.plan_id == "plan_test"
+        assert len(plan2.nodes) == 1
+        assert plan2.nodes[0].strategy_primary.method == "ocr"
+        assert len(plan2.nodes[0].strategy_fallbacks) == 1
+        assert plan2.nodes[0].success_condition.method == "title_match"
+
+    def test_save_load(self):
+        tmpdir = tempfile.mkdtemp()
+        try:
+            plan = ExecutionPlan(plan_id="plan_save", workflow_id="wf_1")
+            plan.nodes.append(ExecutionNode(node_id="n1", action_type="click"))
+            path = plan.save(tmpdir)
+
+            plan2 = ExecutionPlan.load(str(path))
+            assert plan2.plan_id == "plan_save"
+            assert len(plan2.nodes) == 1
+        finally:
+            shutil.rmtree(tmpdir)
+
+    def test_node_avec_variable(self):
+        node = ExecutionNode(
+            node_id="n1",
+            action_type="type",
+            text="{patient}",
+            variable_name="patient",
+        )
+        d = node.to_dict()
+        assert d["variable_name"] == "patient"
+        assert d["text"] == "{patient}"
+
+    def test_node_timeout_et_retry(self):
+        node = ExecutionNode(
+            node_id="n1",
+            action_type="click",
+            timeout_ms=5000,
+            max_retries=3,
+            recovery_action="undo",
+        )
+        assert node.timeout_ms == 5000
+        assert node.max_retries == 3
+        assert node.recovery_action == "undo"
+
+
+# =========================================================================
+# ExecutionCompiler — compilation WorkflowIR → ExecutionPlan
+# =========================================================================
+
+
+class TestExecutionCompiler:
+
+    def _make_ir(self):
+        """Créer un WorkflowIR de test."""
+        ir = WorkflowIR.new("Test workflow", domain="generic")
+        ir.add_step(
+            "Ouvrir le fichier",
+            actions=[
+                {"type": "click", "target": "bouton Ouvrir", "anchor_hint": "Ouvrir"},
+                {"type": "wait", "duration_ms": 2000},
+            ],
+            precondition="L'application est ouverte",
+            postcondition="La fenêtre Ouvrir est affichée",
+        )
+        ir.add_step(
+            "Saisir le nom",
+            actions=[
+                {"type": "type", "text": "{nom_fichier}", "variable": True},
+                {"type": "key_combo", "keys": ["enter"]},
+            ],
+        )
+        ir.add_variable("nom_fichier", description="Nom du fichier", default="rapport.pdf")
+        return ir
+
+    def test_compilation_basique(self):
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        assert plan.workflow_id == ir.workflow_id
+        assert plan.total_nodes == 4  # click + wait + type + key_combo
+        assert plan.domain == "generic"
+
+    def test_click_a_strategie_resolution(self):
+        """Un clic doit avoir une stratégie primaire et des fallbacks."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        click_nodes = [n for n in plan.nodes if n.action_type == "click"]
+        assert len(click_nodes) == 1
+
+        click = click_nodes[0]
+        assert click.strategy_primary is not None
+        assert click.strategy_primary.method in ("ocr", "template", "vlm")
+        assert len(click.strategy_fallbacks) >= 1
+
+    def test_ocr_est_prioritaire(self):
+        """Quand du texte est disponible, OCR est la stratégie primaire."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        click = [n for n in plan.nodes if n.action_type == "click"][0]
+        assert click.strategy_primary.method == "ocr"
+        assert click.strategy_primary.target_text == "Ouvrir"
+
+    def test_vlm_est_fallback(self):
+        """Le VLM est toujours en dernier fallback (exception handler)."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        click = [n for n in plan.nodes if n.action_type == "click"][0]
+        vlm_fallbacks = [f for f in click.strategy_fallbacks if f.method == "vlm"]
+        assert len(vlm_fallbacks) >= 1
+
+    def test_type_a_variable(self):
+        """Une action type avec variable a le bon variable_name."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        type_nodes = [n for n in plan.nodes if n.action_type == "type"]
+        assert len(type_nodes) == 1
+        assert type_nodes[0].variable_name == "nom_fichier"
+
+    def test_wait_a_duration(self):
+        """Un wait a la bonne durée."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        wait_nodes = [n for n in plan.nodes if n.action_type == "wait"]
+        assert len(wait_nodes) == 1
+        assert wait_nodes[0].duration_ms == 2000
+
+    def test_click_a_recovery(self):
+        """Un clic a une politique de recovery."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        click = [n for n in plan.nodes if n.action_type == "click"][0]
+        assert click.recovery_action in ("escape", "undo", "close", "none")
+        assert click.max_retries >= 1
+
+    def test_postcondition_devient_success_condition(self):
+        """La postcondition du WorkflowIR devient la condition de succès du nœud."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        click = [n for n in plan.nodes if n.action_type == "click"][0]
+        assert click.success_condition is not None
+        assert "Ouvrir" in click.success_condition.description
+
+    def test_statistiques_compilation(self):
+        """Les statistiques de compilation sont correctes."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        assert plan.total_nodes == 4
+        assert plan.nodes_with_ocr >= 1
+        assert plan.estimated_duration_s > 0
+
+    def test_variables_dans_le_plan(self):
+        """Les variables du WorkflowIR sont dans le plan avec leurs valeurs par défaut."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        assert "nom_fichier" in plan.variables
+        assert plan.variables["nom_fichier"] == "rapport.pdf"
+
+    def test_params_override_defaults(self):
+        """Les params passés au compile écrasent les valeurs par défaut."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir, params={"nom_fichier": "facture_mars.pdf"})
+
+        assert plan.variables["nom_fichier"] == "facture_mars.pdf"
+
+    def test_plan_json_roundtrip(self):
+        """Compiler → JSON → recharger → même plan."""
+        compiler = ExecutionCompiler()
+        ir = self._make_ir()
+        plan = compiler.compile(ir)
+
+        json_str = plan.to_json()
+        plan2 = ExecutionPlan.from_json(json_str)
+
+        assert plan2.total_nodes == plan.total_nodes
+        assert plan2.workflow_id == plan.workflow_id
+        assert len(plan2.nodes) == len(plan.nodes)
+
+    def test_compilation_workflow_vide(self):
+        """Un workflow vide produit un plan vide."""
+        compiler = ExecutionCompiler()
+        ir = WorkflowIR.new("Vide")
+        plan = compiler.compile(ir)
+
+        assert plan.total_nodes == 0
+        assert plan.nodes == []
+
+    def test_plusieurs_domaines(self):
+        """Le compilateur fonctionne pour différents domaines."""
+        compiler = ExecutionCompiler()
+        for domain in ["tim_codage", "comptabilite", "rh_paie", "generic"]:
+            ir = WorkflowIR.new("Test", domain=domain)
+            ir.add_step("Action", actions=[{"type": "click", "target": "bouton"}])
+            plan = compiler.compile(ir)
+            assert plan.domain == domain