feat(cognition): dataclasses Trace + SceneExpected + Precondition (Phase 2.1)
Crée les 3 dataclasses du modèle Mandat/Protocoles/Scènes v0.3 dans core/cognition/, standalone (aucun branchement runtime), avec sérialisation JSON explicite et tests offline. Préparation des phases : - Phase 2.1 plan : objet Trace (mandate_id, intention_id, scene_id, affordance_signature, expected_retour, level_of_delegation) - Workpack A : SceneExpected (monitor_index, app_name, title_patterns, title_anti, window_rect_hint, scene_role, accepted_transitions, stability_ms) + helper matches_title() - Workpack B : Precondition (kind, window_title_must_contain/anti, critic_question, verify_timeout_ms) + PreconditionRecovery (max_attempts, on_recovery_fail, actions) Toutes les dataclasses sont frozen, immutables, avec to_dict/from_dict tolérants (champs vides/None -> instance vide). Validation au __post_init__ pour Precondition.kind et PreconditionRecovery.on_recovery_fail. Aucune dépendance runtime obligatoire : si l'objet n'est pas posé sur une action, fallback comportement actuel. Aucune modif executor / api_stream / replay_engine / grounding. Tests : 22/22 passent (sérialisation JSON, contrats from_dict tolérants, validation kinds, helpers matches_title/check_title, anti-intention). Tag rollback : rollback/pre-cognition-dataclasses-2026-05-25_0610
This commit is contained in:
Dom
@@ -0,0 +1,10 @@
|
||||
from .trace import Trace
|
||||
from .scene_expected import SceneExpected
|
||||
from .precondition import Precondition, PreconditionRecovery
|
||||
|
||||
__all__ = [
|
||||
"Trace",
|
||||
"SceneExpected",
|
||||
"Precondition",
|
||||
"PreconditionRecovery",
|
||||
]
|
||||
|
||||
124
core/cognition/precondition.py
Normal file
124
core/cognition/precondition.py
Normal file
@@ -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
|
||||
100
core/cognition/scene_expected.py
Normal file
100
core/cognition/scene_expected.py
Normal file
@@ -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)
|
||||
59
core/cognition/trace.py
Normal file
59
core/cognition/trace.py
Normal file
@@ -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
|
||||
)
|
||||
Reference in New Issue
Block a user