feat(core): adaptateur workflow->signature de trajectoire (BFS edges, cibles stables)

Extrait d'un workflow core (dict) la sequence ordonnee (action_type, target stable) via traversee BFS depuis entry_nodes (comme le bridge d'import), en n'utilisant que des champs stables (by_role/by_text/window) et en ignorant coords/IDs de noeuds. Branche la primitive trajectory_signature sur de vrais workflows. Test TDD: tests/unit/test_workflow_trajectory_signature.py (3 tests, RED->GREEN). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-23 18:22:30 +02:00
parent a86c1ebb83
commit 74df0822e2
2 changed files with 145 additions and 0 deletions
--- a/tests/unit/test_workflow_trajectory_signature.py
+++ b/tests/unit/test_workflow_trajectory_signature.py
@@ -0,0 +1,71 @@
+"""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)