mirror of
https://github.com/R0m1k3/Loki.git
synced 2026-10-22 14:38:12 +02:00
Go: rangement des 45 fichiers par domaine (cli_/web_/chat_/llm_/backend_/relay_/sys_) + doc.go carte du code - renommages purs (git mv) + maj des noms cites en commentaires, zero changement de code
This commit is contained in:
nathaninline committed
1 parent
7af70c3d2e
commit
79e8d322a9
47 files changed
+58
-40
No files matched your search
@@ -0,0 +1,430 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// État de conversation CÔTÉ SERVEUR — une seule conversation partagée par tous
|
||||
// les appareils. Avant, l'historique vivait dans le localStorage de chaque
|
||||
// navigateur : refresh = perte des détails (outils/vitesses/raisonnement),
|
||||
// contexte différent par appareil, et fermer l'onglet coupait la génération
|
||||
// (liée à r.Context()). Ici l'état est possédé par le serveur, persisté sur
|
||||
// disque, et la génération tourne dans une goroutine détachée : fermer le
|
||||
// navigateur ne l'arrête plus, et se reconnecter rejoue tout le fil.
|
||||
//
|
||||
// Idée clé : l'UI reconstruit déjà tout l'affichage à partir d'une suite
|
||||
// d'événements SSE `delta` (content, reasoning_content, tool_used, stats…). On
|
||||
// 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
|
||||
|
||||
// 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
|
||||
// calculer la vitesse (tok/s) à partir du temps RÉEL de génération — correct
|
||||
// aussi bien en direct qu'au replay (où tout arrive d'un bloc côté client).
|
||||
type LogEvent struct {
|
||||
Seq int `json:"seq"`
|
||||
TS int64 `json:"ts"`
|
||||
Delta map[string]any `json:"delta"`
|
||||
}
|
||||
|
||||
// Conversation est le fil unique partagé. Protégé par mu ; cond réveille les
|
||||
// abonnés (aucun canal par abonné : les abonnés lisent Log au-delà de leur
|
||||
// dernier Seq puis attendent cond — replay et direct sont le même chemin).
|
||||
type Conversation struct {
|
||||
mu sync.Mutex
|
||||
cond *sync.Cond
|
||||
|
||||
Messages []Message `json:"messages"` // vue « modèle » (nourrit runChat)
|
||||
Log []LogEvent `json:"log"` // vue « UI » rejouable
|
||||
Seq int `json:"seq"`
|
||||
CtxUsed int `json:"ctx_used"` // taille réelle du contexte au dernier tour
|
||||
|
||||
Generating bool `json:"-"`
|
||||
cancel context.CancelFunc // annule la génération en cours (/stop)
|
||||
epoch int // incrémenté à chaque reset → invalide les abonnés
|
||||
}
|
||||
|
||||
var conv = func() *Conversation {
|
||||
c := &Conversation{}
|
||||
c.cond = sync.NewCond(&c.mu)
|
||||
return c
|
||||
}()
|
||||
|
||||
// convPath = fichier de persistance sur la box jean (JeanHome = /etc/jean en
|
||||
// prod). En clair : la box déchiffre déjà pour lancer le modèle, le relais reste
|
||||
// aveugle — le persister ici ne change rien à la posture E2E.
|
||||
func convPath() string { return filepath.Join(JeanHome(), "conversation.json") }
|
||||
|
||||
// LoadConversation recharge l'état persisté au démarrage du process. Sans fichier
|
||||
// (première fois) on part d'une conversation vide.
|
||||
func LoadConversation() {
|
||||
b, err := os.ReadFile(convPath())
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
conv.mu.Lock()
|
||||
defer conv.mu.Unlock()
|
||||
_ = json.Unmarshal(b, conv)
|
||||
// Une génération n'a pas pu survivre à l'arrêt du process : on repart propre.
|
||||
conv.Generating = false
|
||||
conv.cancel = nil
|
||||
}
|
||||
|
||||
// persist écrit l'état sur disque (appelé en fin de tour et sur reset, pas à
|
||||
// chaque delta). L'appelant NE doit PAS détenir mu.
|
||||
func (c *Conversation) persist() {
|
||||
c.mu.Lock()
|
||||
b, err := json.Marshal(c)
|
||||
c.mu.Unlock()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
_ = os.MkdirAll(JeanHome(), 0o755)
|
||||
tmp := convPath() + ".tmp"
|
||||
if err := os.WriteFile(tmp, b, 0o600); err != nil {
|
||||
return
|
||||
}
|
||||
_ = os.Rename(tmp, convPath())
|
||||
}
|
||||
|
||||
// appendDelta journalise un événement d'affichage et réveille les abonnés.
|
||||
func (c *Conversation) appendDelta(delta map[string]any) {
|
||||
c.mu.Lock()
|
||||
c.Seq++
|
||||
c.Log = append(c.Log, LogEvent{Seq: c.Seq, TS: time.Now().UnixMilli(), Delta: delta})
|
||||
if len(c.Log) > maxLogEvents {
|
||||
c.Log = c.Log[len(c.Log)-maxLogEvents:]
|
||||
}
|
||||
c.cond.Broadcast()
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
// convState renvoie un instantané léger (pour /api/chat/state).
|
||||
func (c *Conversation) state() map[string]any {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return map[string]any{"seq": c.Seq, "generating": c.Generating, "ctx_used": c.CtxUsed}
|
||||
}
|
||||
|
||||
// ErrBusy : une génération est déjà en cours (un seul tour à la fois).
|
||||
var ErrBusy = fmt.Errorf("génération en cours")
|
||||
|
||||
// StartTurn ajoute le message utilisateur et lance la génération EN ARRIÈRE-PLAN
|
||||
// (context.Background, détaché de toute connexion HTTP). Renvoie ErrBusy si un
|
||||
// tour est déjà en cours, ou une erreur si le modèle n'est pas prêt.
|
||||
func (c *Conversation) StartTurn(text string, caps Caps, temperature float64) error {
|
||||
if !healthCheck() {
|
||||
return fmt.Errorf("⏳ Le modèle est encore en train de charger — réessaie dans quelques secondes.")
|
||||
}
|
||||
c.mu.Lock()
|
||||
if c.Generating {
|
||||
c.mu.Unlock()
|
||||
return ErrBusy
|
||||
}
|
||||
c.Generating = true
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
c.cancel = cancel
|
||||
c.Messages = append(c.Messages, Message{Role: "user", Content: text})
|
||||
c.mu.Unlock()
|
||||
|
||||
// Borne de tour + bulle utilisateur (rejouables).
|
||||
c.appendDelta(map[string]any{"user": text})
|
||||
if temperature == 0 {
|
||||
temperature = 0.7
|
||||
}
|
||||
go c.generate(ctx, caps, temperature)
|
||||
return nil
|
||||
}
|
||||
|
||||
// generate exécute un tour complet et journalise chaque événement. Détaché : la
|
||||
// fermeture du navigateur n'a aucun effet ici, seul /stop (cancel) l'interrompt.
|
||||
func (c *Conversation) generate(ctx context.Context, caps Caps, temperature float64) {
|
||||
defer func() {
|
||||
c.mu.Lock()
|
||||
c.Generating = false
|
||||
c.cancel = nil
|
||||
c.mu.Unlock()
|
||||
c.appendDelta(map[string]any{"turn_done": true})
|
||||
c.persist()
|
||||
}()
|
||||
|
||||
// Snapshot de la vue modèle.
|
||||
c.mu.Lock()
|
||||
msgs := append([]Message(nil), c.Messages...)
|
||||
ctxUsed := c.CtxUsed
|
||||
c.mu.Unlock()
|
||||
|
||||
// Compaction proactive (façon Hermes) sur la vue MODÈLE uniquement ; le journal
|
||||
// d'affichage garde le fil complet. On signale juste un toast au client.
|
||||
if compacted, changed := MaybeCompact(ctx, msgs, caps, ctxUsed); changed {
|
||||
msgs = compacted
|
||||
c.mu.Lock()
|
||||
c.Messages = compacted
|
||||
c.mu.Unlock()
|
||||
c.appendDelta(map[string]any{"compacted": true})
|
||||
}
|
||||
|
||||
var content strings.Builder
|
||||
extra, _ := runChat(ctx, InjectSkills(msgs, caps), temperature, caps, func(ev StreamEvent) bool {
|
||||
switch {
|
||||
case ev.Err != nil:
|
||||
c.appendDelta(map[string]any{"error": ev.Err.Error()})
|
||||
case ev.ToolUsed != nil:
|
||||
c.appendDelta(map[string]any{"tool_used": map[string]any{
|
||||
"name": ev.ToolUsed.Name, "label": ev.ToolUsed.Label,
|
||||
"result": ev.ToolUsed.Result, "done": ev.ToolUsed.Done, "typing": ev.ToolUsed.Typing,
|
||||
}})
|
||||
case ev.Stats != nil:
|
||||
// Taille réelle du contexte (usage.prompt_tokens + généré) pour le compteur
|
||||
// et la décision de compactage au tour suivant.
|
||||
if ev.Stats.PromptTokensTotal > 0 {
|
||||
c.mu.Lock()
|
||||
c.CtxUsed = ev.Stats.PromptTokensTotal + ev.Stats.GenTokens
|
||||
c.mu.Unlock()
|
||||
}
|
||||
c.appendDelta(map[string]any{"stats": ev.Stats})
|
||||
case ev.DropReasoning:
|
||||
c.appendDelta(map[string]any{"drop_reasoning": true})
|
||||
case ev.Reasoning != "":
|
||||
c.appendDelta(map[string]any{"reasoning_content": ev.Reasoning})
|
||||
case ev.Content != "":
|
||||
content.WriteString(ev.Content)
|
||||
c.appendDelta(map[string]any{"content": ev.Content})
|
||||
}
|
||||
return true // génération détachée : on ne s'interrompt jamais sur un abonné
|
||||
})
|
||||
|
||||
// Persiste la vue modèle : messages d'outils (assistant tool_calls + résultats)
|
||||
// PUIS la réponse finale — même ordre que l'ancien client, pour que le modèle
|
||||
// garde la trace de ce qu'il a fait.
|
||||
c.mu.Lock()
|
||||
c.Messages = append(c.Messages, extra...)
|
||||
if s := content.String(); strings.TrimSpace(s) != "" {
|
||||
c.Messages = append(c.Messages, Message{Role: "assistant", Content: s})
|
||||
}
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
// Stop interrompt la génération en cours (le cas échéant).
|
||||
func (c *Conversation) Stop() {
|
||||
c.mu.Lock()
|
||||
cancel := c.cancel
|
||||
c.mu.Unlock()
|
||||
if cancel != nil {
|
||||
cancel()
|
||||
}
|
||||
}
|
||||
|
||||
// Reset démarre une nouvelle conversation (vide) pour TOUS les appareils. On
|
||||
// interrompt une éventuelle génération, on vide tout et on bump epoch pour que
|
||||
// les abonnés reçoivent l'ordre de nettoyer leur affichage.
|
||||
func (c *Conversation) Reset() {
|
||||
c.Stop()
|
||||
c.mu.Lock()
|
||||
c.Messages = nil
|
||||
c.Log = nil
|
||||
c.Seq = 0
|
||||
c.CtxUsed = 0
|
||||
c.epoch++
|
||||
c.cond.Broadcast()
|
||||
c.mu.Unlock()
|
||||
c.persist()
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
// 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
|
||||
// replay, seul l'état FINAL de chaque bulle compte (les intermédiaires ne servent
|
||||
// qu'à l'affichage progressif en direct). Sans ça, un long fil rejoue des milliers
|
||||
// de tool_used → 2-3 s de rendu inutile sur mobile. On ne garde donc que les
|
||||
// done=true, en mémorisant le dernier non-done pour le cas d'un outil interrompu.
|
||||
var pendingTool map[string]any
|
||||
flushTool := func() {
|
||||
if pendingTool != nil {
|
||||
out = append(out, pendingTool)
|
||||
pendingTool = nil
|
||||
}
|
||||
}
|
||||
for _, ev := range events {
|
||||
if ev.Seq <= from {
|
||||
continue
|
||||
}
|
||||
_, isTool := ev.Delta["tool_used"].(map[string]any)
|
||||
// Tout événement NON-outil clôt une éventuelle bulle d'outil en attente, pour
|
||||
// préserver l'ordre (l'outil non terminé s'affiche avant ce qui le suit).
|
||||
if !isTool {
|
||||
flushTool()
|
||||
}
|
||||
// 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
|
||||
}
|
||||
if isTool {
|
||||
tu, _ := ev.Delta["tool_used"].(map[string]any)
|
||||
done, _ := tu["done"].(bool)
|
||||
if done {
|
||||
pendingTool = nil // les intermédiaires de cet outil sont superflus
|
||||
out = append(out, m)
|
||||
} else {
|
||||
pendingTool = m // on retient le dernier état non terminé, sans l'émettre
|
||||
}
|
||||
continue
|
||||
}
|
||||
out = append(out, m)
|
||||
}
|
||||
flush()
|
||||
flushTool()
|
||||
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.
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
c.mu.Lock()
|
||||
c.cond.Broadcast()
|
||||
c.mu.Unlock()
|
||||
}()
|
||||
|
||||
// 0. Amorçage anti-buffering. Le chat E2E d'app.ajean.link traverse Cloudflare
|
||||
// (ajean.link, proxy orange) : tant qu'un proxy intermédiaire n'a pas reçu assez
|
||||
// d'octets, il bufferise la réponse et ne la relaie qu'en retard (symptôme :
|
||||
// derniers messages qui arrivent 20-30 s après le reste, de façon intermittente
|
||||
// sur Safari). Envoyer d'emblée un gros événement de padding force le proxy à
|
||||
// basculer en mode streaming tout de suite. Le client ignore la clé `pad`.
|
||||
if !emit(map[string]any{"pad": strings.Repeat("·", 2048)}) {
|
||||
return
|
||||
}
|
||||
|
||||
// 1. Replay coalescé (snapshot hors verrou pour ne pas bloquer la génération).
|
||||
c.mu.Lock()
|
||||
snapshot := append([]LogEvent(nil), c.Log...)
|
||||
epoch := c.epoch
|
||||
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
|
||||
}
|
||||
// Le padding doit venir APRÈS caught_up, pas dessus. Un proxy (Cloudflare) garde
|
||||
// toujours le DERNIER bout de flux en tampon jusqu'au prochain flush (~2-3 s). En
|
||||
// envoyant un gros pad juste après, ce sont ses octets — et non le dernier vrai
|
||||
// message — qui deviennent la « queue » qui attend : le dernier message et
|
||||
// caught_up, eux, sont poussés dehors immédiatement. Le client ignore `pad`.
|
||||
// 16 Ko : largement au-dessus du tampon de coalescence d'un proxy courant.
|
||||
if !emit(map[string]any{"pad": strings.Repeat("·", 16384)}) {
|
||||
return
|
||||
}
|
||||
|
||||
// 2. Direct : événements granulaires au-delà de `last`.
|
||||
c.mu.Lock()
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
c.mu.Unlock()
|
||||
return
|
||||
}
|
||||
if c.epoch != epoch { // reset → on ordonne au client de nettoyer et on repart
|
||||
epoch = c.epoch
|
||||
last = 0
|
||||
c.mu.Unlock()
|
||||
if !emit(map[string]any{"reset": true}) {
|
||||
return
|
||||
}
|
||||
c.mu.Lock()
|
||||
continue
|
||||
}
|
||||
sent := false
|
||||
for _, ev := range c.Log {
|
||||
if ev.Seq <= last {
|
||||
continue
|
||||
}
|
||||
last = ev.Seq
|
||||
delta := ev.Delta
|
||||
ts := ev.TS
|
||||
c.mu.Unlock()
|
||||
out := map[string]any{"seq": ev.Seq, "ts": ts}
|
||||
for k, v := range delta {
|
||||
out[k] = v
|
||||
}
|
||||
if !emit(out) {
|
||||
return
|
||||
}
|
||||
c.mu.Lock()
|
||||
sent = true
|
||||
if ctx.Err() != nil {
|
||||
c.mu.Unlock()
|
||||
return
|
||||
}
|
||||
}
|
||||
if sent {
|
||||
continue
|
||||
}
|
||||
c.cond.Wait()
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user