rpa_vision_v3/agent_v0/server_v1/replay_memory.py

# agent_v0/server_v1/replay_memory.py
"""
replay_memory — Greffe de TargetMemoryStore (Fiche #18) sur le pipeline V4.

Phase 1 du plan apprentissage Léa (docs/PLAN_APPRENTISSAGE_LEA.md).

Le runtime V4 appelle :
  - `memory_lookup()` AVANT la cascade coûteuse (OCR/template/VLM)
  - `memory_record_success()` APRÈS validation post-condition (`title_match` strict)
  - `memory_record_failure()` sur les échecs

Fingerprint léger V4 : les coordonnées clic (x_pct, y_pct) sont stockées dans
les deux premières valeurs de `TargetFingerprint.bbox`, et la méthode de
résolution ayant réussi dans le champ `etype`.

Signature d'écran V4 : `sha256(normalize(window_title))[:16]`. Simple et
robuste aux données variables car les titres de fenêtre restent stables.
Les faux positifs (même titre, écrans différents) sont rattrapés par la
post-condition qui décrémentera la fiabilité via `record_failure()`.

Critère de fiabilité : 2 succès minimum et < 30% d'échecs pour déclencher
un hit (paramètres de `TargetMemoryStore.lookup`). C'est exactement la
cristallisation par répétition que l'on veut — Léa est un stagiaire qui
apprend de l'observation.

Auteur : Dom, Alice — avril 2026
"""

from __future__ import annotations

import hashlib
import logging
import os
import unicodedata
from typing import Any, Dict, Optional

logger = logging.getLogger(__name__)

# =========================================================================
# Singleton du store persistant
# =========================================================================

_MEMORY_SINGLETON: Optional[Any] = None
_MEMORY_DISABLED = False


def get_memory_store():
    """Retourne le `TargetMemoryStore` partagé, ou None si indisponible.

    Lazy-init : le store n'est créé qu'au premier appel, ce qui évite
    d'importer `core.learning.target_memory_store` à l'import du module
    (et donc d'éviter les effets de bord sur le démarrage du serveur).
    """
    global _MEMORY_SINGLETON, _MEMORY_DISABLED

    if _MEMORY_DISABLED:
        return None
    if _MEMORY_SINGLETON is not None:
        return _MEMORY_SINGLETON

    try:
        from core.learning.target_memory_store import TargetMemoryStore

        base_path = os.environ.get("RPA_LEARNING_DIR", "data/learning")
        _MEMORY_SINGLETON = TargetMemoryStore(base_path=base_path)
        logger.info(
            "replay_memory: TargetMemoryStore initialisé (base=%s)", base_path,
        )
        return _MEMORY_SINGLETON
    except Exception as exc:
        logger.warning(
            "replay_memory: TargetMemoryStore indisponible (%s) — "
            "l'apprentissage persistant est désactivé", exc,
        )
        _MEMORY_DISABLED = True
        return None


# =========================================================================
# Normalisation de texte et hash
# =========================================================================


def _norm_text(s: str) -> str:
    """Normalise un texte pour un hash stable (accents, casse, NBSP, espaces)."""
    if not s:
        return ""
    s = s.replace("\u00A0", " ").strip().lower()
    s = unicodedata.normalize("NFKD", s)
    s = "".join(ch for ch in s if not unicodedata.combining(ch))
    return " ".join(s.split())


def compute_screen_sig(window_title: str) -> str:
    """Calcule la signature d'écran V4 à partir du titre de fenêtre.

    Le `window_title` est strict depuis la phase "controle des étapes"
    (post-condition `title_match` obligatoire). C'est notre clé naturelle.
    """
    norm = _norm_text(window_title)
    if not norm:
        return ""
    return hashlib.sha256(norm.encode("utf-8")).hexdigest()[:16]


class _TargetSpecLike:
    """Adaptateur dict → objet pour `TargetMemoryStore._hash_target_spec()`.

    Le hash interne de TargetMemoryStore utilise `getattr(spec, "by_role", ...)`
    qui ne fonctionne pas avec un dict brut. On expose les attributs nécessaires.

    On intègre aussi `resolve_order` et `vlm_description` dans `context_hints`
    pour qu'ils entrent dans le hash — deux actions avec le même `by_text`
    mais un `resolve_order` différent doivent avoir des hashes distincts.
    """

    __slots__ = ("by_role", "by_text", "by_position", "context_hints")

    def __init__(self, d: Dict[str, Any]):
        self.by_role = d.get("by_role", "") or ""
        self.by_text = d.get("by_text", "") or ""
        self.by_position = d.get("by_position")

        hints = dict(d.get("context_hints") or {})
        resolve_order = d.get("resolve_order")
        if resolve_order:
            hints["_resolve_order"] = "|".join(resolve_order) if isinstance(
                resolve_order, list
            ) else str(resolve_order)
        if d.get("vlm_description"):
            hints["_vlm_desc"] = str(d["vlm_description"])
        if d.get("anchor_hint"):
            hints["_anchor_hint"] = str(d["anchor_hint"])
        self.context_hints = hints


# =========================================================================
# Lookup — consulté AVANT la cascade coûteuse
# =========================================================================


def memory_lookup(
    window_title: str,
    target_spec: Dict[str, Any],
) -> Optional[Dict[str, Any]]:
    """Cherche une résolution apprise pour cette cible sur cet écran.

    Returns:
        Dict compatible avec le format de sortie de `_resolve_target_sync`
        (resolved, method, x_pct, y_pct, score, ...) si une entrée fiable
        est trouvée. None sinon.
    """
    store = get_memory_store()
    if store is None:
        return None

    screen_sig = compute_screen_sig(window_title)
    if not screen_sig:
        return None

    try:
        spec_shim = _TargetSpecLike(target_spec)
        fp = store.lookup(screen_sig, spec_shim)
    except Exception as exc:
        logger.debug("memory_lookup: erreur lookup (%s)", exc)
        return None

    if fp is None:
        return None

    # Fingerprint léger : bbox = (x_pct, y_pct, 0, 0)
    try:
        x_pct = float(fp.bbox[0])
        y_pct = float(fp.bbox[1])
    except (TypeError, IndexError, ValueError):
        logger.debug("memory_lookup: fingerprint bbox invalide")
        return None

    # Sanity check : les pourcentages doivent être dans [0, 1]
    if not (0.0 <= x_pct <= 1.0 and 0.0 <= y_pct <= 1.0):
        logger.warning(
            "memory_lookup: coords invalides (%.3f, %.3f) pour sig=%s — "
            "entrée ignorée",
            x_pct, y_pct, screen_sig,
        )
        return None

    method = fp.etype or "memory"
    confidence = float(getattr(fp, "confidence", 0.9) or 0.9)

    logger.info(
        "memory_lookup HIT : sig=%s method=%s coords=(%.4f, %.4f) conf=%.2f "
        "target='%s'",
        screen_sig, method, x_pct, y_pct, confidence,
        (target_spec.get("by_text") or "")[:60],
    )

    return {
        "resolved": True,
        "method": f"memory_{method}",
        "x_pct": x_pct,
        "y_pct": y_pct,
        "score": confidence,
        "from_memory": True,
        "screen_sig": screen_sig,
    }


# =========================================================================
# Record — appelé APRÈS validation post-condition
# =========================================================================


def memory_record_success(
    window_title: str,
    target_spec: Dict[str, Any],
    x_pct: float,
    y_pct: float,
    method: str,
    confidence: float = 0.9,
) -> bool:
    """Enregistre une résolution réussie dans la mémoire persistante.

    À appeler APRÈS validation de la post-condition (`title_match` strict).
    """
    store = get_memory_store()
    if store is None:
        return False

    screen_sig = compute_screen_sig(window_title)
    if not screen_sig:
        return False

    # Sanity check : coordonnées dans [0, 1]
    try:
        x_pct = float(x_pct)
        y_pct = float(y_pct)
    except (TypeError, ValueError):
        logger.debug("memory_record_success: coords non numériques, skip")
        return False
    if not (0.0 <= x_pct <= 1.0 and 0.0 <= y_pct <= 1.0):
        logger.debug(
            "memory_record_success: coords hors [0,1] (%.3f, %.3f), skip",
            x_pct, y_pct,
        )
        return False

    try:
        from core.learning.target_memory_store import TargetFingerprint

        # Stripper les préfixes "memory_" empilés pour ne garder que
        # la méthode de résolution originale (ex: template_matching).
        # Sans ça, le cycle lookup → record → lookup empile "memory_"
        # indéfiniment : memory_memory_memory_template_matching.
        method_clean = method or "v4_unknown"
        while method_clean.startswith("memory_"):
            method_clean = method_clean[len("memory_"):]
        method_clean = method_clean or "v4_unknown"
        fingerprint = TargetFingerprint(
            element_id=f"v4_{method_clean}",
            bbox=(x_pct, y_pct, 0.0, 0.0),
            role=target_spec.get("by_role", "") or None,
            etype=method_clean,
            label=(target_spec.get("by_text") or "")[:200] or None,
            confidence=float(confidence),
        )

        spec_shim = _TargetSpecLike(target_spec)
        store.record_success(
            screen_signature=screen_sig,
            target_spec=spec_shim,
            fingerprint=fingerprint,
            strategy_used=method_clean,
            confidence=float(confidence),
        )

        logger.info(
            "memory_record_success: sig=%s method=%s coords=(%.4f, %.4f) "
            "target='%s'",
            screen_sig, method_clean, x_pct, y_pct,
            (target_spec.get("by_text") or "")[:60],
        )
        return True

    except Exception as exc:
        logger.warning("memory_record_success: échec (%s)", exc)
        return False


def memory_record_failure(
    window_title: str,
    target_spec: Dict[str, Any],
    error_message: str,
) -> bool:
    """Incrémente le `fail_count` pour cette (signature, target).

    Appelé quand l'action échoue OU quand la post-condition n'est pas
    satisfaite. Le `TargetMemoryStore.lookup()` ignorera cette entrée
    si le ratio d'échecs dépasse 30%.
    """
    store = get_memory_store()
    if store is None:
        return False

    screen_sig = compute_screen_sig(window_title)
    if not screen_sig:
        return False

    try:
        spec_shim = _TargetSpecLike(target_spec)
        store.record_failure(
            screen_signature=screen_sig,
            target_spec=spec_shim,
            error_message=(error_message or "unknown")[:200],
        )
        logger.debug(
            "memory_record_failure: sig=%s error='%s'",
            screen_sig, (error_message or "")[:80],
        )
        return True
    except Exception as exc:
        logger.debug("memory_record_failure: échec (%s)", exc)
        return False