perf + robustesse: replay coalesce + envoi resilient

- Subscribe rejoue le journal COALESCE (deltas texte consecutifs fusionnes en un
  evenement, avec toks + bornes ts) : chargement rapide meme sur longue conv, fini
  le rendu markdown quadratique token par token
- send() resilient : retry, 409/generating traite comme succes -> plus de 'network
  error' alarmiste alors que l'IA repond quand meme ; reconnexion du flux plus rapide

conversation.go

@@ -244,13 +244,66 @@ func (c *Conversation) Reset() {
 	c.persist()
 }
 
-// Subscribe diffuse les événements au client via emit, en commençant par le
+// coalesceReplay fusionne les deltas texte consécutifs (content / reasoning_content)
-// replay de Log[from:] puis en suivant le direct. Bloque jusqu'à ce que ctx (la
+// d'un même bloc en UN seul événement, pour que le replay au chargement soit léger
-// connexion HTTP) soit annulé — la génération, elle, continue indépendamment.
+// (quelques événements par tour au lieu de milliers de tokens). On conserve le
-// emit renvoie false si l'écriture échoue (client parti) → on sort.
+// nombre de tokens fusionnés (toks) et les bornes d'horodatage (ts0→ts) pour que le
+// client reconstitue le compteur ET la vitesse. Les événements non-texte (user,
+// tool_used, stats, turn_done…) passent tels quels.
+func coalesceReplay(events []LogEvent, from int) []map[string]any {
+	var out []map[string]any
+	var buf strings.Builder
+	bufKey := ""
+	var bufSeq, bufToks int
+	var bufTs0, bufTs int64
+	flush := func() {
+		if bufKey == "" {
+			return
+		}
+		out = append(out, map[string]any{bufKey: buf.String(), "seq": bufSeq, "ts": bufTs, "ts0": bufTs0, "toks": bufToks})
+		buf.Reset()
+		bufKey, bufSeq, bufToks, bufTs0, bufTs = "", 0, 0, 0, 0
+	}
+	for _, ev := range events {
+		if ev.Seq <= from {
+			continue
+		}
+		// Delta texte ? (une seule clé content ou reasoning_content, valeur string)
+		key := ""
+		if s, ok := ev.Delta["content"].(string); ok {
+			key, _ = "content", s
+		} else if s, ok := ev.Delta["reasoning_content"].(string); ok {
+			key, _ = "reasoning_content", s
+		}
+		if key != "" {
+			if bufKey != "" && bufKey != key {
+				flush()
+			}
+			if bufKey == "" {
+				bufKey, bufTs0 = key, ev.TS
+			}
+			buf.WriteString(ev.Delta[key].(string))
+			bufSeq, bufTs = ev.Seq, ev.TS
+			bufToks++
+			continue
+		}
+		flush()
+		m := map[string]any{"seq": ev.Seq, "ts": ev.TS}
+		for k, v := range ev.Delta {
+			m[k] = v
+		}
+		out = append(out, m)
+	}
+	flush()
+	return out
+}
+
+// Subscribe diffuse les événements au client via emit : d'abord un REPLAY coalescé
+// de Log[from:] (léger), puis un caught_up, puis le DIRECT événement par événement.
+// Bloque jusqu'à ce que ctx (la connexion HTTP) soit annulé — la génération, elle,
+// continue indépendamment. emit renvoie false si l'écriture échoue (client parti).
 func (c *Conversation) Subscribe(ctx context.Context, from int, emit func(map[string]any) bool) {
-	// Réveille les attentes de cond quand la connexion se ferme (sinon cond.Wait
+	// Réveille les attentes de cond quand la connexion se ferme.
-	// resterait bloqué faute de nouvel événement).
 	go func() {
 		<-ctx.Done()
 		c.mu.Lock()
@@ -258,17 +311,35 @@ func (c *Conversation) Subscribe(ctx context.Context, from int, emit func(map[st
 		c.mu.Unlock()
 	}()
 
+	// 1. Replay coalescé (snapshot hors verrou pour ne pas bloquer la génération).
 	c.mu.Lock()
-	last := from
+	snapshot := append([]LogEvent(nil), c.Log...)
 	epoch := c.epoch
-	announced := false // a-t-on signalé la fin du replay (caught_up) ?
+	c.mu.Unlock()
+	last := from
+	for _, ev := range coalesceReplay(snapshot, from) {
+		if ctx.Err() != nil {
+			return
+		}
+		if !emit(ev) {
+			return
+		}
+		if s, ok := ev["seq"].(int); ok {
+			last = s
+		}
+	}
+	if !emit(map[string]any{"caught_up": true}) {
+		return
+	}
+
+	// 2. Direct : événements granulaires au-delà de `last`.
+	c.mu.Lock()
 	for {
 		if ctx.Err() != nil {
 			c.mu.Unlock()
 			return
 		}
-		// Un reset a eu lieu : on ordonne au client de nettoyer et on repart de 0.
+		if c.epoch != epoch { // reset → on ordonne au client de nettoyer et on repart
-		if c.epoch != epoch {
 			epoch = c.epoch
 			last = 0
 			c.mu.Unlock()
@@ -278,7 +349,6 @@ func (c *Conversation) Subscribe(ctx context.Context, from int, emit func(map[st
 			c.mu.Lock()
 			continue
 		}
-		// Envoie tout ce qui est plus récent que `last`.
 		sent := false
 		for _, ev := range c.Log {
 			if ev.Seq <= last {
@@ -303,20 +373,8 @@ func (c *Conversation) Subscribe(ctx context.Context, from int, emit func(map[st
 			}
 		}
 		if sent {
-			continue // il peut y avoir eu de nouveaux événements pendant l'envoi
-		}
-		// Replay drainé : on le signale UNE fois au client (il replie alors
-		// instantanément les vieilles bulles, sans animation), avant de suivre le
-		// direct. Événement synthétique, non journalisé.
-		if !announced {
-			announced = true
-			c.mu.Unlock()
-			if !emit(map[string]any{"caught_up": true}) {
-				return
-			}
-			c.mu.Lock()
 			continue
 		}
-		c.cond.Wait() // rien de neuf : dort jusqu'au prochain Broadcast (ou ctx annulé)
+		c.cond.Wait()
 	}
 }

ui/index.html

@@ -1458,14 +1458,16 @@ function handleDelta(d){
     killTyping();
     if(!T.reasonEl){ collapseAll(T.turnCollapsibles); T.reasonEl=addMsg('reasoning',''); if(REPLAYING) collapseInstant(T.reasonEl); T.fullReason=''; T.turnCollapsibles.push(T.reasonEl); }
     showTyping(); T.fullReason+=d.reasoning_content; renderBody(T.reasonEl, T.fullReason);
-    if(!T.reasonFirstTs) T.reasonFirstTs=d.ts||0; T.reasonLastTs=d.ts||T.reasonLastTs; T.reasonTok++;
+    // d.toks/d.ts0 présents quand l'événement est coalescé (replay) : plusieurs
+    // tokens d'un coup. Sinon (direct), 1 token, ts0=ts.
+    if(!T.reasonFirstTs) T.reasonFirstTs=d.ts0||d.ts||0; T.reasonLastTs=d.ts||T.reasonLastTs; T.reasonTok+=(d.toks||1);
     labelTokens(T.reasonEl, 'reasoning', T.reasonTok, T.reasonFirstTs, T.reasonLastTs);
     return; }
   if(d.content){
     removeTyping();
     if(!T.contentEl){ collapseAll(T.turnCollapsibles); T.contentEl=addMsg('assistant',''); T.fullContent=''; }
     T.fullContent+=d.content; renderBody(T.contentEl, T.fullContent);
-    if(!T.contentFirstTs) T.contentFirstTs=d.ts||0; T.contentLastTs=d.ts||T.contentLastTs; T.contentTok++;
+    if(!T.contentFirstTs) T.contentFirstTs=d.ts0||d.ts||0; T.contentLastTs=d.ts||T.contentLastTs; T.contentTok+=(d.toks||1);
     labelTokens(T.contentEl, 'assistant', T.contentTok, T.contentFirstTs, T.contentLastTs);
     return; }
 }
@@ -1489,22 +1491,32 @@ async function connectStream(){
           }
         }
       }
-    }catch(e){ /* coupure : on reconnecte */ }
+    }catch(e){ /* coupure : on reconnecte silencieusement */ }
-    await new Promise(res=>setTimeout(res, 1200));
+    await new Promise(res=>setTimeout(res, 600));
   }
 }
-function stopGen(){ jfetch('/api/chat/stop',{method:'POST'}).catch(()=>{}); toast('stop'); }
+// Envoi RÉSILIENT : sur le tunnel E2E, un aller-retour peut échouer transitoirement
+// alors qu'il a en fait abouti (la génération démarre). On réessaie, et un 409
+// (« déjà en cours ») = succès (c'est notre envoi qui est passé). On ne montre une
+// erreur qu'après plusieurs échecs ET vérification que rien ne tourne — plus de
+// « network error » alarmiste alors que l'IA répond quand même.
 async function send(){
   if(busy) return;
   const ta=document.getElementById('input'); const text=ta.value.trim();
   if(!text) return;
   ta.value='';
-  try{
+  for(let attempt=0; attempt<3; attempt++){
-    const r=await jfetch('/api/chat/send',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({message:text,ctx_used:CTX_USED})});
+    try{
-    if(r.status===409){ toast('génération déjà en cours'); ta.value=text; return; }
+      const r=await jfetch('/api/chat/send',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({message:text,ctx_used:CTX_USED})});
-    if(!r.ok){ let m='erreur'; try{ m=(await r.json()).error||m; }catch(_){} toast(m); ta.value=text; return; }
+      if(r.status===409) return;               // déjà en cours (notre envoi a abouti) → OK
-    // La bulle utilisateur et les tokens arrivent par le flux d'abonnement.
+      if(r.ok) return;                          // la bulle + les tokens arrivent par le flux
-  }catch(e){ toast('échec envoi: '+e.message); ta.value=text; }
+      if(r.status<500){ let m='erreur'; try{ m=(await r.json()).error||m; }catch(_){} toast(m); ta.value=text; return; }
+    }catch(e){ /* réseau : on retente */ }
+    await new Promise(res=>setTimeout(res, 600));
+  }
+  // Après plusieurs échecs : le serveur a peut-être quand même reçu le message.
+  try{ const s=await (await jfetch('/api/chat/state')).json(); if(s.generating) return; }catch(_){}
+  toast('échec de l\'envoi — réessaie'); ta.value=text;
 }
 loadAll();
 setInterval(loadStatus, 5000);