diff --git a/backend/app/engine/qualifier.py b/backend/app/engine/qualifier.py new file mode 100644 index 0000000..9bc0d37 --- /dev/null +++ b/backend/app/engine/qualifier.py @@ -0,0 +1,228 @@ +"""Decision qualification engine. + +Pure functions — no database, no I/O. +Takes a QualificationInput + QualificationConfig and returns a QualificationResult. + +LLM integration (suggest_modalities_from_context) is stubbed pending local Qwen deployment. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum + + +class DecisionType(str, Enum): + INDIVIDUAL = "individual" + COLLECTIVE = "collective" + + +# --------------------------------------------------------------------------- +# Configuration (thresholds — stored as a QualificationProtocol in DB) +# --------------------------------------------------------------------------- + + +@dataclass +class QualificationConfig: + """Configurable thresholds for the qualification engine. + + These defaults will be seeded as a QualificationProtocol record so they + can be adjusted through the admin interface without code changes. + + individual_max: affected_count <= this → always individual + small_group_max: affected_count <= this → individual recommended, collective available + collective_wot_min: affected_count > this → collective required (WoT formula applies) + + Default modalities shown when collective is chosen (ordered by relevance). + """ + individual_max: int = 1 + small_group_max: int = 5 + collective_wot_min: int = 50 + + default_modalities: list[str] = field(default_factory=lambda: [ + "vote_wot", + "vote_smith", + "consultation_avis", + "election", + ]) + + +# --------------------------------------------------------------------------- +# Input / Output +# --------------------------------------------------------------------------- + + +@dataclass +class QualificationInput: + within_mandate: bool = False + affected_count: int | None = None + is_structural: bool = False + context_description: str | None = None # reserved for LLM suggestion + + +@dataclass +class QualificationResult: + decision_type: DecisionType + process: str + recommended_modalities: list[str] + recommend_onchain: bool + confidence: str # "required" | "recommended" | "optional" + collective_available: bool + reasons: list[str] + onchain_reason: str | None = None + + +# --------------------------------------------------------------------------- +# LLM stub +# --------------------------------------------------------------------------- + + +def suggest_modalities_from_context( + context: str, + config: QualificationConfig, +) -> list[str]: + """Suggest voting modalities based on a natural-language context description. + + Stub — returns empty list until local Qwen (qwen3.6) is integrated. + When implemented, this will call the LLM API and return an ordered list + of modality slugs from config.default_modalities. + """ + return [] + + +# --------------------------------------------------------------------------- +# Core engine +# --------------------------------------------------------------------------- + + +def qualify(inp: QualificationInput, config: QualificationConfig) -> QualificationResult: + """Qualify a decision and recommend a type, process, and modalities. + + Rules (in priority order): + R1/R2 within_mandate → individual + consultation_avis, no modalities + R3 affected_count == 1 → individual + personal + R4 affected_count ≤ small_group_max → individual recommended, collective available + R5 small_group_max < affected_count ≤ collective_wot_min → collective recommended + R6 affected_count > collective_wot_min → collective required (WoT) + R7/R8 is_structural → recommend_onchain with reason + """ + reasons: list[str] = [] + + # ── R1/R2: mandate scope overrides everything ─────────────────────────── + if inp.within_mandate: + reasons.append("Décision dans le périmètre d'un mandat existant.") + return QualificationResult( + decision_type=DecisionType.INDIVIDUAL, + process="consultation_avis", + recommended_modalities=[], + recommend_onchain=_onchain(inp, reasons), + confidence="required", + collective_available=False, + reasons=reasons, + onchain_reason=_onchain_reason(inp), + ) + + count = inp.affected_count if inp.affected_count is not None else 1 + + # ── R3: single person ─────────────────────────────────────────────────── + if count <= config.individual_max: + reasons.append("Une seule personne concernée.") + return QualificationResult( + decision_type=DecisionType.INDIVIDUAL, + process="personal", + recommended_modalities=[], + recommend_onchain=_onchain(inp, reasons), + confidence="required", + collective_available=False, + reasons=reasons, + onchain_reason=_onchain_reason(inp), + ) + + # ── R4: small group → individual recommended, collective available ─────── + if count <= config.small_group_max: + reasons.append( + f"{count} personnes concernées : décision individuelle recommandée, " + "vote collectif possible." + ) + modalities = _collect_modalities(inp, config) + return QualificationResult( + decision_type=DecisionType.INDIVIDUAL, + process="personal", + recommended_modalities=[], + recommend_onchain=_onchain(inp, reasons), + confidence="recommended", + collective_available=True, + reasons=reasons, + onchain_reason=_onchain_reason(inp), + ) + + # ── R5/R6: medium or large group → collective ──────────────────────────── + modalities = _collect_modalities(inp, config) + + if count <= config.collective_wot_min: + reasons.append( + f"{count} personnes concernées : vote collectif recommandé." + ) + confidence = "recommended" + else: + reasons.append( + f"{count} personnes concernées : vote collectif obligatoire " + "(formule WoT applicable)." + ) + confidence = "required" + if "vote_wot" not in modalities: + modalities = ["vote_wot"] + modalities + + return QualificationResult( + decision_type=DecisionType.COLLECTIVE, + process="vote_collective", + recommended_modalities=modalities, + recommend_onchain=_onchain(inp, reasons), + confidence=confidence, + collective_available=True, + reasons=reasons, + onchain_reason=_onchain_reason(inp), + ) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _onchain(inp: QualificationInput, reasons: list[str]) -> bool: + if inp.is_structural: + reasons.append( + "Décision structurante : gravure on-chain recommandée " + "(a force de loi ou déclenche une action machine)." + ) + return inp.is_structural + + +def _onchain_reason(inp: QualificationInput) -> str | None: + if not inp.is_structural: + return None + return ( + "Cette décision est structurante : elle a valeur de loi au sein de la " + "communauté ou déclenche une action machine (ex : runtime upgrade). " + "La gravure on-chain (IPFS + system.remark) garantit son immuabilité " + "et sa vérifiabilité publique." + ) + + +def _collect_modalities( + inp: QualificationInput, + config: QualificationConfig, +) -> list[str]: + """Combine default modalities with any LLM suggestions (stub for now).""" + llm_suggestions = [] + if inp.context_description: + llm_suggestions = suggest_modalities_from_context(inp.context_description, config) + + seen: set[str] = set() + result: list[str] = [] + for m in llm_suggestions + config.default_modalities: + if m not in seen: + seen.add(m) + result.append(m) + return result diff --git a/backend/app/tests/test_qualifier.py b/backend/app/tests/test_qualifier.py new file mode 100644 index 0000000..38d917b --- /dev/null +++ b/backend/app/tests/test_qualifier.py @@ -0,0 +1,307 @@ +"""TDD — Moteur de qualification des décisions. + +Ce fichier est la source de vérité exécutable des règles métier du tunnel "Décider". +Chaque test est nommé d'après la règle qu'il encode. + +Règles implémentées dans ce fichier (RED → GREEN au fur et à mesure) : + R1 within_mandate → individual + consultation_avis + R2 within_mandate → aucune modalité de vote n'est proposée + R3 affected_count == 1 (hors mandat) → individual + R4 affected_count ≤ small_group_max → individual recommandé / collective possible + R5 small_group_max < affected_count ≤ collective_wot_min → collective recommandé + R6 affected_count > collective_wot_min → collective WoT obligatoire + R7 is_structural=True → recommend_onchain=True + raison explicite + R8 is_structural=False → recommend_onchain=False (on ne pollue pas avec l'option) + R9 decision_type est toujours dans l'enum autorisé + R10 individual n'expose jamais de modalités de vote + R11 collective expose au moins une modalité + R12 les seuils sont lus depuis QualificationConfig (configurables) +""" + +from __future__ import annotations + +import pytest + +from app.engine.qualifier import ( + DecisionType, + QualificationConfig, + QualificationInput, + QualificationResult, + qualify, +) + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +DEFAULT_CONFIG = QualificationConfig() # valeurs par défaut + + +# --------------------------------------------------------------------------- +# R1 — within_mandate → individual + consultation_avis +# --------------------------------------------------------------------------- + + +def test_r1_within_mandate_gives_individual(): + """Une décision dans le périmètre d'un mandat est toujours individuelle.""" + result = qualify(QualificationInput(within_mandate=True), DEFAULT_CONFIG) + assert result.decision_type == DecisionType.INDIVIDUAL + + +def test_r1_within_mandate_gives_consultation_avis(): + """La décision dans un mandat utilise le processus 'consultation_avis'.""" + result = qualify(QualificationInput(within_mandate=True), DEFAULT_CONFIG) + assert result.process == "consultation_avis" + + +def test_r1_within_mandate_overrides_large_affected_count(): + """Même si de nombreuses personnes sont concernées, un mandat impose 'individual'.""" + result = qualify( + QualificationInput(within_mandate=True, affected_count=500), + DEFAULT_CONFIG, + ) + assert result.decision_type == DecisionType.INDIVIDUAL + assert result.process == "consultation_avis" + + +# --------------------------------------------------------------------------- +# R2 — within_mandate → aucune modalité de vote +# --------------------------------------------------------------------------- + + +def test_r2_within_mandate_no_modalities(): + """Le mandataire décide seul après consultation — pas de vote collectif.""" + result = qualify(QualificationInput(within_mandate=True), DEFAULT_CONFIG) + assert result.recommended_modalities == [] + + +# --------------------------------------------------------------------------- +# R3 — affected_count == 1 hors mandat → individual +# --------------------------------------------------------------------------- + + +def test_r3_single_person_is_individual(): + result = qualify( + QualificationInput(within_mandate=False, affected_count=1), + DEFAULT_CONFIG, + ) + assert result.decision_type == DecisionType.INDIVIDUAL + + +def test_r3_single_person_process_is_personal(): + result = qualify( + QualificationInput(within_mandate=False, affected_count=1), + DEFAULT_CONFIG, + ) + assert result.process == "personal" + + +# --------------------------------------------------------------------------- +# R4 — 2 ≤ affected_count ≤ small_group_max → individual recommandé +# --------------------------------------------------------------------------- + + +def test_r4_small_group_recommends_individual(): + """2 à small_group_max personnes : individual recommandé, collective possible.""" + for count in range(2, DEFAULT_CONFIG.small_group_max + 1): + result = qualify( + QualificationInput(within_mandate=False, affected_count=count), + DEFAULT_CONFIG, + ) + assert result.decision_type == DecisionType.INDIVIDUAL, ( + f"affected_count={count} devrait recommander individual" + ) + assert result.collective_available is True, ( + f"affected_count={count} : collective doit rester disponible" + ) + + +def test_r4_small_group_confidence_is_recommended(): + result = qualify( + QualificationInput(within_mandate=False, affected_count=3), + DEFAULT_CONFIG, + ) + assert result.confidence == "recommended" + + +# --------------------------------------------------------------------------- +# R5 — small_group_max < affected_count ≤ collective_wot_min → collective recommandé +# --------------------------------------------------------------------------- + + +def test_r5_medium_group_recommends_collective(): + mid = DEFAULT_CONFIG.small_group_max + 1 + result = qualify( + QualificationInput(within_mandate=False, affected_count=mid), + DEFAULT_CONFIG, + ) + assert result.decision_type == DecisionType.COLLECTIVE + + +def test_r5_medium_group_confidence_is_recommended(): + mid = DEFAULT_CONFIG.small_group_max + 1 + result = qualify( + QualificationInput(within_mandate=False, affected_count=mid), + DEFAULT_CONFIG, + ) + assert result.confidence == "recommended" + + +# --------------------------------------------------------------------------- +# R6 — affected_count > collective_wot_min → collective WoT requis +# --------------------------------------------------------------------------- + + +def test_r6_large_group_requires_collective(): + result = qualify( + QualificationInput(within_mandate=False, affected_count=DEFAULT_CONFIG.collective_wot_min + 1), + DEFAULT_CONFIG, + ) + assert result.decision_type == DecisionType.COLLECTIVE + + +def test_r6_large_group_confidence_is_required(): + result = qualify( + QualificationInput(within_mandate=False, affected_count=DEFAULT_CONFIG.collective_wot_min + 1), + DEFAULT_CONFIG, + ) + assert result.confidence == "required" + + +def test_r6_large_group_includes_vote_wot_modality(): + result = qualify( + QualificationInput(within_mandate=False, affected_count=DEFAULT_CONFIG.collective_wot_min + 1), + DEFAULT_CONFIG, + ) + assert "vote_wot" in result.recommended_modalities + + +# --------------------------------------------------------------------------- +# R7 — is_structural=True → recommend_onchain + raison +# --------------------------------------------------------------------------- + + +def test_r7_structural_recommends_onchain(): + """Décision structurante (force de loi ou action machine) → on-chain proposé.""" + result = qualify( + QualificationInput(within_mandate=False, affected_count=100, is_structural=True), + DEFAULT_CONFIG, + ) + assert result.recommend_onchain is True + + +def test_r7_structural_provides_onchain_reason(): + result = qualify( + QualificationInput(within_mandate=False, affected_count=100, is_structural=True), + DEFAULT_CONFIG, + ) + assert result.onchain_reason is not None + assert len(result.onchain_reason) > 0 + + +def test_r7_structural_individual_can_also_recommend_onchain(): + """Même une décision individuelle structurante peut être gravée on-chain.""" + result = qualify( + QualificationInput(within_mandate=False, affected_count=1, is_structural=True), + DEFAULT_CONFIG, + ) + assert result.recommend_onchain is True + + +# --------------------------------------------------------------------------- +# R8 — is_structural=False → recommend_onchain=False +# --------------------------------------------------------------------------- + + +def test_r8_non_structural_never_recommends_onchain(): + """On ne pollue pas les décisions ordinaires avec l'option on-chain.""" + for count in [1, 3, 10, 100]: + result = qualify( + QualificationInput(within_mandate=False, affected_count=count, is_structural=False), + DEFAULT_CONFIG, + ) + assert result.recommend_onchain is False, ( + f"affected_count={count}, is_structural=False : on-chain ne doit pas être proposé" + ) + + +# --------------------------------------------------------------------------- +# R9 — decision_type toujours dans l'enum autorisé +# --------------------------------------------------------------------------- + + +def test_r9_decision_type_always_valid(): + valid_types = set(DecisionType) + inputs = [ + QualificationInput(within_mandate=True), + QualificationInput(within_mandate=False, affected_count=1), + QualificationInput(within_mandate=False, affected_count=3), + QualificationInput(within_mandate=False, affected_count=10), + QualificationInput(within_mandate=False, affected_count=100), + QualificationInput(within_mandate=False, affected_count=100, is_structural=True), + ] + for inp in inputs: + result = qualify(inp, DEFAULT_CONFIG) + assert result.decision_type in valid_types, ( + f"Type invalide '{result.decision_type}' pour input {inp}" + ) + + +# --------------------------------------------------------------------------- +# R10 — individual n'expose jamais de modalités +# --------------------------------------------------------------------------- + + +def test_r10_individual_no_modalities(): + for inp in [ + QualificationInput(within_mandate=True), + QualificationInput(within_mandate=False, affected_count=1), + QualificationInput(within_mandate=False, affected_count=3), + ]: + result = qualify(inp, DEFAULT_CONFIG) + if result.decision_type == DecisionType.INDIVIDUAL: + assert result.recommended_modalities == [], ( + f"Individual ne doit pas exposer de modalités : {inp}" + ) + + +# --------------------------------------------------------------------------- +# R11 — collective expose au moins une modalité +# --------------------------------------------------------------------------- + + +def test_r11_collective_has_at_least_one_modality(): + result = qualify( + QualificationInput(within_mandate=False, affected_count=20), + DEFAULT_CONFIG, + ) + assert result.decision_type == DecisionType.COLLECTIVE + assert len(result.recommended_modalities) >= 1 + + +# --------------------------------------------------------------------------- +# R12 — les seuils sont lus depuis QualificationConfig +# --------------------------------------------------------------------------- + + +def test_r12_custom_config_overrides_thresholds(): + """Les seuils viennent de QualificationConfig, pas de constantes hardcodées.""" + custom = QualificationConfig( + individual_max=1, + small_group_max=2, # au lieu de 5 + collective_wot_min=10, + ) + # affected_count=3 est au-delà de small_group_max=2 → collective + result = qualify( + QualificationInput(within_mandate=False, affected_count=3), + custom, + ) + assert result.decision_type == DecisionType.COLLECTIVE + + +def test_r12_default_config_thresholds_are_expected(): + """Valeurs par défaut documentées et vérifiables.""" + cfg = QualificationConfig() + assert cfg.individual_max == 1 + assert cfg.small_group_max == 5 + assert cfg.collective_wot_min == 50