From 74df0822e288995a03a4f14160acb3f47e6f5c26 Mon Sep 17 00:00:00 2001 From: Dom Date: Tue, 23 Jun 2026 18:22:30 +0200 Subject: [PATCH] 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) --- core/execution/trajectory_signature.py | 74 +++++++++++++++++++ .../test_workflow_trajectory_signature.py | 71 ++++++++++++++++++ 2 files changed, 145 insertions(+) create mode 100644 tests/unit/test_workflow_trajectory_signature.py diff --git a/core/execution/trajectory_signature.py b/core/execution/trajectory_signature.py index 3076335d1..e57c688c3 100644 --- a/core/execution/trajectory_signature.py +++ b/core/execution/trajectory_signature.py @@ -32,3 +32,77 @@ def trajectory_signature(steps: Iterable[Mapping[str, Any]]) -> str: """ 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 (by_role/by_text, fallback window).""" + if not isinstance(target, Mapping): + return "" + by_role = str(target.get("by_role") or "").strip() + by_text = str(target.get("by_text") or "").strip() + base = f"{by_role}:{by_text}".strip(":") + if base: + return base + 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)) diff --git a/tests/unit/test_workflow_trajectory_signature.py b/tests/unit/test_workflow_trajectory_signature.py new file mode 100644 index 000000000..5e265bc6d --- /dev/null +++ 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)