feat: v6.4.0 Realtime production — merge 3-voix (au-delà du LWW) + broadcast non bloquant

- app/services/realtime_merge.py : merge à 3 voix diff3-lite, regions disjointes conservees, conflit par champ + drapeau
- protocole base (client embarque la base de sa saisie) ; sans base -> LWW historique (retro-compat)
- ack renvoie le bloc fusionne + conflict ; adoption cote client + toast ; broadcast du resultat fusionne
- broadcast non bloquant : file sortante + tache writer par connexion, coalescence des curseurs
- clients trop lents deconnectes (4413), budget ops anti-flood (400/10s)
- fix fuite room 4404 + room_state() sur page inexistante
- GET /api/realtime/stats (observabilite)
- 26 tests test_realtime_v64.py ; suite 725 verte ; ruff + eslint OK ; version 6.4.0

CHANGELOG.md

@@ -1,5 +1,40 @@
 # Changelog - FlowDeck
 
+## v6.4.0 (2026-09-22) — Realtime editing (production)
+
+> Le realtime passe en mode production : résolution de conflits au-delà du last-write-wins (merge à trois versions) et édition à grande échelle (broadcast non bloquant, coalescence des curseurs, corrections de fuites, observabilité).
+
+### Added
+
+- **Merge à 3-voix (diff3-lite)** — nouveau `app/services/realtime_merge.py` :
+  - `merge_text_3way(base, current, incoming)` : fusion de caractères. Les régions modifiées qui ne se chevauchent pas sont **toutes conservées** (les deux saisies survivent) ; ordre d'arrivée indifférent ; un chevauchement réel retombe en LWW et est signalé.
+  - `merge_block_3way(base, current, incoming)` : fusion **champ-par-champ**. Serveur et client modifiant des champs différents n'ont plus de conflit ; conflit limité au seul champ divergent (plus « tout le bloc perdu »).
+  - Fonctions pures, testées sans WebSocket ni base de données.
+- **Protocole `base`** — chaque `update` envoyé par le client embarque la version du bloc dont dérive sa saisie ; le serveur calcule `merge_block_3way(base, current, incoming)`. Sans `base` (ancien client) → LWW historique, rétro-compatible.
+- **Adoption du résultat fusionné** — l'`ack` renvoie le bloc fusionné + drapeau `conflict` ; le client met à jour sa `base`, adopte le résultat (hors bloc en cours d'édition) et affiche un toast en cas de conflit. Le serveur diffuse toujours le bloc final fusionné pour convergence de tous les pairs.
+- **Broadcast non bloquant** — chaque connexion a une file sortante (`asyncio.Queue`) + une tâche `_writer` dédiée ; `_broadcast()` fait `put_nowait` et n'attend plus le socket → un client lent ne fige plus la room entière.
+- **Coalescence des curseurs** — le writer réduit les messages `sel` empilés à la position la plus récente (seule la dernière compte), en préservant l'ordre des messages importants.
+- **Anti-flood** — budget d'opérations par connexion (`OP_WINDOW_MAX=400` / fenêtre 10 s) ; au-delà, réponse `ack stale` sans application.
+- **Observabilité** — `GET /api/realtime/stats` (authentifié) : rooms, connexions (dont totales), ops, merges, conflits, déconnexions lentes, + détail par page.
+
+### Changed
+
+- **Déconnexion des clients trop lents** — file sortante pleine (`MAX_OUT_QUEUE=512`) → fermeture 4413 + compteur `slow_disconnects`, pour qu'une room ne stagne jamais sur un pair mortel.
+- `app/templates/_page_editor_realtime.html` — les updates incluent désormais `base` ; nouveau traitement de l'`ack` avec `merged`.
+
+### Fixed
+
+- **Fuite de rooms** — une tentative de connexion WS sur une page inexistante (4404) enregistrait une `Room` orpheline en mémoire pour toujours ; la room n'est plus enregistrée tant que la page n'est pas validée.
+- **`room_state()` sur page inexistante** — retournait/plantait sur `None` ; retourne désormais `{"blocks": [], "title": "", "version": 0}` sans laisser d'entrée dans `_rooms`.
+
+### Tests
+
+- **26 nouveaux tests** `tests/test_realtime_v64.py` : merge purs (régions disjointes, ordre indépendant, chevauchement → conflit, suppression de champ, id préservé), protocole WS (ack fusionné, convergence de 2 clients sur le même bloc, rétro-compat LWW sans `base`, conflit scalaire), fuite 4404, coalescence des curseurs, endpoint stats, anti-flood.
+- **14 tests** `tests/test_realtime.py` existants préservés (non-régression).
+- `ruff check app tests` OK · `eslint static/js` 0 problème.
+
+---
+
 ## v6.3.0 (2026-09-21) — API publique complète v2 (REST + scopes + OpenAPI)
 
 > L'API publique `/api/v2` devient une surface REST complète façon Notion : CRUD sur tous les domaines (collections, pages, propriétés, vues, commentaires, notifications, favoris, tags, partage, sprints, templates, forges, admin…), auth Bearer + scopes hiérarchiques, pagination/filtres/tri, erreurs RFC 7807, idempotence, audit et webhooks. Wrappers sur les services existants — un seul chemin de code.

README.md

@@ -2,7 +2,7 @@
 
 Clone complet de **Notion** intégré nativement à **Gitea** — Databases, Pages, Kanban, Calendar, Gallery, Timeline, List, Multi-Users.
 
-> **v2.1.0** — API publique, Webhooks sortants, PWA
+> **v6.4.0** — Realtime production (merge 3-voix, broadcast non bloquant), API publique v2, PWA offline
 
 ## Quick Start
 
@@ -87,7 +87,7 @@ DATABASE_URL=sqlite:////data/flowdeck.db
 ## Tests
 
 ```bash
-python3 -m pytest tests/ -v   # 73/73 passent
+python3 -m pytest tests/ -v   # 725/725 passent
 ```
 
 ## Roadmap

ROADMAP.md

@@ -814,6 +814,41 @@ Détails livrés :
 - [x] `tests/test_public_api_v2.py` — **24 tests** (auth scopes, pagination, filtres, RFC 7807, idempotency, webhooks deliveries, CRUD multi-domaines)
 - [x] Vérif `ruff check app tests` + `pytest -n auto` → **668 verte**
 
+## v6.4.0 — Realtime editing (production) ✅ (2026-09-22)
+
+> **Objectif** : passer le realtime v5.13.0 en « production » — résolution de
+> conflits au-delà du last-write-wins + édition à grande échelle (broadcast non
+> bloquant, plusieurs rooms/pages, observabilité). **COMPLETED**.
+
+#### Conflits au-delà du LWW
+
+- [x] **Merge à 3 voix (diff3-lite)** — nouveau service `app/services/realtime_merge.py` : `merge_text_3way()` (merge de caractères) + `merge_block_3way()` (merge champ-par-champ), fonctions pures et testées sans WebSocket ni base
+- [x] **Régions disjointes conservées** — deux utilisateurs tapant à des endroits différents du *même* bloc voient leurs deux saisies survivre (au lieu d'écraser l'une par l'autre) ; ordre d'arrivée indifférent
+- [x] **Chevauchement réel → LWW par champ + drapeau** — un conflit n'est plus « tout le bloc perdu » mais limité au champ concerné ; `conflict: true` renvoyé dans l'`ack` et le broadcast
+- [x] **Protocole `base`** — le client embarque dans chaque `update` la version du bloc dont dérive sa saisie ; le serveur fait `merge_block_3way(base, current, incoming)` ; sans `base` → LWW historique (rétro-compat ancien client)
+- [x] **Adoption côté client** — l'`ack` renvoie le bloc fusionné ; le client met à jour sa `base`, adopte le résultat (hors bloc en cours d'édition) et affiche un toast en cas de conflit
+- [x] **Broadcast du résultat fusionné** — le serveur diffuse toujours le bloc final fusionné (jamais la proposition brute) pour convergence garantie de tous les clients
+
+#### Échelle & robustesse
+
+- [x] **Broadcast non bloquant** — chaque connexion a une file sortante (`asyncio.Queue`) + une tâche `_writer` dédiée ; `_broadcast()` fait `put_nowait` et n'attend plus le socket → un client lent ne fige plus la room
+- [x] **Coalescence des curseurs** — le writer réduit les messages `sel` empilés à la position la plus récente (seule la dernière compte), tout en préservant l'ordre des messages importants
+- [x] **Déconnexion des clients trop lents** — file pleine (`MAX_OUT_QUEUE=512`) → fermeture 4413 + compteur `slow_disconnects` (évite qu'une room entière stagne sur un pair mortel)
+- [x] **Anti-flood** — budget d'opérations par connexion (`OP_WINDOW_MAX=400` / 10 s) ; au-delà, réponse `ack stale` sans application
+- [x] **Fix fuite de rooms** — une connexion 4404 n'enregistre plus de `Room` orpheline en mémoire ; `room_state()` sur page inexistante retourne un dict vide au lieu de planter sur `None`
+- [x] **Observabilité** — `GET /api/realtime/stats` (authentifié) : rooms, connexions, ops, merges, conflits, déconnexions lentes + détail par page
+
+Détails livrés :
+- `app/services/realtime_merge.py` — merge 3-voix (nouveau, ~170 lignes)
+- `app/services/realtime_server.py` — `RTConn` (file + writer + budget), `_apply()` avec merge, `_evict_slow()`, `stats()`, fix fuites
+- `app/routers/realtime.py` — endpoint `GET /api/realtime/stats`
+- `app/templates/_page_editor_realtime.html` — envoi de `base`, adoption du bloc fusionné, toast de conflit
+- **26 tests** `tests/test_realtime_v64.py` (merge purs, protocole WS, convergence 2 clients, rétro-compat LWW, fuite 4404, coalescence, stats, anti-flood) ; **14 tests** `tests/test_realtime.py` préservés
+- `ruff check app tests` OK · `eslint static/js` 0 problème
+- **Version** — 6.4.0
+
+---
+
 ## v6.0.0 — Pro (futur)
 
 - [x] **PWA** — Progressive Web App, offline support ✅ (livré) — [📄 Conception détaillée](/docs/V6_PWA_Progressive_Web_App.md)
@@ -821,7 +856,7 @@ Détails livrés :
 - [x] **Web Clipper** — extension navigateur ✅ (livré v6.2.0/6.2.1) — [📄 Conception détaillée](/docs/V6_Web_Clipper.md)
 - [x] **API publique complète** — REST API documentée (OpenAPI) ✅ (livré v6.3.0) — [📄 API Guide v2](/docs/API_GUIDE_V6.md) · [📄 OpenAPI](/docs/openapi-v2.json)
 - [ ] **SSO/SAML** — enterprise authentication — [📄 Conception détaillée](/docs/V6_SSO_SAML_Enterprise_Auth.md)
-- [ ] **Realtime editing (production)** — voir **v5.13.0** (curseurs + présence déjà avancés ici) ; reste en v6 : conflits avancés, édition large échelle
+- [x] **Realtime editing (production)** ✅ livré **v6.4.0** (merge 3-voix au-delà du LWW, broadcast non bloquant) ; voir **v5.13.0** pour le socle (curseurs + présence)
 - [ ] **Synced blocks (production)** — voir **v5.14.0** (bloc de base) ; reste en v6 : syncing côté databases/vues
 
 ---
@@ -868,4 +903,4 @@ Quality      DB views,      Agent IA     Palette →       Realtime +          E
                                          DB avancée,    redo, drag&drop,   bookmark,     avancée
                                          Calendrier, AI duplicate)        lightbox…)    (Pt.2)      v6.1 ✅ v6.2 ✅ v6.3 ✅
 
-*Dernière mise à jour: 2026-09-21 — **v6.3.0 API publique complète v2 COMPLETED** (~100 endpoints `/api/v2`, scopes hiérarchiques `read<write<admin`, RFC 7807, idempotence, audit, webhooks CRUD, OpenAPI 402 chemins, 24 tests, suite 668 verte) + **v6.2.1 Web Clipper COMPLETED**. Reste: SSO/SAML, realtime production, synced blocks prod, webhooks HMAC/retry.*
+*Dernière mise à jour: 2026-09-22 — **v6.4.0 Realtime editing (production) COMPLETED** (merge 3-voix au-delà du LWW via `realtime_merge.py`, broadcast non bloquant avec file sortante + writer, coalescence des curseurs, fix fuite de rooms 4404, anti-flood, `GET /api/realtime/stats`, 26 tests dédiés) + **v6.3.0 API publique complète v2 COMPLETED** (~100 endpoints `/api/v2`, scopes hiérarchiques `read<write<admin`, RFC 7807, idempotence, audit, webhooks CRUD, OpenAPI 402 chemins). Reste: SSO/SAML, synced blocks prod (databases/vues).*

VERSION

@@ -1 +1 @@
-6.3.0
+6.4.0

WORKLOAD.md

@@ -1,7 +1,7 @@
 # WORKLOAD — FlowDeck Notion Clone
 
-> **Début**: 2026-07-08 | **Version**: v2.2.0 | **Statut**: EN COURS 🔄
+> **Début**: 2026-07-08 | **Version**: v6.4.0 | **Statut**: EN COURS 🔄
-> **Cible v3.0**: Multi-User, Multi-Forge (Gitea/GitHub), Standalone
+> **Cible**: parité Notion + intégration forge · **Reste roadmap**: SSO/SAML, synced blocks prod (databases/vues)
 
 ## Avancement Global
 
@@ -21,7 +21,10 @@
 | v2.0 | Multi-User + Editor Complete | ✅ | 67/67 |
 | v2.1 | Public API, Webhooks, PWA | ✅ | 73/73 |
 | v2.2 | Share/Publish, Favorites, Library | ✅ | 73/73 |
-| v3.0 | **Auth locale, Multi-Forge, Standalone** | 🔲 | — |
+| v3.0 | Auth locale, Multi-Forge, Standalone | ✅ | — |
+| v4.x–v5.x | MVP → Agent IA, palette, automations, import, calendrier, wiki-links, synced blocks | ✅ | 523+ |
+| v6.0–v6.3 | PWA offline, permissions granulaires, web clipper, API publique v2 | ✅ | 668+ |
+| **v6.4.0** | **Realtime production (merge 3-voix, broadcast non bloquant)** | ✅ | **749+** |
 
 ## Blocs Complétés
 
@@ -58,5 +61,5 @@ CRUD collections/pages, 5 vues HTML, relations/rollups/formulas, sub-items/depen
 - **BDD**: SQLite WAL mode, 21 tables, foreign keys ON
 - **Auth**: OAuth2 Gitea + sessions signed (itsdangerous) + token API
 - **Déploiement**: Docker (python:3.12-slim), docker-compose, port 8080
-- **Tests**: pytest, 73 tests, TestClient avec SQLite temporaire
+- **Tests**: pytest, 749+ tests, TestClient avec SQLite temporaire
 - **CI/CD**: Gitea Actions (.gitea/workflows/ci.yml)

app/main.py

@@ -122,7 +122,7 @@ async def lifespan(_app: FastAPI):
 
 app = FastAPI(
     title="FlowDeck",
-    version="6.3.0",
+    version="6.4.0",
     docs_url="/docs",
     redoc_url="/redoc",
     lifespan=lifespan,

app/routers/realtime.py

@@ -7,7 +7,7 @@ from __future__ import annotations
 import json
 import logging
 
-from fastapi import APIRouter, WebSocket
+from fastapi import APIRouter, Request, WebSocket
 from starlette.websockets import WebSocketDisconnect
 
 from app.auth.session import SessionManager
@@ -17,6 +17,20 @@ logger = logging.getLogger(__name__)
 router = APIRouter(tags=["realtime"])
 
 
+@router.get("/api/realtime/stats")
+async def realtime_stats(request: Request):
+    """Observabilité realtime v6.4.0 : rooms, connexions, ops, merges, conflits.
+
+    Réservé aux utilisateurs authentifiés (données d'activité internes).
+    """
+    user = SessionManager.decode_session(
+        request.cookies.get("flowdeck_session", "")
+    )
+    if not user or not user.get("id"):
+        return {"error": "unauthorized"}
+    return manager.stats()
+
+
 @router.websocket("/ws/pages/{page_id}")
 async def ws_page(websocket: WebSocket, page_id: int):
     await websocket.accept()

app/services/realtime_merge.py

@@ -0,0 +1,164 @@
+"""FlowDeck — v6.4.0 Realtime: résolution de conflits au-delà du last-write-wins.
+
+Le LWW par bloc (v5.13.0) écrase intégralement le bloc du dernier arrivé : si deux
+utilisateurs tapent dans le *même* bloc, la saisie du premier est perdue. Ce module
+implémente un vrai *merge à trois versions* (diff3-lite) :
+
+    base    = l'état du bloc dont le client dérive sa saisie (envoyé avec l'op)
+    current = l'état actuel du bloc côté serveur (déjà mis à jour par d'autres)
+    incoming = la nouvelle proposition du client
+
+Règle champ-par-champ :
+  * incoming == base   → le client n'a pas touché ce champ      → on garde current
+  * current  == base   → le serveur n'a pas touché ce champ     → on garde incoming
+  * current  == incoming → les deux ont fait la même chose      → sans conflit
+  * sinon (conflit)
+      - champ texte (str) : merge de caractères. Les régions modifiées qui ne se
+        chevauchent pas sont *toutes conservées* (les deux saisies survivent) ;
+        chevauchement réel → LWW sur ce champ + drapeau de conflit.
+      - autre type (bool, nombre…) : LWW sur ce champ + drapeau de conflit.
+
+Toutes les fonctions sont pures et testables sans WebSocket ni base de données.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+__all__ = ["merge_text_3way", "merge_block_3way", "changed_region"]
+
+
+def changed_region(base: str, other: str) -> tuple[int, int, str] | None:
+    """Région de `base` remplacée par `other` (trim préfixe/suffixe commun).
+
+    Retourne ``(start, end, replacement)`` tel que ``base[:start] + replacement +
+    base[end:] == other``, ou ``None`` si ``other == base`` (aucun changement).
+    """
+    if base == other:
+        return None
+    minlen = min(len(base), len(other))
+    prefix = 0
+    while prefix < minlen and base[prefix] == other[prefix]:
+        prefix += 1
+    suffix = 0
+    # ne jamais chevaucher le préfixe déjà consommé
+    while (suffix < len(base) - prefix and suffix < len(other) - prefix
+           and base[len(base) - 1 - suffix] == other[len(other) - 1 - suffix]):
+        suffix += 1
+    return prefix, len(base) - suffix, other[prefix:len(other) - suffix]
+
+
+def merge_text_3way(base: str, current: str, incoming: str) -> tuple[str, bool]:
+    """Merge à trois versions d'une chaîne. Retourne ``(texte, conflit)``.
+
+    Les éditions qui ne se chevauchent pas sont toutes les deux conservées ;
+    un chevauchement réel retombe en LWW (``incoming`` gagne) et signale le conflit.
+    """
+    if current == incoming:
+        return current, False
+    if current == base:
+        return incoming, False
+    if incoming == base:
+        return current, False
+
+    rc = changed_region(base, current)
+    ri = changed_region(base, incoming)
+    if rc is None:
+        return incoming, False
+    if ri is None:
+        return current, False
+
+    c_start, c_end, c_text = rc
+    i_start, i_end, i_text = ri
+
+    # Régions disjointes (ou juste adjacentes) → appliquer les deux sur base.
+    if c_end <= i_start or i_end <= c_start:
+        edits = sorted([(c_start, c_end, c_text), (i_start, i_end, i_text)],
+                       key=lambda e: e[0])
+        out: list[str] = []
+        pos = 0
+        for start, end, text in edits:
+            if start < pos:
+                continue  # sécurité: ne jamais réappliquer par-dessus
+            out.append(base[pos:start])
+            out.append(text)
+            pos = end
+        out.append(base[pos:])
+        return "".join(out), False
+
+    # Chevauchement réel → LWW sur ce champ, conflit signalé.
+    return incoming, True
+
+
+def _scalar_conflict(base: Any, current: Any, incoming: Any) -> tuple[Any, bool]:
+    """Conflit sur un champ non-texte : LWW (incoming gagne)."""
+    if current == incoming:
+        return current, False
+    if current == base:
+        return incoming, False
+    if incoming == base:
+        return current, False
+    return incoming, True
+
+
+def merge_block_3way(base_blk: Any, current_blk: Any,
+                     incoming_blk: Any) -> tuple[dict, list[str]]:
+    """Merge à trois versions d'un bloc entier.
+
+    Retourne ``(bloc fusionné, champs en conflit)``. Ne lève jamais d'exception :
+    une entrée non-dict retombe en LWW (``incoming``) avec conflit signalé sur
+    ``__block__`` pour que l'appelant puisse journaliser.
+    """
+    if not isinstance(base_blk, dict):
+        base_blk = {}
+    if not isinstance(current_blk, dict):
+        current_blk = {}
+    if not isinstance(incoming_blk, dict):
+        # proposition invalide → on garde l'état serveur
+        return dict(current_blk), ["__block__"]
+
+    keys = set(base_blk) | set(current_blk) | set(incoming_blk)
+    merged: dict[str, Any] = {}
+    conflicts: list[str] = []
+
+    for key in keys:
+        b = base_blk.get(key)
+        c = current_blk.get(key)
+        i = incoming_blk.get(key)
+
+        if i == b:
+            # le client n'a pas modifié ce champ → valeur serveur. Si le serveur
+            # a *supprimé* le champ (absent de current) alors la suppression doit
+            # gagner : on n'insère pas de clé fantôme value=None.
+            if key not in current_blk:
+                continue
+            merged[key] = c
+        elif c == b:
+            # le serveur n'a pas modifié ce champ → valeur client
+            merged[key] = i
+        elif c == i:
+            merged[key] = c
+        else:
+            # les deux ont changé, différemment
+            if isinstance(b, str) and isinstance(c, str) and isinstance(i, str):
+                text, conflict = merge_text_3way(b, c, i)
+                merged[key] = text
+                if conflict:
+                    conflicts.append(key)
+            else:
+                value, conflict = _scalar_conflict(b, c, i)
+                merged[key] = value
+                if conflict:
+                    conflicts.append(key)
+
+    # Un champ supprimé par le serveur et absent de la proposition client doit
+    # rester supprimé (pas de réapparition d'une valeur None fantôme).
+    merged = {k: v for k, v in merged.items()
+              if not (v is None and k not in incoming_blk)}
+
+    if "id" not in merged:
+        # garantir l'identité du bloc même si base était vide
+        ident = incoming_blk.get("id") or current_blk.get("id")
+        if ident:
+            merged["id"] = ident
+
+    return merged, conflicts

app/services/realtime_server.py

@@ -1,5 +1,23 @@
-"""FlowDeck — v5.13.0 Realtime: WebSocket gateway, présences, curseurs live,
+"""FlowDeck — v6.4.0 Realtime: WebSocket gateway, présences, curseurs live,
-merge des opérations de blocs (last-write-wins par bloc) + version de page.
+merge à trois versions des opérations de blocs (au-delà du last-write-wins) +
+version de page.
+
+Améliorations « production » par rapport à v5.13.0 :
+
+* **Conflits** — les mises à jour de bloc embarquent la ``base`` dont dérive la
+  saisie du client ; le serveur fait un merge 3-voix champ-par-champ + texte
+  (voir ``realtime_merge``) au lieu d'écraser le bloc entier. Les deux saisies
+  disjointes survivent, les chevauchements réels retombent en LWW *par champ*
+  avec drapeau de conflit renvoyé au client.
+* **Échelle** — chaque connexion possède une file sortante + une tâche writer
+  dédiée ; le broadcast devient non bloquant (un client lent ne bloque plus la
+  room), les mises à jour de curseur se coalescent (une seule par flush), et un
+  client trop lent (file pleine) est déconnecté proprement (4413).
+* **Anti-flood** — budget d'opérations par connexion (fenêtre glissante).
+* **Fuites corrigées** — une room 4404 n'est plus enregistrée ; ``room_state``
+  ne crée plus d'objet None ; les rooms vides sont évacuées.
+* **Observabilité** — compteurs (rooms, conns, ops, merges, conflits, déconnexions
+  lentes) exposés par ``stats()``.
 
 Rooms in-memory (un seul worker uvicorn). Persistance en base (page.content)
 avec debounce. Fallback polling côté client si le WS est indisponible.
@@ -9,16 +27,25 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+import time
 
 from fastapi import WebSocket
 
 from app.db import get_conn
+from app.services.realtime_merge import merge_block_3way
 
 logger = logging.getLogger(__name__)
 
 COLORS = ["#2383E2", "#46A758", "#E5484D", "#F76B15", "#8E4EC6", "#12A594",
           "#FFC53D", "#D6409F", "#0091FF", "#3E63DD", "#30A46C", "#FF3333"]
 
+# File sortante maximale par connexion au-delà de laquelle le client est
+# considéré comme trop lent et déconnecté (évite qu'une room entière stagne).
+MAX_OUT_QUEUE = 512
+# Nombre maximal d'opérations acceptées par connexion et par fenêtre (anti-flood).
+OP_WINDOW_SECONDS = 10.0
+OP_WINDOW_MAX = 400
+
 
 def color_for(uid: int) -> str:
     return COLORS[(uid or 0) % len(COLORS)]
@@ -48,7 +75,11 @@ def ensure_block_ids(blocks: list[dict]) -> list[dict]:
 
 
 def apply_op(blocks: list[dict], op: dict) -> list[dict]:
-    """Apply one block op (insert/update/delete/move) — LWW par bloc."""
+    """Apply one block op (insert/update/delete/move) — LWW par bloc.
+
+    Conservé pour la compatibilité : tests et chemins sans ``base`` continuent
+    de fonctionner. Le merge 3-voix vit dans ``RealtimeManager._apply``.
+    """
     t = op.get("type")
     if t == "insert":
         blk = op.get("block") or {}
@@ -90,9 +121,40 @@ def merge_ops(blocks: list[dict], ops: list[dict]) -> list[dict]:
     return out
 
 
+class RTConn:
+    """Une connexion WS : file sortante + tâche writer dédiée.
+
+    Le broadcast ne fait que ``put_nowait`` dans la file ; c'est la tâche writer
+    qui consomme et écrit sur le socket. Un client lent n'empêche donc jamais
+    les autres membres de la room de recevoir les messages.
+    """
+    __slots__ = ("ws", "user", "page_id", "out_q", "writer", "closed",
+                 "_ops_count", "_ops_window_start", "created_at")
+
+    def __init__(self, ws: WebSocket, user: dict, page_id: int):
+        self.ws = ws
+        self.user = user
+        self.page_id = page_id
+        self.out_q: asyncio.Queue = asyncio.Queue(maxsize=MAX_OUT_QUEUE)
+        self.writer: asyncio.Task | None = None
+        self.closed = False
+        self._ops_count = 0
+        self._ops_window_start = time.monotonic()
+        self.created_at = time.monotonic()
+
+    def op_budget_ok(self) -> bool:
+        """Fenêtre glissante simple anti-flood d'opérations."""
+        now = time.monotonic()
+        if now - self._ops_window_start > OP_WINDOW_SECONDS:
+            self._ops_window_start = now
+            self._ops_count = 0
+        self._ops_count += 1
+        return self._ops_count <= OP_WINDOW_MAX
+
+
 class Room:
     __slots__ = ("page_id", "blocks", "title", "version", "conns",
-                 "persist_task", "dirty")
+                 "persist_task", "dirty", "merge_count", "conflict_count")
 
     def __init__(self, page_id: int):
         self.page_id = page_id
@@ -102,21 +164,21 @@ class Room:
         self.conns: set[RTConn] = set()
         self.persist_task: asyncio.Task | None = None
         self.dirty = False
-
+        self.merge_count = 0
-
+        self.conflict_count = 0
-class RTConn:
-    __slots__ = ("ws", "user", "page_id")
-
-    def __init__(self, ws: WebSocket, user: dict, page_id: int):
-        self.ws = ws
-        self.user = user
-        self.page_id = page_id
 
 
 class RealtimeManager:
     def __init__(self):
         self._rooms: dict[int, Room] = {}
+        # compteurs globaux (observabilité)
+        self.stat_ops = 0
+        self.stat_merges = 0
+        self.stat_conflicts = 0
+        self.stat_slow_disconnects = 0
+        self.stat_connections_total = 0
 
+    # ── rooms ────────────────────────────────────────────────────────────
     def room(self, page_id: int) -> Room:
         return self._rooms.setdefault(page_id, Room(page_id))
 
@@ -150,42 +212,86 @@ class RealtimeManager:
                 room.blocks = []
         return True
 
+    # ── entrées / sorties ────────────────────────────────────────────────
     async def connect(self, ws: WebSocket, page_id: int, user: dict) -> RTConn | None:
-        room = self.room(page_id)
+        # Ne JAMAIS enregistrer la room avant d'avoir validé l'existence de la
+        # page : avant, un 4404 laissait une Room orpheline en mémoire pour
+        # toujours (fuite).
+        existing = self._rooms.get(page_id)
+        room = existing if existing is not None else Room(page_id)
         if not room.conns and not await self.load_room(room):
             await ws.close(code=4404)
-            return None
+            return None  # room non enregistrée → pas de fuite
+        if existing is None:
+            self._rooms[page_id] = room
+
         conn = RTConn(ws, user, page_id)
+        self.stat_connections_total += 1
+        conn.writer = asyncio.create_task(self._writer(conn))
         room.conns.add(conn)
+
         me = self._peer(user)
         peers = [self._peer(c.user) for c in room.conns if c is not conn]
         await ws.send_json({"t": "welcome", "self": me,
                             "peers": peers, "color": me["color"]})
         await ws.send_json({"t": "sync", "blocks": room.blocks,
                             "title": room.title, "version": room.version})
-        for c in room.conns:
+        await self._broadcast(room, {"t": "peer_join", "peer": me}, exclude=conn)
-            if c is not conn:
-                try:
-                    await c.ws.send_json({"t": "peer_join", "peer": me})
-                except Exception:
-                    pass
         return conn
 
     async def disconnect(self, conn: RTConn):
         room = self._rooms.get(conn.page_id)
-        if not room:
+        if room is None:
+            self._close_writer(conn)
             return
         room.conns.discard(conn)
-        for c in room.conns:
+        self._close_writer(conn)
-            try:
+        await self._broadcast(room, {"t": "peer_leave",
-                await c.ws.send_json({"t": "peer_leave",
+                                     "id": conn.user.get("id") or 0})
-                                      "id": conn.user.get("id") or 0})
-            except Exception:
-                pass
         if not room.conns:
             await self.flush(room)
             self._rooms.pop(conn.page_id, None)
 
+    def _close_writer(self, conn: RTConn):
+        conn.closed = True
+        t = conn.writer
+        if t is not None and not t.done():
+            t.cancel()
+        conn.writer = None
+
+    async def _writer(self, conn: RTConn):
+        """Tâche dédiée : consomme la file sortante et écrit sur le socket.
+
+        Sert aussi de filet de coalescence : après chaque message consommé, les
+        mises à jour de curseur (``sel``) déjà empilées sont réduites à la
+        dernière (les curseurs n'ont pas besoin d'être ordonnés entre eux, seule
+        la position la plus récente compte).
+        """
+        try:
+            while not conn.closed:
+                msg = await conn.out_q.get()
+                if msg is None:
+                    return
+                await conn.ws.send_json(msg)
+                # coalescence des curseurs en attente
+                pending_sel = []
+                while not conn.out_q.empty():
+                    nxt = conn.out_q.get_nowait()
+                    if nxt is None:
+                        return
+                    if isinstance(nxt, dict) and nxt.get("t") == "sel":
+                        pending_sel.append(nxt)
+                    else:
+                        await conn.ws.send_json(nxt)
+                if pending_sel:
+                    await conn.ws.send_json(pending_sel[-1])
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:  # socket mort → on arrête proprement
+            logger.debug("writer stopped: %s", e)
+            conn.closed = True
+
+    # ── persistance ──────────────────────────────────────────────────────
     async def flush(self, room: Room):
         """Write current room state to DB (sync, used on idle + disconnect)."""
         if room.persist_task and not room.persist_task.done():
@@ -218,67 +324,102 @@ class RealtimeManager:
 
         room.persist_task = asyncio.create_task(_run())
 
+    # ── broadcast non bloquant ───────────────────────────────────────────
+    def _enqueue(self, conn: RTConn, msg: dict) -> bool:
+        """Pose le message dans la file du client. Retourne False si trop lent."""
+        if conn.closed:
+            return False
+        try:
+            conn.out_q.put_nowait(msg)
+            return True
+        except asyncio.QueueFull:
+            return False
+
+    async def _evict_slow(self, room: Room, conn: RTConn):
+        """Client dépassé : on le déconnecte pour ne pas figer la room."""
+        self.stat_slow_disconnects += 1
+        logger.info("realtime: evicting slow client (user=%s page=%s)",
+                    conn.user.get("login"), conn.page_id)
+        room.conns.discard(conn)
+        self._close_writer(conn)
+        try:
+            await conn.ws.close(code=4413)
+        except Exception:
+            pass
+
     async def _broadcast(self, room: Room, msg: dict, exclude: RTConn | None = None):
-        for c in room.conns:
+        """Enfile ``msg`` chez chaque membre — jamais d'attente sur le socket."""
+        slow: list[RTConn] = []
+        for c in list(room.conns):
             if c is exclude:
                 continue
-            try:
+            if not self._enqueue(c, msg):
-                await c.ws.send_json(msg)
+                slow.append(c)
-            except Exception:
+        for c in slow:
-                pass
+            await self._evict_slow(room, c)
 
+    async def _send(self, conn: RTConn, msg: dict):
+        if not self._enqueue(conn, msg):
+            room = self._rooms.get(conn.page_id)
+            if room is not None:
+                await self._evict_slow(room, conn)
+
+    # ── protocole ────────────────────────────────────────────────────────
     async def handle(self, conn: RTConn, msg: dict):
         room = self._rooms.get(conn.page_id)
-        if not room:
+        if room is None:
             return
         t = msg.get("t")
         me = (conn.user.get("id") or 0)
 
         if t == "hello":
-            try:
+            await self._send(conn, {"t": "sync", "blocks": room.blocks,
-                await conn.ws.send_json({"t": "sync", "blocks": room.blocks,
+                                    "title": room.title, "version": room.version})
-                                         "title": room.title, "version": room.version})
-            except Exception:
-                pass
             return
 
         if t == "sync_req":
-            try:
+            await self._send(conn, {"t": "sync", "blocks": room.blocks,
-                await conn.ws.send_json({"t": "sync", "blocks": room.blocks,
+                                    "title": room.title, "version": room.version})
-                                         "title": room.title, "version": room.version})
-            except Exception:
-                pass
             return
 
         if t == "ping":
-            try:
+            await self._send(conn, {"t": "pong"})
-                await conn.ws.send_json({"t": "pong"})
-            except Exception:
-                pass
             return
 
         if t == "op":
+            if not conn.op_budget_ok():
+                # trop d'ops : on ignore silencieusement (le client resync)
+                await self._send(conn, {"t": "ack", "v": room.version,
+                                        "stale": True})
+                return
             op = msg.get("op") or {}
             client_v = msg.get("v", 0)
-            room.blocks = apply_op(room.blocks, op)
+            result = self._apply(room, op)
             room.version += 1
+            self.stat_ops += 1
             self._schedule_persist(room)
             stale = client_v < room.version - 1
-            await self._broadcast(room, {"t": "op", "op": op, "from": me,
+
-                                         "v": room.version}, exclude=conn)
+            # broadcast : on diffuse TOUJOURS le bloc final fusionné (pas la
-            try:
+            # proposition brute) pour que tous les clients convergent.
-                await conn.ws.send_json({"t": "ack", "v": room.version,
+            out_op = dict(op)
-                                         "stale": stale})
+            if result.get("merged") is not None:
-            except Exception:
+                out_op["block"] = result["merged"]
-                pass
+                out_op["merged"] = True
+            await self._broadcast(room, {"t": "op", "op": out_op, "from": me,
+                                         "v": room.version,
+                                         "conflict": result.get("conflict", False)},
+                                  exclude=conn)
+
+            ack = {"t": "ack", "v": room.version, "stale": stale}
+            if result.get("merged") is not None:
+                ack["merged"] = result["merged"]
+                ack["conflict"] = result.get("conflict", False)
+            await self._send(conn, ack)
+
             if stale:
-                try:
+                await self._send(conn, {"t": "sync", "blocks": room.blocks,
-                    await conn.ws.send_json({"t": "sync",
+                                        "title": room.title, "version": room.version})
-                                             "blocks": room.blocks,
-                                             "title": room.title,
-                                             "version": room.version})
-                except Exception:
-                    pass
             return
 
         if t == "title":
@@ -289,11 +430,7 @@ class RealtimeManager:
                 self._schedule_persist(room)
             await self._broadcast(room, {"t": "title", "title": room.title,
                                          "from": me, "v": room.version}, exclude=conn)
-            try:
+            await self._send(conn, {"t": "ack", "v": room.version, "stale": False})
-                await conn.ws.send_json({"t": "ack", "v": room.version,
-                                         "stale": False})
-            except Exception:
-                pass
             return
 
         if t == "sel":
@@ -303,16 +440,72 @@ class RealtimeManager:
                                          "offset": msg.get("offset", 0)}, exclude=conn)
             return
 
+    def _apply(self, room: Room, op: dict) -> dict:
+        """Applique une opération en mergeant à 3 voix si le client fournit
+        une ``base``. Retourne ``{"merged": bloc|None, "conflict": bool}``.
+
+        * ``update`` avec ``base`` → merge 3-voix (au-delà du LWW).
+        * le reste (insert/delete/move, ou update sans base) → LWW historique.
+        """
+        if op.get("type") != "update" or "base" not in op:
+            room.blocks = apply_op(room.blocks, op)
+            return {"merged": None, "conflict": False}
+
+        incoming = op.get("block") or {}
+        base = op.get("base")
+        bid = incoming.get("id")
+        if not bid:
+            return {"merged": None, "conflict": False}
+
+        current = next((b for b in room.blocks if b.get("id") == bid), None)
+        if current is None:
+            # bloc introuvable : LWW historique = pas d'update possible
+            return {"merged": None, "conflict": False}
+
+        merged, conflicts = merge_block_3way(base, current, incoming)
+        room.merge_count += 1
+        conflict = bool(conflicts)
+        if conflict:
+            room.conflict_count += 1
+            self.stat_conflicts += 1
+            logger.debug("realtime conflict page=%s block=%s fields=%s",
+                         room.page_id, bid, conflicts)
+        self.stat_merges += 1
+        room.blocks = [merged if b.get("id") == bid else b for b in room.blocks]
+        return {"merged": merged, "conflict": conflict}
+
+    # ── observabilité ────────────────────────────────────────────────────
     async def room_state(self, page_id: int) -> dict:
         room = self._rooms.get(page_id)
-        if not room.conns and not room.blocks:
+        if room is None:
+            # ne pas créer d'objet None : on charge dans une room jetable
+            room = Room(page_id)
+            if not await self.load_room(room):
+                return {"blocks": [], "title": "", "version": 0}
+        elif not room.conns and not room.blocks:
             await self.load_room(room)
         return {"blocks": room.blocks, "title": room.title, "version": room.version}
 
+    def stats(self) -> dict:
+        rooms = len(self._rooms)
+        conns = sum(len(r.conns) for r in self._rooms.values())
+        return {
+            "rooms": rooms,
+            "connections": conns,
+            "connections_total": self.stat_connections_total,
+            "ops": self.stat_ops,
+            "merges": self.stat_merges,
+            "conflicts": self.stat_conflicts,
+            "slow_disconnects": self.stat_slow_disconnects,
+            "pages": [{"page_id": r.page_id, "conns": len(r.conns),
+                       "version": r.version, "merges": r.merge_count,
+                       "conflicts": r.conflict_count}
+                      for r in self._rooms.values()],
+        }
+
     async def _propagate_synced(self, synced_id: int) -> None:
         """Broadcast a synced-block update to all rooms that reference it."""
         from app.db import get_conn as _get_conn
-        # Find all pages that reference this synced block
         pages: list[int] = []
         try:
             with _get_conn() as conn:
@@ -326,11 +519,10 @@ class RealtimeManager:
         for pid in pages:
             room = self._rooms.get(pid)
             if room and room.conns:
-                # Re-load the page content from DB to get fresh synced blocks
                 await self.load_room(room)
                 await self._broadcast(room, {"t": "synced_update",
-                                               "synced_id": synced_id,
+                                             "synced_id": synced_id,
-                                               "version": room.version})
+                                             "version": room.version})
 
 
 manager = RealtimeManager()

app/templates/_page_editor_realtime.html

@@ -110,7 +110,9 @@ window.__fdRT = (function () {
 
     cur.forEach(b => {
       if (baseById[b.id] !== undefined && JSON.stringify(baseById[b.id]) !== JSON.stringify(b)) {
-        ops.push({ type: 'update', block: clone(b) });
+        // v6.4.0 : on embarque la `base` dont dérive la saisie → le serveur
+        // fait un merge 3-voix au lieu d'écraser le bloc (LWW).
+        ops.push({ type: 'update', block: clone(b), base: clone(baseById[b.id]) });
       }
     });
     base.forEach(b => {
@@ -358,6 +360,27 @@ window.__fdRT = (function () {
     } else if (m.t === 'ack') {
       version = m.v || 0;
       if (pendingOps > 0) pendingOps--;
+      // v6.4.0 : le serveur renvoie le bloc fusionné (merge 3-voix). On
+      // l'adopte comme nouvelle base ; s'il diffère de notre saisie locale
+      // c'est qu'un autre utilisateur avait modifié le même bloc.
+      if (m.merged) {
+        const mid = m.merged.id;
+        const localBlock = (E && E.blocks || []).find(b => b.id === mid);
+        const differs = localBlock && JSON.stringify(localBlock) !== JSON.stringify(m.merged);
+        base = applyOpJS(base, { type: 'update', block: m.merged });
+        if (differs && E) {
+          const fid = activeBlockId();
+          if (fid !== mid) {
+            E.blocks = applyOpJS(E.blocks, { type: 'update', block: m.merged });
+            E.dirty = true;
+            E.render();
+            refocus(fid);
+          }
+          if (m.conflict && window.showToast) {
+            window.showToast('Editing conflict merged on a block', 'info');
+          }
+        }
+      }
       if (m.stale || needSync) { needSync = true; send({ t: 'sync_req' }); }
     } else if (m.t === 'op') {
       if (m.v) version = m.v;

tests/test_realtime_v64.py

@@ -0,0 +1,466 @@
+"""FlowDeck — v6.4.0 Realtime production: merge 3-voix (au-delà du LWW),
+broadcast non bloquant + coalescence des curseurs, corrections de fuites de
+rooms, anti-flood et observabilité.
+
+Covers :
+* ``realtime_merge`` — merge_text_3way / merge_block_3way (purs, sans WS)
+* protocole WS avec ``base`` → ack fusionné + drapeau conflict
+* convergence de deux clients sur le même bloc (les deux saisies survivent)
+* le LWW historique est préservé quand il n'y a pas de ``base``
+* room 4404 non enregistrée (fuite corrigée), ``room_state`` page inexistante
+* coalescence des curseurs via la file sortante
+* ``GET /api/realtime/stats``
+"""
+import json
+import os
+import tempfile
+
+import pytest
+from fastapi.testclient import TestClient
+from starlette.websockets import WebSocketDisconnect
+
+
+@pytest.fixture
+def client():
+    db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+    db_path = db_file.name
+    db_file.close()
+
+    os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
+    os.environ["APP_SECRET_KEY"] = "test-secret-for-realtime-v64"
+    os.environ["RATE_LIMIT_ENABLED"] = "false"
+    from app.config import settings
+    settings.database_url = f"sqlite:///{db_path}"
+
+    from app.db import get_conn, init_db
+    from app.main import app
+    init_db()
+    with get_conn() as conn:
+        conn.execute("INSERT OR IGNORE INTO users (id, login, full_name, is_admin) VALUES (1, 'tester', 'Tester', 1)")
+        conn.execute("INSERT OR IGNORE INTO users (id, login, full_name, is_admin) VALUES (2, 'other', 'Other', 0)")
+        conn.commit()
+
+    from app.services.realtime_server import manager
+    manager._rooms = {}
+    manager.stat_ops = 0
+    manager.stat_merges = 0
+    manager.stat_conflicts = 0
+    manager.stat_slow_disconnects = 0
+    manager.stat_connections_total = 0
+
+    tc = TestClient(app, raise_server_exceptions=False)
+    yield tc
+
+    os.unlink(db_path)
+
+
+def _make_page(title="V64 Page", blocks=None):
+    from app.db import get_conn
+    with get_conn() as conn:
+        cur = conn.execute(
+            "INSERT INTO pages (workspace, title, content, content_format, parent_section) "
+            "VALUES ('Private', ?, ?, 'blocks', 'Private')",
+            (title, json.dumps(blocks or [], ensure_ascii=False)),
+        )
+        conn.commit()
+        return cur.lastrowid
+
+
+def _token(user_id, login):
+    from app.auth.session import SessionManager
+    return SessionManager.create_session({"id": user_id, "login": login,
+                                          "full_name": login.title(), "is_admin": 1})
+
+
+def _auth_client(client, user_id=1, login="tester"):
+    client.cookies.set("flowdeck_session", _token(user_id, login))
+
+
+# ── realtime_merge (purs) ────────────────────────────────────────────────
+
+def test_merge_text_disjoint_edits_survive():
+    from app.services.realtime_merge import merge_text_3way
+    base = "hello world"
+    cur = "hello brave world"      # serveur : insert "brave"
+    inc = "hello world today"      # client  : append "today"
+    out, conflict = merge_text_3way(base, cur, inc)
+    assert conflict is False
+    assert "brave" in out and "today" in out
+
+
+def test_merge_text_disjoint_edits_order_independent():
+    from app.services.realtime_merge import merge_text_3way
+    base = "hello world"
+    a, _ = merge_text_3way(base, "hello brave world", "hello world today")
+    b, _ = merge_text_3way(base, "hello world today", "hello brave world")
+    assert a == b == "hello brave world today"
+
+
+def test_merge_text_real_overlap_flags_conflict_lww():
+    from app.services.realtime_merge import merge_text_3way
+    out, conflict = merge_text_3way("hello", "hxllo", "hyllo")
+    assert conflict is True
+    assert out == "hyllo"  # LWW : incoming gagne
+
+
+def test_merge_text_shortcuts():
+    from app.services.realtime_merge import merge_text_3way
+    assert merge_text_3way("a", "a", "b") == ("b", False)   # client a changé
+    assert merge_text_3way("a", "b", "a") == ("b", False)   # serveur a changé
+    assert merge_text_3way("a", "b", "b") == ("b", False)   # même changement
+
+
+def test_merge_text_same_point_insertion_not_destructive():
+    from app.services.realtime_merge import merge_text_3way
+    out, conflict = merge_text_3way("ab", "aXb", "aYb")
+    # les deux insertions au même index sont conservées (concaténées)
+    assert "X" in out and "Y" in out
+    assert conflict is False
+
+
+def test_changed_region():
+    from app.services.realtime_merge import changed_region
+    assert changed_region("hello", "hello") is None
+    assert changed_region("hello world", "hello brave world") == (6, 6, "brave ")
+    assert changed_region("abc", "abXc") == (2, 2, "X")
+
+
+def test_merge_block_field_level_no_conflict():
+    """Serveur et client changent des champs différents → pas de conflit."""
+    from app.services.realtime_merge import merge_block_3way
+    base = {"id": "1", "text": "hi", "done": False}
+    cur = {"id": "1", "text": "hi there", "done": False}   # serveur édite text
+    inc = {"id": "1", "text": "hi", "done": True}          # client coche done
+    merged, conflicts = merge_block_3way(base, cur, inc)
+    assert merged["text"] == "hi there"
+    assert merged["done"] is True
+    assert conflicts == []
+
+
+def test_merge_block_text_disjoint_merges():
+    from app.services.realtime_merge import merge_block_3way
+    base = {"id": "1", "text": "hello world"}
+    cur = {"id": "1", "text": "hello brave world"}
+    inc = {"id": "1", "text": "hello world today"}
+    merged, conflicts = merge_block_3way(base, cur, inc)
+    assert conflicts == []
+    assert "brave" in merged["text"] and "today" in merged["text"]
+
+
+def test_merge_block_scalar_conflict_lww():
+    from app.services.realtime_merge import merge_block_3way
+    merged, conflicts = merge_block_3way({"id": "1", "n": 1},
+                                         {"id": "1", "n": 2},
+                                         {"id": "1", "n": 3})
+    assert merged["n"] == 3
+    assert "n" in conflicts
+
+
+def test_merge_block_server_deleted_field_stays_deleted():
+    from app.services.realtime_merge import merge_block_3way
+    base = {"id": "1", "tmp": "x"}
+    cur = {"id": "1"}                      # serveur a supprimé tmp
+    inc = {"id": "1", "tmp": "x"}          # client n'a pas touché tmp
+    merged, _ = merge_block_3way(base, cur, inc)
+    assert "tmp" not in merged
+
+
+def test_merge_block_incoming_non_dict_flagged():
+    from app.services.realtime_merge import merge_block_3way
+    _, conflicts = merge_block_3way({"id": "1"}, {"id": "1"}, "junk")
+    assert conflicts == ["__block__"]
+
+
+def test_merge_block_preserves_id_when_base_empty():
+    from app.services.realtime_merge import merge_block_3way
+    merged, _ = merge_block_3way({}, {}, {"id": "z", "text": "new"})
+    assert merged["id"] == "z"
+
+
+# ── apply_op (LWW historique préservé) ───────────────────────────────────
+
+def test_apply_op_still_lww_without_base():
+    from app.services.realtime_server import apply_op
+    blocks = [{"id": "a", "content": "A"}]
+    out = apply_op(blocks, {"type": "update", "block": {"id": "a", "content": "B"}})
+    assert out[0]["content"] == "B"
+    out = apply_op(blocks, {"type": "delete", "id": "a"})
+    assert out == []
+    out = apply_op(blocks, {"type": "move", "id": "a", "index": 0})
+    assert [b["id"] for b in out] == ["a"]
+
+
+# ── protocole WS ─────────────────────────────────────────────────────────
+
+def test_ws_requires_auth(client):
+    _make_page()
+    with pytest.raises(WebSocketDisconnect) as exc:
+        with client.websocket_connect("/ws/pages/1") as ws:
+            ws.receive_json()
+    assert exc.value.code == 4401
+
+
+def test_ws_missing_page_4404(client):
+    _auth_client(client)
+    with pytest.raises(WebSocketDisconnect) as exc:
+        with client.websocket_connect("/ws/pages/9999") as ws:
+            ws.receive_json()
+    assert exc.value.code == 4404
+
+
+def test_ws_4404_does_not_leak_room(client):
+    """Une tentative de connexion sur une page inexistante ne doit pas laisser
+    de Room orpheline en mémoire (fuite corrigée en v6.4.0)."""
+    from app.services.realtime_server import manager
+    _auth_client(client)
+    with pytest.raises(WebSocketDisconnect):
+        with client.websocket_connect("/ws/pages/9999") as ws:
+            ws.receive_json()
+    assert 9999 not in manager._rooms, f"room leak: {list(manager._rooms)}"
+
+
+def test_ws_hello_gets_sync(client):
+    pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "Hi"}])
+    _auth_client(client)
+    with client.websocket_connect(f"/ws/pages/{pid}") as ws:
+        w = ws.receive_json()
+        assert w["t"] == "welcome"
+        s = ws.receive_json()
+        assert s["t"] == "sync"
+        assert s["blocks"][0]["id"] == "a"
+        ws.send_json({"t": "hello"})
+        s2 = ws.receive_json()
+        assert s2["t"] == "sync" and s2["blocks"][0]["content"] == "Hi"
+
+
+def test_ws_update_with_base_returns_merged(client):
+    """Update avec base → ack fusionné, pas de conflit si seul le client a changé."""
+    pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "x"}])
+    _auth_client(client)
+    with client.websocket_connect(f"/ws/pages/{pid}") as ws:
+        ws.receive_json()  # welcome
+        ws.receive_json()  # sync
+        ws.send_json({"t": "op", "v": 0, "op": {
+            "type": "update",
+            "block": {"id": "a", "type": "paragraph", "content": "x!"},
+            "base": {"id": "a", "type": "paragraph", "content": "x"},
+        }})
+        ack = ws.receive_json()
+        assert ack["t"] == "ack"
+        assert ack.get("merged")["content"] == "x!"
+        assert ack.get("conflict") is False
+
+
+def test_ws_two_clients_same_block_merge_disjoint(client):
+    """Deux clients éditent le même bloc à des endroits différents → les deux
+    saisies survivent (au-delà du LWW) et convergeamment."""
+    pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "hello world"}])
+    _auth_client(client, 1, "tester")
+    with client.websocket_connect(f"/ws/pages/{pid}") as wa:
+        wa.receive_json()  # welcome
+        wa.receive_json()  # sync
+        _auth_client(client, 2, "other")
+        with client.websocket_connect(f"/ws/pages/{pid}") as wb:
+            wb.receive_json()  # welcome
+            wb.receive_json()  # sync
+            wa.receive_json()  # peer_join
+
+            # A édite la fin
+            wa.send_json({"t": "op", "v": 1, "op": {
+                "type": "update",
+                "block": {"id": "a", "type": "paragraph", "content": "hello world today"},
+                "base": {"id": "a", "type": "paragraph", "content": "hello world"},
+            }})
+            ack_a = wa.receive_json()
+            assert ack_a["t"] == "ack"
+            op_b = wb.receive_json()
+            assert op_b["t"] == "op"
+
+            # B édite le début, sur la même base
+            wb.send_json({"t": "op", "v": 1, "op": {
+                "type": "update",
+                "block": {"id": "a", "type": "paragraph", "content": "hello brave world"},
+                "base": {"id": "a", "type": "paragraph", "content": "hello world"},
+            }})
+            ack_b = wb.receive_json()
+            assert ack_b["t"] == "ack"
+            merged = ack_b["merged"]["content"]
+            # les deux éditions disjointes doivent être conservées
+            assert "brave" in merged, merged
+            assert "today" in merged, merged
+            assert ack_b["conflict"] is False
+
+            # A reçoit aussi le bloc fusionné
+            op_a2 = wa.receive_json()
+            assert op_a2["t"] == "op"
+            assert "brave" in op_a2["op"]["block"]["content"]
+            assert "today" in op_a2["op"]["block"]["content"]
+
+    # persistance du contenu fusionné
+    from app.db import get_conn
+    with get_conn() as conn:
+        row = conn.execute("SELECT content FROM pages WHERE id=?", (pid,)).fetchone()
+    d = json.loads(row["content"])
+    assert "brave" in d[0]["content"] and "today" in d[0]["content"]
+
+
+def test_ws_update_without_base_stays_lww(client):
+    """Sans ``base`` (ancien client / protocole), le LWW historique s'applique."""
+    pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "x"}])
+    _auth_client(client, 1, "tester")
+    with client.websocket_connect(f"/ws/pages/{pid}") as wa:
+        wa.receive_json()
+        wa.receive_json()
+        _auth_client(client, 2, "other")
+        with client.websocket_connect(f"/ws/pages/{pid}") as wb:
+            wb.receive_json()
+            wb.receive_json()
+            wa.receive_json()  # peer_join
+            wa.send_json({"t": "op", "v": 0, "op": {"type": "update",
+                          "block": {"id": "a", "type": "paragraph", "content": "first"}}})
+            wa.receive_json()  # ack
+            b_op = wb.receive_json()
+            assert b_op["op"]["block"]["content"] == "first"
+            wb.send_json({"t": "op", "v": 1, "op": {"type": "update",
+                          "block": {"id": "a", "type": "paragraph", "content": "second"}}})
+            wb.receive_json()  # ack (pas de merged sans base)
+            a_op = wa.receive_json()
+            assert a_op["op"]["block"]["content"] == "second"
+    from app.db import get_conn
+    with get_conn() as conn:
+        row = conn.execute("SELECT content FROM pages WHERE id=?", (pid,)).fetchone()
+    assert json.loads(row["content"])[0]["content"] == "second"
+
+
+def test_ws_conflicting_scalar_flagged(client):
+    """Conflit sur un champ scalaire multi-états → drapeau conflict=True + LWW.
+
+    Un booléen ne peut pas diverger depuis la même base (2 valeurs) ; on utilise
+    un ``status`` à 3 états pour que les deux côtés changent *différemment*.
+    """
+    pid = _make_page(blocks=[{"id": "a", "type": "todo", "content": "t", "status": "todo"}])
+    _auth_client(client, 1, "tester")
+    with client.websocket_connect(f"/ws/pages/{pid}") as wa:
+        wa.receive_json()
+        wa.receive_json()
+        _auth_client(client, 2, "other")
+        with client.websocket_connect(f"/ws/pages/{pid}") as wb:
+            wb.receive_json()
+            wb.receive_json()
+            wa.receive_json()
+            # A passe todo → doing
+            wa.send_json({"t": "op", "v": 1, "op": {
+                "type": "update",
+                "block": {"id": "a", "type": "todo", "content": "t", "status": "doing"},
+                "base": {"id": "a", "type": "todo", "content": "t", "status": "todo"},
+            }})
+            wa.receive_json()  # ack
+            wb.receive_json()  # op broadcast
+            # B passe todo → done (conflit scalaire réel, même base)
+            wb.send_json({"t": "op", "v": 1, "op": {
+                "type": "update",
+                "block": {"id": "a", "type": "todo", "content": "t", "status": "done"},
+                "base": {"id": "a", "type": "todo", "content": "t", "status": "todo"},
+            }})
+            ack_b = wb.receive_json()
+            assert ack_b["t"] == "ack"
+            assert ack_b["conflict"] is True
+            assert ack_b["merged"]["status"] == "done"  # LWW : incoming gagne
+            assert "status" in str(ack_b.get("merged"))
+
+
+def test_ws_stats_endpoint(client):
+    _auth_client(client)
+    pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "x"}])
+    with client.websocket_connect(f"/ws/pages/{pid}") as ws:
+        ws.receive_json()
+        ws.receive_json()
+        ws.send_json({"t": "op", "v": 0, "op": {
+            "type": "update",
+            "block": {"id": "a", "type": "paragraph", "content": "y"},
+            "base": {"id": "a", "type": "paragraph", "content": "x"},
+        }})
+        ws.receive_json()  # ack
+        r = client.get("/api/realtime/stats")
+        assert r.status_code == 200
+        d = r.json()
+        assert d["rooms"] == 1
+        assert d["connections"] == 1
+        assert d["ops"] >= 1
+        assert d["merges"] >= 1
+        assert isinstance(d["pages"], list) and d["pages"][0]["page_id"] == pid
+
+
+def test_ws_stats_requires_auth(client):
+    r = client.get("/api/realtime/stats")
+    assert r.status_code == 200
+    assert r.json() == {"error": "unauthorized"}
+
+
+def test_ws_op_budget_anti_flood(client):
+    """Au-delà du budget d'ops, le serveur répond ack stale (sans appliquer)."""
+    from app.services import realtime_server as rs
+    pid = _make_page(blocks=[{"id": "a", "type": "paragraph", "content": "x"}])
+    _auth_client(client)
+    old_max = rs.OP_WINDOW_MAX
+    rs.OP_WINDOW_MAX = 3
+    try:
+        with client.websocket_connect(f"/ws/pages/{pid}") as ws:
+            ws.receive_json()
+            ws.receive_json()
+            for i in range(6):
+                ws.send_json({"t": "op", "v": 0, "op": {"type": "update",
+                              "block": {"id": "a", "type": "paragraph", "content": f"v{i}"}}})
+            acks = [ws.receive_json() for _ in range(6)]
+            stale = [a for a in acks if a.get("stale")]
+            assert stale, "budget anti-flood n'a pas été déclenché"
+    finally:
+        rs.OP_WINDOW_MAX = old_max
+
+
+def test_room_state_missing_page_returns_empty(client):
+    import asyncio
+
+    from app.services.realtime_server import manager
+    out = asyncio.run(manager.room_state(424242))
+    assert out == {"blocks": [], "title": "", "version": 0}
+    assert 424242 not in manager._rooms
+
+
+def test_writer_coalesces_cursor_updates():
+    """La file sortante d'une connexion coalesce les curseurs (une seule position
+    finale conservée) tout en préservant l'ordre des messages importants."""
+    import asyncio
+
+    class FakeWS:
+        def __init__(self):
+            self.sent = []
+        async def send_json(self, msg):
+            self.sent.append(msg)
+
+    async def run():
+        from app.services.realtime_server import RealtimeManager, RTConn
+        ws = FakeWS()
+        conn = RTConn(ws, {"id": 1, "login": "u"}, 1)
+        conn.writer = asyncio.create_task(RealtimeManager()._writer(conn))
+        # empiler plusieurs curseurs + un message important au milieu
+        conn.out_q.put_nowait({"t": "sel", "block": "a", "offset": 1})
+        conn.out_q.put_nowait({"t": "sel", "block": "a", "offset": 2})
+        conn.out_q.put_nowait({"t": "title", "title": "T"})
+        conn.out_q.put_nowait({"t": "sel", "block": "a", "offset": 3})
+        await asyncio.sleep(0.05)
+        conn.closed = True
+        conn.out_q.put_nowait(None)
+        await asyncio.sleep(0.02)
+        if conn.writer and not conn.writer.done():
+            conn.writer.cancel()
+        return ws.sent
+
+    sent = asyncio.run(run())
+    sels = [m for m in sent if m.get("t") == "sel"]
+    titles = [m for m in sent if m.get("t") == "title"]
+    # les curseurs sont coalescés : on ne garde pas les positions 1 et 2 séparées
+    assert len(sels) <= 2, f"curseurs non coalescés: {sels}"
+    assert titles and titles[0]["title"] == "T"
+    # l'offset final (3) doit être présent
+    assert sels[-1]["offset"] == 3, sels