diff --git a/core/cognition/__init__.py b/core/cognition/__init__.py index e69de29bb..c4300b58a 100644 --- a/core/cognition/__init__.py +++ b/core/cognition/__init__.py @@ -0,0 +1,10 @@ +from .trace import Trace +from .scene_expected import SceneExpected +from .precondition import Precondition, PreconditionRecovery + +__all__ = [ + "Trace", + "SceneExpected", + "Precondition", + "PreconditionRecovery", +] diff --git a/core/cognition/precondition.py b/core/cognition/precondition.py new file mode 100644 index 000000000..b82af36a2 --- /dev/null +++ b/core/cognition/precondition.py @@ -0,0 +1,124 @@ +"""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 diff --git a/core/cognition/scene_expected.py b/core/cognition/scene_expected.py new file mode 100644 index 000000000..e0e75e289 --- /dev/null +++ b/core/cognition/scene_expected.py @@ -0,0 +1,100 @@ +"""Scène d'intention attendue — workpack A attention scope multi-écrans. + +Cf. docs/coordination/inbox_codex/2026-05-25_0610_claude-to-codex_workpack-A-attention-scope-multi-ecrans.md +""" + +from __future__ import annotations + +from dataclasses import dataclass, field, asdict +from typing import Any, Dict, List, Optional, Tuple + + +@dataclass(frozen=True) +class SceneExpected: + """Description du périmètre visuel attendu pour servir l'intention. + + Construit au build serveur, transporté additif jusqu'au client, consommé + par une garde `_assert_scene_active()` avant tout geste — surtout les + raccourcis clavier qui partent sinon dans la fenêtre active globale. + + Attributs + scene_id : ID stable de la scène + app_name : nom de l'application attendue (ex 'Notepad') + title_patterns : patterns de titre acceptables (substrings) + title_anti : patterns de titre interdits (anti-intention) + monitor_index : index du moniteur (1-based mss). None = quelconque + monitor_geometry : (left, top, width, height) en pixels. Optionnel. + window_rect_hint : (left, top, right, bottom) zone attendue. Optionnel. + scene_role : 'editor' | 'dialog' | 'menu' | 'browser_tab' | ... + required : True si le geste DOIT être bloqué si scène absente + stability_ms : durée min de stabilité avant le geste + accepted_transitions: scènes vers lesquelles transition est attendue + """ + + scene_id: str = "" + app_name: str = "" + title_patterns: Tuple[str, ...] = field(default_factory=tuple) + title_anti: Tuple[str, ...] = field(default_factory=tuple) + monitor_index: Optional[int] = None + monitor_geometry: Optional[Tuple[int, int, int, int]] = None + window_rect_hint: Optional[Tuple[int, int, int, int]] = None + scene_role: str = "" + required: bool = True + stability_ms: int = 0 + accepted_transitions: Tuple[str, ...] = field(default_factory=tuple) + + def to_dict(self) -> Dict[str, Any]: + d = asdict(self) + d["title_patterns"] = list(self.title_patterns) + d["title_anti"] = list(self.title_anti) + d["accepted_transitions"] = list(self.accepted_transitions) + if self.monitor_geometry is not None: + d["monitor_geometry"] = list(self.monitor_geometry) + if self.window_rect_hint is not None: + d["window_rect_hint"] = list(self.window_rect_hint) + return d + + @classmethod + def from_dict(cls, data: Optional[Dict[str, Any]]) -> "SceneExpected": + if not data: + return cls() + + def _tuple_of_4(v): + if v is None: + return None + try: + lst = list(v) + if len(lst) != 4: + return None + return tuple(int(x) for x in lst) + except (TypeError, ValueError): + return None + + return cls( + scene_id=str(data.get("scene_id", "") or ""), + app_name=str(data.get("app_name", "") or ""), + title_patterns=tuple(str(x) for x in (data.get("title_patterns") or [])), + title_anti=tuple(str(x) for x in (data.get("title_anti") or [])), + monitor_index=(int(data["monitor_index"]) if data.get("monitor_index") is not None else None), + monitor_geometry=_tuple_of_4(data.get("monitor_geometry")), + window_rect_hint=_tuple_of_4(data.get("window_rect_hint")), + scene_role=str(data.get("scene_role", "") or ""), + required=bool(data.get("required", True)), + stability_ms=int(data.get("stability_ms", 0) or 0), + accepted_transitions=tuple(str(x) for x in (data.get("accepted_transitions") or [])), + ) + + def matches_title(self, observed_title: str) -> bool: + """Vrai si le titre observé est cohérent avec la scène (patterns + anti).""" + if not observed_title: + return False + norm = observed_title.lower() + for anti in self.title_anti: + if anti and anti.lower() in norm: + return False + if not self.title_patterns: + return True + return any(p and p.lower() in norm for p in self.title_patterns) + + def is_empty(self) -> bool: + return not (self.scene_id or self.app_name or self.title_patterns) diff --git a/core/cognition/trace.py b/core/cognition/trace.py new file mode 100644 index 000000000..287f69b7d --- /dev/null +++ b/core/cognition/trace.py @@ -0,0 +1,59 @@ +"""Trace causale d'une action — modèle Mandat/Protocoles/Scènes v0.3. + +Cf. docs/architecture/MODELE_MANDAT_PROTOCOLS_LEA_2026-05-25_v0.3_ARBITRAGES_DOM.md +""" + +from __future__ import annotations + +from dataclasses import dataclass, field, asdict +from typing import Any, Dict, Optional + + +@dataclass(frozen=True) +class Trace: + """Contrat unificateur transporté du build au runtime à la preuve. + + Tous les champs sont optionnels (str vide / None) pour permettre une + introduction progressive sans casser les actions existantes qui n'en + portent pas. Fallback : comportement actuel si trace absente. + + Attributs + mandate_id : ID du mandat humain de niveau supérieur + intention_id : ID du sous-but courant servant le mandat + scene_id : ID de la scène d'intention pertinente + affordance_signature: signature stable de l'affordance ciblée + expected_retour : description courte du retour attendu + level_of_delegation : N0..N4 (cf v0.3 arbitrage 3) + """ + + mandate_id: str = "" + intention_id: str = "" + scene_id: str = "" + affordance_signature: str = "" + expected_retour: str = "" + level_of_delegation: int = 0 + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, data: Optional[Dict[str, Any]]) -> "Trace": + if not data: + return cls() + return cls( + mandate_id=str(data.get("mandate_id", "") or ""), + intention_id=str(data.get("intention_id", "") or ""), + scene_id=str(data.get("scene_id", "") or ""), + affordance_signature=str(data.get("affordance_signature", "") or ""), + expected_retour=str(data.get("expected_retour", "") or ""), + level_of_delegation=int(data.get("level_of_delegation", 0) or 0), + ) + + def is_empty(self) -> bool: + return not ( + self.mandate_id + or self.intention_id + or self.scene_id + or self.affordance_signature + or self.expected_retour + ) diff --git a/tests/unit/test_cognition_dataclasses.py b/tests/unit/test_cognition_dataclasses.py new file mode 100644 index 000000000..3281a0c5b --- /dev/null +++ b/tests/unit/test_cognition_dataclasses.py @@ -0,0 +1,225 @@ +"""Tests offline des dataclasses cognition (Phase 2.1 + workpacks A/B). + +Pas de runtime, pas de réseau. Sérialisation JSON + invariants. +""" + +from __future__ import annotations + +import json + +import pytest + +from core.cognition import ( + Precondition, + PreconditionRecovery, + SceneExpected, + Trace, +) + + +# ============================================================================= +# Trace +# ============================================================================= + + +def test_trace_empty_default(): + t = Trace() + assert t.is_empty() + assert t.mandate_id == "" + assert t.intention_id == "" + assert t.level_of_delegation == 0 + + +def test_trace_roundtrip_json(): + t = Trace( + mandate_id="m1", + intention_id="ouvrir_notepad", + scene_id="bloc_notes_vierge", + affordance_signature="btn:Enregistrer@SaveAs", + expected_retour="dialog Enregistrer sous apparaît", + level_of_delegation=2, + ) + payload = json.dumps(t.to_dict()) + loaded = Trace.from_dict(json.loads(payload)) + assert loaded == t + assert not loaded.is_empty() + + +def test_trace_from_dict_none_returns_empty(): + assert Trace.from_dict(None) == Trace() + + +def test_trace_from_dict_partial_tolerated(): + loaded = Trace.from_dict({"mandate_id": "m1"}) + assert loaded.mandate_id == "m1" + assert loaded.intention_id == "" + assert loaded.level_of_delegation == 0 + + +def test_trace_from_dict_level_must_be_int_parseable(): + # Contrat documenté : from_dict tolère absent/None/0, pas une string non-int. + # Le caller serveur est responsable de produire un int valide. + with pytest.raises(ValueError): + Trace.from_dict({"level_of_delegation": "not_a_number"}) + + +# ============================================================================= +# SceneExpected +# ============================================================================= + + +def test_scene_expected_empty_default(): + s = SceneExpected() + assert s.is_empty() + assert s.monitor_index is None + + +def test_scene_expected_roundtrip_json(): + s = SceneExpected( + scene_id="bloc_notes_editor", + app_name="Notepad", + title_patterns=("Sans titre", "Untitled"), + title_anti=(".txt", ".md"), + monitor_index=1, + monitor_geometry=(0, 0, 1920, 1080), + window_rect_hint=(100, 100, 1000, 800), + scene_role="editor", + required=True, + stability_ms=200, + accepted_transitions=("save_as_dialog", "confirm_save"), + ) + payload = json.dumps(s.to_dict()) + loaded = SceneExpected.from_dict(json.loads(payload)) + assert loaded == s + + +def test_scene_expected_matches_title_anti_blocks(): + s = SceneExpected( + app_name="Notepad", + title_patterns=("Sans titre",), + title_anti=(".txt",), + ) + assert s.matches_title("Sans titre – Bloc-notes") is True + assert s.matches_title("monfichier.txt – Bloc-notes") is False + assert s.matches_title("Sans titre.txt") is False # anti l'emporte + + +def test_scene_expected_matches_title_no_patterns_accepts_all_except_anti(): + s = SceneExpected(app_name="Notepad", title_anti=("Chrome",)) + assert s.matches_title("Bloc-notes") is True + assert s.matches_title("Google Chrome") is False + + +def test_scene_expected_matches_title_empty_observed_false(): + s = SceneExpected(title_patterns=("Notepad",)) + assert s.matches_title("") is False + + +def test_scene_expected_from_dict_bad_geometry_falls_back_none(): + s = SceneExpected.from_dict({"monitor_geometry": [1, 2, 3]}) # pas 4 valeurs + assert s.monitor_geometry is None + + +# ============================================================================= +# Precondition +# ============================================================================= + + +def test_precondition_noop_default(): + p = Precondition() + assert p.is_noop() + assert p.kind == "noop" + + +def test_precondition_invalid_kind_raises(): + with pytest.raises(ValueError): + Precondition(kind="unknown_kind") + + +def test_precondition_window_title_check_must_contain(): + p = Precondition( + kind="window_title", + window_title_must_contain=("Sans titre", "Untitled"), + ) + assert p.check_title("Sans titre – Bloc-notes") is True + assert p.check_title("Untitled - Notepad") is True + assert p.check_title("monfichier.txt – Bloc-notes") is False + + +def test_precondition_window_title_anti_blocks_even_if_must_present(): + p = Precondition( + kind="window_title", + window_title_must_contain=("Bloc-notes",), + window_title_must_not_contain=(".txt",), + ) + assert p.check_title("Sans titre – Bloc-notes") is True + assert p.check_title("rapport.txt – Bloc-notes") is False + + +def test_precondition_check_title_noop_returns_true(): + """Précondition noop ne bloque jamais — comportement par défaut.""" + p = Precondition(kind="noop") + assert p.check_title("n'importe quoi") is True + + +def test_precondition_roundtrip_json(): + p = Precondition( + kind="window_title", + window_title_must_contain=("Sans titre",), + window_title_must_not_contain=(".txt",), + critic_question="Le document est-il vide et non nommé ?", + verify_timeout_ms=3000, + ) + payload = json.dumps(p.to_dict()) + loaded = Precondition.from_dict(json.loads(payload)) + assert loaded == p + + +# ============================================================================= +# PreconditionRecovery +# ============================================================================= + + +def test_recovery_default_empty(): + r = PreconditionRecovery() + assert r.is_empty() + assert r.max_attempts == 1 + assert r.on_recovery_fail == "pause" + + +def test_recovery_invalid_fail_action_raises(): + with pytest.raises(ValueError): + PreconditionRecovery(on_recovery_fail="continue_aveuglement") + + +def test_recovery_negative_attempts_raises(): + with pytest.raises(ValueError): + PreconditionRecovery(max_attempts=-1) + + +def test_recovery_with_actions_serializable(): + r = PreconditionRecovery( + max_attempts=1, + on_recovery_fail="pause", + actions=( + {"type": "key_combo", "keys": ["ctrl", "n"]}, + {"type": "wait", "duration_ms": 500}, + ), + ) + payload = json.dumps(r.to_dict()) + loaded = PreconditionRecovery.from_dict(json.loads(payload)) + assert loaded == r + assert not loaded.is_empty() + + +def test_recovery_from_dict_filters_non_dict_actions(): + r = PreconditionRecovery.from_dict({ + "actions": [ + {"type": "key_combo"}, + "ignored_string", + None, + 42, + {"type": "wait"}, + ], + }) + assert len(r.actions) == 2