rpa_vision_v3/core/cognition/precondition.py

"""Précondition vérifiable + recovery — workpack B mandat/objectif.

Cf. docs/coordination/inbox_codex/2026-05-25_0610_claude-to-codex_workpack-B-mandat-objectif-preconditions.md

Précondition = l'état attendu vérifiable AVANT de tenter une action.
Recovery = mini-séquence opt-in pour rattraper l'état si non atteint.
"""

from __future__ import annotations

from dataclasses import dataclass, field, asdict
from typing import Any, Dict, List, Optional, Tuple


_VALID_KINDS = {"window_title", "scene_visible", "critic_question", "noop"}
_VALID_FAIL_ACTIONS = {"pause", "abort", "continue_with_warning"}


@dataclass(frozen=True)
class Precondition:
    """État attendu à vérifier AVANT l'action.

    Attributs
        kind                            : 'window_title' | 'scene_visible' | 'critic_question' | 'noop'
        window_title_must_contain       : substrings dont au moins une doit être présente
        window_title_must_not_contain   : substrings interdites (anti-intention)
        critic_question                 : question fermée pour le Critic Ollama
        verify_timeout_ms               : timeout de vérif
    """

    kind: str = "noop"
    window_title_must_contain: Tuple[str, ...] = field(default_factory=tuple)
    window_title_must_not_contain: Tuple[str, ...] = field(default_factory=tuple)
    critic_question: str = ""
    verify_timeout_ms: int = 2000

    def __post_init__(self):
        if self.kind not in _VALID_KINDS:
            raise ValueError(f"Precondition.kind invalide: {self.kind!r} (attendu {_VALID_KINDS})")

    def to_dict(self) -> Dict[str, Any]:
        d = asdict(self)
        d["window_title_must_contain"] = list(self.window_title_must_contain)
        d["window_title_must_not_contain"] = list(self.window_title_must_not_contain)
        return d

    @classmethod
    def from_dict(cls, data: Optional[Dict[str, Any]]) -> "Precondition":
        if not data:
            return cls()
        return cls(
            kind=str(data.get("kind", "noop") or "noop"),
            window_title_must_contain=tuple(
                str(x) for x in (data.get("window_title_must_contain") or [])
            ),
            window_title_must_not_contain=tuple(
                str(x) for x in (data.get("window_title_must_not_contain") or [])
            ),
            critic_question=str(data.get("critic_question", "") or ""),
            verify_timeout_ms=int(data.get("verify_timeout_ms", 2000) or 2000),
        )

    def is_noop(self) -> bool:
        return self.kind == "noop"

    def check_title(self, observed_title: str) -> bool:
        """Vrai si le titre observé satisfait les contraintes (must/anti)."""
        if self.kind != "window_title":
            return True
        if not observed_title:
            return False
        norm = observed_title.lower()
        for anti in self.window_title_must_not_contain:
            if anti and anti.lower() in norm:
                return False
        if not self.window_title_must_contain:
            return True
        return any(p and p.lower() in norm for p in self.window_title_must_contain)


@dataclass(frozen=True)
class PreconditionRecovery:
    """Mini-séquence opt-in de rattrapage si la précondition n'est pas atteinte.

    Attributs
        max_attempts        : nombre max d'essais de recovery (par défaut 1)
        on_recovery_fail    : 'pause' | 'abort' | 'continue_with_warning'
        actions             : liste d'actions (même schéma que les actions du replay)
    """

    max_attempts: int = 1
    on_recovery_fail: str = "pause"
    actions: Tuple[Dict[str, Any], ...] = field(default_factory=tuple)

    def __post_init__(self):
        if self.on_recovery_fail not in _VALID_FAIL_ACTIONS:
            raise ValueError(
                f"PreconditionRecovery.on_recovery_fail invalide: {self.on_recovery_fail!r} "
                f"(attendu {_VALID_FAIL_ACTIONS})"
            )
        if self.max_attempts < 0:
            raise ValueError(f"max_attempts doit être >= 0, got {self.max_attempts}")

    def to_dict(self) -> Dict[str, Any]:
        return {
            "max_attempts": self.max_attempts,
            "on_recovery_fail": self.on_recovery_fail,
            "actions": [dict(a) for a in self.actions],
        }

    @classmethod
    def from_dict(cls, data: Optional[Dict[str, Any]]) -> "PreconditionRecovery":
        if not data:
            return cls()
        raw_actions = data.get("actions") or []
        actions = tuple(dict(a) for a in raw_actions if isinstance(a, dict))
        return cls(
            max_attempts=int(data.get("max_attempts", 1) or 0),
            on_recovery_fail=str(data.get("on_recovery_fail", "pause") or "pause"),
            actions=actions,
        )

    def is_empty(self) -> bool:
        return not self.actions