Sprint 1 : scaffolding complet de Glibredecision

Plateforme de decisions collectives pour Duniter/G1.
Backend FastAPI async + PostgreSQL (14 tables, 8 routers, 6 services,
moteur de vote avec formule d'inertie WoT/Smith/TechComm).
Frontend Nuxt 4 + Nuxt UI v3 + Pinia (9 pages, 5 stores).
Infrastructure Docker + Woodpecker CI + Traefik.
Documentation technique et utilisateur (15 fichiers).
Seed : Licence G1, Engagement Forgeron v2.0.0, 4 protocoles de vote.
30 tests unitaires (formules, mode params, vote nuance) -- tous verts.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Yvv
2026-02-28 12:46:11 +01:00
commit 25437f24e3
100 changed files with 10236 additions and 0 deletions

View File

@@ -0,0 +1,199 @@
"""Vote service: compute results and verify vote signatures."""
from __future__ import annotations
import uuid
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.engine.mode_params import parse_mode_params
from app.engine.nuanced_vote import evaluate_nuanced
from app.engine.smith_threshold import smith_threshold
from app.engine.techcomm_threshold import techcomm_threshold
from app.engine.threshold import wot_threshold
from app.models.protocol import FormulaConfig, VotingProtocol
from app.models.vote import Vote, VoteSession
async def compute_result(session_id: uuid.UUID, db: AsyncSession) -> dict:
"""Load a vote session, its protocol and formula, compute thresholds, and tally.
Parameters
----------
session_id:
UUID of the VoteSession to tally.
db:
Async database session.
Returns
-------
dict
Result dict with keys: threshold, votes_for, votes_against,
votes_total, adopted, smith_ok, techcomm_ok, details.
"""
# Load session with votes eagerly
result = await db.execute(
select(VoteSession)
.options(selectinload(VoteSession.votes))
.where(VoteSession.id == session_id)
)
session = result.scalar_one_or_none()
if session is None:
raise ValueError(f"Session de vote introuvable : {session_id}")
# Load protocol + formula config
proto_result = await db.execute(
select(VotingProtocol)
.options(selectinload(VotingProtocol.formula_config))
.where(VotingProtocol.id == session.voting_protocol_id)
)
protocol = proto_result.scalar_one_or_none()
if protocol is None:
raise ValueError(f"Protocole de vote introuvable pour la session {session_id}")
formula: FormulaConfig = protocol.formula_config
# If mode_params is set on the protocol, it overrides formula_config values
if protocol.mode_params:
params = parse_mode_params(protocol.mode_params)
else:
params = {
"majority_pct": formula.majority_pct,
"base_exponent": formula.base_exponent,
"gradient_exponent": formula.gradient_exponent,
"constant_base": formula.constant_base,
"smith_exponent": formula.smith_exponent,
"techcomm_exponent": formula.techcomm_exponent,
}
# Separate vote types
active_votes: list[Vote] = [v for v in session.votes if v.is_active]
if protocol.vote_type == "nuanced":
return await _compute_nuanced(session, active_votes, formula, params, db)
# --- Binary vote ---
votes_for = sum(1 for v in active_votes if v.vote_value == "for")
votes_against = sum(1 for v in active_votes if v.vote_value == "against")
total = votes_for + votes_against
# WoT threshold
threshold = wot_threshold(
wot_size=session.wot_size,
total_votes=total,
majority_pct=params.get("majority_pct", 50),
base_exponent=params.get("base_exponent", 0.1),
gradient_exponent=params.get("gradient_exponent", 0.2),
constant_base=params.get("constant_base", 0.0),
)
# Smith criterion (optional)
smith_ok = True
smith_required = None
if params.get("smith_exponent") is not None and session.smith_size > 0:
smith_required = smith_threshold(session.smith_size, params["smith_exponent"])
smith_votes = sum(1 for v in active_votes if v.voter_is_smith and v.vote_value == "for")
smith_ok = smith_votes >= smith_required
# TechComm criterion (optional)
techcomm_ok = True
techcomm_required = None
if params.get("techcomm_exponent") is not None and session.techcomm_size > 0:
techcomm_required = techcomm_threshold(session.techcomm_size, params["techcomm_exponent"])
techcomm_votes = sum(1 for v in active_votes if v.voter_is_techcomm and v.vote_value == "for")
techcomm_ok = techcomm_votes >= techcomm_required
adopted = votes_for >= threshold and smith_ok and techcomm_ok
vote_result = "adopted" if adopted else "rejected"
# Update session tallies
session.votes_for = votes_for
session.votes_against = votes_against
session.votes_total = total
session.threshold_required = float(threshold)
session.result = vote_result
session.status = "tallied"
await db.commit()
return {
"threshold": threshold,
"votes_for": votes_for,
"votes_against": votes_against,
"votes_total": total,
"adopted": adopted,
"smith_ok": smith_ok,
"smith_required": smith_required,
"techcomm_ok": techcomm_ok,
"techcomm_required": techcomm_required,
"result": vote_result,
}
async def _compute_nuanced(
session: VoteSession,
active_votes: list[Vote],
formula: FormulaConfig,
params: dict,
db: AsyncSession,
) -> dict:
"""Compute a nuanced vote result."""
vote_levels = [v.nuanced_level for v in active_votes if v.nuanced_level is not None]
threshold_pct = formula.nuanced_threshold_pct or 80
min_participants = formula.nuanced_min_participants or 59
evaluation = evaluate_nuanced(
votes=vote_levels,
threshold_pct=threshold_pct,
min_participants=min_participants,
)
vote_result = "adopted" if evaluation["adopted"] else "rejected"
session.votes_total = evaluation["total"]
session.votes_for = evaluation["positive_count"]
session.votes_against = evaluation["total"] - evaluation["positive_count"]
session.threshold_required = float(threshold_pct)
session.result = vote_result
session.status = "tallied"
await db.commit()
return {
"vote_type": "nuanced",
"result": vote_result,
**evaluation,
}
async def verify_vote_signature(address: str, signature: str, payload: str) -> bool:
"""Verify an Ed25519 signature from a Duniter V2 address.
Parameters
----------
address:
SS58 address of the voter.
signature:
Hex-encoded Ed25519 signature.
payload:
The original message that was signed.
Returns
-------
bool
True if the signature is valid.
TODO
----
Implement actual Ed25519 verification using substrate-interface:
from substrateinterface import Keypair
keypair = Keypair(ss58_address=address, crypto_type=KeypairType.ED25519)
return keypair.verify(payload.encode(), bytes.fromhex(signature))
"""
# TODO: Implement real Ed25519 verification with substrate-interface
# For now, accept all signatures in development mode
if not address or not signature or not payload:
return False
return True