feat(gui): garde-fou runtime — désactive un moteur optionnel non embarqué
Condition du GO-CONDITIONNEL Qwen sur le lot engine capabilities (cb3b767/890edb3/5e5f0bd) : un profil YAML forçant enable_eds/enable_gliner ne doit pas déclencher un chargement voué à l'échec silencieux. NerManagers.ensure_loaded() applique désormais un garde-fou via la sonde engine_capabilities.capabilities_map() (injectable) AVANT toute tentative de load EDS/GLiNER : si le moteur optionnel demandé est indisponible dans le build courant → warning + désactivation forcée dans les réglages runtime. Best-effort (sonde en échec ⇒ réglages inchangés, les try/except de load protègent déjà). Sonde légère (find_spec), aucun import lourd. CamemBERT (requis) inchangé. Diff limité au garde-fou + tests cibles. TDD : 4 tests (test_gui_v6_engine_bridge.py) — eds/gliner indispo désactivés et jamais chargés, moteur dispo conservé, fail-safe sonde. 282 unit passed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -18,13 +18,20 @@ Aucune logique de détection ici : on orchestre uniquement.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Optional
|
||||
from typing import Any, Callable, Dict, Optional
|
||||
|
||||
from engine_capabilities import capabilities_map
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
ProcessFn = Callable[[Path, Path], dict]
|
||||
ManagerFactory = Callable[[], Any]
|
||||
# Fournisseur de capabilities : () -> {key: objet exposant `.available`/`.reason`}.
|
||||
CapsProvider = Callable[[], Dict[str, Any]]
|
||||
|
||||
|
||||
class ManagerState(str, Enum):
|
||||
@@ -86,9 +93,12 @@ class NerManagers:
|
||||
self,
|
||||
settings: EngineSettings,
|
||||
factories: Optional[dict[str, ManagerFactory]] = None,
|
||||
caps_provider: Optional[CapsProvider] = None,
|
||||
) -> None:
|
||||
self._settings = settings
|
||||
self._factories = factories if factories is not None else _default_factories()
|
||||
# Sonde « moteurs réellement embarqués » (injectable pour les tests).
|
||||
self._caps_provider = caps_provider if caps_provider is not None else capabilities_map
|
||||
self._camembert: Any = None
|
||||
self._eds: Any = None
|
||||
self._gliner: Any = None
|
||||
@@ -109,6 +119,43 @@ class NerManagers:
|
||||
"camembert_manager": self._camembert,
|
||||
}
|
||||
|
||||
def _apply_capability_guard(self) -> None:
|
||||
"""Désactive un moteur optionnel demandé mais non embarqué (fail-safe).
|
||||
|
||||
Empêche qu'un profil YAML forçant ``enable_eds``/``enable_gliner`` ne
|
||||
déclenche un chargement voué à l'échec silencieux : si la sonde
|
||||
``engine_capabilities`` indique le moteur indisponible, on log un
|
||||
warning et on force la désactivation dans les réglages runtime.
|
||||
|
||||
Best-effort : toute erreur de sonde laisse les réglages inchangés (les
|
||||
``try/except`` de chargement protègent déjà contre un crash). La sonde
|
||||
reste légère (``find_spec``) — aucun import lourd ici.
|
||||
"""
|
||||
requested = []
|
||||
if self._settings.enable_eds:
|
||||
requested.append(("eds", "EDS-Pseudo"))
|
||||
if self._settings.enable_gliner:
|
||||
requested.append(("gliner", "GLiNER"))
|
||||
if not requested:
|
||||
return
|
||||
try:
|
||||
caps = self._caps_provider()
|
||||
except Exception: # noqa: BLE001 — best-effort, ne jamais bloquer le load
|
||||
return
|
||||
for key, label in requested:
|
||||
cap = caps.get(key) if hasattr(caps, "get") else None
|
||||
if cap is not None and not getattr(cap, "available", False):
|
||||
log.warning(
|
||||
"%s demandé par la configuration mais non embarqué dans "
|
||||
"cette version — désactivation forcée (%s)",
|
||||
label,
|
||||
getattr(cap, "reason", ""),
|
||||
)
|
||||
if key == "eds":
|
||||
self._settings.enable_eds = False
|
||||
else:
|
||||
self._settings.enable_gliner = False
|
||||
|
||||
def ensure_loaded(self) -> ManagerState:
|
||||
"""Charge les managers requis si nécessaire. Idempotent, sans crash."""
|
||||
if not self._settings.use_local_ner:
|
||||
@@ -118,6 +165,8 @@ class NerManagers:
|
||||
return self._state
|
||||
|
||||
self._state = ManagerState.LOADING
|
||||
# Garde-fou : ne jamais tenter de charger un moteur optionnel non embarqué.
|
||||
self._apply_capability_guard()
|
||||
try:
|
||||
# CamemBERT-bio est le NER local principal (obligatoire si NER actif).
|
||||
self._camembert = self._factories["camembert"]()
|
||||
|
||||
@@ -126,6 +126,89 @@ def test_optional_manager_failure_is_tolerated():
|
||||
assert managers.use_hf is True
|
||||
|
||||
|
||||
# -- garde-fou capabilities runtime ----------------------------------------
|
||||
|
||||
|
||||
class _FakeCap:
|
||||
"""Capability minimale pour injecter une sonde dans les tests."""
|
||||
|
||||
def __init__(self, available, reason="(test)"):
|
||||
self.available = available
|
||||
self.reason = reason
|
||||
|
||||
|
||||
def _caps_provider(eds_ok, gliner_ok):
|
||||
def provider():
|
||||
return {
|
||||
"camembert": _FakeCap(True),
|
||||
"eds": _FakeCap(eds_ok),
|
||||
"gliner": _FakeCap(gliner_ok),
|
||||
}
|
||||
|
||||
return provider
|
||||
|
||||
|
||||
def test_guard_disables_unavailable_eds_before_load():
|
||||
# Profil/config forçant EDS alors que le moteur n'est pas embarqué.
|
||||
settings = EngineSettings(enable_eds=True)
|
||||
counter = {"camembert": 0, "eds": 0, "gliner": 0}
|
||||
managers = NerManagers(
|
||||
settings,
|
||||
factories=_counting_factories(counter),
|
||||
caps_provider=_caps_provider(eds_ok=False, gliner_ok=True),
|
||||
)
|
||||
assert managers.ensure_loaded() == ManagerState.READY
|
||||
assert settings.enable_eds is False # désactivation forcée
|
||||
assert counter["eds"] == 0 # jamais tenté de charger
|
||||
assert managers.as_kwargs()["ner_manager"] is None
|
||||
|
||||
|
||||
def test_guard_disables_unavailable_gliner_before_load():
|
||||
settings = EngineSettings(enable_gliner=True)
|
||||
counter = {"camembert": 0, "eds": 0, "gliner": 0}
|
||||
managers = NerManagers(
|
||||
settings,
|
||||
factories=_counting_factories(counter),
|
||||
caps_provider=_caps_provider(eds_ok=True, gliner_ok=False),
|
||||
)
|
||||
assert managers.ensure_loaded() == ManagerState.READY
|
||||
assert settings.enable_gliner is False
|
||||
assert counter["gliner"] == 0
|
||||
assert managers.as_kwargs()["gliner_manager"] is None
|
||||
|
||||
|
||||
def test_guard_keeps_available_engine_enabled():
|
||||
settings = EngineSettings(enable_eds=True, enable_gliner=True)
|
||||
counter = {"camembert": 0, "eds": 0, "gliner": 0}
|
||||
managers = NerManagers(
|
||||
settings,
|
||||
factories=_counting_factories(counter),
|
||||
caps_provider=_caps_provider(eds_ok=True, gliner_ok=True),
|
||||
)
|
||||
assert managers.ensure_loaded() == ManagerState.READY
|
||||
assert settings.enable_eds is True
|
||||
assert settings.enable_gliner is True
|
||||
assert counter["eds"] == 1
|
||||
assert counter["gliner"] == 1
|
||||
|
||||
|
||||
def test_guard_failsafe_when_probe_raises():
|
||||
settings = EngineSettings(enable_eds=True)
|
||||
counter = {"camembert": 0, "eds": 0, "gliner": 0}
|
||||
|
||||
def boom():
|
||||
raise RuntimeError("probe ko")
|
||||
|
||||
managers = NerManagers(
|
||||
settings, factories=_counting_factories(counter), caps_provider=boom
|
||||
)
|
||||
# Best-effort : une sonde en échec ne bloque pas le chargement et ne
|
||||
# modifie pas les réglages (les try/except de load protègent déjà).
|
||||
assert managers.ensure_loaded() == ManagerState.READY
|
||||
assert settings.enable_eds is True
|
||||
assert counter["eds"] == 1
|
||||
|
||||
|
||||
# -- make_process_fn -------------------------------------------------------
|
||||
|
||||
def test_process_fn_calls_engine_with_kwargs(tmp_path):
|
||||
|
||||
Reference in New Issue
Block a user