diff --git a/conversation.go b/conversation.go index e4d299c..b6bf3c5 100644 --- a/conversation.go +++ b/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 -// replay de Log[from:] puis en suivant le direct. 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) → on sort. +// coalesceReplay fusionne les deltas texte consécutifs (content / reasoning_content) +// d'un même bloc en UN seul événement, pour que le replay au chargement soit léger +// (quelques événements par tour au lieu de milliers de tokens). On conserve le +// 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 - // resterait bloqué faute de nouvel événement). + // Réveille les attentes de cond quand la connexion se ferme. 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 { + if c.epoch != epoch { // reset → on ordonne au client de nettoyer et on repart 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() } } diff --git a/ui/index.html b/ui/index.html index 6fe9999..cf9ab9e 100644 --- a/ui/index.html +++ b/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 */ } - await new Promise(res=>setTimeout(res, 1200)); + }catch(e){ /* coupure : on reconnecte silencieusement */ } + 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{ - 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.status===409){ toast('génération déjà en cours'); ta.value=text; return; } - if(!r.ok){ let m='erreur'; try{ m=(await r.json()).error||m; }catch(_){} toast(m); ta.value=text; return; } - // La bulle utilisateur et les tokens arrivent par le flux d'abonnement. - }catch(e){ toast('échec envoi: '+e.message); ta.value=text; } + for(let attempt=0; attempt<3; attempt++){ + try{ + 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.status===409) return; // déjà en cours (notre envoi a abouti) → OK + if(r.ok) return; // la bulle + les tokens arrivent par le flux + 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);