Historique chat : coalescer le journal en fin de tour (fini la perte du debut)

Le journal d'affichage stockait un evenement par TOKEN et etait plafonne a 5000
-> il se remplissait en quelques reponses et perdait le debut de la conversation
a l'ecran (impossible de remonter jusqu'au deput). Fix : compactLogLocked()
coalesce les suites content/reasoning en un seul evenement chacun a la fin de
chaque tour (le replay le faisait deja, pas le stockage), en preservant toks/ts0
pour le compteur tok/s. Plafond monte a 20000 pour absorber un gros tour en cours
avant sa coalescence. Un tour terminé pese desormais une poignee d'evenements au
lieu d'un par token -> des centaines de tours tiennent dans le journal.
This commit is contained in:
nathaninline committed 2026-07-23 20:01:54 +02:00
1 parent ac8095024f
commit 7cfc548474
1 file changed
+94 -7
+94 -7
View File
@@ -24,11 +24,13 @@ import (
// JOURNALISE ces événements (Log) horodatés par Seq. Se reconnecter = rejouer
// Log[from:] puis suivre les événements en direct — aucun code de rendu nouveau.
// maxLogEvents plafonne le journal d'AFFICHAGE (pas la vue modèle). Une très
// longue conversation ne fait pas exploser la mémoire/le disque ; la troncature
// ne retire que de vieilles bulles à l'écran, jamais du contexte du modèle
// (celui-ci vit dans Messages, réduit séparément par le compactage).
const maxLogEvents = 5000
// maxLogEvents plafonne le journal d'AFFICHAGE (pas la vue modèle). Depuis la
// coalescence en fin de tour (compactLogLocked), un tour terminé ne pèse plus
// qu'une poignée d'événements au lieu d'un par token → 20000 couvre des
// centaines de tours. La marge sert surtout à absorber UN tour en cours (streamé
// token par token) avant sa coalescence : un très gros tour (long raisonnement +
// réponse) ne doit pas se faire tronquer le début avant d'être compacté.
const maxLogEvents = 20000
// LogEvent = un événement d'affichage rejouable (un delta SSE + son numéro de
// séquence monotone + un horodatage serveur en ms). Le TS permet au client de
@@ -119,6 +121,88 @@ func (c *Conversation) appendDelta(epoch int, delta map[string]any) {
c.mu.Unlock()
}
// evToks lit le nombre de tokens porté par un événement texte : 1 pour un delta
// brut (streaming), ou la valeur `toks` accumulée pour un événement déjà coalescé.
func evToks(d map[string]any) int {
switch v := d["toks"].(type) {
case int:
return v
case float64: // relu depuis le JSON persisté
return int(v)
}
return 1
}
// evTS0 lit l'horodatage de DÉBUT d'un événement texte (présent seulement sur les
// événements coalescés) ; sinon on retombe sur `fallback` (le TS de l'événement).
func evTS0(d map[string]any, fallback int64) int64 {
switch v := d["ts0"].(type) {
case int64:
return v
case float64:
return int64(v)
}
return fallback
}
// compactLogLocked coalesce EN PLACE les suites d'événements content /
// reasoning_content du journal d'affichage en un seul événement chacun (même
// logique que coalesceReplay). Sans ça, le journal grossit token par token et
// atteint maxLogEvents en quelques réponses → on perd le DÉBUT de la conversation
// à l'affichage. On l'appelle en fin de tour (les événements sont alors figés).
// On préserve `toks` (somme) et `ts0` (premier) pour que le compteur de vitesse
// (tok/s) reste correct au replay. Verrou détenu par l'appelant.
func (c *Conversation) compactLogLocked() {
if len(c.Log) < 2 {
return
}
textKey := func(d map[string]any) string {
if _, ok := d["content"].(string); ok {
return "content"
}
if _, ok := d["reasoning_content"].(string); ok {
return "reasoning_content"
}
return ""
}
out := make([]LogEvent, 0, len(c.Log))
var buf strings.Builder
bufKey := ""
var cur LogEvent
var toks int
var ts0 int64
flush := func() {
if bufKey == "" {
return
}
cur.Delta = map[string]any{bufKey: buf.String(), "toks": toks, "ts0": ts0}
out = append(out, cur)
buf.Reset()
bufKey, toks, ts0 = "", 0, 0
}
for _, ev := range c.Log {
key := textKey(ev.Delta)
if key == "" {
flush()
out = append(out, ev)
continue
}
if bufKey != "" && bufKey != key {
flush()
}
if bufKey == "" {
bufKey = key
cur = LogEvent{Seq: ev.Seq, TS: ev.TS}
ts0 = evTS0(ev.Delta, ev.TS)
}
buf.WriteString(ev.Delta[key].(string))
cur.Seq, cur.TS = ev.Seq, ev.TS
toks += evToks(ev.Delta)
}
flush()
c.Log = out
}
// convState renvoie un instantané léger (pour /api/chat/state).
func (c *Conversation) state() map[string]any {
c.mu.Lock()
@@ -176,6 +260,9 @@ func (c *Conversation) generate(ctx context.Context, caps Caps, temperature floa
return // Reset pendant le tour : Reset a déjà persisté l'état vide
}
c.appendDelta(epoch, map[string]any{"turn_done": true})
c.mu.Lock()
c.compactLogLocked() // le tour est fini : coalesce ses tokens pour garder le journal petit
c.mu.Unlock()
c.persist()
}()
@@ -409,11 +496,11 @@ func coalesceReplay(events []LogEvent, from int) []map[string]any {
flush()
}
if bufKey == "" {
bufKey, bufTs0 = key, ev.TS
bufKey, bufTs0 = key, evTS0(ev.Delta, ev.TS)
}
buf.WriteString(ev.Delta[key].(string))
bufSeq, bufTs = ev.Seq, ev.TS
bufToks++
bufToks += evToks(ev.Delta) // 1 pour un delta brut, N pour un événement déjà coalescé
continue
}
flush()