fix(core): signature de trajectoire stable malgre le moteur de grounding (by_text)

Le champ by_role remontait la methode de detection (yolo/ocr/vlm), instable entre
sessions : deux apprentissages du meme parcours detectes differemment produisaient
deux signatures -> fusion (create-or-update) ratee. On sort by_role de la signature
et on s'appuie sur le texte semantique de la cible (by_text), independant du moteur
de grounding. Fallback quand by_text vide : titre de fenetre / description VLM.

Test TDD: test_signature_stable_despite_grounding_role_difference (RED->GREEN).

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

core/execution/trajectory_signature.py

@@ -0,0 +1,112 @@
+"""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
+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
+
+
+def _normalize_step(step: Mapping[str, Any]) -> str:
+    action_type = str(step.get("action_type", "unknown")).strip().lower()
+    target = str(step.get("target", "")).strip()
+    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))

tests/unit/test_trajectory_signature.py

@@ -0,0 +1,59 @@
+"""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)

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)