mirror of
https://github.com/R0m1k3/Loki.git
synced 2026-10-11 17:26:57 +02:00
correctif : reponse affichee en double quand la compaction de fin de tour fusionne un bloc que le client suivait en direct
This commit is contained in:
1 parent
d22e8c8622
commit
36070387d0
4 files changed
+113
-11
No files matched your search
@@ -171,14 +171,16 @@ func (c *Conversation) compactLogLocked() {
|
||||
var cur LogEvent
|
||||
var toks int
|
||||
var ts0 int64
|
||||
var seq0 int
|
||||
flush := func() {
|
||||
if bufKey == "" {
|
||||
return
|
||||
}
|
||||
cur.Delta = map[string]any{bufKey: buf.String(), "toks": toks, "ts0": ts0}
|
||||
// seq0 = seq du PREMIER delta fusionné (cur.Seq, lui, vaut celui du dernier).
|
||||
cur.Delta = map[string]any{bufKey: buf.String(), "toks": toks, "ts0": ts0, "seq0": seq0}
|
||||
out = append(out, cur)
|
||||
buf.Reset()
|
||||
bufKey, toks, ts0 = "", 0, 0
|
||||
bufKey, toks, ts0, seq0 = "", 0, 0, 0
|
||||
}
|
||||
for _, ev := range c.Log {
|
||||
key := textKey(ev.Delta)
|
||||
@@ -194,6 +196,7 @@ func (c *Conversation) compactLogLocked() {
|
||||
bufKey = key
|
||||
cur = LogEvent{Seq: ev.Seq, TS: ev.TS}
|
||||
ts0 = evTS0(ev.Delta, ev.TS)
|
||||
seq0 = evSeq0(ev.Delta, ev.Seq)
|
||||
}
|
||||
buf.WriteString(ev.Delta[key].(string))
|
||||
cur.Seq, cur.TS = ev.Seq, ev.TS
|
||||
@@ -441,6 +444,50 @@ func (c *Conversation) Reset() {
|
||||
c.persist()
|
||||
}
|
||||
|
||||
// evSeq0 lit le seq de DÉBUT d'un bloc texte : sur un événement coalescé (dont le
|
||||
// Seq vaut celui du DERNIER delta fusionné), c'est le seq du PREMIER delta du bloc ;
|
||||
// sur un delta brut, c'est son propre Seq.
|
||||
//
|
||||
// ⚠️ Sans ça : un client qui a déjà affiché une PARTIE d'un bloc (streaming en
|
||||
// cours) et qui reçoit ensuite ce bloc coalescé — parce que la compaction de fin de
|
||||
// tour a remplacé la suite de deltas par un seul événement de Seq plus grand que
|
||||
// son dernier seq vu — le concatène à ce qu'il affichait déjà : la réponse
|
||||
// apparaissait DEUX FOIS (la 1re copie tronquée à l'endroit exact où le client en
|
||||
// était). Cf. `replace` ci-dessous.
|
||||
func evSeq0(d map[string]any, fallback int) int {
|
||||
switch v := d["seq0"].(type) {
|
||||
case int:
|
||||
return v
|
||||
case float64: // relu depuis le JSON persisté
|
||||
return int(v)
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
|
||||
// decorateEvent aplatit un événement du journal pour l'émission SSE et marque
|
||||
// `replace` quand le client a DÉJÀ vu le début du bloc (from tombe à l'intérieur) :
|
||||
// le texte envoyé est alors le bloc ENTIER, donc le client doit remplacer sa bulle
|
||||
// au lieu d'y concaténer.
|
||||
func decorateEvent(ev LogEvent, from int) map[string]any {
|
||||
out := map[string]any{"seq": ev.Seq, "ts": ev.TS}
|
||||
for k, v := range ev.Delta {
|
||||
out[k] = v
|
||||
}
|
||||
if isTextDelta(ev.Delta) && evSeq0(ev.Delta, ev.Seq) <= from {
|
||||
out["replace"] = true
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// isTextDelta : événement porteur de texte (content / reasoning_content).
|
||||
func isTextDelta(d map[string]any) bool {
|
||||
if _, ok := d["content"].(string); ok {
|
||||
return true
|
||||
}
|
||||
_, ok := d["reasoning_content"].(string)
|
||||
return ok
|
||||
}
|
||||
|
||||
// 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
|
||||
@@ -451,15 +498,22 @@ func coalesceReplay(events []LogEvent, from int) []map[string]any {
|
||||
var out []map[string]any
|
||||
var buf strings.Builder
|
||||
bufKey := ""
|
||||
var bufSeq, bufToks int
|
||||
var bufSeq, bufToks, bufSeq0 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})
|
||||
m := map[string]any{bufKey: buf.String(), "seq": bufSeq, "ts": bufTs, "ts0": bufTs0, "toks": bufToks, "seq0": bufSeq0}
|
||||
// Le client a déjà affiché le début de ce bloc (from tombe dedans, ce qui
|
||||
// arrive quand la compaction de fin de tour l'a fusionné) : on renvoie le
|
||||
// bloc entier et on lui demande de REMPLACER sa bulle, pas d'y ajouter.
|
||||
if bufSeq0 <= from {
|
||||
m["replace"] = true
|
||||
}
|
||||
out = append(out, m)
|
||||
buf.Reset()
|
||||
bufKey, bufSeq, bufToks, bufTs0, bufTs = "", 0, 0, 0, 0
|
||||
bufKey, bufSeq, bufToks, bufTs0, bufTs, bufSeq0 = "", 0, 0, 0, 0, 0
|
||||
}
|
||||
// Coalescence des outils : un appel d'outil génère plein d'événements tool_used
|
||||
// intermédiaires (streaming des arguments) jusqu'à un dernier avec done=true. Au
|
||||
@@ -496,7 +550,7 @@ func coalesceReplay(events []LogEvent, from int) []map[string]any {
|
||||
flush()
|
||||
}
|
||||
if bufKey == "" {
|
||||
bufKey, bufTs0 = key, evTS0(ev.Delta, ev.TS)
|
||||
bufKey, bufTs0, bufSeq0 = key, evTS0(ev.Delta, ev.TS), evSeq0(ev.Delta, ev.Seq)
|
||||
}
|
||||
buf.WriteString(ev.Delta[key].(string))
|
||||
bufSeq, bufTs = ev.Seq, ev.TS
|
||||
@@ -609,14 +663,14 @@ func (c *Conversation) Subscribe(ctx context.Context, from int, emit func(map[st
|
||||
c.cond.Wait()
|
||||
continue
|
||||
}
|
||||
lastEmitted := last
|
||||
last = pending[len(pending)-1].Seq
|
||||
c.mu.Unlock()
|
||||
for _, ev := range pending {
|
||||
out := map[string]any{"seq": ev.Seq, "ts": ev.TS}
|
||||
for k, v := range ev.Delta {
|
||||
out[k] = v
|
||||
}
|
||||
if !emit(out) {
|
||||
// `lastEmitted` (avant mise à jour) sert de repère : si la compaction de
|
||||
// fin de tour vient de fusionner un bloc que le client suivait en direct,
|
||||
// l'événement fusionné arrive avec un Seq supérieur au sien → `replace`.
|
||||
if !emit(decorateEvent(ev, lastEmitted)) {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
@@ -140,3 +140,41 @@ func waitSeq(t *testing.T, ch <-chan int, want int) {
|
||||
t.Fatalf("timeout en attendant seq %d", want)
|
||||
}
|
||||
}
|
||||
|
||||
// Régression : un client qui a déjà affiché une PARTIE d'un bloc de texte et qui se
|
||||
// reconnecte APRÈS la compaction de fin de tour recevait le bloc entier sans savoir
|
||||
// qu'il en avait déjà le début → la réponse s'affichait en double (1re copie
|
||||
// tronquée à l'endroit exact où le client en était). Le bloc coalescé doit porter
|
||||
// `seq0` et, quand `from` tombe dedans, l'indicateur `replace`.
|
||||
func TestReplayMarksReplaceWhenClientSawPartOfBlock(t *testing.T) {
|
||||
c := newTestConv()
|
||||
c.appendDelta(c.epoch, map[string]any{"user": "salut"}) // seq 1
|
||||
c.appendDelta(c.epoch, map[string]any{"content": "Oui, "}) // seq 2
|
||||
c.appendDelta(c.epoch, map[string]any{"content": "j'ai "}) // seq 3
|
||||
c.appendDelta(c.epoch, map[string]any{"content": "accès"}) // seq 4
|
||||
c.mu.Lock()
|
||||
c.compactLogLocked() // fin de tour : les 3 deltas deviennent UN événement seq 4
|
||||
c.mu.Unlock()
|
||||
|
||||
// Le client avait vu jusqu'au seq 3 (milieu du bloc) : il doit recevoir le bloc
|
||||
// entier AVEC replace, pour remplacer sa bulle au lieu d'y concaténer.
|
||||
out := coalesceReplay(c.Log, 3)
|
||||
if len(out) != 1 {
|
||||
t.Fatalf("attendu 1 événement rejoué, obtenu %d (%v)", len(out), out)
|
||||
}
|
||||
if got := out[0]["content"]; got != "Oui, j'ai accès" {
|
||||
t.Fatalf("texte rejoué = %q, attendu le bloc entier", got)
|
||||
}
|
||||
if r, _ := out[0]["replace"].(bool); !r {
|
||||
t.Fatalf("replace absent alors que le client avait déjà affiché le début du bloc")
|
||||
}
|
||||
|
||||
// Client qui n'a rien vu du bloc : pas de replace (il doit juste l'ajouter).
|
||||
out = coalesceReplay(c.Log, 1)
|
||||
if len(out) != 1 {
|
||||
t.Fatalf("attendu 1 événement, obtenu %d", len(out))
|
||||
}
|
||||
if _, ok := out[0]["replace"]; ok {
|
||||
t.Fatalf("replace ne doit PAS être posé quand le client n'a rien vu du bloc")
|
||||
}
|
||||
}
|
||||
@@ -2231,6 +2231,10 @@ function handleDelta(d){
|
||||
if(d.reasoning_content){
|
||||
killTyping();
|
||||
if(!T.reasonEl){ collapseAll(T.turnCollapsibles); T.reasonEl=addMsg('reasoning',''); if(REPLAYING) collapseInstant(T.reasonEl); T.fullReason=''; T.turnCollapsibles.push(T.reasonEl); }
|
||||
// d.replace : le serveur renvoie le bloc ENTIER alors qu'on en affichait déjà
|
||||
// le début (voir decorateEvent/coalesceReplay côté serveur) → on repart de zéro
|
||||
// au lieu de concaténer, sinon le texte apparaît en double.
|
||||
if(d.replace){ T.fullReason=''; T.reasonTok=0; T.reasonFirstTs=0; }
|
||||
showTyping(); T.fullReason+=d.reasoning_content; renderBody(T.reasonEl, T.fullReason);
|
||||
// d.toks/d.ts0 présents quand l'événement est coalescé (replay) : plusieurs
|
||||
// tokens d'un coup. Sinon (direct), 1 token, ts0=ts.
|
||||
@@ -2240,6 +2244,7 @@ function handleDelta(d){
|
||||
if(d.content){
|
||||
removeTyping();
|
||||
if(!T.contentEl){ collapseAll(T.turnCollapsibles); T.contentEl=addMsg('assistant',''); T.fullContent=''; }
|
||||
if(d.replace){ T.fullContent=''; T.contentTok=0; T.contentFirstTs=0; }
|
||||
T.fullContent+=d.content; renderBody(T.contentEl, T.fullContent);
|
||||
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);
|
||||
|
||||
@@ -80,6 +80,10 @@ function handleDelta(d){
|
||||
if(d.reasoning_content){
|
||||
killTyping();
|
||||
if(!T.reasonEl){ collapseAll(T.turnCollapsibles); T.reasonEl=addMsg('reasoning',''); if(REPLAYING) collapseInstant(T.reasonEl); T.fullReason=''; T.turnCollapsibles.push(T.reasonEl); }
|
||||
// d.replace : le serveur renvoie le bloc ENTIER alors qu'on en affichait déjà
|
||||
// le début (voir decorateEvent/coalesceReplay côté serveur) → on repart de zéro
|
||||
// au lieu de concaténer, sinon le texte apparaît en double.
|
||||
if(d.replace){ T.fullReason=''; T.reasonTok=0; T.reasonFirstTs=0; }
|
||||
showTyping(); T.fullReason+=d.reasoning_content; renderBody(T.reasonEl, T.fullReason);
|
||||
// d.toks/d.ts0 présents quand l'événement est coalescé (replay) : plusieurs
|
||||
// tokens d'un coup. Sinon (direct), 1 token, ts0=ts.
|
||||
@@ -89,6 +93,7 @@ function handleDelta(d){
|
||||
if(d.content){
|
||||
removeTyping();
|
||||
if(!T.contentEl){ collapseAll(T.turnCollapsibles); T.contentEl=addMsg('assistant',''); T.fullContent=''; }
|
||||
if(d.replace){ T.fullContent=''; T.contentTok=0; T.contentFirstTs=0; }
|
||||
T.fullContent+=d.content; renderBody(T.contentEl, T.fullContent);
|
||||
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);
|
||||
|
||||
Reference in new issue
Block a user