"""Registre déclaratif des détecteurs narratifs (Sprint 29). Avant le Sprint 29, ajouter un nouveau type de fait imposait de toucher **quatre** fichiers : 1. ``facts.py`` — ajouter une valeur à ``FactType`` ; 2. ``detectors.py`` — écrire ``def detect_xxx(data) -> list[Fact]`` ; 3. ``detectors.py`` — l'inscrire dans le dict ``DETECTORS_BY_TYPE`` ; 4. ``arbiter.py`` — ajouter le type à la séquence ``DEFAULT_TYPE_ORDER`` au bon endroit pour la priorité éditoriale. Sprint 29 ramène le nombre de modifications à **deux** : 1. ``facts.py`` — toujours nécessaire pour le type énuméré ; 2. ``detectors.py`` — décorer la fonction avec ``@register_detector(...)``. Le décorateur : - enregistre la fonction dans un registre global trié par ``priority`` ; - vérifie qu'aucun détecteur ne se réenregistre sur le même ``FactType`` ; - laisse la fonction utilisable telle quelle (rétrocompatibilité) ; - alimente automatiquement ``arbiter.DEFAULT_TYPE_ORDER``. Conventions de priorité (« politique éditoriale » du rapport) ------------------------------------------------------------- Plus la valeur est petite, plus le fait remonte tôt en synthèse à importance égale. Pour conserver l'ordre historique du Sprint 23, on utilise un pas de 10 pour laisser de la place à des insertions futures : 10 GLOBAL_LEADER_CER qui gagne globalement 20 STATISTICAL_TIE y a-t-il un ex-aequo 30 SIGNIFICANT_GAP à quel point l'écart est solide 40 STRATUM_WINNER qui domine sur quel sous-corpus 50 STRATUM_COLLAPSE qui s'effondre sur quoi 60 ERROR_PROFILE_OUTLIER qui se trompe différemment 70 LLM_HALLUCINATION_FLAG hallucinations VLM 80 ROBUSTNESS_FRAGILE sensibilité aux dégradations 90 PARETO_ALTERNATIVE compromis coût/qualité 100 SPEED_WINNER vitesse 110 COST_OUTLIER coût aberrant 120 CONFIDENCE_WARNING mise en garde sur la fiabilité Le décorateur n'impose **pas** de pas — un détecteur tiers peut très bien utiliser ``priority=42`` pour s'insérer entre STRATUM_WINNER et STRATUM_COLLAPSE par exemple. """ from __future__ import annotations import logging import threading from dataclasses import dataclass from typing import Callable, Optional from picarones.core.narrative.facts import ( DetectorFn, DetectorRegistry, FactImportance, FactType, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Métadonnées d'un détecteur # --------------------------------------------------------------------------- @dataclass(frozen=True) class DetectorEntry: """Métadonnées d'un détecteur enregistré.""" fact_type: FactType fn: DetectorFn priority: int importance: FactImportance # --------------------------------------------------------------------------- # Registre global # --------------------------------------------------------------------------- _REGISTRY: dict[FactType, DetectorEntry] = {} _REGISTRY_LOCK = threading.Lock() def register_detector( fact_type: FactType, *, priority: int, importance: FactImportance = FactImportance.MEDIUM, ) -> Callable[[DetectorFn], DetectorFn]: """Décorateur d'enregistrement. Usage:: @register_detector(FactType.GLOBAL_LEADER_CER, priority=10, importance=FactImportance.CRITICAL) def detect_global_leader_cer(data: dict) -> list[Fact]: ... Le décorateur : - vérifie qu'aucun autre détecteur n'est déjà enregistré sur ``fact_type`` (sinon ``ValueError``) ; - vérifie que ``priority`` est un entier ; - retourne la fonction inchangée pour ne pas casser les imports existants. L'``importance`` mémorisée ici sert de **métadonnée** au registre : chaque détecteur reste libre d'émettre des ``Fact`` avec une importance différente selon le contexte (ex. CRITICAL si l'écart est gigantesque, HIGH sinon). """ def _decorator(fn: DetectorFn) -> DetectorFn: with _REGISTRY_LOCK: if fact_type in _REGISTRY: raise ValueError( f"Détecteur déjà enregistré pour {fact_type.value!r} : " f"{_REGISTRY[fact_type].fn.__name__}. Désenregistrer " "explicitement avant de réassigner." ) entry = DetectorEntry( fact_type=fact_type, fn=fn, priority=int(priority), importance=importance, ) _REGISTRY[fact_type] = entry logger.debug( "[narrative.registry] enregistré %s priority=%s importance=%s", fact_type.value, priority, importance.name, ) return fn return _decorator def unregister(fact_type: FactType) -> None: """Retire un détecteur du registre — utilisé par les tests.""" with _REGISTRY_LOCK: _REGISTRY.pop(fact_type, None) def iter_detectors() -> list[DetectorEntry]: """Retourne tous les détecteurs enregistrés, triés par ``priority``. Le tri est stable : à ``priority`` égale, l'ordre d'enregistrement est préservé (utile en présence d'extensions tierces). """ with _REGISTRY_LOCK: entries = list(_REGISTRY.values()) entries.sort(key=lambda e: e.priority) return entries def detector_for(fact_type: FactType) -> Optional[DetectorEntry]: with _REGISTRY_LOCK: return _REGISTRY.get(fact_type) def clear_registry() -> None: """Vide le registre — réservé aux tests d'isolation.""" with _REGISTRY_LOCK: _REGISTRY.clear() def default_type_order() -> tuple[FactType, ...]: """Calcule l'ordre canonique des types depuis le registre courant. Source de vérité de ``arbiter.DEFAULT_TYPE_ORDER`` depuis le Sprint 29. """ return tuple(e.fact_type for e in iter_detectors()) # --------------------------------------------------------------------------- # Pont avec ``DetectorRegistry`` historique # --------------------------------------------------------------------------- def populate_legacy_registry(registry: DetectorRegistry) -> None: """Synchronise le ``DetectorRegistry`` historique depuis le décorateur. L'objet ``DetectorRegistry`` reste l'API publique pour les consommateurs externes (cf. ``DetectorRegistry.run``) ; cette fonction l'alimente depuis le registre déclaratif courant. """ for entry in iter_detectors(): registry.register(entry.fact_type, entry.fn) __all__ = [ "DetectorEntry", "register_detector", "unregister", "iter_detectors", "detector_for", "clear_registry", "default_type_order", "populate_legacy_registry", ] # --------------------------------------------------------------------------- # Sentinel — sans usage direct ; vérifie au build qu'on n'introduit pas # de valeur ``priority`` dupliquée par accident parmi les builtins. # --------------------------------------------------------------------------- def _verify_unique_priorities() -> None: seen: dict[int, FactType] = {} for entry in iter_detectors(): if entry.priority in seen: logger.warning( "[narrative.registry] priority %s dupliquée : " "%s et %s — ordre indéterministe à priorité égale.", entry.priority, seen[entry.priority].value, entry.fact_type.value, ) else: seen[entry.priority] = entry.fact_type