Solveur de suggestions (OR-Tools CP-SAT) — ne place jamais personne

Nouveau module `backend/app/solver/` (mêmes conventions que
`validation/` : models purs, engine pur testé, loader seul lien avec la
base). Pour chaque créneau encore sous son effectif idéal, calcule une
liste ordonnée de bénévoles candidats — c'est la coordination qui
affecte, dans l'écran habituel ; le solveur fait juste remonter ces
candidats en tête de liste (voir le câblage frontend à suivre).

- Seule contrainte dure : aucun chevauchement de créneaux pour un même
  bénévole (règle SHIFT_OVERLAP), à la fois contre l'existant et entre
  deux suggestions du même calcul. L'effectif idéal est un plafond dur.
- Tout le reste est un terme de l'objectif, pas un blocage : priorité
  forte aux créneaux sous l'effectif minimal, bonus vœu de poste, malus
  disponibilité manquante (réutilise `validation.engine.uncovered_minutes`),
  cumul de 8 h et repos (contre les affectations déjà décidées),
  répartition de la charge.
- `POST /editions/{id}/solver/suggestions` (COORDINATION/référent) :
  recalcule à la demande, ne persiste jamais d'affectation.

Dépendance `ortools`. 18 nouveaux tests (moteur + API), 172 tests au
vert, `ruff` et `alembic check` verts. Testé sur le jeu de démo (120
bénévoles, 100 créneaux) : ~0,3 s.

CLAUDE.md : nouvelle section 4 (solveur), renumérotation 5-10,
compétences retirées de « hors périmètre », journal étape 15.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017uUsRg7XzopkCPWimwg9ZM

backend/app/api/schemas.py

@@ -264,6 +264,21 @@ class AssignmentValidateResult(ApiModel):
     violations: list[Violation] = Field(default_factory=list)
 
 
+# --------------------------------------------------------------------------- #
+# Solveur (suggestions, CP-SAT) — ne place jamais personne                     #
+# --------------------------------------------------------------------------- #
+class SolverSuggestion(ApiModel):
+    shift_id: uuid.UUID
+    # Bénévoles suggérés pour ce créneau, meilleur en tête.
+    volunteer_ids: list[uuid.UUID]
+
+
+class SolverRunResult(ApiModel):
+    suggestions: list[SolverSuggestion]
+    shifts_considered: int
+    volunteers_considered: int
+
+
 # --------------------------------------------------------------------------- #
 # Matrice poste × créneau                                                      #
 # --------------------------------------------------------------------------- #

backend/app/api/solver.py

@@ -0,0 +1,38 @@
+"""Suggestions du solveur (OR-Tools CP-SAT). Ne persiste rien : la
+coordination affecte elle-même, dans l'écran habituel, en tenant compte des
+bénévoles remontés en tête de liste."""
+
+import uuid
+
+from fastapi import APIRouter, Depends
+from sqlmodel import Session
+
+from app.api.deps import get_edition
+from app.api.schemas import SolverRunResult, SolverSuggestion
+from app.auth.deps import require_staff
+from app.db import get_session
+from app.models import Account, Edition
+from app.solver import build_solver_input, solve
+
+router = APIRouter(prefix="/editions/{edition_id}/solver", tags=["solver"])
+
+
+@router.post("/suggestions", response_model=SolverRunResult)
+def run_solver(
+    edition: Edition = Depends(get_edition),
+    _guard: Account = Depends(require_staff),
+    session: Session = Depends(get_session),
+) -> SolverRunResult:
+    data = build_solver_input(session=session, edition=edition)
+    solution = solve(data)
+    return SolverRunResult(
+        suggestions=[
+            SolverSuggestion(
+                shift_id=uuid.UUID(s.shift_id),
+                volunteer_ids=[uuid.UUID(v) for v in s.volunteer_ids],
+            )
+            for s in solution.suggestions
+        ],
+        shifts_considered=len(data.shifts),
+        volunteers_considered=len(data.volunteers),
+    )

backend/app/main.py

@@ -14,6 +14,7 @@ from app.api.matrix import router as matrix_router
 from app.api.notifications import router as notification_router
 from app.api.positions import router as position_router
 from app.api.shifts import router as shift_router
+from app.api.solver import router as solver_router
 from app.api.volunteers import router as volunteer_router
 from app.config import get_settings
 from app.logging_config import configure_logging
@@ -44,5 +45,6 @@ app.include_router(availability_router)
 app.include_router(position_router)
 app.include_router(shift_router)
 app.include_router(assignment_router)
+app.include_router(solver_router)
 app.include_router(matrix_router)
 app.include_router(notification_router)

backend/app/solver/__init__.py

@@ -0,0 +1,22 @@
+"""Solveur de suggestions (OR-Tools CP-SAT) — voir `engine.py` pour le détail
+des règles. Ne place jamais personne : seule la coordination affecte."""
+
+from app.solver.engine import solve
+from app.solver.loader import build_solver_input
+from app.solver.models import (
+    ShiftSuggestion,
+    SolverInput,
+    SolverShift,
+    SolverSolution,
+    SolverVolunteer,
+)
+
+__all__ = [
+    "ShiftSuggestion",
+    "SolverInput",
+    "SolverShift",
+    "SolverSolution",
+    "SolverVolunteer",
+    "build_solver_input",
+    "solve",
+]

backend/app/solver/engine.py

@@ -0,0 +1,190 @@
+"""Solveur de suggestions (OR-Tools CP-SAT) — cœur du produit, second moteur.
+
+Ne place jamais personne : il propose, pour chaque créneau encore sous son
+effectif idéal, une liste ordonnée de bénévoles candidats. C'est la
+coordination qui affecte (dans l'app, via l'écran habituel — le solveur se
+contente de faire remonter ces candidats en tête de liste).
+
+Contrainte dure : aucun chevauchement de créneaux pour un même bénévole (règle
+`SHIFT_OVERLAP`, cf. `app.validation`). Tout le reste (disponibilité, cumul de
+8 h, repos, vœu de poste, équité de charge) est un terme de l'objectif, pas
+une contrainte — comme le moteur de validation, qui reste seul juge au moment
+où la coordination confirme une affectation.
+"""
+
+import datetime as dt
+
+from ortools.sat.python import cp_model
+
+from app.solver.models import (
+    ShiftSuggestion,
+    SolverInput,
+    SolverShift,
+    SolverSolution,
+    SolverVolunteer,
+)
+from app.validation.engine import uncovered_minutes
+from app.validation.models import DayPartWindow, ShiftInfo
+
+_EIGHT_HOURS_MIN = 8 * 60
+
+# Poids de l'objectif (unités arbitraires, cf. docstring de `_score`).
+# L'écart à un effectif minimal non atteint prime sur tout le reste : le
+# solveur doit toujours proposer quelqu'un pour un créneau sous-staffé, quitte
+# à ce que la disponibilité soit imparfaite (avertissement, pas un blocage —
+# la validation reste seule juge à la confirmation). Au-delà du minimum, un
+# candidat mal assorti peut légitimement ne pas être suggéré du tout.
+_URGENT_BASE = 1000  # créneau encore sous son effectif minimal
+_NORMAL_BASE = 20  # créneau entre effectif minimal et idéal
+_WISH_BONUS = 30
+_AVAIL_PENALTY_PER_HALF_HOUR = 5
+_DAILY_OVERAGE_PENALTY_PER_MIN = 2
+_REST_PENALTY = 60
+_FAIRNESS_PENALTY_PER_HOUR = 2
+
+
+def _overlaps(a: ShiftInfo, b: ShiftInfo) -> bool:
+    return a.starts_at < b.ends_at and b.starts_at < a.ends_at
+
+
+def _day_boundary(day_parts: tuple[DayPartWindow, ...]) -> dt.time:
+    if not day_parts:
+        return dt.time(0, 0)
+    return min(day_parts, key=lambda part: part.order).start_time
+
+
+def _day_key(moment: dt.datetime, boundary: dt.time) -> dt.date:
+    if moment.time() >= boundary:
+        return moment.date()
+    return moment.date() - dt.timedelta(days=1)
+
+
+def _daily_overage_minutes(
+    volunteer: SolverVolunteer, shift: SolverShift, boundary: dt.time
+) -> int:
+    """Minutes qui dépasseraient les 8 h de la journée du créneau, en ne
+    comptant que les affectations déjà décidées (pas les autres suggestions
+    du même calcul — ce cas limite reste sous la vigilance du moteur de
+    validation au moment de la confirmation)."""
+    key = _day_key(shift.starts_at, boundary)
+    total = shift.ends_at - shift.starts_at
+    for other in volunteer.existing_shifts:
+        if _day_key(other.starts_at, boundary) == key:
+            total += other.ends_at - other.starts_at
+    minutes = int(total.total_seconds() // 60)
+    return max(0, minutes - _EIGHT_HOURS_MIN)
+
+
+def _has_insufficient_rest(volunteer: SolverVolunteer, shift: SolverShift) -> bool:
+    candidate = shift.shift_info
+    for other in volunteer.existing_shifts:
+        if _overlaps(candidate, other):
+            continue
+        if other.ends_at <= candidate.starts_at:
+            gap = candidate.starts_at - other.ends_at
+        elif candidate.ends_at <= other.starts_at:
+            gap = other.starts_at - candidate.ends_at
+        else:
+            continue
+        if gap < dt.timedelta(minutes=_EIGHT_HOURS_MIN):
+            return True
+    return False
+
+
+def _score(
+    volunteer: SolverVolunteer,
+    shift: SolverShift,
+    day_boundary: dt.time,
+) -> int:
+    """Score d'affinité (bénévole, créneau) : plus haut = meilleur candidat.
+
+    Purement additif (aucune interaction entre deux affectations proposées),
+    ce qui garde le modèle CP-SAT linéaire simple. Cumul de 8 h et repos ne
+    sont vérifiés que contre les affectations déjà décidées, pas entre deux
+    suggestions du même calcul : la validation finale (à la confirmation)
+    reste seule juge en cas de cas limite entre deux suggestions.
+    """
+    score = _URGENT_BASE if shift.already_assigned < shift.min_staff else _NORMAL_BASE
+    if shift.position_id in volunteer.wished_position_ids:
+        score += _WISH_BONUS
+    missed = uncovered_minutes(shift.shift_info, volunteer.availability)
+    score -= (missed // 30) * _AVAIL_PENALTY_PER_HALF_HOUR
+    score -= _DAILY_OVERAGE_PENALTY_PER_MIN * _daily_overage_minutes(
+        volunteer, shift, day_boundary
+    )
+    if _has_insufficient_rest(volunteer, shift):
+        score -= _REST_PENALTY
+    score -= (_FAIRNESS_PENALTY_PER_HOUR * volunteer.existing_minutes) // 60
+    return score
+
+
+def solve(
+    data: SolverInput,
+    *,
+    time_limit_seconds: float = 8.0,
+    seed: int = 42,
+) -> SolverSolution:
+    """Calcule les suggestions. Fonction pure : ne touche jamais la base."""
+    boundary = _day_boundary(data.day_parts)
+
+    model = cp_model.CpModel()
+    variables: dict[tuple[int, int], cp_model.IntVar] = {}
+    scores: dict[tuple[int, int], int] = {}
+
+    for si, shift in enumerate(data.shifts):
+        gap = shift.ideal_staff - shift.already_assigned
+        if gap <= 0:
+            continue
+        for vi, volunteer in enumerate(data.volunteers):
+            if any(_overlaps(shift.shift_info, other) for other in volunteer.existing_shifts):
+                continue
+            key = (vi, si)
+            variables[key] = model.NewBoolVar(f"x_{vi}_{si}")
+            scores[key] = _score(volunteer, shift, boundary)
+
+    if not variables:
+        return SolverSolution()
+
+    # Effectif idéal comme plafond dur — le solveur ne sur-staffe jamais.
+    for si, shift in enumerate(data.shifts):
+        gap = shift.ideal_staff - shift.already_assigned
+        n_volunteers = len(data.volunteers)
+        shift_vars = [variables[vi, si] for vi in range(n_volunteers) if (vi, si) in variables]
+        if shift_vars and gap > 0:
+            model.Add(sum(shift_vars) <= gap)
+
+    # Aucun chevauchement entre deux créneaux nouvellement suggérés au même
+    # bénévole (règle `SHIFT_OVERLAP`, bloquante).
+    for vi in range(len(data.volunteers)):
+        my_shifts = [si for si in range(len(data.shifts)) if (vi, si) in variables]
+        for a in range(len(my_shifts)):
+            for b in range(a + 1, len(my_shifts)):
+                si_a, si_b = my_shifts[a], my_shifts[b]
+                if _overlaps(data.shifts[si_a].shift_info, data.shifts[si_b].shift_info):
+                    model.Add(variables[vi, si_a] + variables[vi, si_b] <= 1)
+
+    model.Maximize(sum(score * variables[key] for key, score in scores.items()))
+
+    solver = cp_model.CpSolver()
+    solver.parameters.max_time_in_seconds = time_limit_seconds
+    solver.parameters.num_search_workers = 1
+    solver.parameters.random_seed = seed
+    status = solver.Solve(model)
+
+    chosen: dict[int, list[int]] = {}
+    if status in (cp_model.OPTIMAL, cp_model.FEASIBLE):
+        for (vi, si), var in variables.items():
+            if solver.Value(var):
+                chosen.setdefault(si, []).append(vi)
+
+    suggestions: list[ShiftSuggestion] = []
+    for si, shift in enumerate(data.shifts):
+        picked = sorted(chosen.get(si, []), key=lambda vi: -scores[(vi, si)])
+        if picked:
+            suggestions.append(
+                ShiftSuggestion(
+                    shift_id=shift.id,
+                    volunteer_ids=tuple(data.volunteers[vi].id for vi in picked),
+                )
+            )
+    return SolverSolution(suggestions=tuple(suggestions))

backend/app/solver/loader.py

@@ -0,0 +1,97 @@
+"""Construit un `SolverInput` depuis la base pour une édition. Couche de
+liaison : c'est le seul endroit du module `solver` qui connaît la base."""
+
+from collections import defaultdict
+
+from sqlmodel import Session, select
+
+from app.models import (
+    Assignment,
+    Availability,
+    DayPart,
+    Edition,
+    Shift,
+    Volunteer,
+    VolunteerPositionWish,
+)
+from app.solver.models import SolverInput, SolverShift, SolverVolunteer
+from app.validation.models import AvailabilityWindow, DayPartWindow, ShiftInfo
+
+
+def build_solver_input(*, session: Session, edition: Edition) -> SolverInput:
+    day_parts = list(
+        session.exec(select(DayPart).where(DayPart.edition_id == edition.id))
+    )
+
+    shifts = list(session.exec(select(Shift).where(Shift.edition_id == edition.id)))
+    assignments = list(
+        session.exec(select(Assignment).where(Assignment.edition_id == edition.id))
+    )
+    shifts_by_id = {s.id: s for s in shifts}
+
+    assigned_count: dict = defaultdict(int)
+    shifts_by_volunteer: dict = defaultdict(list)
+    for a in assignments:
+        assigned_count[a.shift_id] += 1
+        shift = shifts_by_id.get(a.shift_id)
+        if shift is not None:
+            shifts_by_volunteer[a.volunteer_id].append(shift)
+
+    solver_shifts = tuple(
+        SolverShift(
+            id=str(s.id),
+            position_id=str(s.position_id),
+            starts_at=s.starts_at,
+            ends_at=s.ends_at,
+            min_staff=s.min_staff,
+            ideal_staff=s.ideal_staff,
+            already_assigned=assigned_count.get(s.id, 0),
+        )
+        for s in shifts
+        if assigned_count.get(s.id, 0) < s.ideal_staff
+    )
+
+    availabilities: dict = defaultdict(list)
+    for a in session.exec(
+        select(Availability).where(Availability.edition_id == edition.id)
+    ):
+        availabilities[a.volunteer_id].append(
+            AvailabilityWindow(date=a.date, start_time=a.start_time, end_time=a.end_time)
+        )
+
+    wishes: dict = defaultdict(set)
+    for w in session.exec(
+        select(VolunteerPositionWish).join(
+            Volunteer, Volunteer.id == VolunteerPositionWish.volunteer_id
+        ).where(Volunteer.edition_id == edition.id)
+    ):
+        wishes[w.volunteer_id].add(str(w.position_id))
+
+    volunteers = list(
+        session.exec(select(Volunteer).where(Volunteer.edition_id == edition.id))
+    )
+    solver_volunteers = tuple(
+        SolverVolunteer(
+            id=str(v.id),
+            availability=tuple(availabilities.get(v.id, ())),
+            wished_position_ids=frozenset(wishes.get(v.id, ())),
+            existing_shifts=tuple(
+                ShiftInfo(id=str(s.id), starts_at=s.starts_at, ends_at=s.ends_at)
+                for s in shifts_by_volunteer.get(v.id, ())
+            ),
+            existing_minutes=sum(
+                int((s.ends_at - s.starts_at).total_seconds() // 60)
+                for s in shifts_by_volunteer.get(v.id, ())
+            ),
+        )
+        for v in volunteers
+    )
+
+    return SolverInput(
+        shifts=solver_shifts,
+        volunteers=solver_volunteers,
+        day_parts=tuple(
+            DayPartWindow(dp.code, dp.start_time, dp.end_time, dp.order)
+            for dp in day_parts
+        ),
+    )

backend/app/solver/models.py

@@ -0,0 +1,60 @@
+"""Types d'entrée/sortie du solveur — suggestions, pas d'affectation automatique.
+
+Le solveur ne connaît que ces structures simples (jamais les modèles
+SQLModel), comme le moteur de validation dont il réutilise les briques
+(`app.validation.models`, `app.validation.engine.uncovered_minutes`).
+"""
+
+import datetime as dt
+from dataclasses import dataclass, field
+
+from app.validation.models import AvailabilityWindow, DayPartWindow, ShiftInfo
+
+
+@dataclass(frozen=True)
+class SolverShift:
+    """Un créneau à compléter (`assigned < ideal_staff`)."""
+
+    id: str
+    position_id: str
+    starts_at: dt.datetime
+    ends_at: dt.datetime
+    min_staff: int
+    ideal_staff: int
+    already_assigned: int
+
+    @property
+    def shift_info(self) -> ShiftInfo:
+        return ShiftInfo(id=self.id, starts_at=self.starts_at, ends_at=self.ends_at)
+
+
+@dataclass(frozen=True)
+class SolverVolunteer:
+    id: str
+    availability: tuple[AvailabilityWindow, ...] = ()
+    wished_position_ids: frozenset[str] = frozenset()
+    # Créneaux déjà occupés (affectations existantes, décisions figées) :
+    # comptent pour chevauchement, cumul journalier, repos.
+    existing_shifts: tuple[ShiftInfo, ...] = ()
+    # Total déjà affecté sur l'édition (minutes), pour répartir la charge.
+    existing_minutes: int = 0
+
+
+@dataclass(frozen=True)
+class SolverInput:
+    shifts: tuple[SolverShift, ...] = field(default_factory=tuple)
+    volunteers: tuple[SolverVolunteer, ...] = field(default_factory=tuple)
+    day_parts: tuple[DayPartWindow, ...] = field(default_factory=tuple)
+
+
+@dataclass(frozen=True)
+class ShiftSuggestion:
+    """Bénévoles suggérés pour un créneau, meilleur en tête."""
+
+    shift_id: str
+    volunteer_ids: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class SolverSolution:
+    suggestions: tuple[ShiftSuggestion, ...] = field(default_factory=tuple)